Merge pull request #132 from shwethags/lineage

addons/hive-bridge/src/main/java/org/apache/hadoop/metadata/hive/hook/HiveHook.java

@@ -37,7 +37,6 @@ package org.apache.hadoop.metadata.hive.hook;
 
 
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.apache.commons.lang.StringEscapeUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.ql.QueryPlan;
@@ -248,10 +247,9 @@ public class HiveHook implements ExecuteWithHookContext {
         Referenceable dbReferenceable = dgiBridge.registerDatabase(oldTable.getDbName());
         Referenceable tableReferenceable =
                 dgiBridge.registerTable(dbReferenceable, oldTable.getDbName(), oldTable.getTableName());
-        LOG.info("Updating entity name {}.{} to {}",
-                oldTable.getDbName(), oldTable.getTableName(), newTable.getTableName());
-        dgiBridge.getMetadataServiceClient().updateEntity(tableReferenceable.getId()._getId(), "name",
-                newTable.getTableName().toLowerCase());
+        LOG.info("Updating entity name {}.{} to {}", oldTable.getDbName(), oldTable.getTableName(),
+                newTable.getTableName());
+        dgiBridge.updateTable(tableReferenceable, newTable);
     }
 
     private void handleCreateTable(HiveMetaStoreBridge dgiBridge, HiveEvent event) throws Exception {

addons/hive-bridge/src/main/java/org/apache/hadoop/metadata/hive/model/HiveDataTypes.java

@@ -27,11 +27,9 @@ public enum HiveDataTypes {
     HIVE_OBJECT_TYPE,
     HIVE_PRINCIPAL_TYPE,
     HIVE_RESOURCE_TYPE,
-    HIVE_FUNCTION_TYPE,
 
     // Structs
     HIVE_SERDE,
-    HIVE_SKEWEDINFO,
     HIVE_ORDER,
     HIVE_RESOURCEURI,
 
@@ -42,7 +40,6 @@ public enum HiveDataTypes {
     HIVE_COLUMN,
     HIVE_PARTITION,
     HIVE_INDEX,
-    HIVE_FUNCTION,
     HIVE_ROLE,
     HIVE_TYPE,
     HIVE_PROCESS,

addons/hive-bridge/src/test/java/org/apache/hadoop/metadata/hive/hook/HiveHookIT.java

@@ -121,8 +121,8 @@ public class HiveHookIT {
 
     private String createTable(boolean partition) throws Exception {
         String tableName = tableName();
-        runCommand("create table " + tableName + "(id int, name string) comment 'table comment' " + (partition ? " partitioned by(dt string)"
-                : ""));
+        runCommand("create table " + tableName + "(id int, name string) comment 'table comment' "
+                + (partition ? " partitioned by(dt string)" : ""));
         return tableName;
     }
 
@@ -141,6 +141,9 @@ public class HiveHookIT {
         Referenceable tableRef = dgiCLient.getEntity(tableId);
         Assert.assertEquals(tableRef.get("tableType"), TableType.MANAGED_TABLE.name());
         Assert.assertEquals(tableRef.get(HiveDataModelGenerator.COMMENT), "table comment");
+        String entityName = HiveMetaStoreBridge.getTableName(CLUSTER_NAME, DEFAULT_DB, tableName);
+        Assert.assertEquals(tableRef.get(HiveDataModelGenerator.NAME), entityName);
+
         final Id sdId = (Id) tableRef.get("sd");
         Referenceable sdRef = dgiCLient.getEntity(sdId.id);
         Assert.assertEquals(sdRef.get(HiveDataModelGenerator.STORAGE_IS_STORED_AS_SUB_DIRS),false);
@@ -304,7 +307,7 @@ public class HiveHookIT {
 
     private String assertTableIsRegistered(String dbName, String tableName, boolean registered) throws Exception {
         LOG.debug("Searching for table {}.{}", dbName, tableName);
-        String query = String.format("%s as t where name = '%s', dbName where name = '%s' and clusterName = '%s'"
+        String query = String.format("%s as t where tableName = '%s', db where name = '%s' and clusterName = '%s'"
                 + " select t", HiveDataTypes.HIVE_TABLE.getName(), tableName.toLowerCase(), dbName.toLowerCase(),
                 CLUSTER_NAME);
         return assertEntityIsRegistered(query, registered);
@@ -321,14 +324,13 @@ public class HiveHookIT {
         String typeName = HiveDataTypes.HIVE_PARTITION.getName();
         String dbType = HiveDataTypes.HIVE_DB.getName();
         String tableType = HiveDataTypes.HIVE_TABLE.getName();
-        String datasetType = MetadataServiceClient.DATA_SET_SUPER_TYPE;
 
         LOG.debug("Searching for partition of {}.{} with values {}", dbName, tableName, value);
         //todo replace with DSL
         String gremlinQuery = String.format("g.V.has('__typeName', '%s').has('%s.values', ['%s']).as('p')."
-                        + "out('__%s.tableName').has('%s.name', '%s').out('__%s.dbName').has('%s.name', '%s')"
+                        + "out('__%s.table').has('%s.tableName', '%s').out('__%s.db').has('%s.name', '%s')"
                         + ".has('%s.clusterName', '%s').back('p').toList()", typeName, typeName, value, typeName,
-                datasetType, tableName.toLowerCase(), tableType, dbType, dbName.toLowerCase(), dbType, CLUSTER_NAME);
+                tableType, tableName.toLowerCase(), tableType, dbType, dbName.toLowerCase(), dbType, CLUSTER_NAME);
         JSONObject response = dgiCLient.searchByGremlin(gremlinQuery);
         JSONArray results = response.getJSONArray(MetadataServiceClient.RESULTS);
         Assert.assertEquals(results.length(), 1);
@@ -349,4 +351,29 @@ public class HiveHookIT {
             return null;
         }
     }
+
+    @Test
+    public void testLineage() throws Exception {
+        String table1 = createTable(false);
+
+        String db2 = createDatabase();
+        String table2 = tableName();
+
+        String query = String.format("create table %s.%s as select * from %s", db2, table2, table1);
+        runCommand(query);
+        String table1Id = assertTableIsRegistered(DEFAULT_DB, table1);
+        String table2Id = assertTableIsRegistered(db2, table2);
+
+        String datasetName = HiveMetaStoreBridge.getTableName(CLUSTER_NAME, db2, table2);
+        JSONObject response = dgiCLient.getInputGraph(datasetName);
+        JSONObject vertices = response.getJSONObject("values").getJSONObject("vertices");
+        Assert.assertTrue(vertices.has(table1Id));
+        Assert.assertTrue(vertices.has(table2Id));
+
+        datasetName = HiveMetaStoreBridge.getTableName(CLUSTER_NAME, DEFAULT_DB, table1);
+        response = dgiCLient.getOutputGraph(datasetName);
+        vertices = response.getJSONObject("values").getJSONObject("vertices");
+        Assert.assertTrue(vertices.has(table1Id));
+        Assert.assertTrue(vertices.has(table2Id));
+    }
 }

client/src/main/java/org/apache/hadoop/metadata/MetadataServiceClient.java

@@ -64,6 +64,7 @@ public class MetadataServiceClient {
     public static final String URI_ENTITIES = "entities";
     public static final String URI_TRAITS = "traits";
     public static final String URI_SEARCH = "discovery/search";
+    public static final String URI_LINEAGE = "lineage/hive/table";
 
     public static final String QUERY = "query";
     public static final String QUERY_TYPE = "queryType";
@@ -128,7 +129,12 @@ public class MetadataServiceClient {
         SEARCH(BASE_URI + URI_SEARCH, HttpMethod.GET),
         SEARCH_DSL(BASE_URI + URI_SEARCH + "/dsl", HttpMethod.GET),
         SEARCH_GREMLIN(BASE_URI + URI_SEARCH + "/gremlin", HttpMethod.GET),
-        SEARCH_FULL_TEXT(BASE_URI + URI_SEARCH + "/fulltext", HttpMethod.GET);
+        SEARCH_FULL_TEXT(BASE_URI + URI_SEARCH + "/fulltext", HttpMethod.GET),
+
+        //Lineage operations
+        LINEAGE_INPUTS_GRAPH(BASE_URI + URI_LINEAGE, HttpMethod.GET),
+        LINEAGE_OUTPUTS_GRAPH(BASE_URI + URI_LINEAGE, HttpMethod.GET),
+        LINEAGE_SCHEMA(BASE_URI + URI_LINEAGE, HttpMethod.GET);
 
         private final String method;
         private final String path;
@@ -147,6 +153,16 @@ public class MetadataServiceClient {
         }
     }
 
+    /**
+     * Register the given type(meta model)
+     * @param typeAsJson type definition a jaon
+     * @return result json object
+     * @throws MetadataServiceException
+     */
+    public JSONObject createType(String typeAsJson) throws MetadataServiceException {
+        return callAPI(API.CREATE_TYPE, typeAsJson);
+    }
+
     public List<String> listTypes() throws MetadataServiceException {
         try {
             final JSONObject jsonObject = callAPI(API.LIST_TYPES, null);
@@ -178,16 +194,6 @@ public class MetadataServiceClient {
     }
 
     /**
-     * Register the given type(meta model)
-     * @param typeAsJson type definition a jaon
-     * @return result json object
-     * @throws MetadataServiceException
-     */
-    public JSONObject createType(String typeAsJson) throws MetadataServiceException {
-        return callAPI(API.CREATE_TYPE, typeAsJson);
-    }
-
-    /**
      * Create the given entity
      * @param entityAsJson entity(type instance) as json
      * @return result json object
@@ -257,6 +263,7 @@ public class MetadataServiceClient {
      * @throws MetadataServiceException
      */
     public JSONArray searchByDSL(String query) throws MetadataServiceException {
+        LOG.debug("DSL query: {}", query);
         WebResource resource = getResource(API.SEARCH_DSL);
         resource = resource.queryParam(QUERY, query);
         JSONObject result = callAPIWithResource(API.SEARCH_DSL, resource);
@@ -274,6 +281,7 @@ public class MetadataServiceClient {
      * @throws MetadataServiceException
      */
     public JSONObject searchByGremlin(String gremlinQuery) throws MetadataServiceException {
+        LOG.debug("Gremlin query: " + gremlinQuery);
         WebResource resource = getResource(API.SEARCH_GREMLIN);
         resource = resource.queryParam(QUERY, gremlinQuery);
         return callAPIWithResource(API.SEARCH_GREMLIN, resource);
@@ -291,6 +299,24 @@ public class MetadataServiceClient {
         return callAPIWithResource(API.SEARCH_FULL_TEXT, resource);
     }
 
+    public JSONObject getInputGraph(String datasetName) throws MetadataServiceException {
+        JSONObject response = callAPI(API.LINEAGE_INPUTS_GRAPH, null, datasetName, "/inputs/graph");
+        try {
+            return response.getJSONObject(MetadataServiceClient.RESULTS);
+        } catch (JSONException e) {
+            throw new MetadataServiceException(e);
+        }
+    }
+
+    public JSONObject getOutputGraph(String datasetName) throws MetadataServiceException {
+        JSONObject response = callAPI(API.LINEAGE_OUTPUTS_GRAPH, null, datasetName, "/outputs/graph");
+        try {
+            return response.getJSONObject(MetadataServiceClient.RESULTS);
+        } catch (JSONException e) {
+            throw new MetadataServiceException(e);
+        }
+    }
+
     public String getRequestId(JSONObject json) throws MetadataServiceException {
         try {
             return json.getString(REQUEST_ID);

repository/src/main/java/org/apache/hadoop/metadata/repository/typestore/ITypeStore.java

@@ -29,7 +29,7 @@ public interface ITypeStore {
      * @param typeSystem type system to persist
      * @throws StorageException
      */
-    public void store(TypeSystem typeSystem) throws MetadataException;
+    void store(TypeSystem typeSystem) throws MetadataException;
 
     /**
      * Persist the given type in the type system - insert or update
@@ -37,12 +37,12 @@ public interface ITypeStore {
      * @param types types to persist
      * @throws StorageException
      */
-    public void store(TypeSystem typeSystem, ImmutableList<String> types) throws MetadataException;
+    void store(TypeSystem typeSystem, ImmutableList<String> types) throws MetadataException;
 
     /**
      * Restore all type definitions
      * @return List of persisted type definitions
      * @throws org.apache.hadoop.metadata.MetadataException
      */
-    public TypesDef restore() throws MetadataException;
+    TypesDef restore() throws MetadataException;
 }

repository/src/main/java/org/apache/hadoop/metadata/services/DefaultMetadataService.java

@@ -40,11 +40,14 @@ import org.apache.hadoop.metadata.typesystem.json.TypesSerialization;
 import org.apache.hadoop.metadata.typesystem.types.AttributeDefinition;
 import org.apache.hadoop.metadata.typesystem.types.ClassType;
 import org.apache.hadoop.metadata.typesystem.types.DataTypes;
+import org.apache.hadoop.metadata.typesystem.types.EnumTypeDefinition;
 import org.apache.hadoop.metadata.typesystem.types.HierarchicalTypeDefinition;
 import org.apache.hadoop.metadata.typesystem.types.IDataType;
 import org.apache.hadoop.metadata.typesystem.types.Multiplicity;
+import org.apache.hadoop.metadata.typesystem.types.StructTypeDefinition;
 import org.apache.hadoop.metadata.typesystem.types.TraitType;
 import org.apache.hadoop.metadata.typesystem.types.TypeSystem;
+import org.apache.hadoop.metadata.typesystem.types.TypeUtils;
 import org.apache.hadoop.metadata.typesystem.types.utils.TypesUtil;
 import org.codehaus.jettison.json.JSONException;
 import org.codehaus.jettison.json.JSONObject;
@@ -114,32 +117,28 @@ public class DefaultMetadataService implements MetadataService {
             return; // this is already registered
         }
 
-        Map<String, IDataType> superTypes = new HashMap();
-        HierarchicalTypeDefinition<ClassType> superTypeDefinition =
+        HierarchicalTypeDefinition<ClassType> infraType =
                 TypesUtil.createClassTypeDef(MetadataServiceClient.INFRASTRUCTURE_SUPER_TYPE,
                         ImmutableList.<String>of(), NAME_ATTRIBUTE, DESCRIPTION_ATTRIBUTE);
-        superTypes.put(MetadataServiceClient.INFRASTRUCTURE_SUPER_TYPE, typeSystem.defineClassType
-                (superTypeDefinition));
 
-        superTypeDefinition =
-                TypesUtil.createClassTypeDef(MetadataServiceClient.DATA_SET_SUPER_TYPE,
-                        ImmutableList.<String>of(),
+        HierarchicalTypeDefinition<ClassType> datasetType = TypesUtil
+                .createClassTypeDef(MetadataServiceClient.DATA_SET_SUPER_TYPE, ImmutableList.<String>of(),
                         NAME_ATTRIBUTE, DESCRIPTION_ATTRIBUTE);
-        superTypes.put(MetadataServiceClient.DATA_SET_SUPER_TYPE, typeSystem.defineClassType(superTypeDefinition));
 
-        superTypeDefinition =
-                TypesUtil.createClassTypeDef(MetadataServiceClient.PROCESS_SUPER_TYPE,
-                        ImmutableList.<String>of(),
-                        NAME_ATTRIBUTE, DESCRIPTION_ATTRIBUTE,
-                        new AttributeDefinition("inputs",
+        HierarchicalTypeDefinition<ClassType> processType = TypesUtil
+                .createClassTypeDef(MetadataServiceClient.PROCESS_SUPER_TYPE, ImmutableList.<String>of(),
+                        NAME_ATTRIBUTE, DESCRIPTION_ATTRIBUTE, new AttributeDefinition("inputs",
                                 DataTypes.arrayTypeName(MetadataServiceClient.DATA_SET_SUPER_TYPE),
-                                new Multiplicity(0, Integer.MAX_VALUE, false), false, null),
+                                Multiplicity.OPTIONAL, false, null),
                         new AttributeDefinition("outputs",
                                 DataTypes.arrayTypeName(MetadataServiceClient.DATA_SET_SUPER_TYPE),
-                                new Multiplicity(0, Integer.MAX_VALUE, false), false, null)
-                );
-        superTypes.put(MetadataServiceClient.PROCESS_SUPER_TYPE, typeSystem.defineClassType(superTypeDefinition));
-        onTypesAddedToRepo(superTypes);
+                                Multiplicity.OPTIONAL, false, null));
+
+        TypesDef typesDef = TypeUtils
+                .getTypesDef(ImmutableList.<EnumTypeDefinition>of(), ImmutableList.<StructTypeDefinition>of(),
+                        ImmutableList.<HierarchicalTypeDefinition<TraitType>>of(),
+                        ImmutableList.of(infraType, datasetType, processType));
+        createType(TypesSerialization.toJson(typesDef));
     }
 
     /**

repository/src/main/scala/org/apache/hadoop/metadata/query/QueryProcessor.scala

@@ -20,18 +20,18 @@ package org.apache.hadoop.metadata.query
 
 import com.thinkaurelius.titan.core.TitanGraph
 import org.apache.hadoop.metadata.query.Expressions._
+import org.slf4j.{LoggerFactory, Logger}
 
 object QueryProcessor {
+    val LOG : Logger = LoggerFactory.getLogger("org.apache.hadoop.metadata.query.QueryProcessor")
 
     def evaluate(e: Expression, g: TitanGraph, gP : GraphPersistenceStrategies = GraphPersistenceStrategy1):
     GremlinQueryResult = {
         val e1 = validate(e)
         val q = new GremlinTranslator(e1, gP).translate()
-//            println("---------------------")
-//            println("Query: " + e1)
-//            println("Expression Tree:\n" + e1.treeString)
-//            println("Gremlin Query: " + q.queryStr)
-//            println("---------------------")
+        LOG.debug("Query: " + e1)
+        LOG.debug("Expression Tree:\n" + e1.treeString)
+        LOG.debug("Gremlin Query: " + q.queryStr)
         new GremlinEvaluator(q, gP, g).evaluate()
     }
 

repository/src/test/java/org/apache/hadoop/metadata/repository/typestore/GraphBackedTypeStoreTest.java

@@ -99,13 +99,17 @@ public class GraphBackedTypeStoreTest {
         List<StructTypeDefinition> structTypes = types.structTypesAsJavaList();
         Assert.assertEquals(1, structTypes.size());
 
+        boolean clsTypeFound = false;
         List<HierarchicalTypeDefinition<ClassType>> classTypes = types.classTypesAsJavaList();
-        Assert.assertEquals(3, classTypes.size());
         for (HierarchicalTypeDefinition<ClassType> classType : classTypes) {
+            if (classType.typeName.equals("Manager")) {
                 ClassType expectedType = ts.getDataType(ClassType.class, classType.typeName);
                 Assert.assertEquals(expectedType.immediateAttrs.size(), classType.attributeDefinitions.length);
                 Assert.assertEquals(expectedType.superTypes.size(), classType.superTypes.size());
+                clsTypeFound = true;
             }
+        }
+        Assert.assertTrue("Manager type not restored", clsTypeFound);
 
         //validate trait
         List<HierarchicalTypeDefinition<TraitType>> traitTypes = types.traitTypesAsJavaList();