ATLAS-2120: Import API updated to detect attribute-type change

docs/src/site/twiki/Import-API-Options.twiki

@@ -68,12 +68,12 @@ To use the option, set the contents of _importOptions.json_ to:
 }
 </verbatim>
 
-To use _startIndex_, use the following in the _importOptions.json_:
+To use _startPosition_, use the following in the _importOptions.json_:
 <verbatim>
 {
 
   "options": {
-    "startIndex": "332"
+    "startPosition": "332"
   }
 }
 </verbatim>
@@ -89,7 +89,7 @@ Steps to use the behavior:
 
 The output of Export has _atlas-typedef.json_ that contains the type definitions for the entities exported.
 
-By default (that is if no options is specified), the type definitions are imported and applied to the system. The entity import is then performed.
+By default (that is if no options is specified), the type definitions are imported and applied to the system being imported to. The entity import is performed after this.
 
 In some cases, you would not want to modify the type definitions. Import may be better off failing than the types be modified.
 
@@ -100,7 +100,7 @@ Table below enumerates the conditions that get addressed as part of type definit
 |*Condition*|*Action*|
 | Incoming type does not exist in target system | Type is created. |
 |Type to be imported and type in target system are same | No change |
-|Type to be imported and type in target system differ by some attributes| Target system type is updated to the attributes present in the source. It is possible that the target system will have attributes in addition to the one present in the source. In that case, the target system's type attributes will be an union of the attributes. (Attributes in target system will not be deleted to match the source.)|
+|Type to be imported and type in target system differ by some attributes| Target system type is updated to the attributes present in the source. It is possible that the target system will have attributes in addition to the one present in the source. In that case, the target system's type attributes will be an union of the attributes. Attributes in target system will not be deleted to match the source. If the type of the attribute differ, import process will be aborted and exception logged.|
 
 To use the option, set the contents of _importOptions.json_ to:
 <verbatim>

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

@@ -92,8 +92,9 @@ public enum AtlasErrorCode {
     RELATIONSHIP_INVALID_ENDTYPE(400, "ATLAS-400-00-045", "Invalid entity-type for relationship attribute ‘{0}’: entity specified (guid={1}) is of type ‘{2}’, but expected type is ‘{3}’"),
     UNKNOWN_CLASSIFICATION(400, "ATLAS-400-00-046", "{0}: Unknown/invalid classification"),
     INVALID_SEARCH_PARAMS(400, "ATLAS-400-00-047", "No search parameter was found. One of the following MUST be specified in the request; typeName, classification or queryText"),
-    INVALID_RELATIONSHIP_ATTRIBUTE(400, "ATLAS-400-00-048", "Expected attribute {0} to be a relationship but found type {}"),
+    INVALID_RELATIONSHIP_ATTRIBUTE(400, "ATLAS-400-00-048", "Expected attribute {0} to be a relationship but found type {1}"),
     INVALID_RELATIONSHIP_TYPE(400, "ATLAS-400-00-049", "Invalid entity type '{0}', guid '{1}' in relationship search"),
+    INVALID_IMPORT_ATTRIBUTE_TYPE_CHANGED(400, "ATLAS-400-00-050", "Attribute {0}.{1} is of type {2}. Import has this attribute type as {3}"),
 
     // All Not found enums go here
     TYPE_NAME_NOT_FOUND(404, "ATLAS-404-00-001", "Given typename {0} was invalid"),

repository/src/main/java/org/apache/atlas/repository/impexp/TypeAttributeDifference.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <p>
  * http://www.apache.org/licenses/LICENSE-2.0
- *
+ * <p>
  * 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.
@@ -17,6 +17,7 @@
  */
 package org.apache.atlas.repository.impexp;
 
+import org.apache.atlas.AtlasErrorCode;
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.impexp.AtlasImportResult;
 import org.apache.atlas.model.typedef.AtlasClassificationDef;
@@ -26,11 +27,15 @@ import org.apache.atlas.model.typedef.AtlasStructDef;
 import org.apache.atlas.model.typedef.AtlasTypesDef;
 import org.apache.atlas.store.AtlasTypeDefStore;
 import org.apache.atlas.type.AtlasTypeRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
 
 public class TypeAttributeDifference {
+    private static final Logger LOG = LoggerFactory.getLogger(TypeAttributeDifference.class);
+
     private final AtlasTypeDefStore typeDefStore;
     private final AtlasTypeRegistry typeRegistry;
 
@@ -48,9 +53,9 @@ public class TypeAttributeDifference {
     }
 
     private void updateEntityDef(AtlasTypesDef typeDefinitionMap, AtlasImportResult result) throws AtlasBaseException {
-        for (AtlasEntityDef def: typeDefinitionMap.getEntityDefs()) {
+        for (AtlasEntityDef def : typeDefinitionMap.getEntityDefs()) {
             AtlasEntityDef existing = typeRegistry.getEntityDefByName(def.getName());
-            if(existing != null && addAttributes(existing, def)) {
+            if (existing != null && addAttributes(existing, def)) {
                 typeDefStore.updateEntityDefByName(existing.getName(), existing);
                 result.incrementMeticsCounter("typedef:entitydef:update");
             }
@@ -58,9 +63,9 @@ public class TypeAttributeDifference {
     }
 
     private void updateClassificationDef(AtlasTypesDef typeDefinitionMap, AtlasImportResult result) throws AtlasBaseException {
-        for (AtlasClassificationDef def: typeDefinitionMap.getClassificationDefs()) {
+        for (AtlasClassificationDef def : typeDefinitionMap.getClassificationDefs()) {
             AtlasClassificationDef existing = typeRegistry.getClassificationDefByName(def.getName());
-            if(existing != null && addAttributes(existing, def)) {
+            if (existing != null && addAttributes(existing, def)) {
                 typeDefStore.updateClassificationDefByName(existing.getName(), existing);
                 result.incrementMeticsCounter("typedef:classification:update");
             }
@@ -68,9 +73,9 @@ public class TypeAttributeDifference {
     }
 
     private void updateEnumDef(AtlasTypesDef typeDefinitionMap, AtlasImportResult result) throws AtlasBaseException {
-        for (AtlasEnumDef def: typeDefinitionMap.getEnumDefs()) {
+        for (AtlasEnumDef def : typeDefinitionMap.getEnumDefs()) {
             AtlasEnumDef existing = typeRegistry.getEnumDefByName(def.getName());
-            if(existing != null && addElements(existing, def)) {
+            if (existing != null && addElements(existing, def)) {
                 typeDefStore.updateEnumDefByName(existing.getName(), existing);
                 result.incrementMeticsCounter("typedef:enum:update");
             }
@@ -78,45 +83,62 @@ public class TypeAttributeDifference {
     }
 
     private void updateStructDef(AtlasTypesDef typeDefinitionMap, AtlasImportResult result) throws AtlasBaseException {
-        for (AtlasStructDef def: typeDefinitionMap.getStructDefs()) {
+        for (AtlasStructDef def : typeDefinitionMap.getStructDefs()) {
             AtlasStructDef existing = typeRegistry.getStructDefByName(def.getName());
-            if(existing != null && addAttributes(existing, def)) {
+            if (existing != null && addAttributes(existing, def)) {
                 typeDefStore.updateStructDefByName(existing.getName(), existing);
                 result.incrementMeticsCounter("typedef:struct:update");
             }
         }
     }
 
-    private boolean addElements(AtlasEnumDef existing, AtlasEnumDef incoming) {
+    private boolean addElements(AtlasEnumDef existing, AtlasEnumDef incoming) throws AtlasBaseException {
         return addElements(existing, getElementsAbsentInExisting(existing, incoming));
     }
 
-    private boolean addAttributes(AtlasStructDef existing, AtlasStructDef incoming) {
+    private boolean addAttributes(AtlasStructDef existing, AtlasStructDef incoming) throws AtlasBaseException {
         return addAttributes(existing, getElementsAbsentInExisting(existing, incoming));
     }
 
-    private List<AtlasStructDef.AtlasAttributeDef> getElementsAbsentInExisting(AtlasStructDef existing, AtlasStructDef incoming) {
+    private List<AtlasStructDef.AtlasAttributeDef> getElementsAbsentInExisting(AtlasStructDef existing, AtlasStructDef incoming) throws AtlasBaseException {
         List<AtlasStructDef.AtlasAttributeDef> difference = new ArrayList<>();
         for (AtlasStructDef.AtlasAttributeDef attr : incoming.getAttributeDefs()) {
-            if(existing.getAttribute(attr.getName()) == null) {
-                difference.add(attr);
-            }
+            updateCollectionWithDifferingAttributes(difference, existing, attr);
         }
 
         return difference;
     }
 
-    private List<AtlasEnumDef.AtlasEnumElementDef> getElementsAbsentInExisting(AtlasEnumDef existing, AtlasEnumDef incoming) {
+    private void updateCollectionWithDifferingAttributes(List<AtlasStructDef.AtlasAttributeDef> difference,
+                                                         AtlasStructDef existing,
+                                                         AtlasStructDef.AtlasAttributeDef incoming) throws AtlasBaseException {
+        AtlasStructDef.AtlasAttributeDef existingAttribute = existing.getAttribute(incoming.getName());
+        if (existingAttribute == null) {
+            difference.add(incoming);
+        } else {
+            if (!existingAttribute.getTypeName().equals(incoming.getTypeName())) {
+                LOG.error("Attribute definition difference found: {}, {}", existingAttribute, incoming);
+                throw new AtlasBaseException(AtlasErrorCode.INVALID_IMPORT_ATTRIBUTE_TYPE_CHANGED, existing.getName(), existingAttribute.getName(), existingAttribute.getTypeName(), incoming.getTypeName());
+            }
+        }
+    }
+
+    private List<AtlasEnumDef.AtlasEnumElementDef> getElementsAbsentInExisting(AtlasEnumDef existing, AtlasEnumDef incoming) throws AtlasBaseException {
         List<AtlasEnumDef.AtlasEnumElementDef> difference = new ArrayList<>();
         for (AtlasEnumDef.AtlasEnumElementDef ed : incoming.getElementDefs()) {
-            if(existing.getElement(ed.getValue()) == null) {
-                difference.add(ed);
-            }
+            updateCollectionWithDifferingAttributes(existing, difference, ed);
         }
 
         return difference;
     }
 
+    private void updateCollectionWithDifferingAttributes(AtlasEnumDef existing, List<AtlasEnumDef.AtlasEnumElementDef> difference, AtlasEnumDef.AtlasEnumElementDef ed) throws AtlasBaseException {
+        AtlasEnumDef.AtlasEnumElementDef existingElement = existing.getElement(ed.getValue());
+        if (existingElement == null) {
+            difference.add(ed);
+        }
+    }
+
     private boolean addAttributes(AtlasStructDef def, List<AtlasStructDef.AtlasAttributeDef> list) {
         for (AtlasStructDef.AtlasAttributeDef ad : list) {
             def.addAttribute(ad);

repository/src/test/java/org/apache/atlas/repository/impexp/ImportServiceTest.java

@@ -18,6 +18,7 @@
 package org.apache.atlas.repository.impexp;
 
 import com.google.inject.Inject;
+import org.apache.atlas.AtlasErrorCode;
 import org.apache.atlas.RequestContextV1;
 import org.apache.atlas.TestModules;
 import org.apache.atlas.TestUtilsV2;
@@ -25,6 +26,7 @@ import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.impexp.AtlasImportRequest;
 import org.apache.atlas.repository.store.graph.AtlasEntityStore;
 import org.apache.atlas.store.AtlasTypeDefStore;
+import org.apache.atlas.type.AtlasClassificationType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,6 +43,7 @@ import java.util.Map;
 import static org.apache.atlas.repository.impexp.ZipFileResourceTestUtils.*;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
 
 @Guice(modules = TestModules.TestOnlyModule.class)
 public class ImportServiceTest {
@@ -156,6 +159,29 @@ public class ImportServiceTest {
         runImportWithNoParameters(getImportService(), zipSource);
     }
 
+    @DataProvider(name = "hdfs_path1")
+    public static Object[][] getDataFromHdfsPath1(ITestContext context) throws IOException {
+        return getZipSource("hdfs_path1.zip");
+    }
+
+
+    @Test(dataProvider = "hdfs_path1", expectedExceptions = AtlasBaseException.class)
+    public void importHdfs_path1(ZipSource zipSource) throws IOException, AtlasBaseException {
+        loadModelFromJson("0000-Area0/0010-base_model.json", typeDefStore, typeRegistry);
+        loadModelFromJson("1000-Hadoop/1020-fs_model.json", typeDefStore, typeRegistry);
+        loadModelFromResourcesJson("tag1.json", typeDefStore, typeRegistry);
+
+        try {
+            runImportWithNoParameters(getImportService(), zipSource);
+        } catch (AtlasBaseException e) {
+            assertEquals(e.getAtlasErrorCode(), AtlasErrorCode.INVALID_IMPORT_ATTRIBUTE_TYPE_CHANGED);
+            AtlasClassificationType tag1 = typeRegistry.getClassificationTypeByName("tag1");
+            assertNotNull(tag1);
+            assertEquals(tag1.getAllAttributes().size(), 2);
+            throw e;
+        }
+    }
+
     private ImportService getImportService() {
         return new ImportService(typeDefStore, entityStore, typeRegistry);
     }

repository/src/test/java/org/apache/atlas/repository/impexp/ZipFileResourceTestUtils.java

@@ -49,8 +49,7 @@ public class ZipFileResourceTestUtils {
     public static final Logger LOG = LoggerFactory.getLogger(ZipFileResourceTestUtils.class);
 
     public static FileInputStream getFileInputStream(String fileName) {
-        final String userDir = System.getProperty("user.dir");
-        String filePath = getFilePath(userDir, fileName);
+        String filePath = getFileFromResources(fileName);
         File f = new File(filePath);
         FileInputStream fs = null;
         try {
@@ -61,6 +60,11 @@ public class ZipFileResourceTestUtils {
         return fs;
     }
 
+    private static String getFileFromResources(String fileName) {
+        final String userDir = System.getProperty("user.dir");
+        return getFilePath(userDir, fileName);
+    }
+
     private static String getFilePath(String startPath, String fileName) {
         return startPath + "/src/test/resources/" + fileName;
     }
@@ -75,6 +79,15 @@ public class ZipFileResourceTestUtils {
         return s;
     }
 
+    public static String getModelJsonFromResources(String fileName) throws IOException {
+        String filePath = getFileFromResources(fileName);
+        File f = new File(filePath);
+        String s = FileUtils.readFileToString(f);
+        assertFalse(StringUtils.isEmpty(s), "Model file read correctly from resources!");
+
+        return s;
+    }
+
     public static Object[][] getZipSource(String fileName) throws IOException {
         FileInputStream fs = ZipFileResourceTestUtils.getFileInputStream(fileName);
 
@@ -119,6 +132,11 @@ public class ZipFileResourceTestUtils {
         createTypesAsNeeded(typesFromJson, typeDefStore, typeRegistry);
     }
 
+    public static void loadModelFromResourcesJson(String fileName, AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) throws IOException, AtlasBaseException {
+        AtlasTypesDef typesFromJson = getAtlasTypesDefFromResourceFile(fileName);
+        createTypesAsNeeded(typesFromJson, typeDefStore, typeRegistry);
+    }
+
     private static void createTypesAsNeeded(AtlasTypesDef typesFromJson, AtlasTypeDefStore typeDefStore, AtlasTypeRegistry typeRegistry) throws AtlasBaseException {
         AtlasTypesDef typesToCreate = AtlasTypeDefStoreInitializer.getTypesToCreate(typesFromJson, typeRegistry);
 
@@ -132,6 +150,11 @@ public class ZipFileResourceTestUtils {
         return AtlasType.fromJson(sampleTypes, AtlasTypesDef.class);
     }
 
+    private static AtlasTypesDef getAtlasTypesDefFromResourceFile(String fileName) throws IOException {
+        String sampleTypes = getModelJsonFromResources(fileName);
+        return AtlasType.fromJson(sampleTypes, AtlasTypesDef.class);
+    }
+
     public static AtlasImportRequest getDefaultImportRequest() {
         return new AtlasImportRequest();
     }

repository/src/test/resources/tag1.json

+{
+  "enumDefs": [],
+  "structDefs": [],
+  "classificationDefs": [
+    {
+      "category": "CLASSIFICATION",
+      "guid": "03f95af3-3415-4522-8f4c-95fadcb59636",
+      "createdBy": "admin",
+      "updatedBy": "admin",
+      "createTime": 1504887556449,
+      "updateTime": 1504887556449,
+      "version": 1,
+      "name": "tag1",
+      "description": "tag1",
+      "typeVersion": "1.0",
+      "attributeDefs": [
+        {
+          "name": "attrib1",
+          "typeName": "date",
+          "isOptional": true,
+          "cardinality": "SINGLE",
+          "valuesMinCount": 0,
+          "valuesMaxCount": 1,
+          "isUnique": false,
+          "isIndexable": false
+        },
+        {
+          "name": "attrib3",
+          "typeName": "int",
+          "isOptional": true,
+          "cardinality": "SINGLE",
+          "valuesMinCount": 0,
+          "valuesMaxCount": 1,
+          "isUnique": false,
+          "isIndexable": false
+        }
+      ],
+      "superTypes": []
+    }
+  ],
+  "entityDefs": [],
+  "relationshipDefs": []
+}