ATLAS-2481: entity-classification edges - labelled as 'classifiedAs', with additional properties

common/src/main/java/org/apache/atlas/repository/Constants.java

@@ -116,9 +116,12 @@ public final class Constants {
     public static final String ATTRIBUTE_NAME_VERSION  = "version";
     public static final String TEMP_STRUCT_NAME_PREFIX = "__tempQueryResultStruct";
 
-    public static final String CLASSIFICATION_ENTITY_GUID          = INTERNAL_PROPERTY_KEY_PREFIX + "entityGuid";
-    public static final String CLASSIFICATION_PROPAGATE_KEY        = INTERNAL_PROPERTY_KEY_PREFIX + "propagate";
-    public static final String CLASSIFICATION_VALIDITY_PERIODS_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "validityPeriods";
+    public static final String CLASSIFICATION_ENTITY_GUID                     = INTERNAL_PROPERTY_KEY_PREFIX + "entityGuid";
+    public static final String CLASSIFICATION_VALIDITY_PERIODS_KEY            = INTERNAL_PROPERTY_KEY_PREFIX + "validityPeriods";
+    public static final String CLASSIFICATION_VERTEX_PROPAGATE_KEY            = INTERNAL_PROPERTY_KEY_PREFIX + "propagate";
+    public static final String CLASSIFICATION_EDGE_NAME_PROPERTY_KEY          = INTERNAL_PROPERTY_KEY_PREFIX + "name";
+    public static final String CLASSIFICATION_EDGE_IS_PROPAGATED_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "isPropagated";
+    public static final String CLASSIFICATION_LABEL                           = "classifiedAs";
 
     private Constants() {
     }

graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasGraphManagement.java

@@ -105,6 +105,15 @@ public interface AtlasGraphManagement {
     AtlasGraphIndex getGraphIndex(String indexName);
 
     /**
+     * Checks if a vertex-centric edge exists already.
+     *
+     * @param label
+     * @param indexName
+     * @return
+     */
+    boolean edgeIndexExist(String label, String indexName);
+
+    /**
      * Creates a mixed Vertex index for the graph.
      *
      * @param name the name of the index to create
@@ -123,6 +132,16 @@ public interface AtlasGraphManagement {
     void createEdgeMixedIndex(String index, String backingIndex, List<AtlasPropertyKey> propertyKeys);
 
     /**
+     * Creates a vertex-centric edge index for the graph.
+     *
+     * @param label edge label name
+     * @param indexName name of the edge index
+     * @param edgeDirection direction of the edge to index
+     * @param propertyKeys edge property keys to be added to the index
+     */
+    void createEdgeIndex(String label, String indexName, AtlasEdgeDirection edgeDirection, List<AtlasPropertyKey> propertyKeys);
+
+    /**
      * Creates a full text index for the given property.
      *
      * @param index the name of the index to create

graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasGraphQuery.java

@@ -52,18 +52,32 @@ public interface AtlasGraphQuery<V, E> {
      */
     AtlasGraphQuery<V, E> in(String propertyKey, Collection<?> values);
 
+    /**
+     * Executes the query and returns the matching edges.
+     * @return
+     */
+    Iterable<AtlasEdge<V, E>> edges();
 
     /**
-     * Executes the query and returns the matching vertices.
+     * Executes the query and returns the matching edges till the max limit
+     * @param limit max number of vertices
      * @return
      */
-    Iterable<AtlasVertex<V, E>> vertices();
+    Iterable<AtlasEdge<V, E>> edges(int limit);
 
     /**
-     * Executes the query and returns the matching edges.
+     * Executes the query and returns the matching edges from given offset till the max limit
+     * @param offset starting offset
+     * @param limit max number of vertices
      * @return
      */
-    Iterable<AtlasEdge<V, E>> edges();
+    Iterable<AtlasEdge<V, E>> edges(int offset, int limit);
+
+    /**
+     * Executes the query and returns the matching vertices.
+     * @return
+     */
+    Iterable<AtlasVertex<V, E>> vertices();
 
     /**
      * Executes the query and returns the matching vertices from given offset till the max limit

graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasVertexQuery.java

@@ -67,4 +67,21 @@ public interface AtlasVertexQuery<V, E> {
      * @return
      */
     long count();
+
+    /**
+     * Specifies the edge label that should be queried.
+     *
+     * @param label
+     * @return
+     */
+    AtlasVertexQuery<V, E> label(String label);
+
+    /**
+     * Returns edges that matches property key and value.
+     *
+     * @param key
+     * @param value
+     * @return
+     */
+    AtlasVertexQuery<V, E> has(String key, Object value);
 }

graphdb/common/src/main/java/org/apache/atlas/repository/graphdb/tinkerpop/query/NativeTinkerpopGraphQuery.java

@@ -48,6 +48,21 @@ public interface NativeTinkerpopGraphQuery<V, E> {
 
     /**
      * Executes graph query
+     * @param limit Max edges to return
+     * @return
+     */
+    Iterable<AtlasEdge<V, E>> edges(int limit);
+
+    /**
+     * Executes graph query
+     * @param offset Starting offset
+     * @param limit Max edges to return
+     * @return
+     */
+    Iterable<AtlasEdge<V, E>> edges(int offset, int limit);
+
+    /**
+     * Executes graph query
      * @param limit Max vertices to return
      * @return
      */

graphdb/common/src/main/java/org/apache/atlas/repository/graphdb/tinkerpop/query/TinkerpopGraphQuery.java

@@ -157,6 +157,45 @@ public abstract class TinkerpopGraphQuery<V, E> implements AtlasGraphQuery<V, E>
     }
 
     @Override
+    public Iterable<AtlasEdge<V, E>> edges(int limit) {
+        return edges(0, limit);
+    }
+
+    @Override
+    public Iterable<AtlasEdge<V, E>> edges(int offset, int limit) {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Executing: " + queryCondition);
+        }
+
+        Preconditions.checkArgument(offset >= 0, "Offset must be non-negative");
+        Preconditions.checkArgument(limit >= 0, "Limit must be non-negative");
+
+        // Compute the overall result by combining the results of all the AndConditions (nested within OR) together.
+        Set<AtlasEdge<V, E>> result = new HashSet<>();
+        long resultIdx = 0;
+        for(AndCondition andExpr : queryCondition.getAndTerms()) {
+            if (result.size() == limit) {
+                break;
+            }
+
+            NativeTinkerpopGraphQuery<V, E> andQuery = andExpr.create(getQueryFactory());
+            for(AtlasEdge<V, E> edge : andQuery.edges(offset + limit)) {
+                if (resultIdx >= offset) {
+                    result.add(edge);
+
+                    if (result.size() == limit) {
+                        break;
+                    }
+                }
+
+                resultIdx++;
+            }
+        }
+
+        return result;
+    }
+
+    @Override
     public Iterable<AtlasVertex<V, E>> vertices(int limit) {
         return vertices(0, limit);
     }

graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusGraphManagement.java

@@ -18,6 +18,8 @@
 package org.apache.atlas.repository.graphdb.janus;
 
 import com.google.common.base.Preconditions;
+import org.apache.atlas.repository.graphdb.AtlasEdgeDirection;
+import org.apache.tinkerpop.gremlin.structure.Direction;
 import org.janusgraph.core.Cardinality;
 import org.janusgraph.core.EdgeLabel;
 import org.janusgraph.core.PropertyKey;
@@ -83,6 +85,22 @@ public class AtlasJanusGraphManagement implements AtlasGraphManagement {
     }
 
     @Override
+    public void createEdgeIndex(String label, String indexName, AtlasEdgeDirection edgeDirection, List<AtlasPropertyKey> propertyKeys) {
+        EdgeLabel edgeLabel = management.getEdgeLabel(label);
+
+        if (edgeLabel == null) {
+            edgeLabel = management.makeEdgeLabel(label).make();
+        }
+
+        Direction     direction = AtlasJanusObjectFactory.createDirection(edgeDirection);
+        PropertyKey[] keys      = AtlasJanusObjectFactory.createPropertyKeys(propertyKeys);
+
+        if (management.getRelationIndex(edgeLabel, indexName) == null) {
+            management.buildEdgeIndex(edgeLabel, indexName, direction, keys);
+        }
+    }
+
+    @Override
     public void createFullTextMixedIndex(String indexName, String backingIndex, List<AtlasPropertyKey> propertyKeys) {
         IndexBuilder indexBuilder = management.buildIndex(indexName, Vertex.class);
 
@@ -192,6 +210,13 @@ public class AtlasJanusGraphManagement implements AtlasGraphManagement {
     }
 
     @Override
+    public boolean edgeIndexExist(String label, String indexName) {
+        EdgeLabel edgeLabel = management.getEdgeLabel(label);
+
+        return edgeLabel != null && management.getRelationIndex(edgeLabel, indexName) != null;
+    }
+
+    @Override
     public void createVertexCompositeIndex(String propertyName, boolean isUnique, List<AtlasPropertyKey> propertyKeys) {
         IndexBuilder indexBuilder = management.buildIndex(propertyName, Vertex.class);
 

graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusObjectFactory.java

@@ -25,6 +25,8 @@ import org.apache.tinkerpop.gremlin.structure.Direction;
 import org.janusgraph.core.Cardinality;
 import org.janusgraph.core.PropertyKey;
 
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * Factory that serves up instances of Janus/Tinkerpop classes that correspond to
@@ -61,7 +63,7 @@ public final class AtlasJanusObjectFactory {
     /**
      * Converts a Multiplicity to a Cardinality.
      *
-     * @param multiplicity
+     * @param cardinality
      * @return
      */
     public static Cardinality createCardinality(AtlasCardinality cardinality) {
@@ -82,4 +84,16 @@ public final class AtlasJanusObjectFactory {
         return ((AtlasJanusPropertyKey)key).getWrappedPropertyKey();
     }
 
+    public static PropertyKey[] createPropertyKeys(List<AtlasPropertyKey> keys) {
+        PropertyKey[] ret = new PropertyKey[keys.size()];
+
+        int i = 0;
+        for (AtlasPropertyKey key : keys) {
+            ret[i] = createPropertyKey(key);
+
+            i++;
+        }
+
+        return ret;
+    }
 }

graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusVertexQuery.java

@@ -76,5 +76,15 @@ public class AtlasJanusVertexQuery implements AtlasVertexQuery<AtlasJanusVertex,
         return query.count();
     }
 
+    @Override
+    public AtlasVertexQuery<AtlasJanusVertex, AtlasJanusEdge> label(String label) {
+        query.labels(label);
+        return this;
+    }
 
+    @Override
+    public AtlasVertexQuery<AtlasJanusVertex, AtlasJanusEdge> has(String key, Object value) {
+        query.has(key, value);
+        return this;
+    }
 }

graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/query/NativeJanusGraphQuery.java

@@ -17,6 +17,7 @@
  */
 package org.apache.atlas.repository.graphdb.janus.query;
 
+import org.apache.tinkerpop.gremlin.structure.Edge;
 import org.janusgraph.core.JanusGraphEdge;
 import org.janusgraph.core.JanusGraphQuery;
 import org.janusgraph.core.JanusGraphVertex;
@@ -64,6 +65,28 @@ public class NativeJanusGraphQuery implements NativeTinkerpopGraphQuery<AtlasJan
     }
 
     @Override
+    public Iterable<AtlasEdge<AtlasJanusVertex, AtlasJanusEdge>> edges(int limit) {
+        Iterable<JanusGraphEdge> it = query.limit(limit).edges();
+        return graph.wrapEdges(it);
+    }
+
+    @Override
+    public Iterable<AtlasEdge<AtlasJanusVertex, AtlasJanusEdge>> edges(int offset, int limit) {
+        List<Edge>               result = new ArrayList<>(limit);
+        Iterator<? extends Edge> iter   = query.limit(offset + limit).edges().iterator();
+
+        for (long resultIdx = 0; iter.hasNext() && result.size() < limit; resultIdx++) {
+            if (resultIdx < offset) {
+                continue;
+            }
+
+            result.add(iter.next());
+        }
+
+        return graph.wrapEdges(result);
+    }
+
+    @Override
     public Iterable<AtlasVertex<AtlasJanusVertex, AtlasJanusEdge>> vertices(int limit) {
         Iterable<JanusGraphVertex> it = query.limit(limit).vertices();
         return graph.wrapVertices(it);

graphdb/titan0/src/main/java/org/apache/atlas/repository/graphdb/titan0/Titan0GraphManagement.java

@@ -20,14 +20,15 @@ package org.apache.atlas.repository.graphdb.titan0;
 import com.thinkaurelius.titan.core.Cardinality;
 import com.thinkaurelius.titan.core.EdgeLabel;
 import com.thinkaurelius.titan.core.PropertyKey;
-import com.thinkaurelius.titan.core.schema.Mapping;
 import com.thinkaurelius.titan.core.schema.PropertyKeyMaker;
 import com.thinkaurelius.titan.core.schema.TitanGraphIndex;
 import com.thinkaurelius.titan.core.schema.TitanManagement;
+import com.tinkerpop.blueprints.Direction;
 import com.tinkerpop.blueprints.Edge;
 import com.tinkerpop.blueprints.Element;
 import com.tinkerpop.blueprints.Vertex;
 import org.apache.atlas.repository.graphdb.AtlasCardinality;
+import org.apache.atlas.repository.graphdb.AtlasEdgeDirection;
 import org.apache.atlas.repository.graphdb.AtlasEdgeLabel;
 import org.apache.atlas.repository.graphdb.AtlasGraphIndex;
 import org.apache.atlas.repository.graphdb.AtlasGraphManagement;
@@ -62,6 +63,20 @@ public class Titan0GraphManagement implements AtlasGraphManagement {
     }
 
     @Override
+    public void createEdgeIndex(String label, String indexName, AtlasEdgeDirection edgeDirection, List<AtlasPropertyKey> propertyKeys) {
+        EdgeLabel edgeLabel = management.getEdgeLabel(label);
+
+        if (edgeLabel == null) {
+            edgeLabel = management.makeEdgeLabel(label).make();
+        }
+
+        Direction     direction = TitanObjectFactory.createDirection(edgeDirection);
+        PropertyKey[] keys      = TitanObjectFactory.createPropertyKeys(propertyKeys);
+
+        management.buildEdgeIndex(edgeLabel, indexName, direction, keys);
+    }
+
+    @Override
     public void createFullTextMixedIndex(String index, String backingIndex, List<AtlasPropertyKey> propertyKeys) {
     }
 
@@ -201,4 +216,10 @@ public class Titan0GraphManagement implements AtlasGraphManagement {
         return GraphDbObjectFactory.createGraphIndex(index);
     }
 
+    @Override
+    public boolean edgeIndexExist(String label, String indexName) {
+        EdgeLabel edgeLabel = management.getEdgeLabel(label);
+
+        return edgeLabel != null && management.getRelationIndex(edgeLabel, indexName) != null;
+    }
 }

graphdb/titan0/src/main/java/org/apache/atlas/repository/graphdb/titan0/Titan0VertexQuery.java

@@ -77,4 +77,16 @@ public class Titan0VertexQuery implements AtlasVertexQuery<Titan0Vertex, Titan0E
     public long count() {
         return vertexQuery.count();
     }
+
+    @Override
+    public AtlasVertexQuery<Titan0Vertex, Titan0Edge> label(String label) {
+        vertexQuery.labels(label);
+        return this;
+    }
+
+    @Override
+    public AtlasVertexQuery<Titan0Vertex, Titan0Edge> has(String key, Object value) {
+        vertexQuery.has(key, value);
+        return this;
+    }
 }

graphdb/titan0/src/main/java/org/apache/atlas/repository/graphdb/titan0/TitanObjectFactory.java

@@ -25,6 +25,9 @@ import com.thinkaurelius.titan.core.Cardinality;
 import com.thinkaurelius.titan.core.PropertyKey;
 import com.tinkerpop.blueprints.Direction;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * Factory that serves up instances of Titan/Tinkerpop classes that correspond to
  * graph database abstraction layer/Atlas classes.
@@ -60,7 +63,7 @@ public final class TitanObjectFactory {
     /**
      * Converts a Multiplicity to a Cardinality.
      *
-     * @param multiplicity
+     * @param cardinality
      * @return
      */
     public static Cardinality createCardinality(AtlasCardinality cardinality) {
@@ -81,4 +84,16 @@ public final class TitanObjectFactory {
         return ((Titan0PropertyKey)key).getWrappedPropertyKey();
     }
 
+    public static PropertyKey[] createPropertyKeys(List<AtlasPropertyKey> keys) {
+        PropertyKey[] ret = new PropertyKey[keys.size()];
+
+        int i = 0;
+        for (AtlasPropertyKey key : keys) {
+            ret[i] = createPropertyKey(key);
+
+            i++;
+        }
+
+        return ret;
+    }
 }

graphdb/titan0/src/main/java/org/apache/atlas/repository/graphdb/titan0/query/NativeTitan0GraphQuery.java

@@ -23,6 +23,7 @@ import com.thinkaurelius.titan.core.attribute.Contain;
 import com.thinkaurelius.titan.core.attribute.Text;
 import com.thinkaurelius.titan.graphdb.query.TitanPredicate;
 import com.tinkerpop.blueprints.Compare;
+import com.tinkerpop.blueprints.Edge;
 import com.tinkerpop.blueprints.Vertex;
 import org.apache.atlas.repository.graphdb.AtlasEdge;
 import org.apache.atlas.repository.graphdb.AtlasGraphQuery.ComparisionOperator;
@@ -64,6 +65,29 @@ public class NativeTitan0GraphQuery implements NativeTinkerpopGraphQuery<Titan0V
         Iterable it = query.edges();
         return graph.wrapEdges(it);
     }
+
+    @Override
+    public Iterable<AtlasEdge<Titan0Vertex, Titan0Edge>> edges(int limit) {
+        Iterable it = query.limit(limit).edges();
+        return graph.wrapEdges(it);
+    }
+
+    @Override
+    public Iterable<AtlasEdge<Titan0Vertex, Titan0Edge>> edges(int offset, int limit) {
+        List<Edge>     result = new ArrayList<>(limit);
+        Iterator<Edge> iter   = query.limit(offset + limit).edges().iterator();
+
+        for (long resultIdx = 0; iter.hasNext() && result.size() < limit; resultIdx++) {
+            if (resultIdx < offset) {
+                continue;
+            }
+
+            result.add(iter.next());
+        }
+
+        return graph.wrapEdges(result);
+    }
+
     @Override
     public Iterable<AtlasVertex<Titan0Vertex, Titan0Edge>> vertices(int limit) {
         Iterable it = query.limit(limit).vertices();

intg/src/main/java/org/apache/atlas/AtlasErrorCode.java

@@ -125,6 +125,8 @@ public enum AtlasErrorCode {
     CLASSIFICATION_UPDATE_FROM_PROPAGATED_ENTITY(400, "ATLAS-400-00-06B", "Update to classification {0} is not allowed from propagated entity"),
     CLASSIFICATION_DELETE_FROM_PROPAGATED_ENTITY(400, "ATLAS-400-00-06C", "Delete of classification {0} is not allowed from propagated entity"),
     CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY(400, "ATLAS-400-00-06D", "Classification {0} is not associated with entity"),
+    NO_CLASSIFICATIONS_FOUND_FOR_ENTITY(400, "ATLAS-400-00-06E", "No classifications associated with entity: {0}"),
+    INVALID_CLASSIFICATION_PARAMS(400, "ATLAS-400-00-06F", "Invalid classification parameters passed for {0} operation for entity: {1}"),
 
     UNAUTHORIZED_ACCESS(403, "ATLAS-403-00-001", "{0} is not authorized to perform {1}"),
 

repository/src/main/java/org/apache/atlas/discovery/EntityDiscoveryService.java

@@ -462,7 +462,7 @@ public class EntityDiscoveryService implements AtlasDiscoveryService {
                 AtlasEntityHeader entity = entityRetriever.toAtlasEntityHeader(atlasVertex, resultAttributes);
 
                 if(searchParameters.getIncludeClassificationAttributes()) {
-                    entity.setClassifications(entityRetriever.getClassifications(atlasVertex));
+                    entity.setClassifications(entityRetriever.getAllClassifications(atlasVertex));
                 }
 
                 ret.addEntity(entity);

repository/src/main/java/org/apache/atlas/query/GremlinClause.java

@@ -46,7 +46,7 @@ enum GremlinClause {
     TEXT_CONTAINS("has('%s', org.janusgraph.core.attribute.Text.textRegex(%s))"),
     TEXT_PREFIX("has('%s', org.janusgraph.core.attribute.Text.textPrefix(%s))"),
     TEXT_SUFFIX("has('%s', org.janusgraph.core.attribute.Text.textRegex(\".*\" + %s))"),
-    TRAIT("or(has('__traitNames', within('%s')), has('__propagatedTraitNames', within('%s')))"),
+    TRAIT("outE('classifiedAs').has('__name', within('%s')).outV()"),
     SELECT_NOOP_FN("def f(r){ r }; "),
     SELECT_FN("def f(r){ t=[[%s]]; %s r.each({t.add([%s])}); t.unique(); }; "),
     SELECT_ONLY_AGG_FN("def f(r){ t=[[%s]]; %s t.add([%s]); t;}; "),

repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedSearchIndexer.java

@@ -37,6 +37,7 @@ import org.apache.atlas.repository.Constants;
 import org.apache.atlas.repository.IndexException;
 import org.apache.atlas.repository.RepositoryException;
 import org.apache.atlas.repository.graphdb.AtlasCardinality;
+import org.apache.atlas.repository.graphdb.AtlasEdgeDirection;
 import org.apache.atlas.repository.graphdb.AtlasGraph;
 import org.apache.atlas.repository.graphdb.AtlasGraphIndex;
 import org.apache.atlas.repository.graphdb.AtlasGraphManagement;
@@ -59,6 +60,7 @@ import javax.inject.Inject;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
@@ -66,6 +68,9 @@ import java.util.Set;
 
 import static org.apache.atlas.model.typedef.AtlasBaseTypeDef.*;
 import static org.apache.atlas.repository.Constants.BACKING_INDEX;
+import static org.apache.atlas.repository.Constants.CLASSIFICATION_EDGE_IS_PROPAGATED_PROPERTY_KEY;
+import static org.apache.atlas.repository.Constants.CLASSIFICATION_LABEL;
+import static org.apache.atlas.repository.Constants.CLASSIFICATION_EDGE_NAME_PROPERTY_KEY;
 import static org.apache.atlas.repository.Constants.CREATED_BY_KEY;
 import static org.apache.atlas.repository.Constants.EDGE_INDEX;
 import static org.apache.atlas.repository.Constants.ENTITY_TEXT_PROPERTY_KEY;
@@ -280,6 +285,11 @@ public class GraphBackedSearchIndexer implements SearchIndexer, ActiveStateChang
             createVertexIndex(management, TYPENAME_PROPERTY_KEY, String.class, true, SINGLE, true, true);
             createVertexIndex(management, VERTEX_TYPE_PROPERTY_KEY, String.class, false, SINGLE, true, true);
 
+            // create vertex-centric index
+            createVertexCentricIndex(management, CLASSIFICATION_LABEL, AtlasEdgeDirection.BOTH, CLASSIFICATION_EDGE_NAME_PROPERTY_KEY, String.class, SINGLE);
+            createVertexCentricIndex(management, CLASSIFICATION_LABEL, AtlasEdgeDirection.BOTH, CLASSIFICATION_EDGE_IS_PROPAGATED_PROPERTY_KEY, Boolean.class, SINGLE);
+            createVertexCentricIndex(management, CLASSIFICATION_LABEL, AtlasEdgeDirection.BOTH, Arrays.asList(CLASSIFICATION_EDGE_NAME_PROPERTY_KEY, CLASSIFICATION_EDGE_IS_PROPAGATED_PROPERTY_KEY));
+
             // create edge indexes
             createEdgeIndex(management, RELATIONSHIP_GUID_PROPERTY_KEY, String.class, SINGLE, true);
 
@@ -475,6 +485,55 @@ public class GraphBackedSearchIndexer implements SearchIndexer, ActiveStateChang
         return propertyKey;
     }
 
+    private void createVertexCentricIndex(AtlasGraphManagement management, String edgeLabel, AtlasEdgeDirection edgeDirection,
+                                          String propertyName, Class propertyClass, AtlasCardinality cardinality) {
+        AtlasPropertyKey propertyKey = management.getPropertyKey(propertyName);
+
+        if (propertyKey == null) {
+            propertyKey = management.makePropertyKey(propertyName, propertyClass, cardinality);
+        }
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Creating vertex-centric index for edge label: {} direction: {} for property: {} of type: {} ",
+                        edgeLabel, edgeDirection.name(), propertyName, propertyClass.getName());
+        }
+
+        final String indexName = edgeLabel + propertyKey.getName();
+
+        if (!management.edgeIndexExist(edgeLabel, indexName)) {
+            management.createEdgeIndex(edgeLabel, indexName, edgeDirection, Collections.singletonList(propertyKey));
+
+            LOG.info("Created vertex-centric index for edge label: {} direction: {} for property: {} of type: {}",
+                    edgeLabel, edgeDirection.name(), propertyName, propertyClass.getName());
+        }
+    }
+
+    private void createVertexCentricIndex(AtlasGraphManagement management, String edgeLabel, AtlasEdgeDirection edgeDirection, List<String> propertyNames) {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Creating vertex-centric index for edge label: {} direction: {} for properties: {}",
+                    edgeLabel, edgeDirection.name(), propertyNames);
+        }
+
+        String                 indexName    = edgeLabel;
+        List<AtlasPropertyKey> propertyKeys = new ArrayList<>();
+
+        for (String propertyName : propertyNames) {
+            AtlasPropertyKey propertyKey = management.getPropertyKey(propertyName);
+
+            if (propertyKey != null) {
+                propertyKeys.add(propertyKey);
+                indexName = indexName + propertyKey.getName();
+            }
+        }
+
+        if (!management.edgeIndexExist(edgeLabel, indexName) && CollectionUtils.isNotEmpty(propertyKeys)) {
+            management.createEdgeIndex(edgeLabel, indexName, edgeDirection, propertyKeys);
+
+            LOG.info("Created vertex-centric index for edge label: {} direction: {} for properties: {}", edgeLabel, edgeDirection.name(), propertyNames);
+        }
+    }
+
+
     private AtlasPropertyKey createEdgeIndex(AtlasGraphManagement management, String propertyName, Class propertyClass,
                                              AtlasCardinality cardinality, boolean createCompositeIndex) {
         AtlasPropertyKey propertyKey = management.getPropertyKey(propertyName);

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java

@@ -40,6 +40,7 @@ import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
 import org.apache.atlas.type.AtlasStructType.AtlasAttribute.AtlasRelationshipEdgeDirection;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,12 +48,12 @@ import org.slf4j.LoggerFactory;
 import java.util.*;
 
 import static org.apache.atlas.model.instance.AtlasEntity.Status.DELETED;
+import static org.apache.atlas.repository.Constants.CLASSIFICATION_LABEL;
 import static org.apache.atlas.repository.Constants.PROPAGATED_TRAIT_NAMES_PROPERTY_KEY;
 import static org.apache.atlas.repository.graph.GraphHelper.EDGE_LABEL_PREFIX;
 import static org.apache.atlas.repository.graph.GraphHelper.addListProperty;
-import static org.apache.atlas.repository.graph.GraphHelper.getIncomingEdgesByLabel;
-import static org.apache.atlas.repository.graph.GraphHelper.getPropagatedEdgeLabel;
-import static org.apache.atlas.repository.graph.GraphHelper.getPropagatedTraitNames;
+import static org.apache.atlas.repository.graph.GraphHelper.getPropagatedEdges;
+import static org.apache.atlas.repository.graph.GraphHelper.getTraitNames;
 import static org.apache.atlas.repository.graph.GraphHelper.getTypeName;
 import static org.apache.atlas.repository.graph.GraphHelper.isRelationshipEdge;
 import static org.apache.atlas.repository.graph.GraphHelper.string;
@@ -350,21 +351,20 @@ public abstract class DeleteHandlerV1 {
         }
     }
 
-    public void removeTagPropagation(AtlasVertex classificationVertex) throws AtlasBaseException {
-        if (classificationVertex != null) {
-            String              classificationName = getTypeName(classificationVertex);
-            Iterator<AtlasEdge> iterator           = getIncomingEdgesByLabel(classificationVertex, getPropagatedEdgeLabel(classificationName));
+    public List<AtlasVertex> removeTagPropagation(AtlasVertex classificationVertex) throws AtlasBaseException {
+        List<AtlasVertex> ret = new ArrayList<>();
 
-            // remove classification from propagated entity vertices
-            while (iterator != null && iterator.hasNext()) {
-                AtlasEdge propagatedEdge = iterator.next();
+        if (classificationVertex != null) {
+            String          classificationName = getTypeName(classificationVertex);
+            List<AtlasEdge> propagatedEdges    = getPropagatedEdges(classificationVertex);
 
-                if (propagatedEdge != null) {
+            if (CollectionUtils.isNotEmpty(propagatedEdges)) {
+                for (AtlasEdge propagatedEdge : propagatedEdges) {
                     AtlasVertex propagatedEntityVertex = propagatedEdge.getOutVertex();
 
                     if (LOG.isDebugEnabled()) {
                         LOG.debug("Removing propagated classification: [{}] from: [{}][{}] with edge label: [{}]", classificationName,
-                                getTypeName(propagatedEntityVertex), GraphHelper.getGuid(propagatedEntityVertex), getPropagatedEdgeLabel(classificationName));
+                                getTypeName(propagatedEntityVertex), GraphHelper.getGuid(propagatedEntityVertex), CLASSIFICATION_LABEL);
                     }
 
                     removePropagatedTraitName(propagatedEntityVertex, classificationName);
@@ -372,14 +372,18 @@ public abstract class DeleteHandlerV1 {
                     deleteEdge(propagatedEdge, true);
 
                     updateModificationMetadata(propagatedEntityVertex);
+
+                    ret.add(propagatedEntityVertex);
                 }
             }
         }
+
+        return ret;
     }
 
     private void removePropagatedTraitName(AtlasVertex entityVertex, String classificationName) {
         if (entityVertex != null && StringUtils.isNotEmpty(classificationName)) {
-            List<String> propagatedTraitNames = getPropagatedTraitNames(entityVertex);
+            List<String> propagatedTraitNames = getTraitNames(entityVertex, true);
 
             propagatedTraitNames.remove(classificationName);
 

repository/src/test/java/org/apache/atlas/query/GremlinQueryComposerTest.java

@@ -36,13 +36,13 @@ import static org.testng.Assert.fail;
 public class GremlinQueryComposerTest {
     @Test
     public void classification() {
-        String expected = "g.V().or(has('__traitNames', within('PII')), has('__propagatedTraitNames', within('PII'))).dedup().limit(25).toList()";
+        String expected = "g.V().outE('classifiedAs').has('__name', within('PII')).outV().dedup().limit(25).toList()";
         verify("PII", expected);
     }
 
     @Test()
     public void dimension() {
-        String expected = "g.V().has('__typeName', 'Table').or(has('__traitNames', within('Dimension')), has('__propagatedTraitNames', within('Dimension'))).dedup().limit(25).toList()";
+        String expected = "g.V().has('__typeName', 'Table').outE('classifiedAs').has('__name', within('Dimension')).outV().dedup().limit(25).toList()";
         verify("Table isa Dimension", expected);
         verify("Table is Dimension", expected);
         verify("Table where Table is Dimension", expected);
@@ -295,14 +295,14 @@ public class GremlinQueryComposerTest {
     @Test
     public void keywordsInWhereClause() {
         verify("Table as t where t has name and t isa Dimension",
-                "g.V().has('__typeName', 'Table').as('t').and(__.has('Table.name'),__.or(has('__traitNames', within('Dimension')), has('__propagatedTraitNames', within('Dimension')))).dedup().limit(25).toList()");
+                "g.V().has('__typeName', 'Table').as('t').and(__.has('Table.name'),__.outE('classifiedAs').has('__name', within('Dimension')).outV()).dedup().limit(25).toList()");
         verify("Table as t where t has name and t.name = 'sales_fact'",
                 "g.V().has('__typeName', 'Table').as('t').and(__.has('Table.name'),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
         verify("Table as t where t is Dimension and t.name = 'sales_fact'",
-                "g.V().has('__typeName', 'Table').as('t').and(__.or(has('__traitNames', within('Dimension')), has('__propagatedTraitNames', within('Dimension'))),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
-        verify("Table isa 'Dimension' and name = 'sales_fact'", "g.V().has('__typeName', 'Table').and(__.or(has('__traitNames', within('Dimension')), has('__propagatedTraitNames', within('Dimension'))),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
+                "g.V().has('__typeName', 'Table').as('t').and(__.outE('classifiedAs').has('__name', within('Dimension')).outV(),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
+        verify("Table isa 'Dimension' and name = 'sales_fact'", "g.V().has('__typeName', 'Table').and(__.outE('classifiedAs').has('__name', within('Dimension')).outV(),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
         verify("Table has name and name = 'sales_fact'", "g.V().has('__typeName', 'Table').and(__.has('Table.name'),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
-        verify("Table is 'Dimension' and Table has owner and name = 'sales_fact'", "g.V().has('__typeName', 'Table').and(__.or(has('__traitNames', within('Dimension')), has('__propagatedTraitNames', within('Dimension'))),__.has('Table.owner'),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
+        verify("Table is 'Dimension' and Table has owner and name = 'sales_fact'", "g.V().has('__typeName', 'Table').and(__.outE('classifiedAs').has('__name', within('Dimension')).outV(),__.has('Table.owner'),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
         verify("Table has name and Table has owner and name = 'sales_fact'", "g.V().has('__typeName', 'Table').and(__.has('Table.name'),__.has('Table.owner'),__.has('Table.name', eq('sales_fact'))).dedup().limit(25).toList()");
     }