ATLAS-2456: Implement tag propagation using relationships

Signed-off-by: Madhan Neethiraj <madhan@apache.org>

Author: sarathsubramanian

addons/models/0000-Area0/0010-base_model.json

@@ -213,7 +213,7 @@
                 "cardinality": "SET",
                 "isLegacyAttribute": true
             },
-            "propagateTags": "NONE"
+            "propagateTags": "ONE_TO_TWO"
         },
         {
           "name": "process_dataset_outputs",
@@ -232,7 +232,7 @@
                 "isContainer": false,
                 "cardinality": "SET"
           },
-          "propagateTags": "NONE"
+          "propagateTags": "ONE_TO_TWO"
         }
     ]
 }

addons/models/1000-Hadoop/1030-hive_model.json

@@ -539,7 +539,7 @@
                 "cardinality": "SINGLE",
                 "isLegacyAttribute": true
             },
-            "propagateTags": "ONE_TO_TWO"
+            "propagateTags": "NONE"
         },
         {
             "name": "hive_table_columns",
@@ -559,7 +559,7 @@
                 "cardinality": "SINGLE",
                 "isLegacyAttribute": true
             },
-            "propagateTags": "ONE_TO_TWO"
+            "propagateTags": "NONE"
         },
         {
             "name": "hive_table_partitionkeys",
@@ -579,7 +579,7 @@
                 "cardinality": "SINGLE",
                 "isLegacyAttribute": true
             },
-            "propagateTags": "ONE_TO_TWO"
+            "propagateTags": "NONE"
         },
         {
             "name": "hive_table_storagedesc",
@@ -599,7 +599,7 @@
                 "cardinality": "SINGLE",
                 "isLegacyAttribute": true
             },
-            "propagateTags": "ONE_TO_TWO"
+            "propagateTags": "NONE"
         },
         {
             "name": "hive_process_column_lineage",

addons/models/1000-Hadoop/1060-hbase_model.json

@@ -157,7 +157,7 @@
                 "cardinality": "SINGLE",
                 "isLegacyAttribute": true
             },
-            "propagateTags": "ONE_TO_TWO"
+            "propagateTags": "NONE"
         },
         {
             "name": "hbase_column_family_columns",
@@ -177,7 +177,7 @@
                 "cardinality": "SINGLE",
                 "isLegacyAttribute": true
             },
-            "propagateTags": "ONE_TO_TWO"
+            "propagateTags": "NONE"
         }
     ]
-}
+}

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

@@ -69,6 +69,7 @@ public final class Constants {
      * Trait names property key and index name.
      */
     public static final String TRAIT_NAMES_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "traitNames";
+    public static final String PROPAGATED_TRAIT_NAMES_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "propagatedTraitNames";
 
     public static final String VERSION_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "version";
     public static final String STATE_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "state";
@@ -115,6 +116,9 @@ public final class Constants {
     public static final String ATTRIBUTE_NAME_VERSION  = "version";
     public static final String TEMP_STRUCT_NAME_PREFIX = "__tempQueryResultStruct";
 
+    public static final String CLASSIFICATION_ENTITY_GUID = INTERNAL_PROPERTY_KEY_PREFIX + "entityGuid";
+    public static final String CLASSIFICATION_PROPAGATE_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "propagate";
+
     private Constants() {
     }
 

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

@@ -22,7 +22,6 @@
 import java.util.List;
 import java.util.Set;
 
-import org.apache.atlas.AtlasException;
 import org.codehaus.jettison.json.JSONException;
 import org.codehaus.jettison.json.JSONObject;
 
@@ -85,7 +84,7 @@ public interface AtlasElement {
      * is needed for this because special logic is required to handle this situation
      * in some implementations.
      */
-    void setListProperty(String propertyName, List<String> values) throws AtlasException;
+    void setListProperty(String propertyName, List<String> values);
 
 
     /**

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

@@ -55,6 +55,14 @@ public interface AtlasVertex<V, E> extends AtlasElement {
      */
     <T> void addProperty(String propertyName, T value);
 
+    /**
+     * Adds a value to a multiplicity-many property.
+     * If the property is already present, the value is added to it; if not, the propery is set with the given value
+     *
+     * @param propertyName
+     * @param value
+     */
+    <T> void addListProperty(String propertyName, T value);
 
     /**
      * Creates a vertex query.

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

@@ -52,6 +52,14 @@ public <T> void addProperty(String propertyName, T value) {
         }
     }
 
+    @Override
+    public <T> void addListProperty(String propertyName, T value) {
+        try {
+            getWrappedElement().property(VertexProperty.Cardinality.list, propertyName, value);
+        } catch(SchemaViolationException e) {
+            throw new AtlasSchemaViolationException(e);
+        }
+    }
 
 
     @Override

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

@@ -99,6 +99,20 @@ public <T> void addProperty(String propertyName, T value) {
         }
     }
 
+    @Override
+    public <T> void addListProperty(String propertyName, T value) {
+        try {
+            getAsTitanVertex().addProperty(propertyName, value);
+        } catch (SchemaViolationException e) {
+            if (getPropertyValues(propertyName, value.getClass()).contains(value)) {
+                // follow java set semantics, don't throw an exception if
+                // value is already there.
+                return;
+            }
+            throw new AtlasSchemaViolationException(e);
+        }
+    }
+
     @Override
     public <T> Collection<T> getPropertyValues(String key, Class<T> clazz) {
 

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

@@ -122,6 +122,10 @@ public enum AtlasErrorCode {
     INVALID_DSL_HAS_PROPERTY(400, "ATLAS-400-00-068", "DSL Semantic Error - Property needs to be a primitive type: {0}"),
     RELATIONSHIP_UPDATE_END_CHANGE_NOT_ALLOWED(404, "ATLAS-400-00-069", "change of relationship end is not permitted. relationship-type={}, relationship-guid={}, end-guid={}, updated-end-guid={}"),
     RELATIONSHIP_UPDATE_TYPE_CHANGE_NOT_ALLOWED(404, "ATLAS-400-00-06A", "change of relationship type is not permitted. relationship-guid={}, current-type={}, new-type={}"),
+    CLASSIFICATION_UPDATE_FROM_PROPAGATED_ENTITY(400, "ATLAS-400-00-06B", "Update to classification {0} is not allowed from propagated entity"),
+    CLASSIFICATION_DELETE_FROM_PROPAGATED_ENTITY(400, "ATLAS-400-00-06C", "Delete of classification {0} is not allowed from propagated entity"),
+    CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY(400, "ATLAS-400-00-06D", "Classification {0} is not associated with entity"),
+
     // All Not found enums go here
     TYPE_NAME_NOT_FOUND(404, "ATLAS-404-00-001", "Given typename {0} was invalid"),
     TYPE_GUID_NOT_FOUND(404, "ATLAS-404-00-002", "Given type guid {0} was invalid"),
@@ -137,6 +141,7 @@ public enum AtlasErrorCode {
     RELATIONSHIP_CRUD_INVALID_PARAMS(404, "ATLAS-404-00-00D", "Invalid relationship creation/updation parameters passed : {0}"),
     RELATIONSHIPDEF_END_TYPE_NAME_NOT_FOUND(404, "ATLAS-404-00-00E", "RelationshipDef {0} endDef typename {0} cannot be found"),
     RELATIONSHIP_ALREADY_DELETED(404, "ATLAS-404-00-00F", "Attempting to delete a relationship which is already deleted : {0}"),
+    INVALID_ENTITY_GUID_FOR_CLASSIFICATION_UPDATE(404, "ATLAS-404-00-010", "Updating entityGuid of classification is not allowed."),
 
     // All data conflict errors go here
     TYPE_ALREADY_EXISTS(409, "ATLAS-409-00-001", "Given type {0} already exists"),
@@ -157,17 +162,16 @@ public enum AtlasErrorCode {
     FAILED_TO_OBTAIN_GREMLIN_SCRIPT_ENGINE(500, "ATLAS-500-00-008", "Failed to obtain gremlin script engine: {0}"),
     JSON_ERROR_OBJECT_MAPPER_NULL_RETURNED(500, "ATLAS-500-00-009", "ObjectMapper.readValue returned NULL for class: {0}"),
     GREMLIN_SCRIPT_EXECUTION_FAILED(500, "ATLAS-500-00-00A", "Gremlin script execution failed: {0}"),
-
     CURATOR_FRAMEWORK_UPDATE(500, "ATLAS-500-00-00B", "ActiveInstanceState.update resulted in exception."),
     QUICK_START(500, "ATLAS-500-00-00C", "Failed to run QuickStart: {0}"),
     EMBEDDED_SERVER_START(500, "ATLAS-500-00-00D", "EmbeddedServer.Start: failed!"),
     STORM_TOPOLOGY_UTIL(500, "ATLAS-500-00-00E", "StormToplogyUtil: {0}"),
     SQOOP_HOOK(500, "ATLAS-500-00-00F", "SqoopHook: {0}"),
     HIVE_HOOK(500, "ATLAS-500-00-010", "HiveHook: {0}"),
     HIVE_HOOK_METASTORE_BRIDGE(500, "ATLAS-500-00-011", "HiveHookMetaStoreBridge: {0}"),
-
-    DATA_ACCESS_SAVE_FAILED(500, "ATLAS-500-00-00B", "Save failed: {0}"),
-    DATA_ACCESS_LOAD_FAILED(500, "ATLAS-500-00-00C", "Load failed: {0}");
+    DATA_ACCESS_SAVE_FAILED(500, "ATLAS-500-00-012", "Save failed: {0}"),
+    DATA_ACCESS_LOAD_FAILED(500, "ATLAS-500-00-013", "Load failed: {0}"),
+    ENTITY_NOTIFICATION_FAILED(500, "ATLAS-500-00-014", "Notification failed for operation: {} : {}");
 
     private String errorCode;
     private String errorMessage;

intg/src/main/java/org/apache/atlas/listener/EntityChangeListenerV2.java

@@ -0,0 +1,81 @@
+/**
+ * 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
+ * <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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.atlas.listener;
+
+import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.instance.AtlasClassification;
+import org.apache.atlas.model.instance.AtlasEntity;
+
+import java.util.List;
+
+/**
+ * Entity change notification listener V2.
+ */
+public interface EntityChangeListenerV2 {
+    /**
+     * This is upon adding new entities to the repository.
+     *
+     * @param entities the created entities
+     * @param isImport
+     */
+    void onEntitiesAdded(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException;
+
+    /**
+     * This is upon updating an entity.
+     *
+     * @param entities the updated entities
+     * @param isImport
+     */
+    void onEntitiesUpdated(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException;
+
+    /**
+     * This is upon deleting entities from the repository.
+     *
+     * @param entities the deleted entities
+     * @param isImport
+     */
+    void onEntitiesDeleted(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException;
+
+    /**
+     * This is upon adding new classifications to an entity.
+     *
+     * @param entity          the entity
+     * @param classifications classifications that needs to be added to an entity
+     * @throws AtlasBaseException if the listener notification fails
+     */
+    void onClassificationsAdded(AtlasEntity entity, List<AtlasClassification> classifications) throws AtlasBaseException;
+
+    /**
+     * This is upon updating classifications to an entity.
+     *
+     * @param entity          the entity
+     * @param classifications classifications that needs to be updated for an entity
+     * @throws AtlasBaseException if the listener notification fails
+     */
+    void onClassificationsUpdated(AtlasEntity entity, List<AtlasClassification> classifications) throws AtlasBaseException;
+
+    /**
+     * This is upon deleting classifications from an entity.
+     *
+     * @param entity              the entity
+     * @param classificationNames classifications names for the instance that needs to be deleted from entity
+     * @throws AtlasBaseException if the listener notification fails
+     */
+    void onClassificationsDeleted(AtlasEntity entity, List<String> classificationNames) throws AtlasBaseException;
+}