ATLAS-2965: Duplicate entities are created when when same qualifiedName is given…

intg/src/test/java/org/apache/atlas/TestUtilsV2.java

@@ -1256,6 +1256,39 @@ public final class TestUtilsV2 {
         return newtestEntityDef;
     }
 
+    public static AtlasEntityWithExtInfo createTableEntityDuplicatesV2(AtlasEntity dbEntity) {
+        AtlasEntity tblEntity = new AtlasEntity(TABLE_TYPE);
+
+        tblEntity.setAttribute(NAME, RandomStringUtils.randomAlphanumeric(10));
+        tblEntity.setAttribute("description", "random table");
+        tblEntity.setAttribute("type", "type");
+        tblEntity.setAttribute("tableType", "MANAGED");
+        tblEntity.setAttribute("database", AtlasTypeUtil.getAtlasObjectId(dbEntity));
+
+        AtlasEntity col1 = createColumnEntity(tblEntity);
+        col1.setAttribute(NAME, "col1");
+
+        AtlasEntity col2 = createColumnEntity(tblEntity);
+        col2.setAttribute(NAME, "col1");
+
+        AtlasEntity col3 = createColumnEntity(tblEntity);
+        col3.setAttribute(NAME, "col1");
+
+        // all 3 columns have different guid but same typeName and unique attributes
+        tblEntity.setAttribute(COLUMNS_ATTR_NAME, Arrays.asList(AtlasTypeUtil.getAtlasObjectId(col1),
+                AtlasTypeUtil.getAtlasObjectId(col2),
+                AtlasTypeUtil.getAtlasObjectId(col3)));
+
+        AtlasEntityWithExtInfo ret = new AtlasEntityWithExtInfo(tblEntity);
+
+        ret.addReferredEntity(dbEntity);
+        ret.addReferredEntity(col1);
+        ret.addReferredEntity(col2);
+        ret.addReferredEntity(col3);
+
+        return ret;
+    }
+
     public static AtlasEntity createColumnEntity(AtlasEntity tableEntity) {
         return createColumnEntity(tableEntity, "col" + seq.addAndGet(1));
     }

repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java

@@ -742,10 +742,11 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
         RequestContext              requestContext   = RequestContext.get();
 
         for (String guid : discoveryContext.getReferencedGuids()) {
-            AtlasVertex vertex = discoveryContext.getResolvedEntityVertex(guid);
             AtlasEntity entity = entityStream.getByGuid(guid);
 
             if (entity != null) { // entity would be null if guid is not in the stream but referenced by an entity in the stream
+                AtlasVertex vertex = getResolvedEntityVertex(discoveryContext, entity);
+
                 if (vertex != null) {
                     if (!isPartialUpdate) {
                         graphDiscoverer.validateAndNormalize(entity);
@@ -779,6 +780,8 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
 
                     discoveryContext.addResolvedGuid(guid, vertex);
 
+                    discoveryContext.addResolvedIdByUniqAttribs(getAtlasObjectId(entity), vertex);
+
                     String generatedGuid = AtlasGraphUtilsV2.getIdFromVertex(vertex);
 
                     entity.setGuid(generatedGuid);
@@ -798,6 +801,34 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
         return context;
     }
 
+    private AtlasVertex getResolvedEntityVertex(EntityGraphDiscoveryContext context, AtlasEntity entity) throws AtlasBaseException {
+        AtlasObjectId objectId = getAtlasObjectId(entity);
+        AtlasVertex   ret      = context.getResolvedEntityVertex(entity.getGuid());
+
+        if (ret != null) {
+            context.addResolvedIdByUniqAttribs(objectId, ret);
+        } else {
+            ret = context.getResolvedEntityVertex(objectId);
+
+            if (ret != null) {
+                context.addResolvedGuid(entity.getGuid(), ret);
+            }
+        }
+
+        return ret;
+    }
+
+    private AtlasObjectId getAtlasObjectId(AtlasEntity entity) {
+        AtlasObjectId ret = entityRetriever.toAtlasObjectId(entity);
+
+        if (ret != null && MapUtils.isNotEmpty(ret.getUniqueAttributes())) {
+            // if uniqueAttributes is not empty, reset guid to null.
+            ret.setGuid(null);
+        }
+
+        return ret;
+    }
+
     private EntityMutationResponse deleteVertices(Collection<AtlasVertex> deletionCandidates) throws AtlasBaseException {
         EntityMutationResponse response = new EntityMutationResponse();
         RequestContext         req      = RequestContext.get();

repository/src/main/java/org/apache/atlas/repository/store/graph/v2/EntityGraphRetriever.java

@@ -260,6 +260,27 @@ public final class EntityGraphRetriever {
         return ret;
     }
 
+    public AtlasObjectId toAtlasObjectId(AtlasEntity entity) {
+        AtlasObjectId   ret        = null;
+        AtlasEntityType entityType = typeRegistry.getEntityTypeByName(entity.getTypeName());
+
+        if (entityType != null) {
+            Map<String, Object> uniqueAttributes = new HashMap<>();
+
+            for (String attributeName : entityType.getUniqAttributes().keySet()) {
+                Object attrValue = entity.getAttribute(attributeName);
+
+                if (attrValue != null) {
+                    uniqueAttributes.put(attributeName, attrValue);
+                }
+            }
+
+            ret = new AtlasObjectId(entity.getGuid(), entity.getTypeName(), uniqueAttributes);
+        }
+
+        return ret;
+    }
+
     public AtlasClassification toAtlasClassification(AtlasVertex classificationVertex) throws AtlasBaseException {
         AtlasClassification ret = new AtlasClassification(getTypeName(classificationVertex));
 

repository/src/test/java/org/apache/atlas/repository/store/graph/v2/AtlasComplexAttributesTest.java

@@ -30,6 +30,7 @@ import org.apache.atlas.model.typedef.AtlasTypesDef;
 import org.apache.atlas.repository.graphdb.AtlasEdge;
 import org.apache.atlas.repository.graphdb.AtlasEdgeDirection;
 import org.apache.atlas.repository.graphdb.AtlasVertex;
+import org.apache.atlas.type.AtlasEntityType;
 import org.apache.commons.lang.time.DateUtils;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Guice;
@@ -256,8 +257,9 @@ public class AtlasComplexAttributesTest extends AtlasEntityTestBase {
         init();
         AtlasEntity              complexEntity       = getEntityFromStore(complexCollectionAttrEntity.getEntity().getGuid());
         AtlasEntitiesWithExtInfo complexEntitiesInfo = new AtlasEntitiesWithExtInfo(complexEntity);
+        AtlasEntityType          entityType          = typeRegistry.getEntityTypeByName(ENTITY_TYPE);
 
-        // Modify array of entities
+        // Replace list of entities with new values
         AtlasEntity e0Array = new AtlasEntity(ENTITY_TYPE, new HashMap<String, Object>() {{ put(NAME, "entityArray00"); put("isReplicated", true); }});
         AtlasEntity e1Array = new AtlasEntity(ENTITY_TYPE, new HashMap<String, Object>() {{ put(NAME, "entityArray11"); put("isReplicated", false); }});
         AtlasEntity e2Array = new AtlasEntity(ENTITY_TYPE, new HashMap<String, Object>() {{ put(NAME, "entityArray22"); put("isReplicated", true); }});
@@ -274,11 +276,18 @@ public class AtlasComplexAttributesTest extends AtlasEntityTestBase {
         AtlasEntityHeader      updatedComplexEntity = response.getFirstUpdatedEntityByTypeName(ENTITY_TYPE_WITH_COMPLEX_COLLECTION_ATTR);
         validateEntity(complexEntitiesInfo, getEntityFromStore(updatedComplexEntity));
 
-        // add a new element to array of entities
+        // add a new element to list of entities
         init();
+
+        e0Array = entityStore.getByUniqueAttributes(entityType, new HashMap<String, Object>() {{ put(NAME, "entityArray00"); put("isReplicated", true); }}).getEntity();
+        e1Array = entityStore.getByUniqueAttributes(entityType, new HashMap<String, Object>() {{ put(NAME, "entityArray11"); put("isReplicated", false); }}).getEntity();
+        e2Array = entityStore.getByUniqueAttributes(entityType, new HashMap<String, Object>() {{ put(NAME, "entityArray22"); put("isReplicated", true); }}).getEntity();
         AtlasEntity e3Array = new AtlasEntity(ENTITY_TYPE, new HashMap<String, Object>() {{ put(NAME, "entityArray33"); put("isReplicated", true); }});
-        entityList.add(getAtlasObjectId(e3Array));
+
+        entityList = new ArrayList<>(Arrays.asList(getAtlasObjectId(e0Array), getAtlasObjectId(e1Array), getAtlasObjectId(e2Array), getAtlasObjectId(e3Array)));
+
         complexEntity.setAttribute("listOfEntities", entityList);
+        complexEntitiesInfo.getReferredEntities().clear();
         complexEntitiesInfo.addReferredEntity(e3Array);
 
         response = entityStore.createOrUpdate(new AtlasEntityStream(complexEntitiesInfo), false);
@@ -287,8 +296,10 @@ public class AtlasComplexAttributesTest extends AtlasEntityTestBase {
 
         // remove one of the entity values - entityArray00
         init();
-        entityList.remove(0);
+        e3Array = entityStore.getByUniqueAttributes(entityType, new HashMap<String, Object>() {{ put(NAME, "entityArray33"); put("isReplicated", true); }}).getEntity();
+        entityList = new ArrayList<>(Arrays.asList(getAtlasObjectId(e1Array), getAtlasObjectId(e2Array), getAtlasObjectId(e3Array)));
         complexEntity.setAttribute("listOfEntities", entityList);
+        complexEntitiesInfo.getReferredEntities().clear();
 
         response = entityStore.createOrUpdate(new AtlasEntityStream(complexEntitiesInfo), false);
         updatedComplexEntity = response.getFirstUpdatedEntityByTypeName(ENTITY_TYPE_WITH_COMPLEX_COLLECTION_ATTR);
@@ -308,6 +319,7 @@ public class AtlasComplexAttributesTest extends AtlasEntityTestBase {
         AtlasEntity e3Array_duplicate = new AtlasEntity(ENTITY_TYPE, new HashMap<String, Object>() {{ put(NAME, "entityArray33"); put("isReplicated", true); }});
         entityList.add(getAtlasObjectId(e3Array_duplicate));
         complexEntity.setAttribute("listOfEntities", entityList);
+        complexEntitiesInfo.getReferredEntities().clear();
         complexEntitiesInfo.addReferredEntity(e3Array_duplicate);
 
         response = entityStore.createOrUpdate(new AtlasEntityStream(complexEntitiesInfo), false);
@@ -318,6 +330,7 @@ public class AtlasComplexAttributesTest extends AtlasEntityTestBase {
         init();
         entityList.clear();
         complexEntity.setAttribute("listOfEntities", entityList);
+        complexEntitiesInfo.getReferredEntities().clear();
 
         response = entityStore.createOrUpdate(new AtlasEntityStream(complexEntitiesInfo), false);
         updatedComplexEntity = response.getFirstUpdatedEntityByTypeName(ENTITY_TYPE_WITH_COMPLEX_COLLECTION_ATTR);

repository/src/test/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2Test.java

@@ -923,6 +923,27 @@ public class AtlasEntityStoreV2Test extends AtlasEntityTestBase {
         }
     }
 
+    @Test
+    public void testCreateWithDuplicateGuids() throws Exception {
+        init();
+        AtlasEntityWithExtInfo tblEntity2 = TestUtilsV2.createTableEntityDuplicatesV2(dbEntity.getEntity());
+        EntityMutationResponse response   = entityStore.createOrUpdate(new AtlasEntityStream(tblEntity2), false);
+
+        List<AtlasEntityHeader> createdEntities = response.getCreatedEntities();
+        assertTrue(CollectionUtils.isNotEmpty(createdEntities));
+        assertEquals(createdEntities.size(), 2);
+
+        String tableGuid = createdEntities.get(0).getGuid();
+        AtlasEntityWithExtInfo tableEntity  = entityStore.getById(tableGuid);
+        assertEquals(tableEntity.getReferredEntities().size(), 1);
+
+        List<AtlasObjectId> columns = (List<AtlasObjectId>) tableEntity.getEntity().getAttribute("columns");
+        assertEquals(columns.size(), 1);
+
+        Set<AtlasObjectId> uniqueColumns = new HashSet<>(columns);
+        assertTrue(uniqueColumns.size() == 1);
+    }
+
     @Test(dependsOnMethods = "testCreate")
     public void associateSameTagToMultipleEntities() throws AtlasBaseException {
         final String TAG_NAME            = "tagx";