Skip to content

A
atlas
Commit e1bdafa2 by Suma S
Options

Merge pull request #136 from hortonworks/master

Merging from master to DAL
parents 19630615 3d9b865d
Showing with 200 additions and 199 deletions
addons/hive-bridge/src/main/java/org/apache/hadoop/metadata/hive/bridge/HiveMetaStoreBridge.java
......@@ -101,8 +101,8 @@ public class HiveMetaStoreBridge {
Database hiveDB = hiveClient.getDatabase(databaseName);
dbRef = new Referenceable(HiveDataTypes.HIVE_DB.getName());
dbRef.set("name", hiveDB.getName());
dbRef.set("clusterName", clusterName);
dbRef.set(HiveDataModelGenerator.NAME, hiveDB.getName().toLowerCase());
dbRef.set(HiveDataModelGenerator.CLUSTER_NAME, clusterName);
dbRef.set("description", hiveDB.getDescription());
dbRef.set("locationUri", hiveDB.getLocationUri());
dbRef.set("parameters", hiveDB.getParameters());
......@@ -139,7 +139,7 @@ public class HiveMetaStoreBridge {
// Import Partitions
Referenceable sdReferenceable = getSDForTable(databaseName, tableName);
importPartitions(databaseName, tableName, databaseReferenceable, tableReferenceable, sdReferenceable);
registerPartitions(databaseName, tableName, tableReferenceable, sdReferenceable);
// Import Indexes
importIndexes(databaseName, tableName, databaseReferenceable, tableReferenceable);
......@@ -159,8 +159,9 @@ public class HiveMetaStoreBridge {
LOG.debug("Getting reference for database {}", databaseName);
String typeName = HiveDataTypes.HIVE_DB.getName();
String dslQuery = String.format("%s where name = '%s' and clusterName = '%s'", typeName,
databaseName.toLowerCase(), clusterName);
String dslQuery = String.format("%s where %s = '%s' and %s = '%s'", typeName,
HiveDataModelGenerator.NAME, databaseName.toLowerCase(), HiveDataModelGenerator.CLUSTER_NAME,
clusterName);
return getEntityReferenceFromDSL(typeName, dslQuery);
}
......@@ -194,6 +195,10 @@ public class HiveMetaStoreBridge {
}
}
public static String getTableName(String clusterName, String dbName, String tableName) {
return String.format("%s.%s@%s", dbName.toLowerCase(), tableName.toLowerCase(), clusterName);
}
/**
* Gets reference for the table
*
......@@ -206,17 +211,9 @@ public class HiveMetaStoreBridge {
LOG.debug("Getting reference for table {}.{}", dbName, tableName);
String typeName = HiveDataTypes.HIVE_TABLE.getName();
String dslQuery = String.format(
"%s as t where name = '%s', dbName where name = '%s' and " + "clusterName = '%s' select t",
HiveDataTypes.HIVE_TABLE.getName(), tableName.toLowerCase(), dbName.toLowerCase(), clusterName);
String entityName = getTableName(clusterName, dbName, tableName);
String dslQuery = String.format("%s as t where name = '%s'", typeName, entityName);
return getEntityReferenceFromDSL(typeName, dslQuery);
// String dbType = HiveDataTypes.HIVE_DB.getName();
// String gremlinQuery = String.format("g.V.has('__typeName', '%s').has('%s.name', '%s').as('t').out"
// + "('__%s.dbName').has('%s.name', '%s').has('%s.clusterName', '%s').back('t').toList()",
// typeName, typeName, tableName, typeName, dbType, dbName, dbType, clusterName);
// return getEntityReferenceFromGremlin(typeName, gremlinQuery);
}
private Referenceable getEntityReferenceFromGremlin(String typeName, String gremlinQuery) throws MetadataServiceException,
......@@ -241,13 +238,12 @@ public class HiveMetaStoreBridge {
// + "dbName where name = '%s' and clusterName = '%s' select p", typeName, valuesStr, tableName,
// dbName, clusterName);
String dbType = HiveDataTypes.HIVE_DB.getName();
String tableType = HiveDataTypes.HIVE_TABLE.getName();
String datasetType = MetadataServiceClient.DATA_SET_SUPER_TYPE;
String tableEntityName = getTableName(clusterName, dbName, tableName);
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')"
+ ".has('%s.clusterName', '%s').back('p').toList()", typeName, typeName, valuesStr, typeName,
datasetType, tableName.toLowerCase(), tableType, dbType, dbName.toLowerCase(), dbType, clusterName);
+ "out('__%s.table').has('%s.name', '%s').back('p').toList()", typeName, typeName, valuesStr,
typeName, datasetType, tableEntityName);
return getEntityReferenceFromGremlin(typeName, gremlinQuery);
}
......@@ -278,7 +274,9 @@ public class HiveMetaStoreBridge {
Table hiveTable = hiveClient.getTable(dbName, tableName);
tableRef = new Referenceable(HiveDataTypes.HIVE_TABLE.getName());
tableRef.set("name", hiveTable.getTableName());
tableRef.set(HiveDataModelGenerator.NAME,
getTableName(clusterName, hiveTable.getDbName(), hiveTable.getTableName()));
tableRef.set(HiveDataModelGenerator.TABLE_NAME, hiveTable.getTableName().toLowerCase());
tableRef.set("owner", hiveTable.getOwner());
tableRef.set("createTime", hiveTable.getMetadata().getProperty(hive_metastoreConstants.DDL_TIME));
......@@ -288,7 +286,7 @@ public class HiveMetaStoreBridge {
tableRef.set(HiveDataModelGenerator.COMMENT, hiveTable.getParameters().get(HiveDataModelGenerator.COMMENT));
// add reference to the database
tableRef.set("dbName", dbReference);
tableRef.set(HiveDataModelGenerator.DB, dbReference);
List<Referenceable> colList = getColumns(hiveTable.getCols());
tableRef.set("columns", colList);
......@@ -323,15 +321,13 @@ public class HiveMetaStoreBridge {
return tableRef;
}
private void importPartitions(String db, String tableName,
Referenceable dbReferenceable,
Referenceable tableReferenceable,
Referenceable sdReferenceable) throws Exception {
private void registerPartitions(String db, String tableName, Referenceable tableReferenceable,
Referenceable sdReferenceable) throws Exception {
Set<Partition> tableParts = hiveClient.getAllPartitionsOf(new Table(Table.getEmptyTable(db, tableName)));
if (tableParts.size() > 0) {
for (Partition hivePart : tableParts) {
importPartition(hivePart, dbReferenceable, tableReferenceable, sdReferenceable);
registerPartition(hivePart, tableReferenceable, sdReferenceable);
}
}
}
......@@ -339,17 +335,14 @@ public class HiveMetaStoreBridge {
public Referenceable registerPartition(Partition partition) throws Exception {
String dbName = partition.getTable().getDbName();
String tableName = partition.getTable().getTableName();
Referenceable dbRef = registerDatabase(dbName);
Referenceable tableRef = registerTable(dbName, tableName);
Referenceable sdRef = getSDForTable(dbName, tableName);
return importPartition(partition, dbRef, tableRef, sdRef);
return registerPartition(partition, tableRef, sdRef);
}
private Referenceable importPartition(Partition hivePart,
Referenceable dbReferenceable,
Referenceable tableReferenceable,
Referenceable sdReferenceable) throws Exception {
LOG.info("Importing partition for {}.{} with values {}", dbReferenceable, tableReferenceable,
private Referenceable registerPartition(Partition hivePart, Referenceable tableReferenceable,
Referenceable sdReferenceable) throws Exception {
LOG.info("Registering partition for {} with values {}", tableReferenceable,
StringUtils.join(hivePart.getValues(), ","));
String dbName = hivePart.getTable().getDbName();
String tableName = hivePart.getTable().getTableName();
......@@ -359,8 +352,7 @@ public class HiveMetaStoreBridge {
partRef = new Referenceable(HiveDataTypes.HIVE_PARTITION.getName());
partRef.set("values", hivePart.getValues());
partRef.set("dbName", dbReferenceable);
partRef.set("tableName", tableReferenceable);
partRef.set(HiveDataModelGenerator.TABLE, tableReferenceable);
//todo fix
partRef.set("createTime", hivePart.getLastAccessTime());
......@@ -398,15 +390,15 @@ public class HiveMetaStoreBridge {
LOG.info("Importing index {} for {}.{}", index.getIndexName(), dbReferenceable, tableReferenceable);
Referenceable indexRef = new Referenceable(HiveDataTypes.HIVE_INDEX.getName());
indexRef.set("indexName", index.getIndexName());
indexRef.set(HiveDataModelGenerator.NAME, index.getIndexName());
indexRef.set("indexHandlerClass", index.getIndexHandlerClass());
indexRef.set("dbName", dbReferenceable);
indexRef.set(HiveDataModelGenerator.DB, dbReferenceable);
indexRef.set("createTime", index.getCreateTime());
indexRef.set("lastAccessTime", index.getLastAccessTime());
indexRef.set("origTableName", index.getOrigTableName());
indexRef.set("indexTableName", index.getIndexTableName());
indexRef.set("origTable", index.getOrigTableName());
indexRef.set("indexTable", index.getIndexTableName());
Referenceable sdReferenceable = fillStorageDescStruct(index.getSd(), null);
indexRef.set("sd", sdReferenceable);
......@@ -430,7 +422,7 @@ public class HiveMetaStoreBridge {
String serdeInfoName = HiveDataTypes.HIVE_SERDE.getName();
Struct serdeInfoStruct = new Struct(serdeInfoName);
serdeInfoStruct.set("name", serdeInfo.getName());
serdeInfoStruct.set(HiveDataModelGenerator.NAME, serdeInfo.getName());
serdeInfoStruct.set("serializationLib", serdeInfo.getSerializationLib());
serdeInfoStruct.set("parameters", serdeInfo.getParameters());
......@@ -438,23 +430,6 @@ public class HiveMetaStoreBridge {
sdReferenceable.set(HiveDataModelGenerator.STORAGE_NUM_BUCKETS, storageDesc.getNumBuckets());
sdReferenceable.set(HiveDataModelGenerator.STORAGE_IS_STORED_AS_SUB_DIRS, storageDesc.isStoredAsSubDirectories());
// Will need to revisit this after we fix typesystem.
/*
LOG.info("skewedInfo = " + skewedInfo);
String skewedInfoName = HiveDataTypes.HIVE_SKEWEDINFO.name();
Struct skewedInfoStruct = new Struct(skewedInfoName);
if (skewedInfo.getSkewedColNames().size() > 0) {
skewedInfoStruct.set("skewedColNames", skewedInfo.getSkewedColNames());
skewedInfoStruct.set("skewedColValues", skewedInfo.getSkewedColValues());
skewedInfoStruct.set("skewedColValueLocationMaps",
skewedInfo.getSkewedColValueLocationMaps());
StructType skewedInfotype = (StructType) hiveTypeSystem.getDataType(skewedInfoName);
ITypedStruct skewedInfoStructTyped =
skewedInfotype.convert(skewedInfoStruct, Multiplicity.OPTIONAL);
sdStruct.set("skewedInfo", skewedInfoStructTyped);
}
*/
//Use the passed column list if not null, ex: use same references for table and SD
List<FieldSchema> columns = storageDesc.getCols();
if (columns != null && !columns.isEmpty()) {
......@@ -499,7 +474,7 @@ public class HiveMetaStoreBridge {
for (FieldSchema fs : schemaList) {
LOG.debug("Processing field " + fs);
Referenceable colReferenceable = new Referenceable(HiveDataTypes.HIVE_COLUMN.getName());
colReferenceable.set("name", fs.getName());
colReferenceable.set(HiveDataModelGenerator.NAME, fs.getName());
colReferenceable.set("type", fs.getType());
colReferenceable.set(HiveDataModelGenerator.COMMENT, fs.getComment());
......@@ -526,4 +501,12 @@ public class HiveMetaStoreBridge {
hiveMetaStoreBridge.registerHiveDataModel();
hiveMetaStoreBridge.importHiveMetadata();
}
public void updateTable(Referenceable tableReferenceable, Table newTable) throws MetadataServiceException {
MetadataServiceClient client = getMetadataServiceClient();
client.updateEntity(tableReferenceable.getId()._getId(), HiveDataModelGenerator.TABLE_NAME,
newTable.getTableName().toLowerCase());
client.updateEntity(tableReferenceable.getId()._getId(), HiveDataModelGenerator.NAME,
getTableName(clusterName, newTable.getDbName(), newTable.getTableName()));
}
}
addons/hive-bridge/src/main/java/org/apache/hadoop/metadata/hive/hook/HiveHook.java
......@@ -20,7 +20,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;
......@@ -231,10 +230,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/HiveDataModelGenerator.java
......@@ -57,9 +57,16 @@ public class HiveDataModelGenerator {
private final Map<String, StructTypeDefinition> structTypeDefinitionMap;
public static final String COMMENT = "comment";
public static final String STORAGE_NUM_BUCKETS = "numBuckets";
public static final String STORAGE_IS_STORED_AS_SUB_DIRS = "storedAsSubDirectories";
public static final String NAME = "name";
public static final String TABLE_NAME = "tableName";
public static final String CLUSTER_NAME = "clusterName";
public static final String TABLE = "table";
public static final String DB = "db";
public HiveDataModelGenerator() {
classTypeDefinitions = new HashMap<>();
enumTypeDefinitionMap = new HashMap<>();
......@@ -72,7 +79,6 @@ public class HiveDataModelGenerator {
// enums
createHiveObjectTypeEnum();
createHivePrincipalTypeEnum();
createFunctionTypeEnum();
createResourceTypeEnum();
// structs
......@@ -89,7 +95,6 @@ public class HiveDataModelGenerator {
createPartitionClass();
createTableClass();
createIndexClass();
createFunctionClass();
createRoleClass();
// DDL/DML Process
......@@ -154,17 +159,6 @@ public class HiveDataModelGenerator {
LOG.debug("Created definition for " + HiveDataTypes.HIVE_PRINCIPAL_TYPE.getName());
}
private void createFunctionTypeEnum() throws MetadataException {
EnumValue values[] = {
new EnumValue("JAVA", 1),
};
EnumTypeDefinition definition = new EnumTypeDefinition(
HiveDataTypes.HIVE_FUNCTION_TYPE.getName(), values);
enumTypeDefinitionMap.put(HiveDataTypes.HIVE_FUNCTION_TYPE.getName(), definition);
LOG.debug("Created definition for " + HiveDataTypes.HIVE_FUNCTION_TYPE.getName());
}
private void createResourceTypeEnum() throws MetadataException {
EnumValue values[] = {
new EnumValue("JAR", 1),
......@@ -179,7 +173,7 @@ public class HiveDataModelGenerator {
private void createSerDeStruct() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("name", DataTypes.STRING_TYPE.getName(),
new AttributeDefinition(NAME, DataTypes.STRING_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("serializationLib", DataTypes.STRING_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
......@@ -192,29 +186,6 @@ public class HiveDataModelGenerator {
LOG.debug("Created definition for " + HiveDataTypes.HIVE_SERDE.getName());
}
/*
private static final DataTypes.ArrayType STRING_ARRAY_TYPE =
new DataTypes.ArrayType(DataTypes.STRING_TYPE);
private static Multiplicity ZeroOrMore = new Multiplicity(0, Integer.MAX_VALUE, true);
private void createSkewedInfoStruct() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("skewedColNames",
String.format("array<%s>", DataTypes.STRING_TYPE.getName()),
ZeroOrMore, false, null),
new AttributeDefinition("skewedColValues",
String.format("array<%s>", STRING_ARRAY_TYPE.getName()),
ZeroOrMore, false, null),
new AttributeDefinition("skewedColValueLocationMaps", STRING_MAP_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
};
StructTypeDefinition definition = new StructTypeDefinition(
DefinedTypes.HIVE_SKEWEDINFO.getName(), attributeDefinitions);
structTypeDefinitionMap.put(DefinedTypes.HIVE_SKEWEDINFO.getName(), definition);
LOG.debug("Created definition for " + DefinedTypes.HIVE_SKEWEDINFO.getName());
}
*/
private void createOrderStruct() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("col", DataTypes.STRING_TYPE.getName(),
......@@ -283,9 +254,9 @@ public class HiveDataModelGenerator {
private void createDBClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("name", DataTypes.STRING_TYPE.getName(),
new AttributeDefinition(NAME, DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("clusterName", DataTypes.STRING_TYPE.getName(),
new AttributeDefinition(CLUSTER_NAME, DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("description", DataTypes.STRING_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
......@@ -308,7 +279,7 @@ public class HiveDataModelGenerator {
private void createTypeClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("name", DataTypes.STRING_TYPE.getName(),
new AttributeDefinition(NAME, DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("type1", DataTypes.STRING_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
......@@ -327,7 +298,7 @@ public class HiveDataModelGenerator {
private void createColumnClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("name", DataTypes.STRING_TYPE.getName(),
new AttributeDefinition(NAME, DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("type", DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
......@@ -343,13 +314,10 @@ public class HiveDataModelGenerator {
}
private void createPartitionClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("values", DataTypes.arrayTypeName(DataTypes.STRING_TYPE.getName()),
Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("dbName", HiveDataTypes.HIVE_DB.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("tableName", HiveDataTypes.HIVE_TABLE.getName(),
new AttributeDefinition(TABLE, HiveDataTypes.HIVE_TABLE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("createTime", DataTypes.LONG_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
......@@ -372,7 +340,9 @@ public class HiveDataModelGenerator {
private void createTableClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("dbName", HiveDataTypes.HIVE_DB.getName(),
new AttributeDefinition(TABLE_NAME, DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition(DB, HiveDataTypes.HIVE_DB.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("owner", DataTypes.STRING_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
......@@ -412,19 +382,19 @@ public class HiveDataModelGenerator {
private void createIndexClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("indexName", DataTypes.STRING_TYPE.getName(),
new AttributeDefinition(NAME, DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("indexHandlerClass", DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("dbName", HiveDataTypes.HIVE_DB.getName(),
new AttributeDefinition(DB, HiveDataTypes.HIVE_DB.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("createTime", DataTypes.LONG_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("lastAccessTime", DataTypes.LONG_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("origTableName", HiveDataTypes.HIVE_TABLE.getName(),
new AttributeDefinition("origTable", HiveDataTypes.HIVE_TABLE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("indexTableName", HiveDataTypes.HIVE_TABLE.getName(),
new AttributeDefinition("indexTable", HiveDataTypes.HIVE_TABLE.getName(),
Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("sd", HiveDataTypes.HIVE_STORAGEDESC.getName(),
Multiplicity.REQUIRED, false, null),
......@@ -441,33 +411,6 @@ public class HiveDataModelGenerator {
LOG.debug("Created definition for " + HiveDataTypes.HIVE_INDEX.getName());
}
private void createFunctionClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("functionName", DataTypes.STRING_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("dbName", HiveDataTypes.HIVE_DB.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("className", DataTypes.INT_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("ownerName", DataTypes.INT_TYPE.getName(),
Multiplicity.OPTIONAL, false, null),
new AttributeDefinition("ownerType", HiveDataTypes.HIVE_PRINCIPAL_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("createTime", DataTypes.LONG_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("functionType", HiveDataTypes.HIVE_FUNCTION_TYPE.getName(),
Multiplicity.REQUIRED, false, null),
new AttributeDefinition("resourceUris",
DataTypes.arrayTypeName(HiveDataTypes.HIVE_RESOURCEURI.getName()), Multiplicity.OPTIONAL, false,
null),
};
HierarchicalTypeDefinition<ClassType> definition = new HierarchicalTypeDefinition<>(
ClassType.class, HiveDataTypes.HIVE_FUNCTION.getName(), null, attributeDefinitions);
classTypeDefinitions.put(HiveDataTypes.HIVE_FUNCTION.getName(), definition);
LOG.debug("Created definition for " + HiveDataTypes.HIVE_FUNCTION.getName());
}
private void createRoleClass() throws MetadataException {
AttributeDefinition[] attributeDefinitions = new AttributeDefinition[]{
new AttributeDefinition("roleName", DataTypes.STRING_TYPE.getName(),
......
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) {
ClassType expectedType = ts.getDataType(ClassType.class, classType.typeName);
Assert.assertEquals(expectedType.immediateAttrs.size(), classType.attributeDefinitions.length);
Assert.assertEquals(expectedType.superTypes.size(), classType.superTypes.size());
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();
......
typesystem/src/main/java/org/apache/hadoop/metadata/typesystem/types/DataTypes.java
......@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.metadata.MetadataException;
import org.apache.hadoop.metadata.typesystem.IReferenceableInstance;
import org.apache.hadoop.metadata.typesystem.persistence.Id;
......@@ -455,7 +456,7 @@ public class DataTypes {
@Override
public String convert(Object val, Multiplicity m) throws MetadataException {
if (val != null) {
if (StringUtils.isNotBlank((CharSequence) val)) {
return val.toString();
}
return convertNull(m);
......
webapp/src/main/java/org/apache/hadoop/metadata/web/resources/TypesResource.java
......@@ -45,6 +45,7 @@ import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.Response;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
......@@ -87,18 +88,17 @@ public class TypesResource {
JSONObject typesJson = metadataService.createType(typeDefinition);
final JSONArray typesJsonArray = typesJson.getJSONArray(MetadataServiceClient.TYPES);
List<Map<String, String>> typesAddedList = new ArrayList<>();
JSONArray typesResponse = new JSONArray();
for (int i = 0; i < typesJsonArray.length(); i++) {
final String name = typesJsonArray.getString(i);
typesAddedList.add(
new HashMap<String, String>() {{
put(MetadataServiceClient.NAME, name);
}});
typesResponse.put(new JSONObject() {{
put(MetadataServiceClient.NAME, name);
}});
}
JSONObject response = new JSONObject();
response.put(MetadataServiceClient.REQUEST_ID, Servlets.getRequestId());
response.put(MetadataServiceClient.TYPES, typesAddedList);
response.put(MetadataServiceClient.TYPES, typesResponse);
return Response.status(ClientResponse.Status.CREATED).entity(response).build();
} catch (MetadataException | IllegalArgumentException e) {
LOG.error("Unable to persist types", e);
......
webapp/src/test/java/org/apache/hadoop/metadata/web/resources/EntityJerseyResourceIT.java
......@@ -51,6 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import javax.ws.rs.HttpMethod;
......@@ -95,6 +96,26 @@ public class EntityJerseyResourceIT extends BaseResourceIT {
}
}
@DataProvider
public Object[][] invalidAttrValues() {
return new Object[][]{
{null}, {""}, {" "}};
}
@Test(dataProvider = "invalidAttrValues")
public void testEntityInvalidValue(String value) throws Exception {
Referenceable databaseInstance = new Referenceable(DATABASE_TYPE);
databaseInstance.set("name", randomString());
databaseInstance.set("description", value);
try {
createInstance(databaseInstance);
Assert.fail("Exptected MetadataServiceException");
} catch(MetadataServiceException e) {
Assert.assertEquals(e.getStatus(), ClientResponse.Status.BAD_REQUEST);
}
}
@Test
public void testSubmitEntityWithBadDateFormat() throws Exception {
......
webapp/src/test/java/org/apache/hadoop/metadata/web/resources/TypesJerseyResourceIT.java
......@@ -83,7 +83,9 @@ public class TypesJerseyResourceIT extends BaseResourceIT {
Assert.assertNotNull(responseAsString);
JSONObject response = new JSONObject(responseAsString);
Assert.assertNotNull(response.get(MetadataServiceClient.TYPES));
JSONArray typesAdded = response.getJSONArray(MetadataServiceClient.TYPES);
Assert.assertEquals(typesAdded.length(), 1);
Assert.assertEquals(typesAdded.getJSONObject(0).getString("name"), typeDefinition.typeName);
Assert.assertNotNull(response.get(MetadataServiceClient.REQUEST_ID));
}
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment