atlas-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From suma...@apache.org
Subject incubator-atlas git commit: ATLAS-463 Disconnect inverse references ( dkantor via sumasai)
Date Mon, 07 Mar 2016 23:21:41 GMT
Repository: incubator-atlas
Updated Branches:
  refs/heads/master a77d1ab53 -> 7ebb20136


ATLAS-463 Disconnect inverse references ( dkantor via sumasai)


Project: http://git-wip-us.apache.org/repos/asf/incubator-atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-atlas/commit/7ebb2013
Tree: http://git-wip-us.apache.org/repos/asf/incubator-atlas/tree/7ebb2013
Diff: http://git-wip-us.apache.org/repos/asf/incubator-atlas/diff/7ebb2013

Branch: refs/heads/master
Commit: 7ebb20136cefb6efe325ee1f30455abd9a3d57da
Parents: a77d1ab
Author: Suma Shivaprasad <sumasai.shivaprasad@gmail.com>
Authored: Mon Mar 7 15:21:20 2016 -0800
Committer: Suma Shivaprasad <sumasai.shivaprasad@gmail.com>
Committed: Mon Mar 7 15:21:20 2016 -0800

----------------------------------------------------------------------
 release-log.txt                                 |   1 +
 .../atlas/repository/graph/AtlasEdgeLabel.java  |  97 ++++
 .../graph/TypedInstanceToGraphMapper.java       | 285 ++++++++++--
 .../test/java/org/apache/atlas/TestUtils.java   |   2 +-
 ...kedMetadataRepositoryDeleteEntitiesTest.java | 443 ++++++++++++++++++-
 .../GraphBackedMetadataRepositoryTest.java      |  25 +-
 .../service/DefaultMetadataServiceTest.java     |   5 +-
 7 files changed, 811 insertions(+), 47 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/7ebb2013/release-log.txt
----------------------------------------------------------------------
diff --git a/release-log.txt b/release-log.txt
index d0a5fd5..94e0780 100644
--- a/release-log.txt
+++ b/release-log.txt
@@ -10,6 +10,7 @@ ATLAS-409 Atlas will not import avro tables with schema read from a file (dosset
 ATLAS-379 Create sqoop and falcon metadata addons (venkatnrangan,bvellanki,sowmyaramesh via shwethags)
 
 ALL CHANGES:
+ATLAS-463 Disconnect inverse references ( dkantor via sumasai)
 ATLAS-479 Add description for different types during create time (guptaneeru via shwethags)
 ATLAS-508 Apache nightly build failure - UnsupportedOperationException: Not a single key: __traitNames (shwethags)
 ATLAS-422 JavaDoc NotificationConsumer and NotificationInterface.(tbeerbower via sumasai)

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/7ebb2013/repository/src/main/java/org/apache/atlas/repository/graph/AtlasEdgeLabel.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/AtlasEdgeLabel.java b/repository/src/main/java/org/apache/atlas/repository/graph/AtlasEdgeLabel.java
new file mode 100644
index 0000000..da2ad9a
--- /dev/null
+++ b/repository/src/main/java/org/apache/atlas/repository/graph/AtlasEdgeLabel.java
@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.atlas.repository.graph;
+
+/**
+ * Represents an edge label used in Atlas.
+ * The format of an Atlas edge label is EDGE_LABEL_PREFIX<<typeName>>.<<attributeName>>[.mapKey]
+ *
+ */
+public class AtlasEdgeLabel {
+    private final String typeName_;
+    private final String attributeName_;
+    private final String mapKey_;
+    private final String edgeLabel_;
+    private final String qualifiedMapKey_;
+    private final String qualifiedAttributeName_;
+    
+    public AtlasEdgeLabel(String edgeLabel) {
+        if (!edgeLabel.startsWith(GraphHelper.EDGE_LABEL_PREFIX)) {
+            throw new IllegalArgumentException("Invalid edge label " + edgeLabel + ": missing required prefix " + GraphHelper.EDGE_LABEL_PREFIX);
+        }
+        String labelWithoutPrefix = edgeLabel.substring(GraphHelper.EDGE_LABEL_PREFIX.length());
+        String[] fields = labelWithoutPrefix.split("\\.", 3);
+        if (fields.length < 2 || fields.length > 3) {
+            throw new IllegalArgumentException("Invalid edge label " + edgeLabel + 
+                ": expected 2 or 3 label components but found " + fields.length); 
+        }
+        typeName_ = fields[0];
+        attributeName_ = fields[1];
+        if (fields.length == 3) {
+            mapKey_ = fields[2];
+            qualifiedMapKey_ = labelWithoutPrefix;
+            qualifiedAttributeName_ = typeName_ + '.' + attributeName_;
+        }
+        else {
+            mapKey_ = null;
+            qualifiedMapKey_ = null;
+            qualifiedAttributeName_ = labelWithoutPrefix;
+        }
+        edgeLabel_ = edgeLabel;
+    }
+    
+    public String getTypeName() {
+        return typeName_;
+    }
+    
+    public String getAttributeName() {
+        return attributeName_;
+    }
+    
+    public String getMapKey() {
+        return mapKey_;
+    }
+    
+    public String getEdgeLabel() {
+        return edgeLabel_;
+    }
+
+    public String getQualifiedMapKey() {
+        return qualifiedMapKey_;
+    }
+    
+    
+    public String getQualifiedAttributeName() {
+    
+        return qualifiedAttributeName_;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+        sb.append('(').append("typeName: ").append(typeName_);
+        sb.append(", attributeName: ").append(attributeName_);
+        if (mapKey_ != null) {
+            sb.append(", mapKey: ").append(mapKey_);
+            sb.append(", qualifiedMapKey: ").append(qualifiedMapKey_);
+        }
+        sb.append(", edgeLabel: ").append(edgeLabel_).append(')');
+        return sb.toString();
+    }
+    
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/7ebb2013/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java b/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
index 7b2890c..ed403e7 100644
--- a/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
+++ b/repository/src/main/java/org/apache/atlas/repository/graph/TypedInstanceToGraphMapper.java
@@ -21,6 +21,7 @@ import com.thinkaurelius.titan.core.SchemaViolationException;
 import com.tinkerpop.blueprints.Direction;
 import com.tinkerpop.blueprints.Edge;
 import com.tinkerpop.blueprints.Vertex;
+
 import org.apache.atlas.AtlasException;
 import org.apache.atlas.repository.Constants;
 import org.apache.atlas.repository.RepositoryException;
@@ -35,13 +36,17 @@ import org.apache.atlas.typesystem.persistence.ReferenceableInstance;
 import org.apache.atlas.typesystem.types.AttributeInfo;
 import org.apache.atlas.typesystem.types.ClassType;
 import org.apache.atlas.typesystem.types.DataTypes;
+import org.apache.atlas.typesystem.types.DataTypes.TypeCategory;
 import org.apache.atlas.typesystem.types.EnumValue;
+import org.apache.atlas.typesystem.types.IConstructableType;
 import org.apache.atlas.typesystem.types.IDataType;
 import org.apache.atlas.typesystem.types.Multiplicity;
 import org.apache.atlas.typesystem.types.ObjectGraphWalker;
+import org.apache.atlas.typesystem.types.StructType;
 import org.apache.atlas.typesystem.types.TraitType;
 import org.apache.atlas.typesystem.types.TypeSystem;
 import org.apache.atlas.typesystem.types.TypeUtils;
+import org.apache.atlas.typesystem.types.TypeUtils.Pair;
 import org.apache.atlas.utils.MD5Utils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -64,7 +69,6 @@ public final class TypedInstanceToGraphMapper {
     private final List<String> deletedEntityGuids = new ArrayList<>();
     private final List<ITypedReferenceableInstance> deletedEntities = new ArrayList<>();
     private final GraphToTypedInstanceMapper graphToTypedInstanceMapper;
-
     private static final GraphHelper graphHelper = GraphHelper.getInstance();
 
     private final String SIGNATURE_HASH_PROPERTY_KEY = Constants.INTERNAL_PROPERTY_KEY_PREFIX + "signature";
@@ -174,6 +178,7 @@ public final class TypedInstanceToGraphMapper {
     void mapInstanceToVertex(ITypedInstance typedInstance, Vertex instanceVertex,
                              Map<String, AttributeInfo> fields, boolean mapOnlyUniqueAttributes, Operation operation)
             throws AtlasException {
+        
         LOG.debug("Mapping instance {} of {} to vertex {}", typedInstance, typedInstance.getTypeName(),
                 instanceVertex);
         for (AttributeInfo attributeInfo : fields.values()) {
@@ -184,12 +189,27 @@ public final class TypedInstanceToGraphMapper {
         }
         
         if (operation == Operation.DELETE) {
+            // Remove uni-directional references to the deletion candidate.
+            removeUnidirectionalReferences(instanceVertex);
+            
             // Remove vertex for deletion candidate.
             graphHelper.removeVertex(instanceVertex);
         }
 
     }
 
+    private String getInstanceName(Vertex referencingVertex, IConstructableType referencingType) {
+
+        if (referencingType.getTypeCategory() == TypeCategory.CLASS) {
+            Id idFromVertex = GraphHelper.getIdFromVertex(referencingType.getName(), referencingVertex);
+            String instanceId = referencingType.getName() + ":" + idFromVertex._getId();
+            return instanceId;
+        }
+        else {
+            return referencingType.getName();
+        }
+    }
+
     void mapAttributesToVertex(ITypedInstance typedInstance, Vertex instanceVertex,
                                AttributeInfo attributeInfo, Operation operation) throws AtlasException {
         Object attrValue = typedInstance.get(attributeInfo.name);
@@ -361,7 +381,6 @@ public final class TypedInstanceToGraphMapper {
 
             IDataType elementType = ((DataTypes.ArrayType) attributeInfo.dataType()).getElemType();
             List<String> newEntries = new ArrayList<>();
-
             if (newElements != null && !newElements.isEmpty()) {
                 int index = 0;
                 for (; index < newElements.size(); index++) {
@@ -404,7 +423,7 @@ public final class TypedInstanceToGraphMapper {
         @SuppressWarnings("unchecked") Map<Object, Object> collection =
             (Map<Object, Object>) typedInstance.get(attributeInfo.name);
         boolean empty = (collection == null || collection.isEmpty());
-        if (!empty  || operation == Operation.UPDATE_FULL) {
+        if (!empty  || operation == Operation.UPDATE_FULL || operation == Operation.DELETE) {
 
             String propertyName = GraphHelper.getQualifiedFieldName(typedInstance, attributeInfo);
             IDataType elementType = ((DataTypes.MapType) attributeInfo.dataType()).getValueType();
@@ -420,23 +439,22 @@ public final class TypedInstanceToGraphMapper {
                     //Add/Update/Remove property value
                     GraphHelper.setProperty(instanceVertex, myPropertyName, newEntry);
                 }
-
-                //Remove unused key references
-                List<Object> origKeys = instanceVertex.getProperty(propertyName);
-                if (origKeys != null) {
-                    if (collection != null) {
-                        origKeys.removeAll(collection.keySet());
-                    }
-                    for (Object unusedKey : origKeys) {
-                        String edgeLabel = GraphHelper.getEdgeLabel(typedInstance, attributeInfo) + "." + unusedKey;
-                        if (instanceVertex.getEdges(Direction.OUT, edgeLabel).iterator().hasNext()) {
-                            Edge edge = instanceVertex.getEdges(Direction.OUT, edgeLabel).iterator().next();
-                            removeUnusedReference(edge.getId().toString(), attributeInfo,
-                                    ((DataTypes.MapType) attributeInfo.dataType()).getValueType());
-                        }
+            }
+            
+            //Remove unused key references
+            List<Object> origKeys = instanceVertex.getProperty(propertyName);
+            if (origKeys != null) {
+                if (collection != null) {
+                    origKeys.removeAll(collection.keySet());
+                }
+                for (Object unusedKey : origKeys) {
+                    String edgeLabel = GraphHelper.getEdgeLabel(typedInstance, attributeInfo) + "." + unusedKey;
+                    if (instanceVertex.getEdges(Direction.OUT, edgeLabel).iterator().hasNext()) {
+                        Edge edge = instanceVertex.getEdges(Direction.OUT, edgeLabel).iterator().next();
+                        removeUnusedReference(edge.getId().toString(), attributeInfo,
+                                ((DataTypes.MapType) attributeInfo.dataType()).getValueType());
                     }
                 }
-
             }
 
             // for dereference on way out
@@ -653,42 +671,241 @@ public final class TypedInstanceToGraphMapper {
     }
 
     private Edge removeUnusedReference(String edgeId, AttributeInfo attributeInfo, IDataType<?> elementType) throws AtlasException {
-        //Remove edges for property values which do not exist any more
+        TypeCategory typeCategory = elementType.getTypeCategory();
+        if (typeCategory != TypeCategory.STRUCT && elementType.getTypeCategory() != TypeCategory.CLASS) {
+            // Only class and struct references have edges.
+            return null;
+        }
+        
+        // Remove edge to disconnect struct or class reference.
+        // For struct or composite class reference, also delete the target instance.
         Edge removedRelation = null;
-        switch (elementType.getTypeCategory()) {
-        case STRUCT:
-            removedRelation = graphHelper.removeRelation(edgeId, true);
-            //Remove the vertex from state so that further processing no longer uses this
-            break;
-        case CLASS:
-            // TODO: disconnect inverse reference if attributeInfo.reverseAttributeName is non-null.
+        TypeUtils.Pair<Edge, Vertex> edgeAndVertex = graphHelper.getEdgeAndTargetVertex(edgeId);
+        if (typeCategory == TypeCategory.STRUCT) {
+            graphHelper.removeEdge(edgeAndVertex.left);
+            removedRelation = edgeAndVertex.left;
+
+            // Create an empty instance to use for clearing all struct attributes.
+            StructType structType = (StructType) elementType;
+            ITypedStruct typedInstance = structType.createInstance();
+            
+            //  Delete target vertex and any underlying structs and composite entities owned by this struct.
+            mapInstanceToVertex(typedInstance, edgeAndVertex.right, structType.fieldMapping().fields, false, Operation.DELETE);
+        }
+        else {
+            // Class reference
             if (attributeInfo.isComposite) {
-                // Delete contained entity.
-                TypeUtils.Pair<Edge, Vertex> edgeAndVertex = graphHelper.getEdgeAndTargetVertex(edgeId);
+                // For uni-directional reference, remove the edge.
+                // For bi-directional reference, the edges are removed
+                // when the composite entity is deleted.
+                if (attributeInfo.reverseAttributeName == null) {
+                    graphHelper.removeEdge(edgeAndVertex.left);
+                    removedRelation = edgeAndVertex.left;
+                }
+                // Delete the contained entity.
+                if (LOG.isDebugEnabled()) {
+                    Vertex sourceVertex = edgeAndVertex.left.getVertex(Direction.OUT);
+                    String sourceTypeName = GraphHelper.getTypeName(sourceVertex);
+                    LOG.debug("Deleting composite entity {}:{} contained by {}:{} through reference {}", 
+                        elementType.getName(), GraphHelper.getIdFromVertex(elementType.getName(), edgeAndVertex.right)._getId(),
+                        sourceTypeName, GraphHelper.getIdFromVertex(sourceTypeName, sourceVertex)._getId(),
+                        attributeInfo.name);
+                }
                 deleteEntity(elementType.getName(), edgeAndVertex.right);
-                graphHelper.removeEdge(edgeAndVertex.left);
-                removedRelation = edgeAndVertex.left;
             }
             else {
-                removedRelation = graphHelper.removeRelation(edgeId, false);
+                if (attributeInfo.reverseAttributeName != null) {
+                    // Disconnect both ends of the bi-directional reference
+                    removeReverseReference(edgeAndVertex, attributeInfo);
+                }
+                graphHelper.removeEdge(edgeAndVertex.left);
+                removedRelation = edgeAndVertex.left;
             }
-            break;
         }
         return removedRelation;
     }
+
+    /**
+     * Remove the reverse reference value for the specified edge and vertex.
+     * 
+     * @param edgeAndVertex
+     * @param attributeInfo
+     * @throws AtlasException
+     */
+    private void removeReverseReference(TypeUtils.Pair<Edge, Vertex> edgeAndVertex, 
+        AttributeInfo attributeInfo) throws AtlasException {
+        
+        Vertex sourceVertex = edgeAndVertex.left.getVertex(Direction.OUT);
+        String inverseTypeName = GraphHelper.getTypeName(edgeAndVertex.right);
+        IConstructableType inverseType = typeSystem.getDataType(IConstructableType.class, inverseTypeName);
+        AttributeInfo inverseAttributeInfo = inverseType.fieldMapping().fields.get(attributeInfo.reverseAttributeName);
+        String inverseEdgeLabel = GraphHelper.getEdgeLabel(inverseType, inverseAttributeInfo);
+        TypeCategory inverseTypeCategory = inverseAttributeInfo.dataType().getTypeCategory();
     
+        // Find and remove the edge which represents the inverse reference value.
+        Iterable<Edge> inverseEdges = GraphHelper.getOutGoingEdgesByLabel(edgeAndVertex.right, inverseEdgeLabel);
+        Edge removedEdge = null;
+        // Search for the edge which references the source vertex.
+        for (Edge edge : inverseEdges) {
+            Vertex vertex = edge.getVertex(Direction.IN);
+            if (vertex.equals(sourceVertex)) {
+                // Found the edge which points back at source vertex.
+                // Disconnect the reference by removing the edge and
+                // removing the edge ID from the vertex property.
+                removeReferenceValue(edge, new AtlasEdgeLabel(edge.getLabel()), edgeAndVertex.right, inverseType, inverseTypeCategory);
+                removedEdge = edge;
+                break;
+            }
+        }
+        if (removedEdge != null) {
+            if (LOG.isDebugEnabled()) {
+                String sourceTypeName = GraphHelper.getTypeName(sourceVertex);
+                LOG.debug("Removed edge {} for reverse reference {} from {}:{} to {}:{} ", removedEdge,
+                    GraphHelper.getQualifiedFieldName(inverseType, inverseAttributeInfo.name), 
+                    inverseTypeName, GraphHelper.getIdFromVertex(inverseTypeName, edgeAndVertex.right)._getId(), 
+                    sourceTypeName, GraphHelper.getIdFromVertex(sourceTypeName, sourceVertex)._getId());
+            }
+        }
+        else {
+            // We didn't find the edge for the inverse reference.
+            // Since Atlas currently does not automatically set
+            // the inverse reference when a reference value is updated,
+            // unbalanced references are not unexpected.
+            // The presence of inverse reference values depends on
+            // well behaved client applications which explicitly set
+            // both ends of the reference.
+            // TODO: throw an exception as it indicates a unbalanced reference?
+            String sourceTypeName = GraphHelper.getTypeName(sourceVertex);
+            LOG.warn("No edge found for inverse reference {} on vertex {} for entity instance {}:{} which points back to vertex {} for {}:{}",
+                inverseAttributeInfo.name, edgeAndVertex.right,
+                inverseTypeName, GraphHelper.getIdFromVertex(inverseTypeName, edgeAndVertex.right)._getId(), 
+                sourceVertex, sourceTypeName, GraphHelper.getIdFromVertex(sourceTypeName, sourceVertex)._getId());
+        }
+    }
+
+    /**
+     * Remove any unidirectional map or array reference to a class, struct, or trait vertex.
+     * This involves removing appropriate value from the vertex property which holds the
+     * reference values.
+     * 
+     * @param targetVertex a vertex which represents a class, struct, or trait instance
+     * @throws AtlasException
+     */
+    private void removeUnidirectionalReferences(Vertex targetVertex) throws AtlasException {
+    
+        // Search for any remaining incoming edges that represent unidirectional references
+        // to the target vertex.
+        Iterable<Edge> incomingEdges = targetVertex.getEdges(Direction.IN);
+        for (Edge edge : incomingEdges) {
+            String label = edge.getLabel();
+            AtlasEdgeLabel atlasEdgeLabel = new AtlasEdgeLabel(label);
+            Vertex referencingVertex = edge.getVertex(Direction.OUT);
+            String typeName = atlasEdgeLabel.getTypeName();
+            IConstructableType referencingType = typeSystem.getDataType(IConstructableType.class, typeName);
+            
+            AttributeInfo attributeInfo = referencingType.fieldMapping().fields.get(atlasEdgeLabel.getAttributeName());
+            if (attributeInfo == null) {
+                String instanceId = getInstanceName(referencingVertex, referencingType);
+                throw new AtlasException("Outgoing edge " + edge.getId().toString()  
+                    + " for " + instanceId + "(vertex " + referencingVertex + "): label " + label
+                    + " has an attribute name " + atlasEdgeLabel.getAttributeName() + " that is undefined on "
+                    + referencingType.getTypeCategory() + " " + typeName);
+            }
+            // Remove the appropriate value from the vertex property for this reference.
+            removeReferenceValue(edge, atlasEdgeLabel, referencingVertex, referencingType, attributeInfo.dataType().getTypeCategory());
+        }
+    }
+
+    private Pair<String, Boolean> removeReferenceValue(Edge edge, AtlasEdgeLabel atlasEdgeLabel,
+            Vertex referencingVertex, IConstructableType referencingType, TypeCategory attrTypeCategory) 
+        throws AtlasException {
+    
+        graphHelper.removeEdge(edge);
+        if (attrTypeCategory != TypeCategory.ARRAY && attrTypeCategory != TypeCategory.MAP) {
+            // Multiplicity-one reference is represented by the edge,
+            // there is no vertex property to update. So just remove the edge.
+            return new Pair<String, Boolean>(edge.getId().toString(), Boolean.TRUE);
+        }
+        List<String> currentRefValues = referencingVertex.getProperty(atlasEdgeLabel.getQualifiedAttributeName());
+        List<String> newRefValues = new ArrayList<>(currentRefValues);
+        Pair<String, Boolean> refValueRemoved = null;
+        if (attrTypeCategory == TypeCategory.ARRAY) {
+            refValueRemoved = removeArrayReferenceValue(atlasEdgeLabel, referencingVertex, edge, newRefValues);
+        }
+        else {
+            refValueRemoved = removeMapReferenceValue(atlasEdgeLabel, referencingVertex, edge, newRefValues);
+        }
+        if (refValueRemoved.right) {
+            if (LOG.isDebugEnabled()) {
+                String instanceId = getInstanceName(referencingVertex, referencingType);
+                LOG.debug("Reference value {} removed from reference {} on vertex {} for instance of {} {}",
+                    refValueRemoved.left, atlasEdgeLabel.getAttributeName(), referencingVertex,
+                    referencingType.getTypeCategory(), instanceId);
+            }
+            // If the referencing instance is an entity, update the modification timestamp.
+            if (referencingType instanceof ClassType) {
+                GraphHelper.setProperty(referencingVertex, Constants.MODIFICATION_TIMESTAMP_PROPERTY_KEY, System.currentTimeMillis());
+            }
+        }
+        else {
+            // The expected value is missing from the reference property values - log a warning.
+            String instanceId = getInstanceName(referencingVertex, referencingType);
+            LOG.warn("Reference value {} expected but not found in array reference {} on vertex {} for instance of {} {}",
+                refValueRemoved.left, atlasEdgeLabel.getAttributeName(), referencingVertex,
+                referencingType.getTypeCategory(), instanceId);
+        }
+        return refValueRemoved;
+    }
+
+    private TypeUtils.Pair<String, Boolean> removeArrayReferenceValue(AtlasEdgeLabel atlasEdgeLabel, Vertex referencingVertex, 
+        Edge edge, List<String> newRefValues) {
+        
+        String refValueToRemove = edge.getId().toString();
+        boolean valueRemoved = newRefValues.remove(refValueToRemove);
+        if (valueRemoved) {
+            GraphHelper.setProperty(referencingVertex, atlasEdgeLabel.getQualifiedAttributeName(), newRefValues);
+        }
+        return new TypeUtils.Pair<String, Boolean>(refValueToRemove, Boolean.valueOf(valueRemoved));
+    }
+
+    private TypeUtils.Pair<String, Boolean>  removeMapReferenceValue(AtlasEdgeLabel atlasEdgeLabel, Vertex referencingVertex, 
+        Edge edge, List<String> newRefValues) throws AtlasException {
+        
+        String refValueToRemove = atlasEdgeLabel.getMapKey();
+        if (refValueToRemove == null) {
+            // Edge label is missing the map key - throw an exception.
+            String typeName = atlasEdgeLabel.getTypeName();
+            throw new AtlasException("Outgoing edge " + edge.getId().toString()  
+                + " for vertex " + referencingVertex + "): label " + atlasEdgeLabel.getEdgeLabel()
+                + " for map attribute " + atlasEdgeLabel.getAttributeName() + " on type "
+                + typeName + " is missing the map key");
+        }
+        boolean valueRemoved = newRefValues.remove(refValueToRemove);
+        if (valueRemoved) {
+            GraphHelper.setProperty(referencingVertex, atlasEdgeLabel.getQualifiedAttributeName(), newRefValues);
+            // For maps, also remove the key-value pair property value.
+            GraphHelper.setProperty(referencingVertex, atlasEdgeLabel.getQualifiedMapKey(), null);
+        }
+        return new TypeUtils.Pair<String, Boolean>(refValueToRemove, Boolean.valueOf(valueRemoved));
+    }
+
     void deleteEntity(String typeName, Vertex instanceVertex) throws AtlasException {
+        // Check if this entity has already been processed.
+        Id id = GraphHelper.getIdFromVertex(typeName, instanceVertex);
+        if (deletedEntityGuids.contains(id._getId())) {
+            return;
+        }
+        deletedEntityGuids.add(id._getId());
+        
         // Remove traits owned by this entity.
         deleteAllTraits(instanceVertex);
         
         // Create an empty instance to use for clearing all attributes.
-        Id id = GraphHelper.getIdFromVertex(typeName, instanceVertex);
         ClassType classType = typeSystem.getDataType(ClassType.class, typeName);
         ITypedReferenceableInstance typedInstance = classType.createInstance(id);
         
         //  Remove any underlying structs and composite entities owned by this entity.
         mapInstanceToVertex(typedInstance, instanceVertex, classType.fieldMapping().fields, false, Operation.DELETE);
-        deletedEntityGuids.add(id._getId());
         deletedEntities.add(typedInstance);
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/7ebb2013/repository/src/test/java/org/apache/atlas/TestUtils.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/TestUtils.java b/repository/src/test/java/org/apache/atlas/TestUtils.java
index d54dfab..ce59c43 100755
--- a/repository/src/test/java/org/apache/atlas/TestUtils.java
+++ b/repository/src/test/java/org/apache/atlas/TestUtils.java
@@ -170,7 +170,7 @@ public final class TestUtils {
         max.set("mentor", julius);
         
         john.set("manager", jane);
-
+        john.set("mentor", max);
         hrDept.set("employees", ImmutableList.of(john, jane, julius, max));
 
         jane.set("subordinates", ImmutableList.of(john, max));

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/7ebb2013/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryDeleteEntitiesTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryDeleteEntitiesTest.java b/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryDeleteEntitiesTest.java
index b024152..3681de7 100644
--- a/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryDeleteEntitiesTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryDeleteEntitiesTest.java
@@ -18,6 +18,10 @@
 
 package org.apache.atlas.repository.graph;
 
+import static org.apache.atlas.typesystem.types.utils.TypesUtil.createOptionalAttrDef;
+import static org.apache.atlas.typesystem.types.utils.TypesUtil.createRequiredAttrDef;
+
+import com.google.common.collect.ImmutableList;
 import com.thinkaurelius.titan.core.TitanGraph;
 import com.thinkaurelius.titan.core.util.TitanCleanup;
 import com.tinkerpop.blueprints.Vertex;
@@ -27,13 +31,25 @@ import org.apache.atlas.TestUtils;
 import org.apache.atlas.discovery.graph.GraphBackedDiscoveryService;
 import org.apache.atlas.repository.Constants;
 import org.apache.atlas.repository.RepositoryException;
+import org.apache.atlas.typesystem.IStruct;
 import org.apache.atlas.typesystem.ITypedReferenceableInstance;
+import org.apache.atlas.typesystem.ITypedStruct;
 import org.apache.atlas.typesystem.Referenceable;
+import org.apache.atlas.typesystem.Struct;
+import org.apache.atlas.typesystem.TypesDef;
 import org.apache.atlas.typesystem.exception.EntityNotFoundException;
+import org.apache.atlas.typesystem.persistence.Id;
+import org.apache.atlas.typesystem.types.AttributeDefinition;
 import org.apache.atlas.typesystem.types.ClassType;
+import org.apache.atlas.typesystem.types.DataTypes;
+import org.apache.atlas.typesystem.types.EnumTypeDefinition;
+import org.apache.atlas.typesystem.types.HierarchicalTypeDefinition;
 import org.apache.atlas.typesystem.types.Multiplicity;
+import org.apache.atlas.typesystem.types.StructTypeDefinition;
+import org.apache.atlas.typesystem.types.TraitType;
 import org.apache.atlas.typesystem.types.TypeSystem;
 import org.apache.atlas.typesystem.types.TypeUtils.Pair;
+import org.apache.atlas.typesystem.types.utils.TypesUtil;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -44,7 +60,10 @@ import javax.inject.Inject;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Test for GraphBackedMetadataRepository.deleteEntities
@@ -95,8 +114,12 @@ public class GraphBackedMetadataRepositoryDeleteEntitiesTest {
 
 
 
+    /**
+     * Verify deleting entities with composite references to other entities.
+     * The composite entities should also be deleted.
+     */
     @Test
-    public void testDeleteEntities() throws Exception {
+    public void testDeleteEntitiesWithCompositeArrayReference() throws Exception {
         String hrDeptGuid = createHrDeptGraph();
 
         ITypedReferenceableInstance hrDept = repositoryService.getEntityDefinition(hrDeptGuid);
@@ -136,25 +159,397 @@ public class GraphBackedMetadataRepositoryDeleteEntitiesTest {
         Assert.assertEquals(vertexCount, 0);
     }
 
-    @Test(dependsOnMethods = "testDeleteEntities")
-    public void testDeleteContainedEntity() throws Exception {
+    @Test
+    public void testDeleteEntitiesWithCompositeMapReference() throws Exception {
+        // Define type for map value.
+        HierarchicalTypeDefinition<ClassType> mapValueDef = TypesUtil.createClassTypeDef("CompositeMapValue", 
+            ImmutableList.<String>of(),
+            TypesUtil.createOptionalAttrDef("attr1", DataTypes.STRING_TYPE));
+        
+        // Define type with map where the value is a composite class reference to MapValue.
+        HierarchicalTypeDefinition<ClassType> mapOwnerDef = TypesUtil.createClassTypeDef("CompositeMapOwner", 
+            ImmutableList.<String>of(),
+            new AttributeDefinition("map", DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
+                        "CompositeMapValue"), Multiplicity.OPTIONAL, true, null));
+        TypesDef typesDef = TypesUtil.getTypesDef(ImmutableList.<EnumTypeDefinition>of(),
+            ImmutableList.<StructTypeDefinition>of(), ImmutableList.<HierarchicalTypeDefinition<TraitType>>of(),
+            ImmutableList.of(mapOwnerDef, mapValueDef));
+        typeSystem.defineTypes(typesDef);
+        ClassType mapOwnerType = typeSystem.getDataType(ClassType.class, "CompositeMapOwner");
+        ClassType mapValueType = typeSystem.getDataType(ClassType.class, "CompositeMapValue");
+        
+        // Create instances of MapOwner and MapValue.
+        // Set MapOwner.map with one entry that references MapValue instance.
+        ITypedReferenceableInstance mapOwnerInstance = mapOwnerType.createInstance();
+        ITypedReferenceableInstance mapValueInstance = mapValueType.createInstance();
+        mapOwnerInstance.set("map", Collections.singletonMap("value1", mapValueInstance));
+        List<String> createEntitiesResult = repositoryService.createEntities(mapOwnerInstance, mapValueInstance);
+        Assert.assertEquals(createEntitiesResult.size(), 2);
+        List<String> guids = repositoryService.getEntityList("CompositeMapOwner");
+        Assert.assertEquals(guids.size(), 1);
+        String mapOwnerGuid = guids.get(0);
+        
+        // Verify MapOwner.map attribute has expected value.
+        mapOwnerInstance = repositoryService.getEntityDefinition(mapOwnerGuid);
+        Object object = mapOwnerInstance.get("map");
+        Assert.assertNotNull(object);
+        Assert.assertTrue(object instanceof Map);
+        Map<String, ITypedReferenceableInstance> map = (Map<String, ITypedReferenceableInstance>)object;
+        Assert.assertEquals(map.size(), 1);
+        mapValueInstance = map.get("value1");
+        Assert.assertNotNull(mapValueInstance);
+        String mapValueGuid = mapValueInstance.getId()._getId();
+        String edgeLabel = GraphHelper.getEdgeLabel(mapOwnerType, mapOwnerType.fieldMapping.fields.get("map"));
+        String mapEntryLabel = edgeLabel + "." + "value1";
+        AtlasEdgeLabel atlasEdgeLabel = new AtlasEdgeLabel(mapEntryLabel);
+        Vertex mapOwnerVertex = GraphHelper.getInstance().getVertexForGUID(mapOwnerGuid);
+        object = mapOwnerVertex.getProperty(atlasEdgeLabel.getQualifiedMapKey());
+        Assert.assertNotNull(object);
+        
+        Pair<List<String>, List<ITypedReferenceableInstance>> deleteEntitiesResult = 
+            repositoryService.deleteEntities(Arrays.asList(mapOwnerGuid));
+        Assert.assertEquals(deleteEntitiesResult.left.size(), 2);
+        Assert.assertTrue(deleteEntitiesResult.left.containsAll(guids));
+        verifyEntityDoesNotExist(mapOwnerGuid);
+        verifyEntityDoesNotExist(mapValueGuid);
+    }
+    
+    /**
+     * Verify deleting an entity which is contained by another
+     * entity through a bi-directional composite reference.
+     * 
+     * @throws Exception
+     */
+    @Test(dependsOnMethods = "testDeleteEntitiesWithCompositeArrayReference")
+    public void testDisconnectBidirectionalReferences() throws Exception {
         String hrDeptGuid = createHrDeptGraph();
         ITypedReferenceableInstance hrDept = repositoryService.getEntityDefinition(hrDeptGuid);
         Object refValue = hrDept.get("employees");
         Assert.assertTrue(refValue instanceof List);
         List<Object> employees = (List<Object>)refValue;
         Assert.assertEquals(employees.size(), 4);
-        Object listValue = employees.get(2);
-        Assert.assertTrue(listValue instanceof ITypedReferenceableInstance);
-        ITypedReferenceableInstance employee = (ITypedReferenceableInstance) listValue;
-        String employeeGuid = employee.getId()._getId();
+        String employeeGuid = null;
+        for (Object listValue : employees) {
+            Assert.assertTrue(listValue instanceof ITypedReferenceableInstance);
+            ITypedReferenceableInstance employee = (ITypedReferenceableInstance) listValue;
+            if (employee.get("name").equals("Max")) {
+                employeeGuid = employee.getId()._getId();
+            }
+        }
+        Assert.assertNotNull(employeeGuid);
+        
+        // Verify that Max is one of Jane's subordinates.
+        ITypedReferenceableInstance jane = repositoryService.getEntityDefinition("Manager", "name", "Jane");
+        refValue = jane.get("subordinates");
+        Assert.assertTrue(refValue instanceof List);
+        List<Object> subordinates = (List<Object>)refValue;
+        Assert.assertEquals(subordinates.size(), 2);
+        List<String> subordinateIds = new ArrayList<>(2);
+        for (Object listValue : employees) {
+            Assert.assertTrue(listValue instanceof ITypedReferenceableInstance);
+            ITypedReferenceableInstance employee = (ITypedReferenceableInstance) listValue;
+            subordinateIds.add(employee.getId()._getId());
+        }
+        Assert.assertTrue(subordinateIds.contains(employeeGuid));
         
         Pair<List<String>, List<ITypedReferenceableInstance>> deletedEntities = repositoryService.deleteEntities(Arrays.asList(employeeGuid));
         Assert.assertTrue(deletedEntities.left.contains(employeeGuid));
         verifyEntityDoesNotExist(employeeGuid);
         
+        // Verify that the Department.employees reference to the deleted employee
+        // was disconnected.
+        hrDept = repositoryService.getEntityDefinition(hrDeptGuid);
+        refValue = hrDept.get("employees");
+        Assert.assertTrue(refValue instanceof List);
+        employees = (List<Object>)refValue;
+        Assert.assertEquals(employees.size(), 3);
+        for (Object listValue : employees) {
+            Assert.assertTrue(listValue instanceof ITypedReferenceableInstance);
+            ITypedReferenceableInstance employee = (ITypedReferenceableInstance) listValue;
+            Assert.assertNotEquals(employee.getId()._getId(), employeeGuid);
+        }
+        
+        // Verify that the Manager.subordinates reference to the deleted employee
+        // Max was disconnected.
+        jane = repositoryService.getEntityDefinition("Manager", "name", "Jane");
+        refValue = jane.get("subordinates");
+        Assert.assertTrue(refValue instanceof List);
+        subordinates = (List<Object>)refValue;
+        Assert.assertEquals(subordinates.size(), 1);
+        Object listValue = subordinates.get(0);
+        Assert.assertTrue(listValue instanceof ITypedReferenceableInstance);
+        ITypedReferenceableInstance subordinate = (ITypedReferenceableInstance) listValue;
+        String subordinateGuid = subordinate.getId()._getId();
+        Assert.assertNotEquals(subordinateGuid, employeeGuid);
+        
+        // Verify that max's Person.mentor unidirectional reference to john was disconnected.
+        ITypedReferenceableInstance john = repositoryService.getEntityDefinition("Manager", "name", "John");
+        refValue = john.get("mentor");
+        Assert.assertNull(refValue);
+        
+        // Now delete jane - this should disconnect the manager reference from her
+        // subordinate.
+        String janeGuid = jane.getId()._getId();
+        deletedEntities = repositoryService.deleteEntities(Arrays.asList(janeGuid));
+        Assert.assertTrue(deletedEntities.left.contains(janeGuid));
+        verifyEntityDoesNotExist(janeGuid);
+        subordinate = repositoryService.getEntityDefinition(subordinateGuid);
+        Assert.assertNull(subordinate.get("manager"));
     }
 
+    /**
+     * Verify deleting entity that is the target of a unidirectional class array reference
+     * from a class instance.
+     */
+    @Test
+    public void testDisconnectUnidirectionalArrayReferenceFromClassType() throws Exception {
+        createDbTableGraph();
+        
+        // Get the guid for one of the table's columns.
+        ITypedReferenceableInstance table = repositoryService.getEntityDefinition(TestUtils.TABLE_TYPE, "name", TestUtils.TABLE_NAME);
+        String tableGuid = table.getId()._getId();
+        Object refValues = table.get("columns");
+        Assert.assertTrue(refValues instanceof List);
+        List<Object> refList = (List<Object>) refValues;
+        Assert.assertEquals(refList.size(), 5);
+        Assert.assertTrue(refList.get(0) instanceof ITypedReferenceableInstance);
+        ITypedReferenceableInstance column = (ITypedReferenceableInstance) refList.get(0);
+        String columnGuid = column.getId()._getId();
+        
+        // Delete the column.
+        Pair<List<String>, List<ITypedReferenceableInstance>> deletedEntities = 
+            repositoryService.deleteEntities(Arrays.asList(columnGuid));
+        Assert.assertEquals(deletedEntities.left.size(), 1);
+        Assert.assertEquals(deletedEntities.right.size(), 1);
+        verifyEntityDoesNotExist(columnGuid);
+        
+        // Verify table.columns reference to the deleted column has been disconnected.
+        table = repositoryService.getEntityDefinition(tableGuid);
+        refList = (List<Object>) table.get("columns");
+        Assert.assertEquals(refList.size(), 4);
+        for (Object refValue : refList) {
+            Assert.assertTrue(refValue instanceof ITypedReferenceableInstance);
+            column = (ITypedReferenceableInstance)refValue;
+            Assert.assertFalse(column.getId()._getId().equals(columnGuid));
+        }
+    }
+    
+    /**
+     * Verify deleting entities that are the target of a unidirectional class array reference
+     * from a struct or trait instance.
+     */
+    @Test
+    public void testDisconnectUnidirectionalArrayReferenceFromStructAndTraitTypes() throws Exception {
+        // Define class types.
+        HierarchicalTypeDefinition<ClassType> structTargetDef = TypesUtil.createClassTypeDef("StructTarget", 
+            ImmutableList.<String>of(), TypesUtil.createOptionalAttrDef("attr1", DataTypes.STRING_TYPE));
+        HierarchicalTypeDefinition<ClassType> traitTargetDef = TypesUtil.createClassTypeDef("TraitTarget", 
+            ImmutableList.<String>of(), TypesUtil.createOptionalAttrDef("attr1", DataTypes.STRING_TYPE));
+        HierarchicalTypeDefinition<ClassType> structContainerDef = TypesUtil.createClassTypeDef("StructContainer", 
+            ImmutableList.<String>of(), TypesUtil.createOptionalAttrDef("struct", "TestStruct"));
+        
+        // Define struct and trait types which have a unidirectional array reference
+        // to a class type.
+        StructTypeDefinition structDef = TypesUtil.createStructTypeDef("TestStruct", 
+            new AttributeDefinition("target", DataTypes.arrayTypeName("StructTarget"), Multiplicity.OPTIONAL, false, null),
+            new AttributeDefinition("nestedStructs", DataTypes.arrayTypeName("NestedStruct"), Multiplicity.OPTIONAL, false, null));
+        StructTypeDefinition nestedStructDef = TypesUtil.createStructTypeDef("NestedStruct", 
+            TypesUtil.createOptionalAttrDef("attr1", DataTypes.STRING_TYPE));
+        HierarchicalTypeDefinition<TraitType> traitDef = TypesUtil.createTraitTypeDef("TestTrait", ImmutableList.<String>of(), 
+            new AttributeDefinition("target", DataTypes.arrayTypeName("TraitTarget"), Multiplicity.OPTIONAL, false, null));
+        
+        TypesDef typesDef = TypesUtil.getTypesDef(ImmutableList.<EnumTypeDefinition>of(), ImmutableList.of(structDef, nestedStructDef),
+            ImmutableList.of(traitDef), ImmutableList.of(structTargetDef, traitTargetDef, structContainerDef));
+        typeSystem.defineTypes(typesDef);
+        
+        // Create instances of class, struct, and trait types.
+        Referenceable structTargetEntity = new Referenceable("StructTarget");
+        Referenceable traitTargetEntity = new Referenceable("TraitTarget");
+        Referenceable structContainerEntity = new Referenceable("StructContainer");
+        Referenceable structInstance = new Referenceable("TestStruct");
+        Referenceable nestedStructInstance = new Referenceable("NestedStruct");
+        Referenceable traitInstance = new Referenceable("TestTrait");
+        structContainerEntity.set("struct", structInstance);
+        structInstance.set("target", ImmutableList.of(structTargetEntity));
+        structInstance.set("nestedStructs", ImmutableList.of(nestedStructInstance));
+        
+        ClassType structTargetType = typeSystem.getDataType(ClassType.class, "StructTarget");
+        ClassType traitTargetType = typeSystem.getDataType(ClassType.class, "TraitTarget");
+        ClassType structContainerType = typeSystem.getDataType(ClassType.class, "StructContainer");
+        
+        ITypedReferenceableInstance structTargetConvertedEntity = 
+            structTargetType.convert(structTargetEntity, Multiplicity.REQUIRED);
+        ITypedReferenceableInstance traitTargetConvertedEntity = 
+            traitTargetType.convert(traitTargetEntity, Multiplicity.REQUIRED);
+        ITypedReferenceableInstance structContainerConvertedEntity =
+            structContainerType.convert(structContainerEntity, Multiplicity.REQUIRED);
+        
+        List<String> guids = repositoryService.createEntities(
+            structTargetConvertedEntity, traitTargetConvertedEntity, structContainerConvertedEntity);
+        Assert.assertEquals(guids.size(), 3);
+        
+        guids = repositoryService.getEntityList("StructTarget");
+        Assert.assertEquals(guids.size(), 1);
+        String structTargetGuid = guids.get(0);
+        
+        guids = repositoryService.getEntityList("TraitTarget");
+        Assert.assertEquals(guids.size(), 1);
+        String traitTargetGuid = guids.get(0);
+
+        guids = repositoryService.getEntityList("StructContainer");
+        Assert.assertEquals(guids.size(), 1);
+        String structContainerGuid = guids.get(0);
+        
+        // Add TestTrait to StructContainer instance
+        traitInstance.set("target", ImmutableList.of(new Id(traitTargetGuid, 0, "TraitTarget")));
+        TraitType traitType = typeSystem.getDataType(TraitType.class, "TestTrait");
+        ITypedStruct convertedTrait = traitType.convert(traitInstance, Multiplicity.REQUIRED);
+        repositoryService.addTrait(structContainerGuid, convertedTrait);
+        
+        // Verify that the unidirectional references from the struct and trait instances
+        // are pointing at the target entities.
+        structContainerConvertedEntity = repositoryService.getEntityDefinition(structContainerGuid);
+        Object object = structContainerConvertedEntity.get("struct");
+        Assert.assertNotNull(object);
+        Assert.assertTrue(object instanceof ITypedStruct);
+        ITypedStruct struct = (ITypedStruct) object;
+        object = struct.get("target");
+        Assert.assertNotNull(object);
+        Assert.assertTrue(object instanceof List);
+        List<ITypedReferenceableInstance> refList = (List<ITypedReferenceableInstance>)object;
+        Assert.assertEquals(refList.size(), 1);
+        Assert.assertEquals(refList.get(0).getId()._getId(), structTargetGuid);
+        
+        IStruct trait = structContainerConvertedEntity.getTrait("TestTrait");
+        Assert.assertNotNull(trait);
+        object = trait.get("target");
+        Assert.assertNotNull(object);
+        Assert.assertTrue(object instanceof List);
+        refList = (List<ITypedReferenceableInstance>)object;
+        Assert.assertEquals(refList.size(), 1);
+        Assert.assertEquals(refList.get(0).getId()._getId(), traitTargetGuid);
+        
+        // Delete the entities that are targets of the struct and trait instances.
+        Pair<List<String>, List<ITypedReferenceableInstance>> deleteEntitiesResult = 
+            repositoryService.deleteEntities(Arrays.asList(structTargetGuid, traitTargetGuid));
+        verifyEntityDoesNotExist(structTargetGuid);
+        verifyEntityDoesNotExist(traitTargetGuid);
+        Assert.assertEquals(deleteEntitiesResult.left.size(), 2);
+        Assert.assertTrue(deleteEntitiesResult.left.containsAll(Arrays.asList(structTargetGuid, traitTargetGuid)));
+        
+        // Verify that the unidirectional references from the struct and trait instances
+        // to the deleted entities were disconnected.
+        structContainerConvertedEntity = repositoryService.getEntityDefinition(structContainerGuid);
+        object = structContainerConvertedEntity.get("struct");
+        Assert.assertNotNull(object);
+        Assert.assertTrue(object instanceof ITypedStruct);
+        struct = (ITypedStruct) object;
+        Assert.assertNull(struct.get("target"));
+        trait = structContainerConvertedEntity.getTrait("TestTrait");
+        Assert.assertNotNull(trait);
+        Assert.assertNull(trait.get("target"));
+        
+        // Delete the entity which contains nested structs and has the TestTrait trait.
+        deleteEntitiesResult = 
+            repositoryService.deleteEntities(Arrays.asList(structContainerGuid));
+        verifyEntityDoesNotExist(structContainerGuid);
+        Assert.assertEquals(deleteEntitiesResult.left.size(), 1);
+        Assert.assertTrue(deleteEntitiesResult.left.contains(structContainerGuid));
+        
+        // Verify all TestStruct struct vertices were removed.
+        int vertexCount = countVertices(Constants.ENTITY_TYPE_PROPERTY_KEY, "TestStruct");
+        Assert.assertEquals(vertexCount, 0);
+        
+        // Verify all NestedStruct struct vertices were removed.
+        vertexCount = countVertices(Constants.ENTITY_TYPE_PROPERTY_KEY, "NestedStruct");
+        Assert.assertEquals(vertexCount, 0);
+
+        // Verify all TestTrait trait vertices were removed.
+        vertexCount = countVertices(Constants.ENTITY_TYPE_PROPERTY_KEY, "TestTrait");
+        Assert.assertEquals(vertexCount, 0);
+    }
+    
+    /**
+     * Verify deleting entities that are the target of class map references.
+     */
+    @Test
+    public void testDisconnectMapReferenceFromClassType() throws Exception {
+        // Define type for map value.
+        HierarchicalTypeDefinition<ClassType> mapValueDef = TypesUtil.createClassTypeDef("MapValue", 
+            ImmutableList.<String>of(),
+            new AttributeDefinition("biMapOwner", "MapOwner", Multiplicity.OPTIONAL, false, "biMap"));
+        
+        // Define type with unidirectional and bidirectional map references,
+        // where the map value is a class reference to MapValue.
+        HierarchicalTypeDefinition<ClassType> mapOwnerDef = TypesUtil.createClassTypeDef("MapOwner", 
+            ImmutableList.<String>of(),
+            new AttributeDefinition("map", DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
+                        "MapValue"), Multiplicity.OPTIONAL, false, null),
+            new AttributeDefinition("biMap", DataTypes.mapTypeName(DataTypes.STRING_TYPE.getName(),
+                "MapValue"), Multiplicity.OPTIONAL, false, "biMapOwner"));
+        TypesDef typesDef = TypesUtil.getTypesDef(ImmutableList.<EnumTypeDefinition>of(),
+            ImmutableList.<StructTypeDefinition>of(), ImmutableList.<HierarchicalTypeDefinition<TraitType>>of(),
+            ImmutableList.of(mapOwnerDef, mapValueDef));
+        typeSystem.defineTypes(typesDef);
+        ClassType mapOwnerType = typeSystem.getDataType(ClassType.class, "MapOwner");
+        ClassType mapValueType = typeSystem.getDataType(ClassType.class, "MapValue");
+        
+        // Create instances of MapOwner and MapValue.
+        // Set MapOwner.map and MapOwner.biMap with one entry that references MapValue instance.
+        ITypedReferenceableInstance mapOwnerInstance = mapOwnerType.createInstance();
+        ITypedReferenceableInstance mapValueInstance = mapValueType.createInstance();
+        mapOwnerInstance.set("map", Collections.singletonMap("value1", mapValueInstance));
+        mapOwnerInstance.set("biMap", Collections.singletonMap("value1", mapValueInstance));
+        // Set biMapOwner reverse reference on MapValue.
+        mapValueInstance.set("biMapOwner", mapOwnerInstance);
+        List<String> createEntitiesResult = repositoryService.createEntities(mapOwnerInstance, mapValueInstance);
+        Assert.assertEquals(createEntitiesResult.size(), 2);
+        List<String> guids = repositoryService.getEntityList("MapOwner");
+        Assert.assertEquals(guids.size(), 1);
+        String mapOwnerGuid = guids.get(0);
+        
+        String edgeLabel = GraphHelper.getEdgeLabel(mapOwnerType, mapOwnerType.fieldMapping.fields.get("map"));
+        String mapEntryLabel = edgeLabel + "." + "value1";
+        AtlasEdgeLabel atlasEdgeLabel = new AtlasEdgeLabel(mapEntryLabel);
+        edgeLabel = GraphHelper.getEdgeLabel(mapOwnerType, mapOwnerType.fieldMapping.fields.get("biMap"));
+        mapEntryLabel = edgeLabel + "." + "value1";
+        AtlasEdgeLabel biMapAtlasEdgeLabel = new AtlasEdgeLabel(mapEntryLabel);
+        
+        // Verify MapOwner.map attribute has expected value.
+        String mapValueGuid = null;
+        Vertex mapOwnerVertex = null;
+        mapOwnerInstance = repositoryService.getEntityDefinition(mapOwnerGuid);
+        for (String mapAttrName : Arrays.asList("map", "biMap")) {
+            Object object = mapOwnerInstance.get(mapAttrName);
+            Assert.assertNotNull(object);
+            Assert.assertTrue(object instanceof Map);
+            Map<String, ITypedReferenceableInstance> map = (Map<String, ITypedReferenceableInstance>)object;
+            Assert.assertEquals(map.size(), 1);
+            mapValueInstance = map.get("value1");
+            Assert.assertNotNull(mapValueInstance);
+            mapValueGuid = mapValueInstance.getId()._getId();
+            mapOwnerVertex = GraphHelper.getInstance().getVertexForGUID(mapOwnerGuid);
+            object = mapOwnerVertex.getProperty(atlasEdgeLabel.getQualifiedMapKey());
+            Assert.assertNotNull(object);
+        }
+        
+        // Delete the map value instance.
+        // This should disconnect the references from the map owner instance. 
+        Pair<List<String>, List<ITypedReferenceableInstance>> deleteEntitiesResult = 
+            repositoryService.deleteEntities(Arrays.asList(mapValueGuid));
+        verifyEntityDoesNotExist(mapValueGuid);
+        
+        // Verify map references from mapOwner were disconnected.
+        mapOwnerInstance = repositoryService.getEntityDefinition(mapOwnerGuid);
+        Assert.assertNull(mapOwnerInstance.get("map"));
+        Assert.assertNull(mapOwnerInstance.get("biMap"));
+        mapOwnerVertex = GraphHelper.getInstance().getVertexForGUID(mapOwnerGuid);
+        Object object = mapOwnerVertex.getProperty(atlasEdgeLabel.getQualifiedMapKey());
+        Assert.assertNull(object);
+        object = mapOwnerVertex.getProperty(biMapAtlasEdgeLabel.getQualifiedMapKey());
+        Assert.assertNull(object);
+    }
+    
     private String createHrDeptGraph() throws Exception {
         Referenceable deptEg1 = TestUtils.createDeptEg1(typeSystem);
         ClassType deptType = typeSystem.getDataType(ClassType.class, "Department");
@@ -170,6 +565,36 @@ public class GraphBackedMetadataRepositoryDeleteEntitiesTest {
         return entityList.get(0);
     }
     
+    private void createDbTableGraph() throws Exception {
+        Referenceable databaseInstance = new Referenceable(TestUtils.DATABASE_TYPE);
+        databaseInstance.set("name", TestUtils.DATABASE_NAME);
+        databaseInstance.set("description", "foo database");
+ 
+        ClassType dbType = typeSystem.getDataType(ClassType.class, TestUtils.DATABASE_TYPE);
+        ITypedReferenceableInstance db = dbType.convert(databaseInstance, Multiplicity.REQUIRED);
+        Referenceable tableInstance = new Referenceable(TestUtils.TABLE_TYPE, TestUtils.CLASSIFICATION);
+        tableInstance.set("name", TestUtils.TABLE_NAME);
+        tableInstance.set("description", "bar table");
+        tableInstance.set("type", "managed");
+        Struct traitInstance = (Struct) tableInstance.getTrait(TestUtils.CLASSIFICATION);
+        traitInstance.set("tag", "foundation_etl");
+        tableInstance.set("tableType", 1); // enum
+
+        tableInstance.set("database", databaseInstance);
+        ArrayList<Referenceable> columns = new ArrayList<>();
+        for (int index = 0; index < 5; index++) {
+            Referenceable columnInstance = new Referenceable("column_type");
+            final String name = "column_" + index;
+            columnInstance.set("name", name);
+            columnInstance.set("type", "string");
+            columns.add(columnInstance);
+        }
+        tableInstance.set("columns", columns);
+        ClassType tableType = typeSystem.getDataType(ClassType.class, TestUtils.TABLE_TYPE);
+        ITypedReferenceableInstance table = tableType.convert(tableInstance, Multiplicity.REQUIRED);
+        repositoryService.createEntities(db, table);
+    }
+    
     private int countVertices(String propertyName, Object value) {
         Iterable<Vertex> vertices = graphProvider.get().getVertices(propertyName, value);
         int vertexCount = 0;
@@ -179,9 +604,9 @@ public class GraphBackedMetadataRepositoryDeleteEntitiesTest {
         return vertexCount;
     }
     
-    private void verifyEntityDoesNotExist(String hrDeptGuid) throws RepositoryException {
+    private void verifyEntityDoesNotExist(String guid) throws RepositoryException {
         try {
-            repositoryService.getEntityDefinition(hrDeptGuid);
+            repositoryService.getEntityDefinition(guid);
             Assert.fail("EntityNotFoundException was expected but none thrown");
         } catch(EntityNotFoundException e) {
             // good

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/7ebb2013/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java b/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
index 21789fd..83e4d85 100755
--- a/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
+++ b/repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java
@@ -534,7 +534,6 @@ public class GraphBackedMetadataRepositoryTest {
         Assert.assertTrue(creationTimestamp < modificationTimestampPostUpdate);
 
         // Update max's mentor reference to jane.
-        instance = personType.createInstance(max.getId());
         instance.set("mentor", janeGuid);
         repositoryService.updatePartial(instance);
 
@@ -550,6 +549,30 @@ public class GraphBackedMetadataRepositoryTest {
         Long modificationTimestampPost2ndUpdate = vertex.getProperty(Constants.MODIFICATION_TIMESTAMP_PROPERTY_KEY);
         Assert.assertNotNull(modificationTimestampPost2ndUpdate);
         Assert.assertTrue(modificationTimestampPostUpdate < modificationTimestampPost2ndUpdate);
+        
+        ITypedReferenceableInstance julius = repositoryService.getEntityDefinition("Person", "name", "Julius");
+        Id juliusGuid = julius.getId();
+        instance = personType.createInstance(max.getId());
+        instance.set("manager", juliusGuid);
+        repositoryService.updatePartial(instance);
+
+        // Verify the update was applied correctly - julius should now be max's manager.
+        max = repositoryService.getEntityDefinition(maxGuid);
+        object = max.get("manager");
+        Assert.assertTrue(object instanceof ITypedReferenceableInstance);
+        refTarget = (ITypedReferenceableInstance) object;
+        Assert.assertEquals(refTarget.getId()._getId(), juliusGuid._getId());
+        
+        // Verify that max is no longer a subordinate of jane.
+        jane = repositoryService.getEntityDefinition(janeGuid._getId());
+        Object refValue = jane.get("subordinates");
+        Assert.assertTrue(refValue instanceof List);
+        List<Object> subordinates = (List<Object>)refValue;
+        Assert.assertEquals(subordinates.size(), 1);
+        Object listValue = subordinates.get(0);
+        Assert.assertTrue(listValue instanceof ITypedReferenceableInstance);
+        ITypedReferenceableInstance subordinate = (ITypedReferenceableInstance) listValue;
+        Assert.assertNotEquals(subordinate.getId()._getId(), maxGuid);
     }
 
     private ITypedReferenceableInstance createHiveTableInstance(Referenceable databaseInstance) throws Exception {

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/7ebb2013/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
----------------------------------------------------------------------
diff --git a/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java b/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
index 019e21b..4eacfb2 100644
--- a/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
+++ b/repository/src/test/java/org/apache/atlas/service/DefaultMetadataServiceTest.java
@@ -389,7 +389,7 @@ public class DefaultMetadataServiceTest {
         Assert.assertEquals(arrColumnsList.size(), columns.size());
         assertReferenceables(arrColumnsList.get(1), columns.get(1));
         assertReferenceables(arrColumnsList.get(2), columns.get(2));
-
+        
         //Remove a class reference/Id and insert another reference
         //Also covers isComposite case since columns is a composite
         values.clear();
@@ -788,7 +788,8 @@ public class DefaultMetadataServiceTest {
         for (ITypedReferenceableInstance deletedEntity : deletedEntitiesFromListener) {
             deletedGuidsFromListener.add(deletedEntity.getId()._getId());
         }
-        Assert.assertEquals(deletedGuidsFromListener, deletedGuids);
+        Assert.assertEquals(deletedGuidsFromListener.size(), deletedGuids.size());
+        Assert.assertTrue(deletedGuidsFromListener.containsAll(deletedGuids));
     }
     
     private static class DeleteEntitiesChangeListener implements EntityChangeListener {


Mime
View raw message