This is an automated email from the ASF dual-hosted git repository.
dengzhhu653 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 82edd2a2ef0 HIVE-29694: Reuse DB column descriptors when adding
partitions in HMS (#6577)
82edd2a2ef0 is described below
commit 82edd2a2ef04c7d75431f285f01c85a535ed07c5
Author: InvisibleProgrammer <[email protected]>
AuthorDate: Fri Jul 10 04:45:22 2026 +0200
HIVE-29694: Reuse DB column descriptors when adding partitions in HMS
(#6577)
Co-authored-by: Stamatis Zampetakis <[email protected]>
Co-authored-by: zdeng <[email protected]>
---
.../hadoop/hive/metastore/conf/MetastoreConf.java | 9 +-
.../metastore/directsql/MetaStoreDirectSql.java | 247 ++++++++++++++++-----
.../directsql/MetastoreDirectSqlUtils.java | 6 +
.../metastore/metastore/impl/TableStoreImpl.java | 80 +++++--
.../metastore/TestHMSColumnDescriptorReuse.java | 220 ++++++++++++++++++
5 files changed, 486 insertions(+), 76 deletions(-)
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 e955d1e07d6..0a571587eba 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
@@ -312,9 +312,6 @@ public enum ConfVars {
ACID_HOUSEKEEPER_SERVICE_INTERVAL("metastore.acid.housekeeper.interval",
"hive.metastore.acid.housekeeper.interval", 60, TimeUnit.SECONDS,
"Time interval describing how often the acid housekeeper runs."),
-
COMPACTION_HOUSEKEEPER_SERVICE_INTERVAL("metastore.compaction.housekeeper.interval",
- "hive.metastore.compaction.housekeeper.interval", 300,
TimeUnit.SECONDS,
- "Time interval describing how often the acid compaction housekeeper
runs."),
ACID_TXN_CLEANER_INTERVAL("metastore.acid.txn.cleaner.interval",
"hive.metastore.acid.txn.cleaner.interval", 10, TimeUnit.SECONDS,
"Time interval describing how often aborted and committed txns are
cleaned."),
@@ -432,6 +429,9 @@ public enum ConfVars {
TimeUnit.SECONDS, "MetaStore Client socket timeout in seconds"),
CLIENT_CONNECTION_TIMEOUT("metastore.client.connection.timeout",
"hive.metastore.client.connection.timeout", 600,
TimeUnit.SECONDS, "MetaStore Client connection timeout in
seconds"),
+
COMPACTION_HOUSEKEEPER_SERVICE_INTERVAL("metastore.compaction.housekeeper.interval",
+ "hive.metastore.compaction.housekeeper.interval", 300,
TimeUnit.SECONDS,
+ "Time interval describing how often the acid compaction housekeeper
runs."),
COMPACTOR_HISTORY_RETENTION_DID_NOT_INITIATE("metastore.compactor.history.retention.did.not.initiate",
"hive.compactor.history.retention.did.not.initiate", 2,
new RangeValidator(0, 100), "Determines how many compaction records in
state " +
@@ -1373,6 +1373,9 @@ public enum ConfVars {
"hive.metastore.partition.order.expr", "\"PART_NAME\" asc",
"The default partition order if the metastore does not return all
partitions. \n" +
"It can be sorted based on any column in the PARTITIONS table
(e.g., \"PARTITIONS\".\"CREATE_TIME\" desc, \"PARTITIONS\".\"LAST_ACCESS_TIME\"
desc etc)"),
+
PARTITION_REUSE_COLUMN_DESCRIPTORS("metastore.partition.reuse.column.descriptors",
+ "hive.metastore.partition.reuse.column.descriptors", false,
+ "Add partition reuse existing column descriptors to avoid metadata
bloat on schema evolution."),
PART_INHERIT_TBL_PROPS("metastore.partition.inherit.table.properties",
"hive.metastore.partition.inherit.table.properties", "",
"List of comma separated keys occurring in table properties which will
get inherited to newly created partitions. \n" +
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java
index 713f3ff4e33..7c95577e261 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java
@@ -29,6 +29,8 @@
import static org.apache.hadoop.hive.metastore.ColumnType.TIMESTAMP_TYPE_NAME;
import static org.apache.hadoop.hive.metastore.ColumnType.TINYINT_TYPE_NAME;
import static org.apache.hadoop.hive.metastore.ColumnType.VARCHAR_TYPE_NAME;
+import static
org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.extractSqlClob;
+import static
org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.extractSqlString;
import static
org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.getFullyQualifiedName;
import static
org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.makeParams;
import static
org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.prepareParams;
@@ -107,6 +109,7 @@
import org.apache.hadoop.hive.metastore.client.builder.GetPartitionsArgs;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
+import org.apache.hadoop.hive.metastore.model.MColumnDescriptor;
import org.apache.hadoop.hive.metastore.model.MConstraint;
import org.apache.hadoop.hive.metastore.model.MCreationMetadata;
import org.apache.hadoop.hive.metastore.model.MDatabase;
@@ -432,28 +435,27 @@ public Database getDatabase(String catName, String
dbName) throws MetaException{
queryDbParams.getInnerQuery(), params, queryTextDbParams));
if (!sqlResult2.isEmpty()) {
for (Object[] line : sqlResult2) {
- dbParams.put(MetastoreDirectSqlUtils.extractSqlString(line[0]),
MetastoreDirectSqlUtils
- .extractSqlString(line[1]));
+ dbParams.put(extractSqlString(line[0]), extractSqlString(line[1]));
}
}
Database db = new Database();
- db.setName(MetastoreDirectSqlUtils.extractSqlString(dbline[1]));
- db.setLocationUri(MetastoreDirectSqlUtils.extractSqlString(dbline[2]));
- db.setDescription(MetastoreDirectSqlUtils.extractSqlString(dbline[3]));
- db.setOwnerName(MetastoreDirectSqlUtils.extractSqlString(dbline[4]));
- String type = MetastoreDirectSqlUtils.extractSqlString(dbline[5]);
+ db.setName(extractSqlString(dbline[1]));
+ db.setLocationUri(extractSqlString(dbline[2]));
+ db.setDescription(extractSqlString(dbline[3]));
+ db.setOwnerName(extractSqlString(dbline[4]));
+ String type = extractSqlString(dbline[5]);
db.setOwnerType(
(null == type || type.trim().isEmpty()) ? null :
PrincipalType.valueOf(type));
- db.setCatalogName(MetastoreDirectSqlUtils.extractSqlString(dbline[6]));
+ db.setCatalogName(extractSqlString(dbline[6]));
if (dbline[7] != null) {
db.setCreateTime(MetastoreDirectSqlUtils.extractSqlInt(dbline[7]));
}
-
db.setManagedLocationUri(MetastoreDirectSqlUtils.extractSqlString(dbline[8]));
- String dbType = MetastoreDirectSqlUtils.extractSqlString(dbline[9]);
+ db.setManagedLocationUri(extractSqlString(dbline[8]));
+ String dbType = extractSqlString(dbline[9]);
if (dbType != null &&
dbType.equalsIgnoreCase(DatabaseType.REMOTE.name())) {
db.setType(DatabaseType.REMOTE);
-
db.setConnector_name(MetastoreDirectSqlUtils.extractSqlString(dbline[10]));
-
db.setRemote_dbname(MetastoreDirectSqlUtils.extractSqlString(dbline[11]));
+ db.setConnector_name(extractSqlString(dbline[10]));
+ db.setRemote_dbname(extractSqlString(dbline[11]));
} else {
db.setType(DatabaseType.NATIVE);
}
@@ -534,6 +536,132 @@ public void addPartitions(List<MPartition> parts,
List<List<MPartitionPrivilege>
directSqlInsertPart.addPartitions(parts, partPrivilegesList,
partColPrivilegesList);
}
+ /**
+ * Gets a suitable column descriptor for an existing table, based on the
latest partitions
+ *
+ * @param cols column list of a partition
+ * @param tblId table id
+ * @return an existing column descriptor which columns matches with @cols.
Null if there is no match.
+ * @throws MetaException
+ */
+ public MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, long
tblId)
+ throws MetaException {
+ if (cols == null || cols.isEmpty()) {
+ return null;
+ }
+
+ // Please note! In case you modify any of those methods,
+ // run TestHMSColumnDescriptorReuse with all the supported databases
+ List<Long> cdCandidates =
filterCandidatesByColumnCount(findTheLatestColumnDescriptors(tblId),
cols.size());
+ if (cdCandidates.isEmpty()) {
+ return null;
+ }
+
+ Long matchedColumnDescriptorId =
matchColumnDescriptorWithActualColumns(cdCandidates, cols);
+ if (matchedColumnDescriptorId != null) {
+ return pm.getObjectById(MColumnDescriptor.class,
matchedColumnDescriptorId);
+ }
+ return null;
+ }
+
+ private List<Long> findTheLatestColumnDescriptors(long tblId) throws
MetaException {
+ final int limit = 20;
+ String findLatestDescriptorsSql =
+ """
+ SELECT s."CD_ID"
+ FROM "PARTITIONS" p
+ JOIN "SDS" s ON s."SD_ID" = p."SD_ID"
+ WHERE p."TBL_ID" = ?
+ GROUP BY s."CD_ID"
+ ORDER BY MAX(p."PART_ID") DESC
+ """;
+
+ try (QueryWrapper query = new
QueryWrapper(pm.newQuery("javax.jdo.query.SQL", findLatestDescriptorsSql))) {
+
+ List<Object> sqlResult = executeWithArray(
+ query.getInnerQuery(), new Object[]{tblId},
findLatestDescriptorsSql, limit);
+ if (sqlResult == null || sqlResult.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<Long> latestColumnDescriptorIds = new ArrayList<>();
+ for (Object cdId : sqlResult) {
+
latestColumnDescriptorIds.add(MetastoreDirectSqlUtils.extractSqlLong(cdId));
+ }
+ return latestColumnDescriptorIds;
+ }
+ }
+
+ private List<Long> filterCandidatesByColumnCount(List<Long> cdCandidates,
int size) throws MetaException {
+ if (cdCandidates == null || cdCandidates.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ String placeholders = cdCandidates.stream()
+ .map(c -> "?")
+ .collect(Collectors.joining(", "));
+ String candidatesWithProperColumnCountSql = String.format(
+ """
+ SELECT c."CD_ID"
+ FROM "COLUMNS_V2" c
+ WHERE c."CD_ID" IN (%s)
+ GROUP BY c."CD_ID"
+ HAVING COUNT(*) = ?
+ """, placeholders);
+
+ try (QueryWrapper query =
+ new QueryWrapper(pm.newQuery("javax.jdo.query.SQL",
candidatesWithProperColumnCountSql))) {
+
+ Object[] params = new Object[cdCandidates.size() + 1];
+ for (int i = 0; i < cdCandidates.size(); i++) {
+ params[i] = cdCandidates.get(i);
+ }
+ params[cdCandidates.size()] = size;
+
+ List<Object> result = executeWithArray(query.getInnerQuery(), params,
candidatesWithProperColumnCountSql);
+ if (result == null || result.isEmpty()) {
+ return Collections.emptyList();
+ }
+ List<Long> candidateIds = new ArrayList<>();
+ for (Object cdId : result) {
+ candidateIds.add(MetastoreDirectSqlUtils.extractSqlLong(cdId));
+ }
+ return candidateIds;
+ }
+ }
+
+ private Long matchColumnDescriptorWithActualColumns(List<Long> cdCandidates,
List<FieldSchema> cols)
+ throws MetaException {
+ if (cdCandidates == null || cdCandidates.isEmpty()) {
+ return null;
+ }
+
+ String findColumnSql = "SELECT \"COLUMN_NAME\", \"TYPE_NAME\", \"COMMENT\"
FROM \"COLUMNS_V2\" "
+ + "WHERE \"CD_ID\" = ? ORDER BY \"INTEGER_IDX\"";
+ for (Long candidate: cdCandidates) {
+ try (QueryWrapper query = new
QueryWrapper(pm.newQuery("javax.jdo.query.SQL", findColumnSql))) {
+ List<Object[]> rows = executeWithArray(query.getInnerQuery(), new
Object[]{candidate}, findColumnSql);
+ if (rows != null && rows.size() == cols.size()) {
+ int i = 0;
+ for (; i < cols.size(); i++) {
+ Object[] row = rows.get(i);
+ FieldSchema col = new FieldSchema(
+ extractSqlString(row[0]),
+ extractSqlClob(row[1]),
+ extractSqlString(row[2]));
+ if (!cols.get(i).equals(col)) {
+ break;
+ }
+ }
+ if (i == cols.size()) {
+ return candidate;
+ }
+ }
+ }
+ }
+ return null;
+ }
+
/**
* Alter partitions in batch using direct SQL
* @param table the target table
@@ -1685,17 +1813,17 @@ public List<HiveObjectPrivilege>
getTableAllColumnGrants(String catName, String
// If there is some result convert it into HivePrivilege bag and return.
for (Object[] privLine : queryResult) {
- String privAuthorizer =
MetastoreDirectSqlUtils.extractSqlString(privLine[authorizerIndex]);
- String principalName =
MetastoreDirectSqlUtils.extractSqlString(privLine[principalNameIndex]);
+ String privAuthorizer = extractSqlString(privLine[authorizerIndex]);
+ String principalName = extractSqlString(privLine[principalNameIndex]);
PrincipalType ptype = PrincipalType.valueOf(
-
MetastoreDirectSqlUtils.extractSqlString(privLine[principalTypeIndex]));
- String columnName =
MetastoreDirectSqlUtils.extractSqlString(privLine[columnNameIndex]);
- String privilege =
MetastoreDirectSqlUtils.extractSqlString(privLine[privilegeIndex]);
+ extractSqlString(privLine[principalTypeIndex]));
+ String columnName = extractSqlString(privLine[columnNameIndex]);
+ String privilege = extractSqlString(privLine[privilegeIndex]);
int createTime =
MetastoreDirectSqlUtils.extractSqlInt(privLine[createTimeIndex]);
- String grantor =
MetastoreDirectSqlUtils.extractSqlString(privLine[grantorIndex]);
+ String grantor = extractSqlString(privLine[grantorIndex]);
PrincipalType grantorType =
PrincipalType.valueOf(
-
MetastoreDirectSqlUtils.extractSqlString(privLine[grantorTypeIndex]));
+ extractSqlString(privLine[grantorTypeIndex]));
boolean grantOption =
MetastoreDirectSqlUtils.extractSqlBoolean(privLine[grantOptionIndex]);
HiveObjectRef objectRef = new HiveObjectRef(HiveObjectType.COLUMN,
dbName, tableName, null,
@@ -1927,17 +2055,17 @@ public List<SQLForeignKey> getForeignKeys(String
catName, String parent_db_name,
boolean validate = (enableValidateRely & 2) != 0;
boolean rely = (enableValidateRely & 1) != 0;
SQLForeignKey currKey = new SQLForeignKey(
- MetastoreDirectSqlUtils.extractSqlString(line[0]),
- MetastoreDirectSqlUtils.extractSqlString(line[1]),
- MetastoreDirectSqlUtils.extractSqlString(line[2]),
- MetastoreDirectSqlUtils.extractSqlString(line[3]),
- MetastoreDirectSqlUtils.extractSqlString(line[4]),
- MetastoreDirectSqlUtils.extractSqlString(line[5]),
+ extractSqlString(line[0]),
+ extractSqlString(line[1]),
+ extractSqlString(line[2]),
+ extractSqlString(line[3]),
+ extractSqlString(line[4]),
+ extractSqlString(line[5]),
MetastoreDirectSqlUtils.extractSqlInt(line[6]),
MetastoreDirectSqlUtils.extractSqlInt(line[7]),
MetastoreDirectSqlUtils.extractSqlInt(line[8]),
- MetastoreDirectSqlUtils.extractSqlString(line[9]),
- MetastoreDirectSqlUtils.extractSqlString(line[10]),
+ extractSqlString(line[9]),
+ extractSqlString(line[10]),
enable,
validate,
rely
@@ -1995,14 +2123,14 @@ public List<SQLPrimaryKey> getPrimaryKeys(String
catName, String db_name, String
boolean validate = (enableValidateRely & 2) != 0;
boolean rely = (enableValidateRely & 1) != 0;
SQLPrimaryKey currKey = new SQLPrimaryKey(
- MetastoreDirectSqlUtils.extractSqlString(line[0]),
- MetastoreDirectSqlUtils.extractSqlString(line[1]),
- MetastoreDirectSqlUtils.extractSqlString(line[2]),
- MetastoreDirectSqlUtils.extractSqlInt(line[3]),
MetastoreDirectSqlUtils.extractSqlString(line[4]),
+ extractSqlString(line[0]),
+ extractSqlString(line[1]),
+ extractSqlString(line[2]),
+ MetastoreDirectSqlUtils.extractSqlInt(line[3]),
extractSqlString(line[4]),
enable,
validate,
rely);
-
currKey.setCatName(MetastoreDirectSqlUtils.extractSqlString(line[6]));
+ currKey.setCatName(extractSqlString(line[6]));
ret.add(currKey);
}
}
@@ -2055,10 +2183,10 @@ public List<SQLUniqueConstraint>
getUniqueConstraints(String catName, String db_
boolean rely = (enableValidateRely & 1) != 0;
ret.add(new SQLUniqueConstraint(
catName,
- MetastoreDirectSqlUtils.extractSqlString(line[0]),
- MetastoreDirectSqlUtils.extractSqlString(line[1]),
- MetastoreDirectSqlUtils.extractSqlString(line[2]),
- MetastoreDirectSqlUtils.extractSqlInt(line[3]),
MetastoreDirectSqlUtils.extractSqlString(line[4]),
+ extractSqlString(line[0]),
+ extractSqlString(line[1]),
+ extractSqlString(line[2]),
+ MetastoreDirectSqlUtils.extractSqlInt(line[3]),
extractSqlString(line[4]),
enable,
validate,
rely));
@@ -2113,10 +2241,10 @@ public List<SQLNotNullConstraint>
getNotNullConstraints(String catName, String d
boolean rely = (enableValidateRely & 1) != 0;
ret.add(new SQLNotNullConstraint(
catName,
- MetastoreDirectSqlUtils.extractSqlString(line[0]),
- MetastoreDirectSqlUtils.extractSqlString(line[1]),
- MetastoreDirectSqlUtils.extractSqlString(line[2]),
- MetastoreDirectSqlUtils.extractSqlString(line[3]),
+ extractSqlString(line[0]),
+ extractSqlString(line[1]),
+ extractSqlString(line[2]),
+ extractSqlString(line[3]),
enable,
validate,
rely));
@@ -2175,11 +2303,11 @@ public List<SQLDefaultConstraint>
getDefaultConstraints(String catName, String d
boolean rely = (enableValidateRely & 1) != 0;
SQLDefaultConstraint currConstraint = new SQLDefaultConstraint(
catName,
- MetastoreDirectSqlUtils.extractSqlString(line[0]),
- MetastoreDirectSqlUtils.extractSqlString(line[1]),
- MetastoreDirectSqlUtils.extractSqlString(line[2]),
- MetastoreDirectSqlUtils.extractSqlString(line[5]),
- MetastoreDirectSqlUtils.extractSqlString(line[3]),
+ extractSqlString(line[0]),
+ extractSqlString(line[1]),
+ extractSqlString(line[2]),
+ extractSqlString(line[5]),
+ extractSqlString(line[3]),
enable,
validate,
rely);
@@ -2239,11 +2367,11 @@ public List<SQLCheckConstraint>
getCheckConstraints(String catName, String db_na
boolean rely = (enableValidateRely & 1) != 0;
SQLCheckConstraint currConstraint = new SQLCheckConstraint(
catName,
- MetastoreDirectSqlUtils.extractSqlString(line[0]),
- MetastoreDirectSqlUtils.extractSqlString(line[1]),
- MetastoreDirectSqlUtils.extractSqlString(line[2]),
- MetastoreDirectSqlUtils.extractSqlString(line[5]),
- MetastoreDirectSqlUtils.extractSqlString(line[3]),
+ extractSqlString(line[0]),
+ extractSqlString(line[1]),
+ extractSqlString(line[2]),
+ extractSqlString(line[5]),
+ extractSqlString(line[3]),
enable,
validate,
rely);
@@ -2309,7 +2437,7 @@ public List<String> dropAllPartitionsAndGetLocations(Long
tableId, String baseLo
partIds.add(MetastoreDirectSqlUtils.extractSqlLong(((Object[])result)[0]));
String partitionLocation;
if (baseLocationToNotShow != null &&
- (partitionLocation =
MetastoreDirectSqlUtils.extractSqlString(((Object[])result)[1])) != null &&
+ (partitionLocation = extractSqlString(((Object[])result)[1])) !=
null &&
!FileUtils.isSubdirectory(baseLocationToNotShow,
partitionLocation)) {
locations.add(partitionLocation);
}
@@ -2639,8 +2767,8 @@ public Map<String, List<String>>
getColAndPartNamesWithStats(
String lastPartName = null;
List<String> cols = null;
for (Object[] line : sqlResult) {
- String col = MetastoreDirectSqlUtils.extractSqlString(line[1]);
- String part = MetastoreDirectSqlUtils.extractSqlString(line[0]);
+ String col = extractSqlString(line[1]);
+ String part = extractSqlString(line[0]);
if (!part.equals(lastPartName)) {
if (lastPartName != null) {
result.put(lastPartName, cols);
@@ -2675,8 +2803,7 @@ private void getStatsTableListResult(
.ensureList(executeWithArray(query.getInnerQuery(),
STATS_TABLE_TYPES, queryText));
for (Object[] line : sqlResult) {
result.add(new org.apache.hadoop.hive.common.TableName(
- MetastoreDirectSqlUtils.extractSqlString(line[2]),
MetastoreDirectSqlUtils
- .extractSqlString(line[1]),
MetastoreDirectSqlUtils.extractSqlString(line[0])));
+ extractSqlString(line[2]), extractSqlString(line[1]),
extractSqlString(line[0])));
}
}
}
@@ -2784,11 +2911,11 @@ private List<Function>
getFunctionsFromFunctionIds(List<Long> funcIdList, String
for (Object[] function : queryResult) {
Long funcId =
MetastoreDirectSqlUtils.extractSqlLong(function[funcIdIndex]);
- String funcName =
MetastoreDirectSqlUtils.extractSqlString(function[funcNameIndex]);
- String dbName =
MetastoreDirectSqlUtils.extractSqlString(function[dbNameIndex]);
- String funcClassName =
MetastoreDirectSqlUtils.extractSqlString(function[funcClassNameIndex]);
- String funcOwnerName =
MetastoreDirectSqlUtils.extractSqlString(function[funcOwnerNameIndex]);
- String funcOwnerType =
MetastoreDirectSqlUtils.extractSqlString(function[funcOwnerTypeIndex]);
+ String funcName = extractSqlString(function[funcNameIndex]);
+ String dbName = extractSqlString(function[dbNameIndex]);
+ String funcClassName = extractSqlString(function[funcClassNameIndex]);
+ String funcOwnerName = extractSqlString(function[funcOwnerNameIndex]);
+ String funcOwnerType = extractSqlString(function[funcOwnerTypeIndex]);
int funcCreateTime =
MetastoreDirectSqlUtils.extractSqlInt(function[funcCreateTimeIndex]);
int funcType =
MetastoreDirectSqlUtils.extractSqlInt(function[funcTypeIndex]);
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetastoreDirectSqlUtils.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetastoreDirectSqlUtils.java
index 489a61a0859..b586f75ee6c 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetastoreDirectSqlUtils.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetastoreDirectSqlUtils.java
@@ -582,6 +582,12 @@ public static Boolean extractSqlBoolean(Object value)
throws MetaException {
public static String extractSqlString(Object value) {
if (value == null) return null;
+
+ // Workaround for DataNucleus's Oracle empty-string behavior (it puts
ASCII value 1 instead of an empty string).
+ if ("\u0001".equals(value)) {
+ return "";
+ }
+
return value.toString();
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
index e0e183f93f2..82900c08fa6 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java
@@ -136,6 +136,7 @@
import static
org.apache.hadoop.hive.metastore.ObjectStore.isCurrentStatsValidForTheQuery;
import static
org.apache.hadoop.hive.metastore.ObjectStore.makeParameterDeclarationString;
import static
org.apache.hadoop.hive.metastore.ObjectStore.verifyStatsChangeCtx;
+import static
org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars.PARTITION_REUSE_COLUMN_DESCRIPTORS;
import static
org.apache.hadoop.hive.metastore.metastore.impl.PrivilegeStoreImpl.getPrincipalTypeFromStr;
import static
org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
import static
org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.newMetaException;
@@ -3334,19 +3335,7 @@ private MPartition convertToMPart(Partition part, MTable
mt)
"Partition doesn't have a valid table or database name");
}
- // If this partition's set of columns is the same as the parent table's,
- // use the parent table's, so we do not create a duplicate column
descriptor,
- // thereby saving space
- MStorageDescriptor msd;
- if (mt.getSd() != null && mt.getSd().getCD() != null &&
- mt.getSd().getCD().getCols() != null &&
- part.getSd() != null &&
- convertToFieldSchemas(mt.getSd().getCD().getCols()).
- equals(part.getSd().getCols())) {
- msd = convertToMStorageDescriptor(part.getSd(), mt.getSd().getCD());
- } else {
- msd = convertToMStorageDescriptor(part.getSd());
- }
+ MStorageDescriptor msd = convertToMStorageDescriptor(part.getSd(), mt);
return new MPartition(Warehouse.makePartName(convertToFieldSchemas(mt
.getPartitionKeys()), part.getValues()), mt, part.getValues(), part
@@ -3354,6 +3343,71 @@ private MPartition convertToMPart(Partition part, MTable
mt)
msd, part.getParameters());
}
+ /**
+ * Converts a storage descriptor to a db-backed storage descriptor. Creates
a
+ * new db-backed column descriptor object for this SD, unless a matching one
already
+ * exists for the given table.
+ * @param sd the storage descriptor to wrap in a db-backed object
+ * @param mt the table to search for existing column descriptors, may be null
+ * @return the storage descriptor db-backed object
+ */
+ private MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor sd,
MTable mt)
+ throws MetaException {
+ if (sd == null) {
+ return null;
+ }
+
+ MColumnDescriptor mcd = (mt != null) ? getColumnDescriptor(sd.getCols(),
mt) : null;
+ if (mcd == null) {
+ mcd = createNewMColumnDescriptor(convertToMFieldSchemas(sd.getCols()));
+ }
+ return convertToMStorageDescriptor(sd, mcd);
+ }
+
+ private MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, MTable
mt)
+ throws MetaException {
+ if (cols == null || cols.isEmpty()) {
+ return null;
+ }
+
+ // First check to see if partition and table's column descriptor match;
+ // that's the easy and relatively fast check since it does not require
+ // a round-trip to the database.
+ List<FieldSchema> tableSchema =
+ mt.getSd() != null &&
+ mt.getSd().getCD() != null &&
+ mt.getSd().getCD().getCols() != null ?
+ convertToFieldSchemas(mt.getSd().getCD().getCols()) : null;
+
+ if (cols.equals(tableSchema)) {
+ return mt.getSd().getCD();
+ }
+
+ if (!MetastoreConf.getBoolVar(conf, PARTITION_REUSE_COLUMN_DESCRIPTORS)) {
+ return null;
+ }
+
+ long tblId = mt.getId();
+ try {
+ return new GetHelper<TableName, MColumnDescriptor>(this, null) {
+ @Override
+ protected String describeResult() {
+ return "Matching column descriptor";
+ }
+ @Override
+ protected MColumnDescriptor getSqlResult() throws MetaException {
+ return getDirectSql().getColumnDescriptor(cols, tblId);
+ }
+ @Override
+ protected MColumnDescriptor getJdoResult() {
+ return null;
+ }
+ }.run(false);
+ } catch (NoSuchObjectException e) {
+ return null;
+ }
+ }
+
private Partition convertToPart(String catName, String dbName, String
tblName,
MPartition mpart, boolean isAcidTable, GetPartitionsArgs... args)
throws MetaException {
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java
new file mode 100644
index 00000000000..c1b310c5798
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java
@@ -0,0 +1,220 @@
+/*
+ * 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 org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.InvalidObjectException;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.client.builder.DatabaseBuilder;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.dbinstall.rules.DatabaseRule;
+import org.apache.hadoop.hive.metastore.dbinstall.rules.Derby;
+import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.hive.metastore.Warehouse.DEFAULT_CATALOG_NAME;
+import static org.junit.Assert.assertEquals;
+
+@Category(MetastoreUnitTest.class)
+public class TestHMSColumnDescriptorReuse {
+ private ObjectStore objectStore = null;
+ private Configuration conf;
+
+ // Modify to try out with different databases.
+ // Keep it on Derby once you commit your change. It makes the test execution
faster.
+ private final DatabaseRule db = new Derby(true);
+
+ @Before
+ public void setUp() throws Exception {
+ db.before();
+ db.install();
+
+ conf = MetastoreConf.newMetastoreConf();
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECT_URL_KEY,
db.getJdbcUrl());
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECTION_DRIVER,
db.getJdbcDriver());
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECTION_USER_NAME,
db.getHiveUser());
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.PWD,
db.getHivePassword());
+ MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.AUTO_CREATE_ALL,
false);
+
+ MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.HIVE_IN_TEST, true);
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.PARTITION_REUSE_COLUMN_DESCRIPTORS, true);
+
+ MetaStoreTestUtils.setConfForStandloneMode(conf);
+
+ objectStore = new ObjectStore();
+ objectStore.setConf(conf);
+ HMSHandler.createDefaultCatalog(objectStore, new Warehouse(conf));
+ Database db = new DatabaseBuilder()
+ .setName("default")
+ .setDescription("description")
+ .setLocation("locationurl")
+ .build(conf);
+ objectStore.createDatabase(db);
+ }
+
+ @After
+ public void tearDown() {
+ db.after();
+ }
+
+ @Test
+ public void testReuseAfterSimpleSchemaEvolution() throws Exception {
+ FieldSchema id = new FieldSchema("id", ColumnType.STRING_TYPE_NAME, "");
+ FieldSchema fname = new FieldSchema("fname", ColumnType.STRING_TYPE_NAME,
"");
+ FieldSchema country = new FieldSchema("country",
ColumnType.STRING_TYPE_NAME, "");
+
+ Table tbl1 = newTable(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+
+ objectStore.addPartition(newPart(tbl2, "Hungary"));
+ objectStore.addPartition(newPart(tbl1, "Ukraine"));
+
+ assertEquals(2, countColumnDescriptors());
+ }
+
+ @Test
+ public void testAddPartitionAlterAddPartition() throws MetaException,
InvalidObjectException {
+ FieldSchema id = new FieldSchema("id", ColumnType.STRING_TYPE_NAME, "");
+ FieldSchema fname = new FieldSchema("fname", ColumnType.STRING_TYPE_NAME,
"");
+ FieldSchema country = new FieldSchema("country",
ColumnType.STRING_TYPE_NAME, "");
+
+ Table tbl1 = newTable(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+ objectStore.addPartition(newPart(tbl1, "Greece"));
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+
+ // Mimics replication scenario where we are adding partitions to the
"same" table but with a different schema.
+ // The tbl1 is using the old storage descriptor so "Germany" and "Belgium"
partitions will have the old schema
+ // And will lead to "duplicate" entries in "CDS" and "COLUMNS_V2" tables.
+ objectStore.addPartition(newPart(tbl1, "Germany"));
+ objectStore.addPartition(newPart(tbl1, "Belgium"));
+
+ // On the other hand the addition of a partition to the table with the new
schema is successfully using the
+ // existing storage/column descriptors
+ objectStore.addPartition(newPart(tbl2, "England"));
+
+ assertEquals(2, countColumnDescriptors());
+ }
+
+ @Test
+ public void testNoReusableColumnDescriptors() throws MetaException,
InvalidObjectException {
+ FieldSchema id = new FieldSchema("id", ColumnType.STRING_TYPE_NAME, "");
+ FieldSchema fname = new FieldSchema("fname", ColumnType.STRING_TYPE_NAME,
"");
+ FieldSchema country = new FieldSchema("country",
ColumnType.STRING_TYPE_NAME, "");
+
+ Table tbl1 = newTable(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+
+ FieldSchema address = new FieldSchema("address",
ColumnType.STRING_TYPE_NAME, "");
+ Table tbl3 = newTable(Arrays.asList(id, fname, lname, address),
Collections.singletonList(country));
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl3, null);
+ objectStore.addPartition(newPart(tbl3, "Hungary"));
+
+ assertEquals(3, countColumnDescriptors());
+ }
+
+ private int countColumnDescriptors() {
+ try(Connection c = TestTxnDbUtil.getConnection(conf)){
+ try(ResultSet rs = c.prepareStatement("SELECT COUNT(*) FROM
\"CDS\"").executeQuery()) {
+ rs.next();
+ return rs.getInt(1);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Table newTable(List<FieldSchema> columns, List<FieldSchema>
partCols) {
+ int timeSec = (int) (System.currentTimeMillis() / 1000);
+
+ StorageDescriptor sd = new StorageDescriptor(columns,
+ "/fake/location/person",
+ "org.apache.hadoop.mapred.TextInputFormat",
+ "org.apache.hadoop.mapred.MapFileOutputFormat",
+ false,
+ 0,
+ new SerDeInfo("SerDeName", "serializationLib", null),
+ null,
+ null,
+ null);
+
+ Map<String, String> tableParams = new HashMap<>();
+ tableParams.put("EXTERNAL", "false");
+
+ return
+ new Table("person", "default", "owner", timeSec, timeSec, 3, sd,
partCols,
+ tableParams, null, null, "MANAGED_TABLE");
+ }
+
+ private static Partition newPart(Table tbl, String value) {
+ int timeSec = (int) (System.currentTimeMillis() / 1000);
+ Map<String, String> partitionParams = new HashMap<>();
+
+ partitionParams.put("PARTITION_LEVEL_PRIVILEGE", "true");
+ StorageDescriptor psd = tbl.getSd().deepCopy();
+ psd.setLocation(psd.getLocation() + "/" + value);
+ Partition p = new Partition(
+ Collections.singletonList(value),
+ tbl.getDbName(),
+ tbl.getTableName(),
+ timeSec,
+ timeSec,
+ psd,
+ partitionParams);
+ p.setCatName(DEFAULT_CATALOG_NAME);
+
+ return p;
+ }
+}
