This is an automated email from the ASF dual-hosted git repository.
saihemanth-cloudera 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 58f7341617c HIVE-29724: Provide a Metastore tool for de-duplicating
the columns (#6719)
58f7341617c is described below
commit 58f7341617c1185bafe9b2ee9c30429877b586c7
Author: dengzh <[email protected]>
AuthorDate: Wed Sep 30 01:07:26 2026 +0800
HIVE-29724: Provide a Metastore tool for de-duplicating the columns (#6719)
---
.../hadoop/hive/metastore/conf/MetastoreConf.java | 2 +
.../metastore/metastore/impl/TableStoreImpl.java | 8 +-
.../hive/metastore/tools/ColumnDeduplicator.java | 343 +++++++++++++++++++++
.../hive/metastore/tools/MetaToolObjectStore.java | 88 +++++-
.../metastore/tools/metatool/HiveMetaTool.java | 3 +
.../tools/metatool/HiveMetaToolCommandLine.java | 87 +++++-
.../tools/metatool/MetaToolTaskDedupColumns.java | 120 +++++++
.../metastore/TestHMSColumnDescriptorReuse.java | 64 ++++
.../metatool/TestHiveMetaToolCommandLine.java | 53 +++-
9 files changed, 742 insertions(+), 26 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 83542c02c4e..853139b26e7 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
@@ -2054,6 +2054,8 @@ public enum ConfVars {
"The maximum non-native tables allowed per table type during
collecting the summary."),
METADATA_SUMMARY_NONNATIVE_THREADS("hive.metatool.summary.nonnative.threads",
"hive.metatool.summary.nonnative.threads", 20,
"Number of threads to be allocated for MetaToolTaskMetadataSummary for
collecting the non-native table's summary."),
+ DEDUP_COLUMNS_TIMEOUT("hive.metatool.dedupColumns.timeout",
"hive.metatool.dedupColumns.timeout", 60, TimeUnit.MINUTES,
+ "The maximum time in minutes for the -dedupColumns metatool command to
run before timing out."),
METASTORE_SUPPORT_ACID("metastore.support.acid",
"hive.metastore.support.acid", true,
"Whether to support acid functionality in Hive metastore server."),
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 f1fa5aaac40..069b04043fe 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
@@ -23,6 +23,7 @@
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
+import javax.jdo.PersistenceManager;
import javax.jdo.Query;
import java.util.ArrayList;
import java.util.Arrays;
@@ -146,7 +147,6 @@
@SuppressWarnings("unchecked")
public class TableStoreImpl extends RawStoreBundle implements TableStore {
private final static Logger LOG =
LoggerFactory.getLogger(TableStoreImpl.class);
- private DatabaseProduct dbType;
protected int batchSize = NO_BATCHING;
private boolean areTxnStatsSupported = false;
private PartitionExpressionProxy expressionProxy = null;
@@ -155,7 +155,6 @@ public class TableStoreImpl extends RawStoreBundle
implements TableStore {
@Override
public void setBaseStore(RawStore store) {
super.setBaseStore(store);
- this.dbType = PersistenceManagerProvider.getDatabaseProduct();
this.batchSize = MetastoreConf.getIntVar(store.getConf(),
MetastoreConf.ConfVars.RAWSTORE_PARTITION_BATCH_SIZE);
this.areTxnStatsSupported = MetastoreConf.getBoolVar(baseStore.getConf(),
@@ -442,7 +441,7 @@ private void removeUnusedColumnDescriptor(MColumnDescriptor
oldCD) {
return;
}
LOG.debug("execute removeUnusedColumnDescriptor");
- if (!hasRemainingCDReference(oldCD)) {
+ if (!hasRemainingCDReference(pm, oldCD)) {
// First remove any constraints that may be associated with this CD
Query query = pm.newQuery(MConstraint.class, "parentColumn == inCD ||
childColumn == inCD");
query.declareParameters("MColumnDescriptor inCD");
@@ -464,8 +463,9 @@ private void removeUnusedColumnDescriptor(MColumnDescriptor
oldCD) {
* @param oldCD the column descriptor to check if it has references or not
* @return true if has references
*/
- private boolean hasRemainingCDReference(MColumnDescriptor oldCD) {
+ public static boolean hasRemainingCDReference(PersistenceManager pm,
MColumnDescriptor oldCD) {
assert oldCD != null;
+ DatabaseProduct dbType = PersistenceManagerProvider.getDatabaseProduct();
Query query;
/**
* In order to workaround oracle not supporting limit statement caused
performance issue, HIVE-9447 makes
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/ColumnDeduplicator.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/ColumnDeduplicator.java
new file mode 100644
index 00000000000..b4854c80b3b
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/ColumnDeduplicator.java
@@ -0,0 +1,343 @@
+/*
+ * 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.tools;
+
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.hadoop.hive.metastore.Deadline;
+import org.apache.hadoop.hive.metastore.RawStore;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.metastore.RawStoreBundle;
+import org.apache.hadoop.hive.metastore.model.MColumnDescriptor;
+import org.apache.hadoop.hive.metastore.model.MConstraint;
+import org.apache.hadoop.hive.metastore.model.MPartition;
+import org.apache.hadoop.hive.metastore.model.MStorageDescriptor;
+import org.apache.hadoop.hive.metastore.model.MTable;
+
+import javax.jdo.JDOHelper;
+import javax.jdo.PersistenceManager;
+import javax.jdo.Query;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import static
org.apache.hadoop.hive.metastore.ObjectStore.appendPatternCondition;
+import static
org.apache.hadoop.hive.metastore.metastore.impl.TableStoreImpl.convertToFieldSchemas;
+import static
org.apache.hadoop.hive.metastore.metastore.impl.TableStoreImpl.hasRemainingCDReference;
+import static org.apache.hadoop.hive.metastore.utils.StringUtils.isEmpty;
+
+/**
+ * De-duplicates column descriptors (CDs) for partitioned tables in the
metastore.
+ * Identical column schemas within a table are merged so that partitions share
+ * the same CD, reducing metadata bloat that can accumulate during replication.
+ */
+final class ColumnDeduplicator {
+ private final RawStore store;
+ private final PersistenceManager pm;
+ private final AtomicReference<String> progress;
+ private final boolean isDryRun;
+ private final boolean isVerbose;
+
+ ColumnDeduplicator(RawStoreBundle bundle, AtomicReference<String> progress,
+ boolean isDryRun, boolean isVerbose) {
+ this.store = bundle.getBaseStore();
+ this.pm = bundle.getPersistentManager();
+ this.progress = progress;
+ this.isDryRun = isDryRun;
+ this.isVerbose = isVerbose;
+ }
+
+ MetaToolObjectStore.DedupColumnsResult run(String catalogFilter, String
dbFilter, String tableFilter) {
+ List<TableInfo> tables = findPartitionedTables(catalogFilter, dbFilter,
tableFilter);
+ MetaToolObjectStore.DedupColumnsResult result = new
MetaToolObjectStore.DedupColumnsResult(tables.size());
+
+ long start = System.currentTimeMillis();
+ for (int i = 0; i < tables.size() && result.getException() == null; i++) {
+ boolean committed = false;
+ TableInfo table = tables.get(i);
+ store.openTransaction();
+ try {
+ Deadline.checkTimeout();
+ deduplicateTable(table, result);
+ committed = store.commitTransaction();
+ } catch (Exception ex) {
+ result.catchException(ex);
+ } finally {
+ if (!committed) {
+ store.rollbackTransaction();
+ if (result.getException() == null) {
+ result.catchException(
+ new MetaException("Failed to apply column descriptor
de-duplication updates for table " + table));
+ }
+ }
+ }
+ if (progress != null) {
+ progress.set(String.format(
+ "Finished %d tables in %d total tables, time taken: %d ms, columns
updated: %d, removed: %d",
+ (i + 1),
+ result.getTablesScanned(),
+ (System.currentTimeMillis() - start),
+ result.getStorageDescriptorsUpdated(),
+ result.getColumnDescriptorsRemoved()));
+ }
+ }
+ return result;
+ }
+
+ private void deduplicateTable(TableInfo table,
MetaToolObjectStore.DedupColumnsResult result) {
+ List<PartitionSdInfo> partitionSds =
loadPartitionStorageDescriptors(table.tableId);
+ if (partitionSds.isEmpty()) {
+ return;
+ }
+
+ Set<Long> cdIds = partitionSds.stream().map(p ->
p.cdId).collect(Collectors.toSet());
+ cdIds.add(table.tableCdId);
+
+ Map<Long, List<FieldSchema>> cdColumns = loadColumnSchemas(cdIds);
+ Map<List<FieldSchema>, List<Long>> groups = groupByColumnSchema(cdColumns);
+
+ Map<Long, Long> cdRemap = new HashMap<>();
+ for (List<Long> group : groups.values()) {
+ if (group.size() <= 1) {
+ continue;
+ }
+ long canonicalCdId = pickCanonicalCdId(new HashSet<>(group),
table.tableCdId, partitionSds);
+ for (long cdId : group) {
+ if (cdId != canonicalCdId) {
+ cdRemap.put(cdId, canonicalCdId);
+ }
+ }
+ }
+
+ if (cdRemap.isEmpty()) {
+ return;
+ }
+
+ List<Map.Entry<PartitionSdInfo, Long>> partSdUpdates =
buildPartitionUpdates(partitionSds, cdRemap);
+ if (partSdUpdates.isEmpty()) {
+ return;
+ }
+
+ result.incrementTablesWithDuplicates();
+ for (Map.Entry<PartitionSdInfo, Long> update : partSdUpdates) {
+ result.incrementStorageDescriptorsUpdated();
+ if (isVerbose) {
+ PartitionSdInfo partSd = update.getKey();
+ long newCdId = update.getValue();
+ result.addDetail(String.format("table %s.%s.%s: SD %s CD %d -> %d",
+ table.catalogName, table.dbName, table.tableName,
+ JDOHelper.getObjectId(partSd.sd), partSd.cdId, newCdId));
+ }
+ }
+ if (!isDryRun) {
+ applyTableChanges(partSdUpdates, result);
+ }
+ }
+
+ private void applyTableChanges(List<Map.Entry<PartitionSdInfo, Long>>
partSdUpdates,
+ MetaToolObjectStore.DedupColumnsResult result) {
+ Set<Long> replacedCdIds = new HashSet<>();
+ Map<Long, MColumnDescriptor> newCDs = new HashMap<>();
+ for (Map.Entry<PartitionSdInfo, Long> update : partSdUpdates) {
+ PartitionSdInfo partSd = update.getKey();
+ long newCdId = update.getValue();
+ MColumnDescriptor canonicalCd =
+ newCDs.computeIfAbsent(newCdId, id ->
pm.getObjectById(MColumnDescriptor.class, id));
+ partSd.sd.setCD(canonicalCd);
+ replacedCdIds.add(partSd.cdId);
+ }
+ result.addColumnDescriptorsRemoved(deleteUnusedColumnDescriptors(pm,
replacedCdIds));
+ }
+
+ private List<Map.Entry<PartitionSdInfo, Long>> buildPartitionUpdates(
+ List<PartitionSdInfo> partitionSds, Map<Long, Long> cdRemap) {
+ List<Map.Entry<PartitionSdInfo, Long>> updates = new ArrayList<>();
+ for (PartitionSdInfo partSd : partitionSds) {
+ Long newCdId = cdRemap.get(partSd.cdId);
+ if (newCdId != null && !newCdId.equals(partSd.cdId)) {
+ updates.add(Map.entry(partSd, newCdId));
+ }
+ }
+ return updates;
+ }
+
+ private long pickCanonicalCdId(Set<Long> group, long tableCdId,
List<PartitionSdInfo> partitionSds) {
+ if (group.contains(tableCdId)) {
+ return tableCdId;
+ }
+ Map<Long, Long> usageCount = new HashMap<>();
+ for (PartitionSdInfo partSd : partitionSds) {
+ if (group.contains(partSd.cdId)) {
+ usageCount.merge(partSd.cdId, 1L, Long::sum);
+ }
+ }
+ return group.stream()
+ .max((a, b) -> {
+ int usageCompare = Long.compare(usageCount.getOrDefault(a, 0L),
usageCount.getOrDefault(b, 0L));
+ return usageCompare != 0 ? usageCompare : Long.compare(b, a);
+ })
+ .orElse(group.iterator().next());
+ }
+
+ private List<TableInfo> findPartitionedTables(String catalogFilter, String
dbFilter, String tableFilter) {
+ StringBuilder filter = new StringBuilder();
+ List<String> parameterVals = new ArrayList<>();
+ if (!isEmpty(catalogFilter)) {
+ appendPatternCondition(filter, "table.database.catalogName",
catalogFilter, parameterVals);
+ }
+ if (!isEmpty(dbFilter)) {
+ appendPatternCondition(filter, "table.database.name", dbFilter,
parameterVals);
+ }
+ if (!isEmpty(tableFilter)) {
+ appendPatternCondition(filter, "table.tableName", tableFilter,
parameterVals);
+ }
+
+ Query query = !filter.isEmpty() ?
+ pm.newQuery(MPartition.class, filter.toString()) :
+ pm.newQuery(MPartition.class);
+ query.setResult("DISTINCT this.table");
+ boolean success = false;
+ List<TableInfo> tables = new ArrayList<>();
+ store.openTransaction();
+ try {
+ List<MTable> mTables = (List<MTable>)
query.executeWithArray(parameterVals.toArray(new String[0]));
+ pm.retrieveAll(mTables);
+ for (MTable mTable : mTables) {
+ pm.retrieve(mTable.getDatabase());
+ pm.retrieve(mTable.getSd());
+ pm.retrieve(mTable.getSd().getCD());
+ tables.add(new TableInfo(
+ mTable.getId(),
+ mTable.getSd().getCD().getId(),
+ mTable.getDatabase().getCatalogName(),
+ mTable.getDatabase().getName(),
+ mTable.getTableName()));
+ }
+ success = store.commitTransaction();
+ } finally {
+ query.closeAll();
+ if (!success) {
+ store.rollbackTransaction();
+ }
+ }
+ return tables;
+ }
+
+ private List<PartitionSdInfo> loadPartitionStorageDescriptors(long tableId) {
+ Query query = pm.newQuery(MPartition.class, "table.id == tblId");
+ query.declareParameters("long tblId");
+ query.setResult("sd");
+ List<PartitionSdInfo> partitionSds = new ArrayList<>();
+ try {
+ List<MStorageDescriptor> sds = (List<MStorageDescriptor>)
query.execute(tableId);
+ if (sds == null) {
+ return partitionSds;
+ }
+ pm.retrieveAll(sds);
+ for (MStorageDescriptor sd : sds) {
+ pm.retrieve(sd.getCD());
+ partitionSds.add(new PartitionSdInfo(sd, sd.getCD().getId()));
+ }
+ } finally {
+ query.closeAll();
+ }
+ return partitionSds;
+ }
+
+ private Map<Long, List<FieldSchema>> loadColumnSchemas(Set<Long> cdIds) {
+ Map<Long, List<FieldSchema>> result = new HashMap<>();
+ for (Long cdId : cdIds) {
+ MColumnDescriptor cd = pm.getObjectById(MColumnDescriptor.class, cdId);
+ if (cd != null) {
+ pm.retrieve(cd);
+ result.put(cdId, convertToFieldSchemas(cd.getCols()));
+ }
+ }
+ return result;
+ }
+
+ private static Map<List<FieldSchema>, List<Long>>
groupByColumnSchema(Map<Long, List<FieldSchema>> cdColumns) {
+ Map<List<FieldSchema>, List<Long>> groups = new HashMap<>();
+ for (Map.Entry<Long, List<FieldSchema>> entry : cdColumns.entrySet()) {
+ groups.computeIfAbsent(entry.getValue(), ignored -> new ArrayList<>())
+ .add(entry.getKey());
+ }
+ return groups;
+ }
+
+ private int deleteUnusedColumnDescriptors(PersistenceManager pm, Set<Long>
candidateCdIds) {
+ int removed = 0;
+ for (long cdId : candidateCdIds) {
+ MColumnDescriptor cd = pm.getObjectById(MColumnDescriptor.class, cdId);
+ if (cd == null || hasRemainingCDReference(pm, cd)) {
+ continue;
+ }
+ removeConstraintsForCd(cd);
+ pm.retrieve(cd);
+ pm.deletePersistent(cd);
+ removed++;
+ }
+ return removed;
+ }
+
+ /** Same constraint cleanup as {@code
TableStoreImpl.removeUnusedColumnDescriptor}. */
+ private void removeConstraintsForCd(MColumnDescriptor cd) {
+ Query query = pm.newQuery(MConstraint.class, "parentColumn == inCD ||
childColumn == inCD");
+ query.declareParameters("MColumnDescriptor inCD");
+ try {
+ List<MConstraint> constraints = (List<MConstraint>) query.execute(cd);
+ if (CollectionUtils.isNotEmpty(constraints)) {
+ pm.deletePersistentAll(constraints);
+ }
+ } finally {
+ query.closeAll();
+ }
+ }
+
+ private static final class TableInfo {
+ private final long tableId;
+ private final long tableCdId;
+ private final String catalogName;
+ private final String dbName;
+ private final String tableName;
+
+ private TableInfo(long tableId, long tableCdId, String catalogName, String
dbName, String tableName) {
+ this.tableId = tableId;
+ this.tableCdId = tableCdId;
+ this.catalogName = catalogName;
+ this.dbName = dbName;
+ this.tableName = tableName;
+ }
+
+ @Override
+ public String toString() {
+ return catalogName + "." + dbName + "." + tableName;
+ }
+ }
+
+ private record PartitionSdInfo(MStorageDescriptor sd, long cdId) {
+
+ }
+}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/MetaToolObjectStore.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/MetaToolObjectStore.java
index a65e0b280b1..67a97243bd5 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/MetaToolObjectStore.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/MetaToolObjectStore.java
@@ -32,6 +32,7 @@
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import org.apache.commons.lang3.tuple.Pair;
@@ -672,7 +673,8 @@ public List<MetadataTableSummary> getMetadataSummary(String
catalogFilter,
return metadataTableSummaryList;
}
- private void collectPartitionSummary(Map<Long, MetadataTableSummary>
summaries, Set<Long> partedTabs,
+ private void collectPartitionSummary(Map<Long, MetadataTableSummary>
summaries,
+ Set<Long> partedTabs,
Set<Long> nonPartedTabs) throws MetaException {
String queryText0 = "select \"TBL_ID\", count(1) from \"PARTITION_KEYS\"
where \"TBL_ID\" in (";
runBatched(batchSize, new ArrayList<>(summaries.keySet()), new
Batchable<Long, Void>() {
@@ -724,7 +726,8 @@ public List<Void> run(List<Long> input) throws Exception {
private void collectColumnSummary(Map<Long, MetadataTableSummary> summaries)
throws MetaException {
String queryText0 = "select \"TBL_ID\", count(*), sum(CASE WHEN
\"TYPE_NAME\" like 'array%' THEN 1 ELSE 0 END)," +
" sum(CASE WHEN \"TYPE_NAME\" like 'struct%' THEN 1 ELSE 0 END),
sum(CASE WHEN \"TYPE_NAME\" like 'map%' THEN 1 ELSE 0 END)" +
- " from \"TBLS\" t join \"SDS\" s on t.\"SD_ID\" = s.\"SD_ID\" join
\"CDS\" c on s.\"CD_ID\" = c.\"CD_ID\" join \"COLUMNS_V2\" v on c.\"CD_ID\" =
v.\"CD_ID\"" +
+ " from \"TBLS\" t join \"SDS\" s on t.\"SD_ID\" = s.\"SD_ID\" join
\"CDS\" c" +
+ " on s.\"CD_ID\" = c.\"CD_ID\" join \"COLUMNS_V2\" v on c.\"CD_ID\" =
v.\"CD_ID\"" +
" where \"TBL_ID\" in (";
runBatched(batchSize, new ArrayList<>(summaries.keySet()), new
Batchable<Long, Void>() {
@Override
@@ -750,7 +753,8 @@ public List<Void> run(List<Long> input) throws Exception {
}
private void collectTabFormatSummary(Map<Long, MetadataTableSummary>
summaries) throws MetaException {
- String queryText0 = "select t.\"TBL_ID\", d.\"SLIB\", s.\"IS_COMPRESSED\"
from \"TBLS\" t left join \"SDS\" s on t.\"SD_ID\" = s.\"SD_ID\" left join
\"SERDES\" d on d.\"SERDE_ID\" = s.\"SERDE_ID\"" +
+ String queryText0 = "select t.\"TBL_ID\", d.\"SLIB\", s.\"IS_COMPRESSED\"
from \"TBLS\" t left join \"SDS\" s" +
+ " on t.\"SD_ID\" = s.\"SD_ID\" left join \"SERDES\" d on
d.\"SERDE_ID\" = s.\"SERDE_ID\"" +
" where t.\"TBL_ID\" in (";
String queryText1 = "select p.\"TBL_ID\", " +
dbType.toVarChar("p.\"PARAM_VALUE\"") + " from \"TABLE_PARAMS\" p " +
" where p.\"PARAM_KEY\" = 'transactional_properties' and p.\"TBL_ID\"
in (";
@@ -843,8 +847,10 @@ private void collectTabFormatSummary(List<Long>
transactionalTables, Long tableI
private void collectBasicStats(Map<Long, MetadataTableSummary> summaries,
Set<Long> nonPartedTabs,
Set<Long> partedTabs) throws MetaException {
- String queryText0 = "select \"TBL_ID\", \"PARAM_KEY\", CAST(" +
dbType.toVarChar("\"PARAM_VALUE\"") + " AS decimal(21,0)) from \"TABLE_PARAMS\"
where \"PARAM_KEY\" " +
- "in ('" + StatsSetupConst.TOTAL_SIZE + "', '" +
StatsSetupConst.NUM_FILES + "', '" + StatsSetupConst.ROW_COUNT + "') and
\"TBL_ID\" in (";
+ String queryText0 = "select \"TBL_ID\", \"PARAM_KEY\", CAST(" +
dbType.toVarChar("\"PARAM_VALUE\"") +
+ " AS decimal(21,0)) from \"TABLE_PARAMS\" where \"PARAM_KEY\"" +
+ " in ('" + StatsSetupConst.TOTAL_SIZE + "', '" +
StatsSetupConst.NUM_FILES + "', '" +
+ StatsSetupConst.ROW_COUNT + "') and \"TBL_ID\" in (";
runBatched(batchSize, new ArrayList<>(nonPartedTabs), new Batchable<Long,
Void>() {
@Override
public List<Void> run(List<Long> input) throws Exception {
@@ -853,9 +859,11 @@ public List<Void> run(List<Long> input) throws Exception {
}
});
- String queryText1 = "select \"TBL_ID\", \"PARAM_KEY\", sum(CAST(" +
dbType.toVarChar("\"PARAM_VALUE\"") + " AS decimal(21,0))) from \"PARTITIONS\"
t " +
- "join \"PARTITION_PARAMS\" p on p.\"PART_ID\" = t.\"PART_ID\" where
\"PARAM_KEY\" " +
- "in ('" + StatsSetupConst.TOTAL_SIZE + "', '" +
StatsSetupConst.NUM_FILES + "', '" + StatsSetupConst.ROW_COUNT + "') and
t.\"TBL_ID\" in (";
+ String queryText1 = "select \"TBL_ID\", \"PARAM_KEY\", sum(CAST(" +
dbType.toVarChar("\"PARAM_VALUE\"") +
+ " AS decimal(21,0))) from \"PARTITIONS\" t" +
+ " join \"PARTITION_PARAMS\" p on p.\"PART_ID\" = t.\"PART_ID\" where
\"PARAM_KEY\"" +
+ " in ('" + StatsSetupConst.TOTAL_SIZE + "', '" +
StatsSetupConst.NUM_FILES + "', '" +
+ StatsSetupConst.ROW_COUNT + "') and t.\"TBL_ID\" in (";
runBatched(batchSize, new ArrayList<>(partedTabs), new Batchable<Long,
Void>() {
@Override
public List<Void> run(List<Long> input) throws Exception {
@@ -927,7 +935,8 @@ public List<Long> run(List<Long> input) throws Exception {
int size = input.size();
String queryText =
"\"TBL_ID\" from \"TABLE_PARAMS\" where \"PARAM_KEY\" =
'current-snapshot-timestamp-ms' "
- + (lastUpdatedDays != null ? (" and CAST(" +
dbType.toVarChar("\"PARAM_VALUE\"") + " AS decimal(21,0)) > " +
(System.currentTimeMillis() - lastUpdatedDays * 24 * 3600000L)) : "")
+ + (lastUpdatedDays != null ? (" and CAST(" +
dbType.toVarChar("\"PARAM_VALUE\"") + " AS decimal(21,0)) > "
+ + (System.currentTimeMillis() - lastUpdatedDays * 24 *
3600000L)) : "")
+ " and \"TBL_ID\" in (" + (size == 0 ? "" : repeat(",?",
size).substring(1)) + ") "
+ " order by CAST(" + dbType.toVarChar("\"PARAM_VALUE\"") + "
AS decimal(21,0)) DESC";
if (tablesLimit != null && tablesLimit >= 0) {
@@ -960,4 +969,65 @@ public List<Long> run(List<Long> input) throws Exception {
});
return new HashSet<>(tables);
}
+
+ public static final class DedupColumnsResult {
+ private final int tablesScanned;
+ private int tablesWithDuplicates;
+ private int storageDescriptorsUpdated;
+ private int columnDescriptorsRemoved;
+ private final List<String> details = new ArrayList<>();
+ private Exception exception;
+ DedupColumnsResult(int tablesScanned) {
+ this.tablesScanned = tablesScanned;
+ }
+ public int getTablesScanned() {
+ return tablesScanned;
+ }
+
+ public int getTablesWithDuplicates() {
+ return tablesWithDuplicates;
+ }
+
+ void incrementTablesWithDuplicates() {
+ this.tablesWithDuplicates++;
+ }
+
+ public int getStorageDescriptorsUpdated() {
+ return storageDescriptorsUpdated;
+ }
+
+ void incrementStorageDescriptorsUpdated() {
+ this.storageDescriptorsUpdated++;
+ }
+
+ public int getColumnDescriptorsRemoved() {
+ return columnDescriptorsRemoved;
+ }
+
+ void addColumnDescriptorsRemoved(int count) {
+ this.columnDescriptorsRemoved += count;
+ }
+
+ public List<String> getDetails() {
+ return details;
+ }
+
+ void addDetail(String detail) {
+ details.add(detail);
+ }
+
+ void catchException(Exception exception) {
+ this.exception = exception;
+ }
+
+ public Exception getException() {
+ return exception;
+ }
+ }
+
+ public DedupColumnsResult dedupColumns(String catalogFilter, String
dbFilter, String tableFilter,
+ AtomicReference<String> progress, boolean isDryRun, boolean isVerbose) {
+ return new ColumnDeduplicator(this.createRawStoreBundle(), progress,
isDryRun, isVerbose)
+ .run(catalogFilter, dbFilter, tableFilter);
+ }
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaTool.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaTool.java
index 28814335280..2ab17c67edc 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaTool.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaTool.java
@@ -32,6 +32,7 @@
* - execute JDOQL against the metastore using DataNucleus
* - perform HA name node upgrade
* - summarize the data in HMS
+ * - de-duplicate column descriptors for partitioned tables
*/
public final class HiveMetaTool {
private static final Logger LOGGER =
LoggerFactory.getLogger(HiveMetaTool.class.getName());
@@ -60,6 +61,8 @@ public static void execute(String[] args) throws Exception {
task = new MetaToolTaskDiffExtTblLocs();
} else if (cl.isMetadataSummary()) {
task = new MetaToolTaskMetadataSummary();
+ } else if (cl.isDedupColumns()) {
+ task = new MetaToolTaskDedupColumns();
} else {
throw new IllegalArgumentException("No task was specified!");
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaToolCommandLine.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaToolCommandLine.java
index 29af84c407f..a1c9db7a327 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaToolCommandLine.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/HiveMetaToolCommandLine.java
@@ -86,9 +86,30 @@ class HiveMetaToolCommandLine {
)
.create("diffExtTblLocs");
+ @SuppressWarnings("static-access")
+ private static final Option DEDUP_COLUMNS = OptionBuilder
+ .withArgName("catalog> " + "<db> " + "<table")
+ .hasArgs(3)
+ .hasOptionalArgs(3)
+ .withDescription("De-duplicate column descriptors for partitioned tables
to reduce metadata bloat. "
+ + "Optional catalog, database, and table filters support *
wildcards. Use -dryRun to preview changes, "
+ + "-verbose to print per-partition details, and -timeout to set a
timeout in seconds.")
+ .create("dedupColumns");
+
+ private static final Option VERBOSE = OptionBuilder
+ .withDescription("Print per-partition de-duplication details. Valid only
with -dedupColumns.")
+ .create("verbose");
+
+ @SuppressWarnings("static-access")
+ private static final Option TIMEOUT = OptionBuilder
+ .withArgName("timeout-seconds")
+ .hasArg()
+ .withDescription("Timeout in seconds for -dedupColumns. Defaults to
hive.metatool.dedupColumns.timeout.")
+ .create("timeout");
+
private static final Option DRY_RUN = OptionBuilder
- .withDescription("Perform a dry run of updateLocation changes.When run
with the dryRun option updateLocation " +
- "changes are displayed but not persisted. dryRun is valid only with
the updateLocation option.")
+ .withDescription("Perform a dry run without persisting changes. Valid
with -updateLocation and "
+ + "-dedupColumns.")
.create("dryRun");
@SuppressWarnings("static-access")
@@ -128,6 +149,9 @@ class HiveMetaToolCommandLine {
OPTIONS.addOption(TABLE_PROP_KEY);
OPTIONS.addOption(HELP);
OPTIONS.addOption(METADATA_SUMMARY);
+ OPTIONS.addOption(DEDUP_COLUMNS);
+ OPTIONS.addOption(VERBOSE);
+ OPTIONS.addOption(TIMEOUT);
}
private boolean listFSRoot;
@@ -136,10 +160,13 @@ class HiveMetaToolCommandLine {
private String[] listExtTblLocsParams;
private String[] diffExtTblLocsParams;
private boolean dryRun;
+ private boolean verbose;
+ private Long dedupColumnsTimeoutSeconds;
private String serdePropKey;
private String tablePropKey;
private boolean help;
private String[] metadataSummaryParams;
+ private String[] dedupColumnsParams;
public static HiveMetaToolCommandLine parseArguments(String[] args) {
HiveMetaToolCommandLine cl = null;
@@ -174,16 +201,28 @@ private void parseCommandLine(String[] args) throws
ParseException {
listExtTblLocsParams = cl.getOptionValues(LIST_EXT_TBL_LOCS.getOpt());
diffExtTblLocsParams = cl.getOptionValues(DIFF_EXT_TBL_LOCS.getOpt());
dryRun = cl.hasOption(DRY_RUN.getOpt());
+ verbose = cl.hasOption(VERBOSE.getOpt());
+ if (cl.hasOption(TIMEOUT.getOpt())) {
+ dedupColumnsTimeoutSeconds =
Long.parseLong(cl.getOptionValue(TIMEOUT.getOpt()));
+ if (dedupColumnsTimeoutSeconds <= 0) {
+ throw new IllegalArgumentException("HiveMetaTool:-timeout must be a
positive number of seconds");
+ }
+ }
serdePropKey = cl.getOptionValue(SERDE_PROP_KEY.getOpt());
tablePropKey = cl.getOptionValue(TABLE_PROP_KEY.getOpt());
help = cl.hasOption(HELP.getOpt());
metadataSummaryParams = cl.getOptionValues(METADATA_SUMMARY.getOpt());
+ dedupColumnsParams = cl.getOptionValues(DEDUP_COLUMNS.getOpt());
+ if (cl.hasOption(DEDUP_COLUMNS.getOpt()) && dedupColumnsParams == null) {
+ dedupColumnsParams = new String[0];
+ }
int commandCount = (isListFSRoot() ? 1 : 0) + (isExecuteJDOQL() ? 1 : 0) +
(isUpdateLocation() ? 1 : 0) +
- (isListExtTblLocs() ? 1 : 0) + (isDiffExtTblLocs() ? 1 : 0) +
(isMetadataSummary() ? 1 : 0);
+ (isListExtTblLocs() ? 1 : 0) + (isDiffExtTblLocs() ? 1 : 0) +
(isMetadataSummary() ? 1 : 0) +
+ (isDedupColumns() ? 1 : 0);
if (commandCount != 1) {
throw new IllegalArgumentException("exactly one of -listFSRoot,
-executeJDOQL, -updateLocation, " +
- "-listExtTblLocs, -diffExtTblLocs, -metadataSummary must be
set");
+ "-listExtTblLocs, -diffExtTblLocs, -metadataSummary,
-dedupColumns must be set");
}
if (updateLocationParams != null && updateLocationParams.length != 2) {
@@ -201,9 +240,22 @@ private void parseCommandLine(String[] args) throws
ParseException {
diffExtTblLocsParams.length + " arguments");
}
- if ((dryRun || serdePropKey != null || tablePropKey != null) &&
!isUpdateLocation()) {
- throw new IllegalArgumentException("-dryRun, -serdePropKey,
-tablePropKey may be used only for the " +
- "-updateLocation command");
+ if (dryRun && !isUpdateLocation() && !isDedupColumns()) {
+ throw new IllegalArgumentException(
+ "-dryRun may be used only for the -updateLocation or -dedupColumns
commands");
+ }
+
+ if ((serdePropKey != null || tablePropKey != null) && !isUpdateLocation())
{
+ throw new IllegalArgumentException(
+ "-serdePropKey, -tablePropKey may be used only for the
-updateLocation command");
+ }
+
+ if (verbose && !isDedupColumns()) {
+ throw new IllegalArgumentException("-verbose may be used only for the
-dedupColumns command");
+ }
+
+ if (dedupColumnsTimeoutSeconds != null && !isDedupColumns()) {
+ throw new IllegalArgumentException("-timeout may be used only for the
-dedupColumns command");
}
if (metadataSummaryParams != null && metadataSummaryParams.length < 1) {
@@ -233,9 +285,12 @@ private void printArguments() {
"\tlistExtTblLocs: " + Arrays.toString(listExtTblLocsParams) + "\n" +
"\tdiffExtTblLocs: " + Arrays.toString(diffExtTblLocsParams) + "\n" +
"\tdryRun : " + dryRun + "\n" +
+ "\tverbose : " + verbose + "\n" +
+ "\ttimeout : " + dedupColumnsTimeoutSeconds + "\n" +
"\tserdePropKey : " + serdePropKey + "\n" +
"\ttablePropKey : " + tablePropKey + "\n" +
- "\tmetadataSummary : " + Arrays.toString(metadataSummaryParams));
+ "\tmetadataSummary : " + Arrays.toString(metadataSummaryParams) + "\n"
+
+ "\tdedupColumns : " + Arrays.toString(dedupColumnsParams));
}
boolean isListFSRoot() {
@@ -278,6 +333,14 @@ boolean isDryRun() {
return dryRun;
}
+ boolean isVerbose() {
+ return verbose;
+ }
+
+ Long getDedupColumnsTimeoutSeconds() {
+ return dedupColumnsTimeoutSeconds;
+ }
+
String getSerdePropKey() {
return serdePropKey;
}
@@ -294,6 +357,14 @@ String[] getMetadataSummaryParams() {
return metadataSummaryParams;
}
+ boolean isDedupColumns() {
+ return dedupColumnsParams != null;
+ }
+
+ String[] getDedupColumnsParams() {
+ return dedupColumnsParams == null ? new String[0] : dedupColumnsParams;
+ }
+
private boolean isHelp() {
return help;
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/MetaToolTaskDedupColumns.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/MetaToolTaskDedupColumns.java
new file mode 100644
index 00000000000..06994e5d493
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/MetaToolTaskDedupColumns.java
@@ -0,0 +1,120 @@
+/*
+ * 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.tools.metatool;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hadoop.hive.metastore.Deadline;
+import org.apache.hadoop.hive.metastore.DeadlineException;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.tools.MetaToolObjectStore;
+
+class MetaToolTaskDedupColumns extends MetaToolTask {
+ @Override
+ void execute() {
+ String[] params = getCl().getDedupColumnsParams();
+ String catalogFilter = params.length > 0 ? params[0] : null;
+ String dbFilter = params.length > 1 ? params[1] : null;
+ String tableFilter = params.length > 2 ? params[2] : null;
+ boolean isDryRun = getCl().isDryRun();
+ boolean isVerbose = getCl().isVerbose();
+
+ Thread daemon = null;
+ MetaToolObjectStore.DedupColumnsResult result = null;
+ AtomicReference<String> progress = new AtomicReference<>();
+ AtomicBoolean stopped = new AtomicBoolean(false);
+ long timeoutMs = getCl().getDedupColumnsTimeoutSeconds() != null ?
+ getCl().getDedupColumnsTimeoutSeconds() * 1000L :
+ MetastoreConf.getTimeVar(getObjectStore().getConf(),
MetastoreConf.ConfVars.DEDUP_COLUMNS_TIMEOUT,
+ TimeUnit.MILLISECONDS);
+ Deadline.registerIfNot(timeoutMs);
+ boolean timerStarted = false;
+ try {
+ timerStarted = Deadline.startTimer("dedupColumns");
+ if (isVerbose) {
+ daemon = new Thread(() -> {
+ while (!stopped.get()) {
+ try {
+ Thread.sleep(30 * 1000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ String message = progress.get();
+ if (message != null) {
+ System.out.println(message);
+ }
+ }
+ });
+ daemon.setDaemon(true);
+ daemon.start();
+ }
+ result = getObjectStore().dedupColumns(catalogFilter, dbFilter,
tableFilter, progress, isDryRun, isVerbose);
+ } catch (MetaException ex) {
+ // Deadline.startTimer throws MetaException, this shouldn't happen as
the Deadline has been registered.
+ System.err.println(ex.getMessage());
+ } finally {
+ if (daemon != null) {
+ stopped.set(true);
+ daemon.interrupt();
+ }
+ if (timerStarted) {
+ try {
+ Deadline.stopTimer();
+ } catch (MetaException ignored) {
+ }
+ }
+ }
+
+ if (result != null) {
+ printSummary(result, isDryRun, isVerbose);
+ if (result.getException() != null) {
+ if (result.getException().getCause() instanceof DeadlineException
deadline) {
+ // Just print the timeout message
+ System.err.println(deadline.getMessage());
+ return;
+ }
+ throw new IllegalStateException("HiveMetaTool: failed to de-duplicate
column descriptors for all tables",
+ result.getException());
+ }
+ }
+ }
+
+ private void printSummary(MetaToolObjectStore.DedupColumnsResult result,
boolean isDryRun, boolean isVerbose) {
+ System.out.println(isDryRun ?
+ "Dry run of -dedupColumns.." :
+ "De-duplicated column descriptors..");
+ System.out.println("Tables scanned: " + result.getTablesScanned());
+ System.out.println("Tables with duplicate column descriptors: " +
result.getTablesWithDuplicates());
+ System.out.println("Partition storage descriptors " + (isDryRun ? "to
update" : "updated") + ": "
+ + result.getStorageDescriptorsUpdated());
+ if (!isDryRun) {
+ System.out.println("Column descriptors removed: " +
result.getColumnDescriptorsRemoved());
+ }
+ if (isVerbose) {
+ for (String detail : result.getDetails()) {
+ System.out.println(detail);
+ }
+ }
+ }
+}
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
index 008ee3b5a67..483c15fccba 100644
---
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
@@ -22,6 +22,7 @@
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.GetPartitionsByNamesRequest;
import org.apache.hadoop.hive.metastore.api.InvalidObjectException;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Partition;
@@ -29,9 +30,11 @@
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.client.builder.GetPartitionsArgs;
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.tools.MetaToolObjectStore;
import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
import org.junit.After;
import org.junit.Before;
@@ -45,9 +48,12 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.hadoop.hive.metastore.Warehouse.DEFAULT_CATALOG_NAME;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
@Category(MetastoreUnitTest.class)
public class TestHMSColumnDescriptorReuse {
@@ -166,6 +172,64 @@ public void testNoReusableColumnDescriptors() throws
MetaException, InvalidObjec
assertEquals(3, countColumnDescriptors());
}
+ @Test
+ public void testDeduplicateColumnDescriptorsTool() throws Exception {
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.PARTITION_REUSE_COLUMN_DESCRIPTORS, false);
+
+ 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"));
+ int cdsBeforeDedup = countColumnDescriptors();
+ assertTrue(cdsBeforeDedup == 1);
+
+ AtomicReference<String> progress = new AtomicReference<>();
+ MetaToolObjectStore metaToolStore = new MetaToolObjectStore();
+ metaToolStore.setConf(conf);
+ Deadline.registerIfNot(30 * 1000);
+ Deadline.startTimer("testDeduplicateColumnDescriptorsTool");
+ try {
+ MetaToolObjectStore.DedupColumnsResult result =
metaToolStore.dedupColumns(null, "default", "person", progress,
+ false, false);
+ assertEquals(0, result.getTablesWithDuplicates());
+
+ 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(tbl1, "Germany"));
+ objectStore.addPartition(newPart(tbl1, "Belgium"));
+ objectStore.addPartition(newPart(tbl2, "England"));
+ cdsBeforeDedup = countColumnDescriptors();
+ assertTrue(cdsBeforeDedup > 2);
+
+ result = metaToolStore.dedupColumns(null, "default", "person", progress,
false, false);
+ assertTrue(result.getStorageDescriptorsUpdated() > 0);
+ assertEquals(2, countColumnDescriptors());
+ assertNotNull(progress.get());
+
+ Deadline.registerIfNot(30 * 1000);
+ Deadline.startTimer("testDeduplicateColumnDescriptorsTool");
+ GetPartitionsByNamesRequest request = new
GetPartitionsByNamesRequest("default", "person");
+ request.setNames(List.of("country=Germany", "country=Belgium",
"country=Greece", "country=US"));
+ List<Partition> partitions = objectStore.getPartitionsByNames("hive",
"default", "person",
+ GetPartitionsArgs.from(request));
+ assertEquals(4, partitions.size());
+ assertTrue(partitions.stream().allMatch(p ->
tbl1.getSd().getCols().equals(p.getSd().getCols())));
+
+ request.setNames(List.of("country=Italy", "country=England"));
+ partitions = objectStore.getPartitionsByNames("hive", "default",
"person", GetPartitionsArgs.from(request));
+ assertEquals(2, partitions.size());
+ assertTrue(partitions.stream().allMatch(p ->
tbl2.getSd().getCols().equals(p.getSd().getCols())));
+ } finally {
+ Deadline.stopTimer();
+ }
+ }
+
private int countColumnDescriptors() {
try(Connection c = TestTxnDbUtil.getConnection(conf)){
try(ResultSet rs = c.prepareStatement("SELECT COUNT(*) FROM
\"CDS\"").executeQuery()) {
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/tools/metatool/TestHiveMetaToolCommandLine.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/tools/metatool/TestHiveMetaToolCommandLine.java
index 9297194196e..f28b177eb02 100644
---
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/tools/metatool/TestHiveMetaToolCommandLine.java
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/tools/metatool/TestHiveMetaToolCommandLine.java
@@ -88,7 +88,7 @@ public void testParseUpdateLocation() throws ParseException {
@Test
public void testNoTask() throws ParseException {
exception.expect(IllegalArgumentException.class);
- exception.expectMessage("exactly one of -listFSRoot, -executeJDOQL,
-updateLocation, -listExtTblLocs, -diffExtTblLocs, -metadataSummary must be
set");
+ exception.expectMessage("exactly one of -listFSRoot, -executeJDOQL,
-updateLocation, -listExtTblLocs, -diffExtTblLocs, -metadataSummary,
-dedupColumns must be set");
new HiveMetaToolCommandLine(new String[] {});
}
@@ -96,7 +96,7 @@ public void testNoTask() throws ParseException {
@Test
public void testMultipleTask() throws ParseException {
exception.expect(IllegalArgumentException.class);
- exception.expectMessage("exactly one of -listFSRoot, -executeJDOQL,
-updateLocation, -listExtTblLocs, -diffExtTblLocs, -metadataSummary must be
set");
+ exception.expectMessage("exactly one of -listFSRoot, -executeJDOQL,
-updateLocation, -listExtTblLocs, -diffExtTblLocs, -metadataSummary,
-dedupColumns must be set");
new HiveMetaToolCommandLine(new String[] {"-listFSRoot", "-executeJDOQL",
"select a from b"});
}
@@ -132,15 +132,58 @@ public void testDiffExtTblLocsArgCount() throws
ParseException {
@Test
public void testDryRunNotAllowed() throws ParseException {
exception.expect(IllegalArgumentException.class);
- exception.expectMessage("-dryRun, -serdePropKey, -tablePropKey may be used
only for the -updateLocation command");
+ exception.expectMessage("-dryRun may be used only for the -updateLocation
or -dedupColumns commands");
new HiveMetaToolCommandLine(new String[] {"-listFSRoot", "-dryRun"});
}
+ @Test
+ public void testParseDedupColumns() throws ParseException {
+ HiveMetaToolCommandLine cl = new HiveMetaToolCommandLine(
+ new String[] {"-dedupColumns", "hive", "default", "person", "-dryRun",
"-verbose"});
+ assertTrue(cl.isDedupColumns());
+ assertTrue(cl.isDryRun());
+ assertTrue(cl.isVerbose());
+ assertEquals("hive", cl.getDedupColumnsParams()[0]);
+ assertEquals("default", cl.getDedupColumnsParams()[1]);
+ assertEquals("person", cl.getDedupColumnsParams()[2]);
+
+ cl = new HiveMetaToolCommandLine(new String[] {"-dedupColumns"});
+ assertTrue(cl.isDedupColumns());
+ assertEquals(0, cl.getDedupColumnsParams().length);
+
+ cl = new HiveMetaToolCommandLine(new String[] {"-dedupColumns",
"-timeout", "3600"});
+ assertEquals(Long.valueOf(3600L), cl.getDedupColumnsTimeoutSeconds());
+ }
+
+ @Test
+ public void testTimeoutNotAllowed() throws ParseException {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("-timeout may be used only for the -dedupColumns
command");
+
+ new HiveMetaToolCommandLine(new String[] {"-listFSRoot", "-timeout",
"60"});
+ }
+
+ @Test
+ public void testInvalidTimeout() throws ParseException {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("HiveMetaTool:-timeout must be a positive number
of seconds");
+
+ new HiveMetaToolCommandLine(new String[] {"-dedupColumns", "-timeout",
"0"});
+ }
+
+ @Test
+ public void testVerboseNotAllowed() throws ParseException {
+ exception.expect(IllegalArgumentException.class);
+ exception.expectMessage("-verbose may be used only for the -dedupColumns
command");
+
+ new HiveMetaToolCommandLine(new String[] {"-listFSRoot", "-verbose"});
+ }
+
@Test
public void testSerdePropKeyNotAllowed() throws ParseException {
exception.expect(IllegalArgumentException.class);
- exception.expectMessage("-dryRun, -serdePropKey, -tablePropKey may be used
only for the -updateLocation command");
+ exception.expectMessage("-serdePropKey, -tablePropKey may be used only for
the -updateLocation command");
new HiveMetaToolCommandLine(new String[] {"-listFSRoot", "-serdePropKey",
"abc"});
}
@@ -148,7 +191,7 @@ public void testSerdePropKeyNotAllowed() throws
ParseException {
@Test
public void testTablePropKeyNotAllowed() throws ParseException {
exception.expect(IllegalArgumentException.class);
- exception.expectMessage("-dryRun, -serdePropKey, -tablePropKey may be used
only for the -updateLocation command");
+ exception.expectMessage("-serdePropKey, -tablePropKey may be used only for
the -updateLocation command");
new HiveMetaToolCommandLine(new String[] {"-executeJDOQL", "select a from
b", "-tablePropKey", "abc"});
}
