You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ng...@apache.org on 2019/05/15 18:01:41 UTC
[hive] 01/02: HIVE-21663:[Thrift only] Support for object
translation in HMS (Naveen Gangam, reviewed by Daniel Dai)
This is an automated email from the ASF dual-hosted git repository.
ngangam pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
commit fcd4721591e0ba7d9ba24821f9528dadb7ec16fd
Author: Naveen Gangam <ng...@apache.org>
AuthorDate: Wed May 15 13:51:38 2019 -0400
HIVE-21663:[Thrift only] Support for object translation in HMS (Naveen Gangam, reviewed by Daniel Dai)
---
.../hadoop/hive/metastore/HiveMetaStoreClient.java | 124 ++++
.../hadoop/hive/metastore/IMetaStoreClient.java | 18 +
.../hadoop/hive/metastore/conf/MetastoreConf.java | 6 +
.../src/main/thrift/hive_metastore.thrift | 159 +++--
.../hadoop/hive/metastore/HiveMetaStore.java | 126 +++-
.../metastore/IMetaStoreMetadataTransformer.java | 58 ++
.../metastore/MetastoreDefaultTransformer.java | 404 +++++++++++
.../metastore/HiveMetaStoreClientPreCatalog.java | 9 +
.../hadoop/hive/metastore/TestHiveMetaStore.java | 789 +++++++++++++++++++++
9 files changed, 1629 insertions(+), 64 deletions(-)
diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
index c55a680..f947ee1 100644
--- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
+++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClient.java
@@ -142,6 +142,8 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
private int retries = 5;
private long retryDelaySeconds = 0;
private final ClientCapabilities version;
+ private static String[] processorCapabilities;
+ private static String processorIdentifier;
//copied from ErrorMsg.java
private static final String REPL_EVENTS_MISSING_IN_METASTORE = "Notification events are missing in the meta store.";
@@ -719,6 +721,14 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
}
}
+ public static void setProcessorCapabilities(final String[] capabilities) {
+ processorCapabilities = capabilities;
+ }
+
+ public static void setProcessorIdentifier(final String id) {
+ processorIdentifier = id;
+ }
+
@Override
public void setMetaConf(String key, String value) throws TException {
client.setMetaConf(key, value);
@@ -1652,12 +1662,14 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
@Override
public List<Partition> listPartitions(String db_name, String tbl_name, short max_parts)
throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
return listPartitions(getDefaultCatalog(conf), db_name, tbl_name, max_parts);
}
@Override
public List<Partition> listPartitions(String catName, String db_name, String tbl_name,
int max_parts) throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
List<Partition> parts = client.get_partitions(prependCatalogToDbName(catName, db_name, conf),
tbl_name, shrinkMaxtoShort(max_parts));
return deepCopyPartitions(
@@ -1681,12 +1693,14 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
@Override
public List<Partition> listPartitions(String db_name, String tbl_name,
List<String> part_vals, short max_parts) throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
return listPartitions(getDefaultCatalog(conf), db_name, tbl_name, part_vals, max_parts);
}
@Override
public List<Partition> listPartitions(String catName, String db_name, String tbl_name,
List<String> part_vals, int max_parts) throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
List<Partition> parts = client.get_partitions_ps(prependCatalogToDbName(catName, db_name, conf),
tbl_name, part_vals, shrinkMaxtoShort(max_parts));
return deepCopyPartitions(FilterUtils.filterPartitionsIfEnabled(isClientFilterEnabled, filterHook, parts));
@@ -1696,6 +1710,7 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
public List<Partition> listPartitionsWithAuthInfo(String db_name, String tbl_name,
short max_parts, String user_name,
List<String> group_names) throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
return listPartitionsWithAuthInfo(getDefaultCatalog(conf), db_name, tbl_name, max_parts, user_name,
group_names);
}
@@ -1704,6 +1719,7 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
public List<Partition> listPartitionsWithAuthInfo(String catName, String dbName, String tableName,
int maxParts, String userName,
List<String> groupNames) throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
List<Partition> parts = client.get_partitions_with_auth(prependCatalogToDbName(catName,
dbName, conf), tableName, shrinkMaxtoShort(maxParts), userName, groupNames);
return deepCopyPartitions(FilterUtils.filterPartitionsIfEnabled(isClientFilterEnabled, filterHook, parts));
@@ -1714,6 +1730,7 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
List<String> part_vals, short max_parts,
String user_name, List<String> group_names)
throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
return listPartitionsWithAuthInfo(getDefaultCatalog(conf), db_name, tbl_name, part_vals, max_parts,
user_name, group_names);
}
@@ -1723,6 +1740,7 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
List<String> partialPvals, int maxParts,
String userName, List<String> groupNames)
throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
List<Partition> parts = client.get_partitions_ps_with_auth(prependCatalogToDbName(catName,
dbName, conf), tableName, partialPvals, shrinkMaxtoShort(maxParts), userName, groupNames);
return deepCopyPartitions(FilterUtils.filterPartitionsIfEnabled(isClientFilterEnabled, filterHook, parts));
@@ -1737,6 +1755,7 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
@Override
public List<Partition> listPartitionsByFilter(String catName, String db_name, String tbl_name,
String filter, int max_parts) throws TException {
+ // TODO should we add capabilities here as well as it returns Partition objects
List<Partition> parts =client.get_partitions_by_filter(prependCatalogToDbName(
catName, db_name, conf), tbl_name, filter, shrinkMaxtoShort(max_parts));
return deepCopyPartitions(FilterUtils.filterPartitionsIfEnabled(isClientFilterEnabled, filterHook, parts));
@@ -1852,6 +1871,11 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
tbl_name);
gpbnr.setNames(part_names);
gpbnr.setGet_col_stats(getColStats);
+ if (processorCapabilities != null)
+ gpbnr.setProcessorCapabilities(Arrays.asList(processorCapabilities));
+ if (processorIdentifier != null)
+ gpbnr.setProcessorIdentifier(processorIdentifier);
+
List<Partition> parts = client.get_partitions_by_names_req(gpbnr).getPartitions();
return deepCopyPartitions(FilterUtils.filterPartitionsIfEnabled(isClientFilterEnabled, filterHook, parts));
}
@@ -1908,6 +1932,11 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
req.setCatName(catName);
req.setCapabilities(version);
req.setGetColumnStats(getColumnStats);
+ if (processorCapabilities != null)
+ req.setProcessorCapabilities(Arrays.asList(processorCapabilities));
+ if (processorIdentifier != null)
+ req.setProcessorIdentifier(processorIdentifier);
+
Table t = client.get_table_req(req).getTable();
return deepCopy(FilterUtils.filterTableIfEnabled(isClientFilterEnabled, filterHook, t));
}
@@ -1926,6 +1955,11 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
req.setCapabilities(version);
req.setValidWriteIdList(validWriteIdList);
req.setGetColumnStats(getColumnStats);
+ if (processorCapabilities != null)
+ req.setProcessorCapabilities(Arrays.asList(processorCapabilities));
+ if (processorIdentifier != null)
+ req.setProcessorIdentifier(processorIdentifier);
+
Table t = client.get_table_req(req).getTable();
return deepCopy(FilterUtils.filterTableIfEnabled(isClientFilterEnabled, filterHook, t));
}
@@ -1943,6 +1977,8 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
req.setCatName(catName);
req.setTblNames(tableNames);
req.setCapabilities(version);
+ if (processorCapabilities != null)
+ req.setProcessorCapabilities(Arrays.asList(processorCapabilities));
List<Table> tabs = client.get_table_objects_by_name_req(req).getTables();
return deepCopyTables(FilterUtils.filterTablesIfEnabled(isClientFilterEnabled, filterHook, tabs));
}
@@ -2031,6 +2067,21 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
return FilterUtils.filterTableNamesIfEnabled(isClientFilterEnabled, filterHook, catName, dbName, tables);
}
+ /** {@inheritDoc} */
+ @Override
+ public List<ExtendedTableInfo> getTablesExt(String catName, String dbName, String tablePattern,
+ int requestedFields, int limit) throws MetaException, TException {
+ if (catName == null)
+ catName = getDefaultCatalog(conf);
+ GetTablesExtRequest req = new GetTablesExtRequest(catName, dbName, tablePattern, requestedFields);
+ req.setLimit(limit);
+ if (processorIdentifier != null)
+ req.setProcessorIdentifier(processorIdentifier);
+ if (processorCapabilities != null)
+ req.setProcessorCapabilities(Arrays.asList(processorCapabilities));
+ return client.get_tables_ext(req);
+ }
+
@Override
public List<Table> getAllMaterializedViewObjectsForRewriting() throws TException {
try {
@@ -3813,6 +3864,10 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
@Override
public GetPartitionsResponse getPartitionsWithSpecs(GetPartitionsRequest request)
throws TException {
+ if (processorCapabilities != null)
+ request.setProcessorCapabilities(Arrays.asList(processorCapabilities));
+ if (processorIdentifier != null)
+ request.setProcessorIdentifier(processorIdentifier);
return client.get_partitions_with_specs(request);
}
@@ -3855,4 +3910,73 @@ public class HiveMetaStoreClient implements IMetaStoreClient, AutoCloseable {
public String getServerVersion() throws TException {
return client.getVersion();
}
+
+ /**
+ * Builder for requiredFields bitmask to be sent via GetTablesExtRequest
+ */
+ public static class GetTablesRequestBuilder {
+ private int requestedFields = 0x0;
+ final static GetTablesExtRequestFields defValue = GetTablesExtRequestFields.ALL;
+
+ public GetTablesRequestBuilder() {
+ }
+
+ public GetTablesRequestBuilder with(GetTablesExtRequestFields type) {
+ switch (type) {
+ case ALL :
+ this.requestedFields |= Integer.MAX_VALUE;
+ break;
+ case PROCESSOR_CAPABILITIES :
+ this.requestedFields |= 0x2;
+ break;
+ case ACCESS_TYPE :
+ this.requestedFields |= 0x1;
+ break;
+ default:
+ this.requestedFields |= Integer.MAX_VALUE;
+ break;
+ }
+ return this;
+ }
+
+ public int bitValue() {
+ return this.requestedFields;
+ }
+
+ public static int defaultValue() {
+ return new GetTablesRequestBuilder().with(defValue).bitValue();
+ }
+ }
+
+ /**
+ * Builder for building table capabilities to be includes in TBLPROPERTIES
+ * during createTable
+ */
+ public static class TableCapabilityBuilder {
+ private String capabilitiesString = null;
+ public static final String KEY_CAPABILITIES = "OBJCAPABILITIES";
+
+ public TableCapabilityBuilder() {
+ capabilitiesString = new String();
+ }
+
+ public TableCapabilityBuilder add(String skill) {
+ if (skill != null) {
+ capabilitiesString += skill + ",";
+ }
+ return this;
+ }
+
+ public String build() {
+ return this.capabilitiesString.substring(0, capabilitiesString.length() - 1);
+ }
+
+ public String getDBValue() {
+ return KEY_CAPABILITIES + "=" + build();
+ }
+
+ public static String getKey() {
+ return KEY_CAPABILITIES;
+ }
+ }
}
diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
index 8999d55..aa7e8df 100644
--- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
+++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreClient.java
@@ -50,6 +50,7 @@ import org.apache.hadoop.hive.metastore.api.DataOperationType;
import org.apache.hadoop.hive.metastore.api.DefaultConstraintsRequest;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
+import org.apache.hadoop.hive.metastore.api.ExtendedTableInfo;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.FindSchemasByColsResp;
import org.apache.hadoop.hive.metastore.api.FindSchemasByColsRqst;
@@ -65,6 +66,7 @@ import org.apache.hadoop.hive.metastore.api.GetPrincipalsInRoleRequest;
import org.apache.hadoop.hive.metastore.api.GetPrincipalsInRoleResponse;
import org.apache.hadoop.hive.metastore.api.GetRoleGrantsForPrincipalRequest;
import org.apache.hadoop.hive.metastore.api.GetRoleGrantsForPrincipalResponse;
+import org.apache.hadoop.hive.metastore.api.GetTablesExtRequestFields;
import org.apache.hadoop.hive.metastore.api.HeartbeatTxnRangeResponse;
import org.apache.hadoop.hive.metastore.api.HiveObjectPrivilege;
import org.apache.hadoop.hive.metastore.api.HiveObjectRef;
@@ -345,6 +347,22 @@ public interface IMetaStoreClient {
throws MetaException, TException, UnknownDBException;
/**
+ * Get the names of all the tables along with extended table metadata
+ * @param catName catalog name.
+ * @param dbName Name of the database to fetch tables from.
+ * @param tablePattern pattern to match the tables names.
+ * @param requestedFields An int bitmask to indicate the depth of the returned objects
+ * @param processorCapabilities A list of "capabilities" possessed by the caller, to be matched with table's params
+ * @param processorId Any string id to identify the caller/client, for logging purposes only.
+ * @param limit Maximum size of the result set. <=0 indicates no limit
+ * @return List of ExtendedTableInfo that match the input arguments.
+ * @throws MetaException Thrown if there is error on fetching from DBMS.
+ * @throws TException Thrown if there is a thrift transport exception.
+ */
+ public List<ExtendedTableInfo> getTablesExt(String catName, String dbName, String tablePattern, int requestedFields,
+ int limit) throws MetaException, TException;
+
+ /**
* Get materialized views that have rewriting enabled. This will use the default catalog.
* @param dbName Name of the database to fetch materialized views from.
* @return List of materialized view names.
diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
index bac9d01..c0d8919 100644
--- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
+++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
@@ -750,6 +750,12 @@ public class MetastoreConf {
"metastore.partition.management.table.pattern", "*",
"Automatic partition management will look for tables using the specified table pattern"),
+ METASTORE_METADATA_TRANSFORMER_CLASS("metastore.metadata.transformer.class", "metastore.metadata.transformer.class",
+ "org.apache.hadoop.hive.metastore.MetastoreDefaultTransformer",
+ "Fully qualified class name for the metastore metadata transformer class \n"
+ + "which is used by HMS Server to fetch the extended tables/partitions information \n"
+ + "based on the data processor capabilities \n"
+ + " This class should implement the IMetaStoreMetadataTransformer interface"),
MULTITHREADED("javax.jdo.option.Multithreaded", "javax.jdo.option.Multithreaded", true,
"Set this to true if multiple threads access metastore through JDO concurrently."),
MAX_OPEN_TXNS("metastore.max.open.txns", "hive.max.open.txns", 100000,
diff --git a/standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift b/standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift
index 9dd2580..7d5d5b7 100644
--- a/standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift
+++ b/standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift
@@ -29,6 +29,10 @@ namespace php metastore
namespace cpp Apache.Hadoop.Hive
const string DDL_TIME = "transient_lastDdlTime"
+const byte ACCESSTYPE_NONE = 1;
+const byte ACCESSTYPE_READONLY = 2;
+const byte ACCESSTYPE_WRITEONLY = 4;
+const byte ACCESSTYPE_READWRITE = 8;
struct Version {
1: string version,
@@ -433,6 +437,63 @@ struct CreationMetadata {
6: optional i64 materializationTime
}
+// table information
+struct Table {
+ 1: optional i64 id, // id of the table. It will be ignored if set. It's only for
+ // read purposed
+ 2: string tableName, // name of the table
+ 3: string dbName, // database name ('default')
+ 4: string owner, // owner of this table
+ 5: i32 createTime, // creation time of the table
+ 6: i32 lastAccessTime, // last access time (usually this will be filled from HDFS and shouldn't be relied on)
+ 7: i32 retention, // retention time
+ 8: StorageDescriptor sd, // storage descriptor of the table
+ 9: list<FieldSchema> partitionKeys, // partition keys of the table. only primitive types are supported
+ 10: map<string, string> parameters, // to store comments or any other user level parameters
+ 11: string viewOriginalText, // original view text, null for non-view
+ 12: string viewExpandedText, // expanded view text, null for non-view
+ 13: string tableType, // table type enum, e.g. EXTERNAL_TABLE
+ 14: optional PrincipalPrivilegeSet privileges,
+ 15: optional bool temporary=false,
+ 16: optional bool rewriteEnabled, // rewrite enabled or not
+ 17: optional CreationMetadata creationMetadata, // only for MVs, it stores table names used and txn list at MV creation
+ 18: optional string catName, // Name of the catalog the table is in
+ 19: optional PrincipalType ownerType = PrincipalType.USER, // owner type of this table (default to USER for backward compatibility)
+ 20: optional i64 writeId=-1,
+ 21: optional bool isStatsCompliant,
+ 22: optional ColumnStatistics colStats, // column statistics for table
+ 23: optional byte accessType
+}
+
+struct Partition {
+ 1: list<string> values // string value is converted to appropriate partition key type
+ 2: string dbName,
+ 3: string tableName,
+ 4: i32 createTime,
+ 5: i32 lastAccessTime,
+ 6: StorageDescriptor sd,
+ 7: map<string, string> parameters,
+ 8: optional PrincipalPrivilegeSet privileges,
+ 9: optional string catName,
+ 10: optional i64 writeId=-1,
+ 11: optional bool isStatsCompliant,
+ 12: optional ColumnStatistics colStats // column statistics for partition
+}
+
+struct PartitionWithoutSD {
+ 1: list<string> values // string value is converted to appropriate partition key type
+ 2: i32 createTime,
+ 3: i32 lastAccessTime,
+ 4: string relativePath,
+ 5: map<string, string> parameters,
+ 6: optional PrincipalPrivilegeSet privileges
+}
+
+struct PartitionSpecWithSharedSD {
+ 1: list<PartitionWithoutSD> partitions,
+ 2: StorageDescriptor sd,
+}
+
// column statistics
struct BooleanColumnStatsData {
@@ -531,62 +592,6 @@ struct ColumnStatistics {
// the calling query?
}
-// table information
-struct Table {
- 1: optional i64 id, // id of the table. It will be ignored if set. It's only for
- // read purposed
- 2: string tableName, // name of the table
- 3: string dbName, // database name ('default')
- 4: string owner, // owner of this table
- 5: i32 createTime, // creation time of the table
- 6: i32 lastAccessTime, // last access time (usually this will be filled from HDFS and shouldn't be relied on)
- 7: i32 retention, // retention time
- 8: StorageDescriptor sd, // storage descriptor of the table
- 9: list<FieldSchema> partitionKeys, // partition keys of the table. only primitive types are supported
- 10: map<string, string> parameters, // to store comments or any other user level parameters
- 11: string viewOriginalText, // original view text, null for non-view
- 12: string viewExpandedText, // expanded view text, null for non-view
- 13: string tableType, // table type enum, e.g. EXTERNAL_TABLE
- 14: optional PrincipalPrivilegeSet privileges,
- 15: optional bool temporary=false,
- 16: optional bool rewriteEnabled, // rewrite enabled or not
- 17: optional CreationMetadata creationMetadata, // only for MVs, it stores table names used and txn list at MV creation
- 18: optional string catName, // Name of the catalog the table is in
- 19: optional PrincipalType ownerType = PrincipalType.USER, // owner type of this table (default to USER for backward compatibility)
- 20: optional i64 writeId=-1,
- 21: optional bool isStatsCompliant,
- 22: optional ColumnStatistics colStats // column statistics for table
-}
-
-struct Partition {
- 1: list<string> values // string value is converted to appropriate partition key type
- 2: string dbName,
- 3: string tableName,
- 4: i32 createTime,
- 5: i32 lastAccessTime,
- 6: StorageDescriptor sd,
- 7: map<string, string> parameters,
- 8: optional PrincipalPrivilegeSet privileges,
- 9: optional string catName,
- 10: optional i64 writeId=-1,
- 11: optional bool isStatsCompliant,
- 12: optional ColumnStatistics colStats // column statistics for partition
-}
-
-struct PartitionWithoutSD {
- 1: list<string> values // string value is converted to appropriate partition key type
- 2: i32 createTime,
- 3: i32 lastAccessTime,
- 4: string relativePath,
- 5: map<string, string> parameters,
- 6: optional PrincipalPrivilegeSet privileges
-}
-
-struct PartitionSpecWithSharedSD {
- 1: list<PartitionWithoutSD> partitions,
- 2: StorageDescriptor sd,
-}
-
struct PartitionListComposingSpec {
1: list<Partition> partitions
}
@@ -841,7 +846,9 @@ struct GetPartitionsByNamesRequest {
1: required string db_name,
2: required string tbl_name,
3: optional list<string> names,
- 4: optional bool get_col_stats
+ 4: optional bool get_col_stats,
+ 5: optional list<string> processorCapabilities,
+ 6: optional string processorIdentifier
}
struct GetPartitionsByNamesResult {
@@ -865,6 +872,13 @@ enum TxnType {
COMPACTION = 3
}
+// specifies which info to return with GetTablesExtRequest
+enum GetTablesExtRequestFields {
+ ACCESS_TYPE = 1, // return accessType
+ PROCESSOR_CAPABILITIES = 2, // return ALL Capabilities for each Tables
+ ALL = 2147483647
+}
+
struct ResourceUri {
1: ResourceType resourceType,
2: string uri,
@@ -1331,7 +1345,6 @@ enum ClientCapability {
INSERT_ONLY_TABLES = 2
}
-
struct ClientCapabilities {
1: required list<ClientCapability> values
}
@@ -1342,7 +1355,9 @@ struct GetTableRequest {
3: optional ClientCapabilities capabilities,
4: optional string catName,
6: optional string validWriteIdList,
- 7: optional bool getColumnStats
+ 7: optional bool getColumnStats,
+ 8: optional list<string> processorCapabilities,
+ 9: optional string processorIdentifier
}
struct GetTableResult {
@@ -1354,13 +1369,32 @@ struct GetTablesRequest {
1: required string dbName,
2: optional list<string> tblNames,
3: optional ClientCapabilities capabilities,
- 4: optional string catName
+ 4: optional string catName,
+ 5: optional list<string> processorCapabilities,
+ 6: optional string processorIdentifier
}
struct GetTablesResult {
1: required list<Table> tables
}
+struct GetTablesExtRequest {
+ 1: required string catalog,
+ 2: required string database,
+ 3: required string tableNamePattern, // table name matching pattern
+ 4: required i32 requestedFields, // ORed GetTablesExtRequestFields
+ 5: optional i32 limit, // maximum number of tables returned (0=all)
+ 6: optional list<string> processorCapabilities, // list of capabilities “possessed” by the client
+ 7: optional string processorIdentifier
+}
+
+// response to GetTablesExtRequest call
+struct ExtendedTableInfo {
+ 1: required string tblName, // always returned
+ 2: optional i32 accessType, // if AccessType set
+ 3: optional list<string> processorCapabilities // if ProcessorCapabilities set
+}
+
// Request type for cm_recycle
struct CmRecycleRequest {
1: required string dataPath,
@@ -1789,7 +1823,9 @@ struct GetPartitionsRequest {
5: optional string user,
6: optional list<string> groupNames,
7: GetPartitionsProjectionSpec projectionSpec
- 8: GetPartitionsFilterSpec filterSpec // TODO not yet implemented. Must be present but ignored
+ 8: GetPartitionsFilterSpec filterSpec, // TODO not yet implemented. Must be present but ignored
+ 9: optional list<string> processorCapabilities,
+ 10: optional string processorIdentifier
}
// Exceptions.
@@ -1948,6 +1984,7 @@ service ThriftHiveMetastore extends fb303.FacebookService
Table get_table(1:string dbname, 2:string tbl_name)
throws (1:MetaException o1, 2:NoSuchObjectException o2)
list<Table> get_table_objects_by_name(1:string dbname, 2:list<string> tbl_names)
+ list<ExtendedTableInfo> get_tables_ext(1: GetTablesExtRequest req) throws (1: MetaException o1)
GetTableResult get_table_req(1:GetTableRequest req) throws (1:MetaException o1, 2:NoSuchObjectException o2)
GetTablesResult get_table_objects_by_name_req(1:GetTablesRequest req)
throws (1:MetaException o1, 2:InvalidOperationException o2, 3:UnknownDBException o3)
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
index 1a694fb..7d948e4 100644
--- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HiveMetaStore.java
@@ -33,6 +33,7 @@ import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.prependNotNu
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Modifier;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
@@ -278,6 +279,7 @@ public class HiveMetaStore extends ThriftHiveMetastore {
private FileMetadataManager fileMetadataManager;
private PartitionExpressionProxy expressionProxy;
private StorageSchemaReader storageSchemaReader;
+ private IMetaStoreMetadataTransformer transformer;
// Variables for metrics
// Package visible so that HMSMetricsListener can see them.
@@ -629,6 +631,26 @@ public class HiveMetaStore extends ThriftHiveMetastore {
isServerFilterEnabled = getIfServerFilterenabled();
filterHook = isServerFilterEnabled ? loadFilterHooks() : null;
+
+ String className = MetastoreConf.getVar(conf, MetastoreConf.ConfVars.METASTORE_METADATA_TRANSFORMER_CLASS);
+ Class<?> clazz;
+ try {
+ clazz = conf.getClassByName(className);
+ } catch (ClassNotFoundException e) {
+ LOG.error("Unable to load class " + className, e);
+ throw new IllegalArgumentException(e);
+ }
+ Constructor<?> constructor;
+ try {
+ constructor = clazz.getConstructor(IHMSHandler.class);
+ if (Modifier.isPrivate(constructor.getModifiers()))
+ throw new IllegalArgumentException("Illegal implementation for metadata transformer. Constructor is private");
+ transformer = (IMetaStoreMetadataTransformer) constructor.newInstance(this);
+ } catch (NoSuchMethodException | InstantiationException | IllegalAccessException
+ | IllegalArgumentException | InvocationTargetException e) {
+ LOG.error("Unable to create instance of class " + className, e);
+ throw new IllegalArgumentException(e);
+ }
}
/**
@@ -3029,16 +3051,85 @@ public class HiveMetaStore extends ThriftHiveMetastore {
}
@Override
+ public List<ExtendedTableInfo> get_tables_ext(final GetTablesExtRequest req) throws MetaException {
+ List<String> tables = new ArrayList<String>();
+ List<ExtendedTableInfo> ret = new ArrayList<ExtendedTableInfo>();
+ String catalog = req.getCatalog();
+ String database = req.getDatabase();
+ String pattern = req.getTableNamePattern();
+ List<String> processorCapabilities = req.getProcessorCapabilities();
+ String processorId = req.getProcessorIdentifier();
+ List<Table> tObjects = new ArrayList<>();
+
+ startTableFunction("get_tables_ext", catalog, database, pattern);
+ Exception ex = null;
+ try {
+ tables = getMS().getTables(catalog, database, pattern, null);
+ LOG.debug("get_tables_ext:getTables() returned " + tables.size());
+ tables = FilterUtils.filterTableNamesIfEnabled(isServerFilterEnabled, filterHook,
+ catalog, database, tables);
+
+ if (tables.size() > 0) {
+ tObjects = getMS().getTableObjectsByName(catalog, database, tables);
+ LOG.debug("get_tables_ext:getTableObjectsByName() returned " + tObjects.size());
+ if (processorCapabilities != null && processorCapabilities.size() > 0) {
+ if (transformer != null) {
+ Map<Table, List<String>> retMap = transformer.transform(tObjects, processorCapabilities, processorId);
+
+ for (Map.Entry<Table, List<String>> entry : retMap.entrySet()) {
+ ret.add(convertTableToExtendedTable(entry.getKey(), entry.getValue(), req.getRequestedFields()));
+ }
+ } else {
+ for (Table table : tObjects) {
+ ret.add(convertTableToExtendedTable(table, processorCapabilities, req.getRequestedFields()));
+ }
+ }
+ }
+ }
+ } catch (MetaException e) {
+ ex = e;
+ throw e;
+ } catch (Exception e) {
+ ex = e;
+ throw newMetaException(e);
+ } finally {
+ endFunction("get_tables_ext", ret != null, ex);
+ }
+ return ret;
+ }
+
+ private ExtendedTableInfo convertTableToExtendedTable (Table table,
+ List<String> processorCapabilities, int mask) {
+ ExtendedTableInfo extTable = new ExtendedTableInfo(table.getTableName());
+ if ((mask & GetTablesExtRequestFields.ACCESS_TYPE.getValue()) == GetTablesExtRequestFields.ACCESS_TYPE.getValue())
+ extTable.setAccessType(table.getAccessType());
+
+ if ((mask & GetTablesExtRequestFields.PROCESSOR_CAPABILITIES.getValue())
+ == GetTablesExtRequestFields.PROCESSOR_CAPABILITIES.getValue())
+ extTable.setProcessorCapabilities(processorCapabilities);
+
+ return extTable;
+ }
+
+ @Override
public GetTableResult get_table_req(GetTableRequest req) throws MetaException,
NoSuchObjectException {
String catName = req.isSetCatName() ? req.getCatName() : getDefaultCatalog(conf);
return new GetTableResult(getTableInternal(catName, req.getDbName(), req.getTblName(),
- req.getCapabilities(), req.getValidWriteIdList(), req.isGetColumnStats()));
+ req.getCapabilities(), req.getValidWriteIdList(), req.isGetColumnStats(),
+ req.getProcessorCapabilities(), req.getProcessorIdentifier()));
}
private Table getTableInternal(String catName, String dbname, String name,
ClientCapabilities capabilities, String writeIdList, boolean getColumnStats)
throws MetaException, NoSuchObjectException {
+ return getTableInternal(catName, dbname, name, capabilities, writeIdList, getColumnStats, null, null);
+ }
+
+ private Table getTableInternal(String catName, String dbname, String name,
+ ClientCapabilities capabilities, String writeIdList, boolean getColumnStats,
+ List<String> processorCapabilities, String processorId)
+ throws MetaException, NoSuchObjectException {
if (isInTest) {
assertClientHasCapability(capabilities, ClientCapability.TEST_CAPABILITY,
"Hive tests", "get_table_req");
@@ -3054,6 +3145,17 @@ public class HiveMetaStore extends ThriftHiveMetastore {
"insert-only tables", "get_table_req");
}
+ if (processorCapabilities != null && processorCapabilities.size() > 0 && transformer != null) {
+ List<Table> tList = new ArrayList<>();
+ tList.add(t);
+ Map<Table, List<String>> ret = transformer.transform(tList, processorCapabilities, processorId);
+ if (ret.size() > 1) {
+ LOG.warn("Unexpected resultset size:" + ret.size());
+ throw new MetaException("Unexpected result from metadata transformer:return list size=" + ret.size());
+ }
+ t = (Table)(ret.keySet().iterator().next());
+ }
+
firePreEvent(new PreReadTableEvent(t, this));
} catch (MetaException | NoSuchObjectException e) {
ex = e;
@@ -3064,7 +3166,6 @@ public class HiveMetaStore extends ThriftHiveMetastore {
return t;
}
-
@Override
public List<TableMeta> get_table_meta(String dbnames, String tblNames, List<String> tblTypes)
throws MetaException, NoSuchObjectException {
@@ -4924,6 +5025,13 @@ public class HiveMetaStore extends ThriftHiveMetastore {
List<Partition> partitions = getMS()
.getPartitionSpecsByFilterAndProjection(table, request.getProjectionSpec(),
request.getFilterSpec());
+ List<String> processorCapabilities = request.getProcessorCapabilities();
+ String processorId = request.getProcessorIdentifier();
+ if (processorCapabilities != null && processorCapabilities.size() > 0) {
+ if (transformer != null) {
+ partitions = transformer.transformPartitions(partitions, processorCapabilities, processorId);
+ }
+ }
List<PartitionSpec> partitionSpecs =
MetaStoreServerUtils.getPartitionspecsGroupedByStorageDescriptor(table, partitions);
response = new GetPartitionsResponse();
@@ -6454,12 +6562,19 @@ public class HiveMetaStore extends ThriftHiveMetastore {
throws TException {
List<Partition> partitions = get_partitions_by_names(gpbnr.getDb_name(),
gpbnr.getTbl_name(), gpbnr.getNames(),
- gpbnr.isSetGet_col_stats() && gpbnr.isGet_col_stats());
+ gpbnr.isSetGet_col_stats() && gpbnr.isGet_col_stats(), gpbnr.getProcessorCapabilities(),
+ gpbnr.getProcessorIdentifier());
return new GetPartitionsByNamesResult(partitions);
}
public List<Partition> get_partitions_by_names(final String dbName, final String tblName,
final List<String> partNames, boolean getColStats) throws TException {
+ return get_partitions_by_names(dbName, tblName, partNames, getColStats, null, null);
+ }
+
+ public List<Partition> get_partitions_by_names(final String dbName, final String tblName,
+ final List<String> partNames, boolean getColStats, List<String> processorCapabilities,
+ String processorId) throws TException {
String[] dbNameParts = parseDbName(dbName, conf);
String parsedCatName = dbNameParts[CAT_NAME];
@@ -6498,6 +6613,11 @@ public class HiveMetaStore extends ThriftHiveMetastore {
}
}
+ if (transformer != null) {
+ if (processorCapabilities != null && processorCapabilities.size() > 0) {
+ ret = transformer.transformPartitions(ret, processorCapabilities, processorId);
+ }
+ }
success = getMS().commitTransaction();
} catch (Exception e) {
ex = e;
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreMetadataTransformer.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreMetadataTransformer.java
new file mode 100644
index 0000000..d33c36e
--- /dev/null
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/IMetaStoreMetadataTransformer.java
@@ -0,0 +1,58 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.hadoop.hive.metastore;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+
+/**
+ * This interface provides a way for users to implement a custom metadata transformer for tables/partitions.
+ * This transformer can match and manipulate the data to be returned to the data processor
+ * The classes implementing this interface should be in the HMS classpath, if this configuration is turned on.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface IMetaStoreMetadataTransformer {
+
+ /**
+ * @param table A Table object to be transformed
+ * @param processorCapabilities A array of String capabilities received from the data processor
+ * @param processorId String ID used for logging purpose.
+ * @return Map A Map of transformed objects keyed by Table and value is list of required capabilities
+ * @throws HiveMetaException
+ */
+ // TODO HiveMetaException or MetaException
+ public Map<Table, List<String>> transform(List<Table> tables, List<String> processorCapabilities, String processorId) throws MetaException;
+
+
+ /**
+ * @param parts A list of Partition objects to be transformed
+ * @param processorCapabilities A array of String capabilities received from the data processor
+ * @param processorId String ID used for logging purpose.
+ * @return Map A Map of transformed objects keyed by Partition and value is list of required capabilities
+ * @throws HiveMetaException
+ */
+ // TODO HiveMetaException or MetaException
+ public List<Partition> transformPartitions(List<Partition> parts, List<String> processorCapabilities, String processorId) throws MetaException;
+}
diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetastoreDefaultTransformer.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetastoreDefaultTransformer.java
new file mode 100644
index 0000000..79e27e4
--- /dev/null
+++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/MetastoreDefaultTransformer.java
@@ -0,0 +1,404 @@
+/*
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.hadoop.hive.metastore;
+
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_NONE;
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_READONLY;
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_READWRITE;
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_WRITEONLY;
+
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MetastoreDefaultTransformer implements IMetaStoreMetadataTransformer {
+ public static final Logger LOG = LoggerFactory.getLogger(MetastoreDefaultTransformer.class);
+ private IHMSHandler hmsHandler = null;
+
+ private static final String CONNECTORREAD = "CONNECTORREAD".intern();
+ private static final String CONNECTORWRITE = "CONNECTORWRITE".intern();
+ private static final String EXTWRITE = "EXTWRITE".intern();
+ private static final String EXTREAD = "EXTREAD".intern();
+ private static final String HIVEACIDWRITE = "HIVEACIDWRITE".intern();
+ private static final String HIVEBUCKET2 = "HIVEBUCKET2".intern();
+ private static final String HIVECACHEINVALIDATE = "HIVECACHEINVALIDATE".intern();
+ private static final String HIVEFULLACIDREAD = "HIVEFULLACIDREAD".intern();
+ private static final String HIVEFULLACIDWRITE = "HIVEFULLACIDWRITE".intern();
+ private static final String HIVEMANAGEDINSERTREAD = "HIVEMANAGEDINSERTREAD".intern();
+ private static final String HIVEMANAGEDINSERTWRITE = "HIVEMANAGEDINSERTWRITE".intern();
+ private static final String HIVEMANAGESTATS = "HIVEMANAGESTATS".intern();
+ private static final String HIVEMQT = "HIVEMQT".intern();
+ private static final String HIVEONLYMQTWRITE = "HIVEONLYMQTWRITE".intern();
+ private static final String HIVESQL = "HIVESQL".intern();
+ private static final String OBJCAPABILITIES = "OBJCAPABILITIES".intern();
+
+ public MetastoreDefaultTransformer(IHMSHandler handler) throws HiveMetaException {
+ this.hmsHandler = handler;
+ }
+
+ @Override
+ public Map<Table, List<String>> transform(List<Table> objects, List<String> processorCapabilities, String processorId) throws MetaException {
+ LOG.info("Starting translation for processor " + processorId + " on list " + objects.size());
+ Map<Table, List<String>> ret = new HashMap<Table, List<String>>();
+
+ for (Table table : objects) {
+ Map<String, String> params = table.getParameters();
+ String tableType = table.getTableType();
+ String tCapabilities = params.get(OBJCAPABILITIES);
+ int numBuckets = table.getSd().getNumBuckets();
+ boolean isBucketed = (numBuckets > 0) ? true : false;
+ List<String> generated = new ArrayList<String>();
+
+ LOG.info("Table " + table.getTableName() + ",#bucket=" + numBuckets + ",isBucketed:" + isBucketed + ",tableType=" + tableType + ",tableCapabilities=" + tCapabilities);
+
+ // if the table has no tCapabilities, then generate default ones
+ if (tCapabilities == null) {
+ LOG.debug("Table has no specific required capabilities");
+
+ switch (tableType) {
+ case "EXTERNAL_TABLE":
+ Table newTable = new Table(table);
+ generated.add(EXTREAD);
+ generated.add(EXTWRITE);
+
+ if (numBuckets > 0) {
+ generated.add(HIVEBUCKET2);
+ if (processorCapabilities.contains(HIVEBUCKET2)) {
+ newTable.setAccessType(ACCESSTYPE_READWRITE);
+ } else {
+ newTable.setAccessType(ACCESSTYPE_READONLY);
+ StorageDescriptor newSd = new StorageDescriptor(table.getSd());
+ newSd.setNumBuckets(-1); // remove bucketing info
+ newTable.setSd(newSd);
+ }
+ } else { // Unbucketed, so RW for all
+ newTable.setAccessType(ACCESSTYPE_READWRITE);
+ }
+
+ ret.put(newTable, generated);
+ break;
+ case "MANAGED_TABLE":
+ String txnal = params.get("transactional");
+ if (txnal == null || txnal.equalsIgnoreCase("FALSE")) { // non-ACID MANAGED table
+ table.setAccessType(ACCESSTYPE_READWRITE);
+ // table does not require any capabilities for full RW
+ }
+
+ if (txnal != null && txnal.equalsIgnoreCase("TRUE")) { // ACID table
+ if (!(processorCapabilities.contains(CONNECTORREAD) ||
+ processorCapabilities.contains(HIVEFULLACIDREAD) ||
+ processorCapabilities.contains(HIVEMANAGEDINSERTREAD))) {
+ table.setAccessType(ACCESSTYPE_NONE); // clients have no access to ACID tables without capabilities
+ generated.add(CONNECTORREAD);
+ generated.add(CONNECTORWRITE);
+ generated.add(HIVECACHEINVALIDATE);
+ generated.add(HIVEMANAGESTATS);
+ }
+
+ String txntype = params.get("transactional_properties");
+ if (txntype != null && txntype.equalsIgnoreCase("insert_only")) { // MICRO_MANAGED Tables
+ generated.add(HIVEMANAGEDINSERTREAD);
+ generated.add(HIVEMANAGEDINSERTWRITE);
+ // MGD table is insert only, not full ACID
+ if (processorCapabilities.contains(HIVEMANAGEDINSERTWRITE) || processorCapabilities.contains(CONNECTORWRITE)) {
+ table.setAccessType(ACCESSTYPE_READWRITE); // clients have RW access to INSERT-ONLY ACID tables
+ } else if (processorCapabilities.contains(HIVEMANAGEDINSERTREAD) || processorCapabilities.contains(CONNECTORREAD)) {
+ table.setAccessType(ACCESSTYPE_READONLY); // clients have RO access to INSERT-ONLY ACID tables
+ } else {
+ table.setAccessType(ACCESSTYPE_NONE); // clients have NO access to INSERT-ONLY ACID tables
+ }
+ } else { // FULL ACID MANAGED TABLE
+ generated.add(HIVEFULLACIDWRITE);
+ generated.add(HIVEFULLACIDREAD);
+ if (processorCapabilities.contains(HIVEFULLACIDWRITE) || processorCapabilities.contains(CONNECTORWRITE)) {
+ table.setAccessType(ACCESSTYPE_READWRITE); // clients have RW access to IUD ACID tables
+ } else if (processorCapabilities.contains(HIVEFULLACIDREAD) || processorCapabilities.contains(CONNECTORREAD)) {
+ table.setAccessType(ACCESSTYPE_READONLY); // clients have RO access to IUD ACID tables
+ } else {
+ table.setAccessType(ACCESSTYPE_NONE); // clients have NO access to IUD ACID tables
+ }
+ }
+ }
+ ret.put(table, generated);
+ break;
+ case "VIRTUAL_VIEW":
+ generated.add("HIVESQL");
+ if (processorCapabilities.contains(HIVESQL) ||
+ processorCapabilities.contains(CONNECTORREAD)) {
+ table.setAccessType(ACCESSTYPE_READONLY);
+ } else {
+ table.setAccessType(ACCESSTYPE_NONE);
+ }
+ ret.put(table, generated);
+ break;
+ case "MATERIALIZED_VIEW":
+ generated.add(CONNECTORREAD);
+ generated.add(HIVEFULLACIDREAD);
+ generated.add(HIVEONLYMQTWRITE);
+ generated.add(HIVEMANAGESTATS);
+ generated.add(HIVEMQT);
+ if ((processorCapabilities.contains(CONNECTORREAD) ||
+ processorCapabilities.contains(HIVEFULLACIDREAD)) && processorCapabilities.contains(HIVEMQT)) {
+ LOG.info("Processor has one of the READ abilities and HIVEMQT, AccessType=RO");
+ table.setAccessType(ACCESSTYPE_READONLY);
+ } else {
+ LOG.info("Processor has no READ abilities or HIVEMQT, AccessType=None");
+ table.setAccessType(ACCESSTYPE_NONE);
+ }
+ ret.put(table, generated);
+ break;
+ default:
+ table.setAccessType(ACCESSTYPE_NONE);
+ break;
+ }
+ ret.put(table,generated);
+ continue;
+ }
+
+ // WITH CAPABLITIES ON TABLE
+ tCapabilities = tCapabilities.replaceAll("\\s","").toUpperCase(); // remove spaces between tCapabilities + toUppercase
+ List<String> requiredCapabilities = Arrays.asList(tCapabilities.split(","));
+ switch (tableType) {
+ case "EXTERNAL_TABLE":
+ if (processorCapabilities.containsAll(requiredCapabilities)) {
+ // AccessType is RW
+ LOG.info("Abilities for match: Table type=" + tableType + ",accesstype is RW");
+ table.setAccessType(ACCESSTYPE_READWRITE);
+ ret.put(table, requiredCapabilities);
+ break;
+ }
+
+ Table newTable = new Table(table);
+ boolean removedBucketing = false;
+
+ if (requiredCapabilities.contains(HIVEBUCKET2) && !processorCapabilities.contains(HIVEBUCKET2)) {
+ StorageDescriptor newSd = new StorageDescriptor(table.getSd());
+ newSd.setNumBuckets(-1); // removing bucketing if HIVEBUCKET2 isnt specified
+ newTable.setSd(newSd);
+ removedBucketing = true;
+ newTable.setAccessType(ACCESSTYPE_READONLY);
+ LOG.info("Removed bucketing information from table");
+ }
+
+ if (requiredCapabilities.contains(EXTWRITE) && processorCapabilities.contains(EXTWRITE)) {
+ if (!removedBucketing) {
+ LOG.info("EXTWRITE Matches, accessType=" + ACCESSTYPE_READWRITE);
+ newTable.setAccessType(ACCESSTYPE_READWRITE);
+ }
+ } else if (requiredCapabilities.contains(EXTREAD) && processorCapabilities.contains(EXTREAD)) {
+ LOG.debug("EXTREAD Matches, accessType=" + ACCESSTYPE_READONLY);
+ newTable.setAccessType(ACCESSTYPE_READONLY);
+ } else {
+ LOG.debug("No matches, accessType=" + ACCESSTYPE_NONE);
+ newTable.setAccessType(ACCESSTYPE_NONE);
+ }
+
+ LOG.info("setting required to " + requiredCapabilities);
+ ret.put(newTable, requiredCapabilities);
+ break;
+ case "MANAGED_TABLE":
+ if (processorCapabilities.containsAll(requiredCapabilities)) {
+ // AccessType is RW
+ LOG.info("Abilities for match: Table type=" + tableType + ",accesstype is RW");
+ table.setAccessType(ACCESSTYPE_READWRITE);
+ ret.put(table, requiredCapabilities);
+ continue;
+ }
+
+ String txnal = params.get("transactional");
+ if (txnal == null || txnal.equalsIgnoreCase("FALSE")) { // non-ACID MANAGED table
+ LOG.info("Table is non ACID: ,accesstype is RW");
+ table.setAccessType(ACCESSTYPE_READWRITE);
+ // table does not require any capabilities for full RW
+ }
+
+ if (txnal != null && txnal.equalsIgnoreCase("TRUE")) { // ACID table
+ String txntype = params.get("transactional_properties");
+ if (txntype != null && txntype.equalsIgnoreCase("insert_only")) { // MICRO_MANAGED Tables
+ LOG.info("Table is INSERTONLY ACID: ,accesstype is RW");
+ // MGD table is insert only, not full ACID
+ if ((processorCapabilities.contains(HIVEMANAGEDINSERTWRITE) && requiredCapabilities.contains(HIVEMANAGEDINSERTWRITE))
+ || (processorCapabilities.contains(CONNECTORWRITE) && requiredCapabilities.contains(CONNECTORWRITE))) {
+ table.setAccessType(ACCESSTYPE_READWRITE); // clients have RW access to INSERT-ONLY ACID tables
+ } else if ((processorCapabilities.contains(HIVEMANAGEDINSERTREAD) && requiredCapabilities.contains(HIVEMANAGEDINSERTREAD))
+ || (processorCapabilities.contains(CONNECTORREAD) && requiredCapabilities.contains(CONNECTORREAD))) {
+ table.setAccessType(ACCESSTYPE_READONLY);
+ } else {
+ table.setAccessType(ACCESSTYPE_NONE);
+ }
+ } else { // MANAGED FULL ACID TABLES
+ LOG.info("Table is FULL ACID:");
+ if ((processorCapabilities.contains(HIVEFULLACIDWRITE) && requiredCapabilities.contains(HIVEFULLACIDWRITE))
+ || (processorCapabilities.contains(CONNECTORWRITE) && requiredCapabilities.contains(CONNECTORWRITE))) {
+ table.setAccessType(ACCESSTYPE_READWRITE); // clients have RW access to IUD ACID tables
+ } else if ((processorCapabilities.contains(HIVEFULLACIDREAD) && requiredCapabilities.contains(HIVEFULLACIDREAD))
+ || (processorCapabilities.contains(CONNECTORREAD) && requiredCapabilities.contains(CONNECTORREAD))) {
+ table.setAccessType(ACCESSTYPE_READONLY); // clients have RO access to IUD ACID tables
+ } else {
+ table.setAccessType(ACCESSTYPE_NONE); // clients have NO access to IUD ACID tables
+ }
+ }
+ }
+ LOG.info("setting required to " + requiredCapabilities + ",MANAGED:Access=" + table.getAccessType());
+ ret.put(table, requiredCapabilities);
+ break;
+ case "VIRTUAL_VIEW":
+ if (processorCapabilities.containsAll(requiredCapabilities)) {
+ table.setAccessType(ACCESSTYPE_READONLY);
+ } else {
+ table.setAccessType(ACCESSTYPE_NONE);
+ }
+ ret.put(table, requiredCapabilities);
+ break;
+ case "MATERIALIZED_VIEW":
+ if (processorCapabilities.containsAll(requiredCapabilities)) {
+ table.setAccessType(ACCESSTYPE_READONLY);
+ } else {
+ table.setAccessType(ACCESSTYPE_NONE);
+ }
+ ret.put(table, requiredCapabilities);
+ break;
+ default:
+ table.setAccessType(ACCESSTYPE_NONE);
+ ret.put(table, requiredCapabilities);
+ break;
+ }
+ }
+
+ LOG.info("Transformer return list of " + objects.size());
+ return ret;
+ }
+
+ @Override
+ public List<Partition> transformPartitions(List<Partition> objects, List<String> processorCapabilities, String processorId) throws MetaException {
+ LOG.info("Starting translation for partition for processor " + processorId + " on list " + objects.size());
+ List<Partition> ret = new ArrayList<>();
+ Map<String, Table> tableCache = new HashMap<>();
+ int tableBuckets = 0;
+ int partBuckets = 0;
+
+ for (Partition partition : objects) {
+ String tableName = partition.getTableName();
+ String dbName = partition.getDbName();
+ Table table = null;
+
+ if (tableCache.containsKey(dbName + "." + tableName)) {
+ table = tableCache.get(dbName + "." + tableName);
+ } else {
+ try {
+ table = hmsHandler.get_table_core(MetaStoreUtils.getDefaultCatalog(null), dbName, tableName);
+ } catch (Exception e) {
+ throw new MetaException("Could not load table " + tableName + ":" + e.getMessage());
+ }
+ tableCache.put(dbName + "." + tableName, table);
+ }
+ Map<String, String> params = table.getParameters();
+ String tableType = table.getTableType();
+ String tCapabilities = params.get(OBJCAPABILITIES);
+ tableBuckets = table.getSd().getNumBuckets();
+ partBuckets = partition.getSd().getNumBuckets();
+ LOG.info("Number of original part buckets=" + partBuckets);
+
+ if (tCapabilities == null) {
+ LOG.debug("Table " + table.getTableName() + " has no specific required capabilities");
+
+ switch (tableType) {
+ case "EXTERNAL_TABLE":
+ if (partBuckets > 0 && !processorCapabilities.contains(HIVEBUCKET2)) {
+ Partition newPartition = new Partition(partition);
+ StorageDescriptor newSd = new StorageDescriptor(partition.getSd());
+ newSd.setNumBuckets(-1); // remove bucketing info
+ newPartition.setSd(newSd);
+ ret.add(newPartition);
+ } else {
+ ret.add(partition);
+ }
+ break;
+ case "MANAGED_TABLE":
+ String txnal = params.get("transactional");
+ if (txnal == null || txnal.equalsIgnoreCase("FALSE")) { // non-ACID MANAGED table
+ if (partBuckets > 0 && !processorCapabilities.contains(HIVEBUCKET2)) {
+ Partition newPartition = new Partition(partition);
+ StorageDescriptor newSd = new StorageDescriptor(partition.getSd());
+ newSd.setNumBuckets(-1); // remove bucketing info
+ newPartition.setSd(newSd);
+ ret.add(newPartition);
+ break;
+ }
+ }
+ // INSERT or FULL ACID table, bucketing info to be retained
+ ret.add(partition);
+ break;
+ default:
+ ret.add(partition);
+ break;
+ }
+ } else { // table has capabilities
+ tCapabilities = tCapabilities.replaceAll("\\s","").toUpperCase(); // remove spaces between tCapabilities + toUppercase
+ List<String> requiredCapabilities = Arrays.asList(tCapabilities.split(","));
+ if (partBuckets <= 0 || processorCapabilities.containsAll(requiredCapabilities)) {
+ ret.add(partition);
+ continue;
+ }
+
+ switch (tableType) {
+ case "EXTERNAL_TABLE":
+ if (requiredCapabilities.contains(HIVEBUCKET2) && !processorCapabilities.contains(HIVEBUCKET2)) {
+ Partition newPartition = new Partition(partition);
+ StorageDescriptor newSd = new StorageDescriptor(partition.getSd());
+ newSd.setNumBuckets(-1); // removing bucketing if HIVEBUCKET2 isnt specified
+ newPartition.setSd(newSd);
+ LOG.info("Removed bucketing information from partition");
+ ret.add(newPartition);
+ break;
+ }
+ case "MANAGED_TABLE":
+ String txnal = params.get("transactional");
+ if (txnal == null || txnal.equalsIgnoreCase("FALSE")) { // non-ACID MANAGED table
+ if (!processorCapabilities.contains(HIVEBUCKET2)) {
+ Partition newPartition = new Partition(partition);
+ StorageDescriptor newSd = new StorageDescriptor(partition.getSd());
+ newSd.setNumBuckets(-1); // remove bucketing info
+ newPartition.setSd(newSd);
+ ret.add(newPartition);
+ break;
+ }
+ }
+ ret.add(partition);
+ break;
+ default:
+ ret.add(partition);
+ break;
+ }
+ }
+ }
+ LOG.info("Returning partition set of size " + ret.size());
+ return ret;
+ }
+}
diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java
index 459c7c2..4878a47 100644
--- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java
+++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/HiveMetaStoreClientPreCatalog.java
@@ -1453,6 +1453,15 @@ public class HiveMetaStoreClientPreCatalog implements IMetaStoreClient, AutoClos
/** {@inheritDoc} */
@Override
+ public List<ExtendedTableInfo> getTablesExt(String catName, String dbName, String tablePattern,
+ int requestedFields, int limit) throws MetaException, TException {
+ GetTablesExtRequest req = new GetTablesExtRequest(catName, dbName, tablePattern, requestedFields);
+ req.setLimit(limit);
+ return client.get_tables_ext(req);
+ }
+
+ /** {@inheritDoc} */
+ @Override
public Materialization getMaterializationInvalidationInfo(CreationMetadata cm, String validTxnList)
throws MetaException, InvalidOperationException, UnknownDBException, TException {
return client.get_materialization_invalidation_info(cm, validTxnList);
diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHiveMetaStore.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHiveMetaStore.java
index 0c4c84c..cbca6c3 100644
--- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHiveMetaStore.java
+++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHiveMetaStore.java
@@ -44,14 +44,25 @@ import java.lang.reflect.*;
import static org.mockito.Mockito.mock;
import com.google.common.collect.Sets;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient.GetTablesRequestBuilder;
+import org.apache.hadoop.hive.metastore.TableType;
+import org.apache.hadoop.hive.metastore.api.ExtendedTableInfo;
import org.apache.hadoop.hive.metastore.api.GetPartitionsFilterSpec;
import org.apache.hadoop.hive.metastore.api.GetPartitionsProjectionSpec;
import org.apache.hadoop.hive.metastore.api.GetPartitionsRequest;
import org.apache.hadoop.hive.metastore.api.GetPartitionsResponse;
+import org.apache.hadoop.hive.metastore.api.GetTableRequest;
+import org.apache.hadoop.hive.metastore.api.GetTablesExtRequest;
+import org.apache.hadoop.hive.metastore.api.GetTablesExtRequestFields;
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_NONE;
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_READONLY;
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_READWRITE;
+import static org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.ACCESSTYPE_WRITEONLY;
import org.apache.hadoop.hive.metastore.api.PartitionSpec;
import org.apache.hadoop.hive.metastore.api.PartitionSpecWithSharedSD;
import org.apache.hadoop.hive.metastore.api.PartitionWithoutSD;
import org.apache.hadoop.hive.metastore.client.builder.DatabaseBuilder;
+import org.apache.hadoop.hive.metastore.client.builder.PartitionBuilder;
import org.apache.hadoop.hive.metastore.client.builder.TableBuilder;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
@@ -121,6 +132,7 @@ public abstract class TestHiveMetaStore {
private static final String TEST_DB1_NAME = "testdb1";
private static final String TEST_DB2_NAME = "testdb2";
+ private static final String CAPABILITIES_KEY = "OBJCAPABILITIES";
private static final int DEFAULT_LIMIT_PARTITION_REQUEST = 100;
@@ -143,6 +155,8 @@ public abstract class TestHiveMetaStore {
MetastoreConf.setLongVar(conf, ConfVars.BATCH_RETRIEVE_MAX, 2);
MetastoreConf.setLongVar(conf, ConfVars.LIMIT_PARTITION_REQUEST, DEFAULT_LIMIT_PARTITION_REQUEST);
MetastoreConf.setVar(conf, ConfVars.STORAGE_SCHEMA_READER_IMPL, "no.such.class");
+
+ MetastoreConf.setVar(conf, ConfVars.METASTORE_METADATA_TRANSFORMER_CLASS, "org.apache.hadoop.hive.metastore.MetastoreDefaultTransformer");
}
@Test
@@ -3351,4 +3365,779 @@ public abstract class TestHiveMetaStore {
m.invoke(hms, tbl, part, false, null);
verify(wh, never()).getFileStatusesForLocation(part.getSd().getLocation());
}
+
+ private void resetHMSClient() {
+ client.setProcessorIdentifier(null);
+ client.setProcessorCapabilities(null);
+ }
+
+ private void setHMSClient(String id, String[] caps) {
+ client.setProcessorIdentifier(id);
+ client.setProcessorCapabilities(caps);
+ }
+
+ /**
+ * EXTERNAL_TABLE
+ * 1) Old table with no capabilities
+ * 2a) New table with capabilities with no client requirements
+ * 2b) New table with capabilities with no matching client requirements
+ * 2c) New table with capabilities with partial match requirements
+ * 2d) New table with capabilities with full match requirements
+ */
+ @Test
+ public void testTransformerExternalTable() throws Exception {
+ try {
+ final String dbName = "db1";
+ String basetblName = "oldstyletable";
+ Map<String, Object> tProps = new HashMap<>();
+ int buckets = 32;
+
+ String tblName = basetblName;
+ tProps.put("DBNAME", dbName);
+ tProps.put("TBLNAME", tblName);
+ tProps.put("TBLTYPE", TableType.EXTERNAL_TABLE);
+ tProps.put("BUCKETS", buckets);
+ StringBuilder properties = new StringBuilder();
+ properties.append("EXTERNAL").append("=").append("TRUE");
+ tProps.put("PROPERTIES", properties.toString());
+ Table tbl = createTableWithCapabilities(tProps);
+
+ // retrieve the table
+ Table tbl2 = client.getTable(dbName, tblName);
+ LOG.info("Table=" + tblName + ",Access=" + tbl2.getAccessType());
+ assertEquals("Created and retrieved tables do not match:" + tbl2.getTableName() + ":" + tblName,
+ tbl2.getTableName(), tblName);
+ assertEquals("TableType mismatch", TableType.EXTERNAL_TABLE.name(), tbl2.getTableType());
+ assertEquals(buckets, tbl2.getSd().getNumBuckets()); // no transformation
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType()); // old client, AccessType not set
+
+ tblName = "test_ext_bucketed_wc";
+ properties = new StringBuilder();
+ properties.append("EXTERNAL").append("=").append("TRUE");
+ properties.append(";");
+ properties.append(CAPABILITIES_KEY).append("=").append("HIVEBUCKET2,EXTREAD,EXTWRITE");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType());
+ assertEquals(buckets, tbl2.getSd().getNumBuckets()); // no tranformation
+
+ setHMSClient("testTranformerExternalTable", (new String[] { "HIVEBUCKET2", "EXTREAD", "EXTWRITE"}));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ assertEquals(buckets, tbl2.getSd().getNumBuckets()); // client has the HIVEBUCKET2 capability, retain bucketing info
+
+ resetHMSClient();
+
+ setHMSClient("testTransformerExternalTableRO", (new String[] { "EXTREAD", "EXTWRITE"}));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ assertEquals(-1, tbl2.getSd().getNumBuckets()); // client has no HIVEBUCKET2 capability, remove bucketing info
+
+ resetHMSClient();
+
+ tblName = "test_ext_unbucketed_wc";
+ properties = new StringBuilder();
+ properties.append(CAPABILITIES_KEY).append("=").append("EXTREAD,EXTWRITE");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tProps.remove("BUCKETS");
+ tbl = createTableWithCapabilities(tProps);
+
+ setHMSClient("TestTransformerExternalUnbucketed", (new String[] {"EXTREAD", "EXTWRITE"}));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+
+ resetHMSClient();
+
+ tblName = "test_ext_sparkbucketed_wc";
+ properties = new StringBuilder();
+ properties.append("EXTERNAL").append("=").append("TRUE");
+ properties.append(";");
+ properties.append(CAPABILITIES_KEY).append("=").append("SPARKDECIMAL,SPARKBUCKET,EXTREAD");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tProps.put("BUCKETS", buckets);
+ tbl = createTableWithCapabilities(tProps);
+
+ setHMSClient("testTransformerExternalTableSpark", (new String[] { "HIVEFULLACIDREAD", "CONNECTORREAD"}));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType()); // requires EXTREAD for RO
+ assertEquals(buckets, tbl2.getSd().getNumBuckets());
+
+ setHMSClient("testTransformerExternalTableSpark", (new String[] { "EXTREAD", "CONNECTORREAD"}));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType()); // requires EXTREAD for RO
+ assertEquals(buckets, tbl2.getSd().getNumBuckets());
+
+ setHMSClient("testTransformerExternalTableSpark", (new String[] { "SPARKBUCKET", "SPARKDECIMAL", "EXTREAD" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ assertEquals(buckets, tbl2.getSd().getNumBuckets()); // client has SPARKBUCKET capability
+
+ resetHMSClient();
+
+ } catch (Exception e) {
+ } finally {
+ resetHMSClient();
+ }
+ }
+
+ @Test
+ public void testTransformerManagedTable() throws Exception {
+ try {
+ resetHMSClient();
+
+ final String dbName = "db1";
+ String basetblName = "oldstylemgdtable";
+ Map<String, Object> tProps = new HashMap<>();
+
+ String tblName = basetblName;
+ tProps.put("DBNAME", dbName);
+ tProps.put("TBLNAME", tblName);
+ tProps.put("TBLTYPE", TableType.MANAGED_TABLE);
+ // create unbucketed managed table with no capabilities
+ Table tbl = createTableWithCapabilities(tProps);
+
+ // retrieve the table
+ Table tbl2 = client.getTable(dbName, tblName);
+ LOG.info("Table=" + tblName + ",Access=" + tbl2.getAccessType());
+ assertEquals("Created and retrieved tables do not match:" + tbl2.getTableName() + ":" + tblName,
+ tbl2.getTableName(), tblName);
+ assertEquals("TableType mismatch", TableType.MANAGED_TABLE.name(), tbl2.getTableType());
+ assertEquals(-1, tbl2.getSd().getNumBuckets());
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType()); // no translation to be done, so accessType not set
+
+ // managed table with no capabilities
+ tblName = "test_mgd_insert_woc";
+ StringBuilder properties = new StringBuilder();
+ tProps.put("TBLNAME", tblName);
+ properties.append("transactional=true");
+ properties.append(";");
+ properties.append("transactional_properties='insert_only'");
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType()); // no transformation
+
+ setHMSClient("testMGDwithConnectorRead", new String[] {"CONNECTORREAD"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwithInsertRead", new String[] {"HIVEMANAGEDINSERTREAD"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwithInsertWrite", new String[] {"HIVEMANAGEDINSERTWRITE"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwithConnectorWrite", new String[] {"CONNECTORWRITE"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ // bucketed table with no capabilities
+ tblName = "test_mgd_insert_wc";
+ properties = new StringBuilder();
+ properties.append(CAPABILITIES_KEY).append("=").append("HIVEMANAGEDINSERTREAD,HIVEMANAGEDINSERTWRITE,HIVECACHEINVALIDATE")
+ .append("HIVEMANAGEDSTATS,CONNECTORREAD,CONNECTORWRITE");
+ properties.append(";");
+ properties.append("transactional_properties='insert_only'");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType());
+
+ setHMSClient("testMGDwithAllWrites", new String[] { "HIVEMANAGEDINSERTWRITE", "HIVECACHEINVALIDATE",
+ "HIVEMANAGEDSTATS" , "CONNECTORWRITE" });
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwith1MissingWrite", new String[] {"HIVEMANAGEDINSERTREAD", "HIVEMANAGEDINSERTWRITE",
+ "HIVECACHEINVALIDATE" });
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwith1Write", new String[] {"CONNECTORWRITE"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwithConnectorRead", new String[] {"CONNECTORREAD"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ resetHMSClient();
+
+ // managed tables with no capabilities
+ tblName = "test_mgd_full_woc";
+ properties = new StringBuilder();
+ tProps.put("TBLNAME", tblName);
+ tProps.put("TBLTYPE", TableType.MANAGED_TABLE);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType());
+
+ setHMSClient("testMGDwithConnectorRead", new String[] {"CONNECTORREAD"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwithFullACIDRead", new String[] {"HIVEFULLACIDREAD"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwith1Write", new String[] {"HIVEFULLACIDWRITE"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwithConnectorWrite", new String[] {"CONNECTORWRITE"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ // bucketed table with no capabilities
+ tblName = "test_mgd_full_wc";
+ properties = new StringBuilder();
+ properties.append(CAPABILITIES_KEY).append("=").append("HIVEFULLACIDREAD,HIVEFULLACIDWRITE,HIVECACHEINVALIDATE")
+ .append("HIVEMANAGEDSTATS,CONNECTORREAD,CONNECTORWRITE");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType());
+
+ setHMSClient("testMGDFULLwithAllWrites", new String[] { "HIVEFULLACIDWRITE", "HIVECACHEINVALIDATE",
+ "HIVEMANAGEDSTATS", "CONNECTORWRITE" });
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDFULLwith1MissingWrite", new String[] { "HIVECACHEINVALIDATE",
+ "HIVEMANAGEDSTATS", "CONNECTORWRITE" });
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDFULLwith1Write", new String[] { "CONNECTORWRITE" });
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwith1MissingWrite", new String[] {"HIVEFULLACIDREAD", "HIVEFULLACIDWRITE",
+ "HIVECACHEINVALIDATE" });
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testMGDwithConnectorRead", new String[] {"CONNECTORREAD"});
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType());
+ resetHMSClient();
+
+ setHMSClient("testTranformerExternalTable", (new String[] { "HIVEBUCKET2", "EXTREAD", "EXTWRITE"}));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READWRITE, tbl2.getAccessType());
+ assertEquals(100, tbl2.getSd().getNumBuckets());
+
+ resetHMSClient();
+
+ } catch (Exception e) {
+ } finally {
+ resetHMSClient();
+ }
+ }
+
+ @Test
+ public void testTransformerVirtualView() throws Exception {
+ try {
+ final String dbName = "db1";
+ String basetblName = "oldstyleview";
+ Map<String, Object> tProps = new HashMap<>();
+
+ String tblName = basetblName;
+ tProps.put("DBNAME", dbName);
+ tProps.put("TBLNAME", tblName);
+ tProps.put("TBLTYPE", TableType.VIRTUAL_VIEW);
+ StringBuilder properties = new StringBuilder();
+ tProps.put("PROPERTIES", properties.toString());
+ Table tbl = createTableWithCapabilities(tProps);
+
+ // retrieve the table
+ Table tbl2 = client.getTable(dbName, tblName);
+ LOG.info("View=" + tblName + ",Access=" + tbl2.getAccessType());
+ assertEquals("Created and retrieved views do not match:" + tbl2.getTableName() + ":" + tblName,
+ tbl2.getTableName(), tblName);
+ assertEquals("TableType mismatch", TableType.VIRTUAL_VIEW.name(), tbl2.getTableType());
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType()); // old client, AccessType not set
+
+ // table has capabilities
+ tblName = "test_view_wc";
+ properties = new StringBuilder();
+ properties.append(CAPABILITIES_KEY).append("=").append("HIVESQL");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType()); // old client, no transformation
+
+ setHMSClient("testTransformerVirtualView", (new String[] { "HIVESQL" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType()); // RO accessonly for all views
+
+ setHMSClient("testTransformerVirtualView", (new String[] { "EXTWRITE", "HIVEFULLACIDWRITE" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType()); // missing HIVESQL
+
+ resetHMSClient();
+
+ // table does not capabilities but client is newer
+ tblName = "test_view_woc";
+ properties = new StringBuilder();
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ setHMSClient("testTransformerVirtualView", (new String[] { "HIVESQL" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType()); // RO accessonly for all views
+
+ setHMSClient("testTransformerVirtualView", (new String[] { "EXTWRITE", "HIVEFULLACIDWRITE" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType()); // missing HIVESQL
+ } catch (Exception e) {
+ } finally {
+ resetHMSClient();
+ }
+ }
+
+ @Test
+ public void testTransformerMaterializedView() throws Exception {
+ try {
+ resetHMSClient();
+
+ final String dbName = "db1";
+ String basetblName = "oldstylemqtview";
+ Map<String, Object> tProps = new HashMap<>();
+
+ String tblName = basetblName;
+ tProps.put("DBNAME", dbName);
+ tProps.put("TBLNAME", tblName);
+ tProps.put("TBLTYPE", TableType.MATERIALIZED_VIEW);
+ StringBuilder properties = new StringBuilder();
+ tProps.put("PROPERTIES", properties.toString());
+ Table tbl = createTableWithCapabilities(tProps);
+
+ // retrieve the table
+ Table tbl2 = client.getTable(dbName, tblName);
+ LOG.info("View=" + tblName + ",Access=" + tbl2.getAccessType());
+ assertEquals("Created and retrieved views do not match:" + tbl2.getTableName() + ":" + tblName,
+ tbl2.getTableName(), tblName);
+ assertEquals("TableType mismatch", TableType.MATERIALIZED_VIEW.name(), tbl2.getTableType());
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType()); // old client, AccessType not set
+
+ // table has capabilities
+ tblName = "test_mqtview_wc";
+ properties = new StringBuilder();
+ properties.append(CAPABILITIES_KEY).append("=").append("HIVEFULLACIDREAD,HIVEONLYMQTWRITE,HIVEMANAGESTATS,HIVEMQT,CONNECTORREAD");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match expected value:" + tblName,
+ 0, tbl2.getAccessType()); // View has capabilities, processor doesnt, no tranformation
+
+ setHMSClient("testTransformerMQTFullSet", (new String[] { "HIVEFULLACIDREAD", "HIVEONLYMQTWRITE",
+ "HIVEMANAGESTATS", "HIVEMQT", "CONNECTORREAD" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType()); // RO accessonly for all views
+
+ setHMSClient("testTransformerMQTFullRead", (new String[] { "HIVEFULLACIDREAD", "HIVEMQT" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType());
+
+ setHMSClient("testTransformerMQTConnRead", (new String[] { "CONNECTORREAD", "HIVEMQT" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType()); // RO accessonly for all views
+
+ setHMSClient("testTransformerMQTDummySet", (new String[] { "EXTWRITE", "HIVEFULLACIDWRITE" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType()); // missing HIVEMQT + *READ
+
+ resetHMSClient();
+
+ // table does not capabilities but client is newer
+ tblName = "test_mqtview_woc";
+ properties = new StringBuilder();
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ tbl = createTableWithCapabilities(tProps);
+
+ setHMSClient("testTransformerVirtualView", (new String[] { "HIVEFULLACIDREAD", "HIVEMQT" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType()); // RO accessonly for all views
+
+ setHMSClient("testTransformerMQTConnRead", (new String[] { "CONNECTORREAD", "HIVEMQT" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_READONLY, tbl2.getAccessType()); // RO accessonly for all views
+
+ setHMSClient("testTransformerVirtualView", (new String[] { "EXTWRITE, HIVEFULLACIDWRITE" }));
+ tbl2 = client.getTable(dbName, tblName);
+ assertEquals("Table access type does not match the expected value:" + tblName,
+ ACCESSTYPE_NONE, tbl2.getAccessType()); // missing HIVEMQT + *READ
+ } catch (Exception e) {
+ } finally {
+ resetHMSClient();
+ }
+ }
+
+ @Test
+ public void testGetTablesExt() throws Exception {
+ try {
+ resetHMSClient();
+
+ final String dbName = "db1";
+ String tblName = "test_get_tables_ext";
+ int count = 10;
+ Map<String, Object> tProps = new HashMap<>();
+
+ tProps.put("DBNAME", dbName);
+ tProps.put("TBLNAME", tblName);
+
+ StringBuilder table_params = new StringBuilder();
+ List<String> capabilities = new ArrayList<>();
+ TableType type = TableType.EXTERNAL_TABLE;
+
+ capabilities.add("HIVEFULLACIDWRITE");
+ capabilities.add("HIVEFULLACIDREAD");
+ capabilities.add("HIVECACHEINVALIDATE");
+ capabilities.add("CONNECTORREAD");
+ capabilities.add("CONNECTORWRITE");
+
+ tProps.put("CAPABILITIES", capabilities);
+ tProps.put("TABLECOUNT", count);
+ tProps.put("TBLTYPE", type);
+
+ setHMSClient("test_get_tables_ext", (String[])(capabilities.toArray(new String[0])));
+
+ List<String> tables = createTables(tProps);
+ int requestedFields = (new GetTablesRequestBuilder().with(GetTablesExtRequestFields.PROCESSOR_CAPABILITIES)).bitValue();
+ List<ExtendedTableInfo> extTables = client.getTablesExt(null, dbName, "*", requestedFields, 2000);
+ LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
+ assertEquals("Return list size does not match expected size:extTables", count, extTables.size());
+ for (ExtendedTableInfo tableInfo : extTables) {
+ assertEquals("Capability set size does not match", capabilities.size(), tableInfo.getProcessorCapabilities().size());
+ assertEquals("AccessType not expected to be set", 0, tableInfo.getAccessType());
+ }
+
+ requestedFields = (new GetTablesRequestBuilder().with(GetTablesExtRequestFields.ACCESS_TYPE)).bitValue();
+ extTables = client.getTablesExt(null, dbName, "*", requestedFields, 2000);
+ LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
+ assertEquals("Return list size does not match expected size", count, extTables.size());
+ for (ExtendedTableInfo tableInfo : extTables) {
+ assertNull("Return value should not contain capabilities", tableInfo.getProcessorCapabilities());
+ assertTrue("AccessType expected to be set", tableInfo.getAccessType() > 0);
+ }
+
+ requestedFields = (new GetTablesRequestBuilder().with(GetTablesExtRequestFields.ALL)).bitValue();
+ extTables = client.getTablesExt(null, dbName, "*", requestedFields, 2000);
+ LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
+ assertEquals("Return list size does not match expected size", count, extTables.size());
+ for (ExtendedTableInfo tableInfo : extTables) {
+ assertEquals("Capability set size does not match", capabilities.size(), tableInfo.getProcessorCapabilities().size());
+ assertTrue("AccessType expected to be set", tableInfo.getAccessType() > 0);
+ }
+ } catch (Exception e) {
+ System.err.println(StringUtils.stringifyException(e));
+ System.err.println("testGetTablesExt() failed.");
+ throw e;
+ } finally {
+ resetHMSClient();
+ }
+ }
+
+ @Test
+ public void testGetPartitionsByNames() throws Exception {
+ try {
+ resetHMSClient();
+
+ final String dbName = "db1";
+ String tblName = "test_get_parts_by_names";
+ int count = 1;
+ int pCount = 10;
+ int bucketCount = 100;
+ Map<String, Object> tProps = new HashMap<>();
+
+ tProps.put("DBNAME", dbName);
+ tProps.put("TBLNAME", tblName);
+
+ List<String> capabilities = new ArrayList<>();
+ TableType type = TableType.MANAGED_TABLE;
+
+ capabilities.add("HIVEFULLACIDWRITE");
+ capabilities.add("HIVEFULLACIDREAD");
+ capabilities.add("HIVECACHEINVALIDATE");
+ capabilities.add("CONNECTORREAD");
+ capabilities.add("CONNECTORWRITE");
+
+ // tProps.put("CAPABILITIES", capabilities);
+ tProps.put("TBLTYPE", type);
+ tProps.put("PARTITIONS", pCount);
+
+ setHMSClient("TestGetPartitionByNames#1", (String[])(capabilities.toArray(new String[0])));
+
+ Table table = createTableWithCapabilities(tProps);
+
+ List<String> partNames = new ArrayList<>();
+ List<String> partValues = new ArrayList<>();
+ for (int i = 1; i <= pCount; i++) {
+ partValues.add("partcol=" + i);
+ }
+ List<Partition> parts = client.getPartitionsByNames(dbName, tblName, partValues, false);
+ assertEquals("Return list size does not match expected size", pCount, parts.size());
+
+ tblName = "test_gp_ext_bucketed_wc";
+ StringBuilder properties = new StringBuilder();
+ properties.append("EXTERNAL").append("=").append("TRUE");
+ properties.append(";");
+ properties.append(CAPABILITIES_KEY).append("=").append("HIVEBUCKET2,EXTREAD,EXTWRITE");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("TBLTYPE", TableType.EXTERNAL_TABLE);
+ tProps.put("BUCKETS", bucketCount);
+ tProps.put("PROPERTIES", properties.toString());
+ table = createTableWithCapabilities(tProps);
+
+ parts = client.getPartitionsByNames(dbName, tblName, partValues, false);
+ LOG.debug("Return list size=" + parts.size());
+
+ for (Partition part : parts) {
+ assertEquals("Partition bucket count does not match", -1, part.getSd().getNumBuckets());
+ }
+
+ // processor has capabilities
+ capabilities.clear();
+ capabilities.add("HIVEBUCKET2");
+ setHMSClient("TestGetPartitionByNames#2", (String[])(capabilities.toArray(new String[0])));
+ parts = client.getPartitionsByNames(dbName, tblName, partValues, false);
+
+ for (Partition part : parts) {
+ assertEquals("Partition bucket count does not match", -1, part.getSd().getNumBuckets());
+ }
+
+ tblName = "test_parts_mgd_insert_wc";
+ properties = new StringBuilder();
+ properties.append(CAPABILITIES_KEY).append("=").append("HIVEMANAGEDINSERTREAD,HIVEMANAGEDINSERTWRITE,HIVECACHEINVALIDATE")
+ .append("HIVEMANAGEDSTATS,CONNECTORREAD,CONNECTORWRITE");
+ properties.append(";");
+ properties.append("transactional_properties='insert_only'");
+ tProps.put("TBLNAME", tblName);
+ tProps.put("PROPERTIES", properties.toString());
+ table = createTableWithCapabilities(tProps);
+
+ capabilities.clear();
+ capabilities.add("CONNECTORREAD");
+ setHMSClient("TestGetPartitionByNames#3", (String[])(capabilities.toArray(new String[0])));
+ parts = client.getPartitionsByNames(dbName, tblName, partValues, false);
+ assertEquals("Partition count does not match", pCount, parts.size());
+
+ } catch (Exception e) {
+ System.err.println(StringUtils.stringifyException(e));
+ System.err.println("testGetPartitionsByNames() failed.");
+ throw e;
+ } finally {
+ resetHMSClient();
+ }
+ }
+
+ private List<String> createTables(Map<String, Object> props) throws Exception {
+ int count = ((Integer)props.get("TABLECOUNT")).intValue();
+ String tblName = (String)props.get("TBLNAME");
+ List<String> caps = (List<String>)props.get("CAPABILITIES");
+ StringBuilder table_params = new StringBuilder();
+ table_params.append(CAPABILITIES_KEY).append("=").append(String.join(",", caps));
+ props.put("PROPERTIES", table_params.toString());
+ List<String> ret = new ArrayList<>();
+ String newtblName = null;
+
+ try {
+ for (int i = 0; i < count; i++) {
+ newtblName = tblName + i;
+ props.put("TBLNAME", newtblName);
+ props.put("DROPDB", Boolean.FALSE);
+ createTableWithCapabilities(props);
+ ret.add(newtblName);
+ }
+ } catch (Exception e) {
+ LOG.warn("Create table failed for " + newtblName);
+ }
+
+ return ret;
+ }
+
+ private Table createTableWithCapabilities(Map<String, Object> props) throws Exception {
+ String catalog = (String)props.getOrDefault("CATALOG", "testcat");
+ String dbName = (String)props.getOrDefault("DBNAME", "simpdb");
+ String tblName = (String)props.getOrDefault("TBLNAME", "test_table");
+ TableType type = (TableType)props.getOrDefault("TBLTYPE", TableType.MANAGED_TABLE);
+ int buckets = ((Integer)props.getOrDefault("BUCKETS", -1)).intValue();
+ String properties = (String)props.getOrDefault("PROPERTIES", "");
+ boolean dropDb = ((Boolean)props.getOrDefault("DROPDB", Boolean.TRUE)).booleanValue();
+ int partitionCount = ((Integer)props.getOrDefault("PARTITIONS", 0)).intValue();
+
+ final String typeName = "Person";
+
+ Map<String,String> table_params = new HashMap();
+ if (properties.length() > 0) {
+ String[] propArray = properties.split(";");
+ for (String prop : propArray) {
+ String[] keyValue = prop.split("=");
+ table_params.put(keyValue[0], keyValue[1]);
+ }
+ }
+
+ try {
+ client.dropTable(dbName, tblName);
+ } catch (Exception e) {
+ LOG.info("Drop table failed for " + dbName + "." + tblName);
+ }
+
+ try {
+ if (dropDb)
+ silentDropDatabase(dbName);
+ } catch (Exception e) {
+ LOG.info("Drop database failed for " + dbName);
+ }
+
+ if (dropDb)
+ new DatabaseBuilder()
+ .setName(dbName)
+ .create(client, conf);
+
+ try {
+ client.dropType(typeName);
+ } catch (Exception e) {
+ LOG.info("Drop type failed for " + typeName);
+ }
+
+ Type typ1 = new Type();
+ typ1.setName(typeName);
+ typ1.setFields(new ArrayList<>(2));
+ typ1.getFields().add(
+ new FieldSchema("name", ColumnType.STRING_TYPE_NAME, ""));
+ typ1.getFields().add(
+ new FieldSchema("income", ColumnType.INT_TYPE_NAME, ""));
+ client.createType(typ1);
+
+ TableBuilder builder = new TableBuilder()
+ .setDbName(dbName)
+ .setTableName(tblName)
+ .setCols(typ1.getFields())
+ .setType(type.name())
+ .setNumBuckets(buckets)
+ .setTableParams(table_params)
+ .addBucketCol("name")
+ .addStorageDescriptorParam("test_param_1", "Use this for comments etc");
+
+ if (buckets > 0)
+ builder.setNumBuckets(buckets).addBucketCol("name");
+
+ if (partitionCount > 0) {
+ builder.addPartCol("partcol", "string");
+ }
+
+ Table tbl = builder.create(client, conf);
+ LOG.info("Table " + tblName + " created:type=" + type.name());
+
+ if (partitionCount > 0) {
+ List<Partition> partitions = new ArrayList<>();
+
+ List<List<String>> partValues = new ArrayList<>();
+ for (int i = 1; i <= partitionCount; i++) {
+ partValues.add(Lists.newArrayList("" + i));
+ }
+
+ for(List<String> vals : partValues){
+ addPartition(client, tbl, vals);
+ }
+ }
+
+ if (isThriftClient) {
+ // the createTable() above does not update the location in the 'tbl'
+ // object when the client is a thrift client and the code below relies
+ // on the location being present in the 'tbl' object - so get the table
+ // from the metastore
+ tbl = client.getTable(dbName, tblName);
+ }
+ return tbl;
+ }
+
+ private void addPartition(IMetaStoreClient client, Table table, List<String> values)
+ throws TException {
+ PartitionBuilder partitionBuilder = new PartitionBuilder().inTable(table);
+ values.forEach(val -> partitionBuilder.addValue(val));
+ Partition p = partitionBuilder.build(conf);
+ p.getSd().setNumBuckets(-1); // PartitionBuilder uses 0 as default whereas we use -1 for Tables.
+ client.add_partition(p);
+ }
}