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 26f76bc92f0b..9aa120b7237a 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 000000000000..7a3b031314a9 --- /dev/null +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/ColumnDeduplicator.java @@ -0,0 +1,341 @@ +/* + * 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.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 progress; + private final boolean isDryRun; + private final boolean isVerbose; + + ColumnDeduplicator(RawStoreBundle bundle, AtomicReference 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 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 { + 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 partitionSds = loadPartitionStorageDescriptors(table.tableId); + if (partitionSds.isEmpty()) { + return; + } + + Set cdIds = partitionSds.stream().map(p -> p.cdId).collect(Collectors.toSet()); + cdIds.add(table.tableCdId); + + Map> cdColumns = loadColumnSchemas(cdIds); + Map, List> groups = groupByColumnSchema(cdColumns); + + Map cdRemap = new HashMap<>(); + for (List 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> partSdUpdates = buildPartitionUpdates(partitionSds, cdRemap); + if (partSdUpdates.isEmpty()) { + return; + } + + result.incrementTablesWithDuplicates(); + for (Map.Entry 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> partSdUpdates, + MetaToolObjectStore.DedupColumnsResult result) { + Set replacedCdIds = new HashSet<>(); + Map newCDs = new HashMap<>(); + for (Map.Entry 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> buildPartitionUpdates( + List partitionSds, Map cdRemap) { + List> 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 group, long tableCdId, List partitionSds) { + if (group.contains(tableCdId)) { + return tableCdId; + } + Map 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 findPartitionedTables(String catalogFilter, String dbFilter, String tableFilter) { + StringBuilder filter = new StringBuilder(); + List 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 tables = new ArrayList<>(); + store.openTransaction(); + try { + List mTables = (List) 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 loadPartitionStorageDescriptors(long tableId) { + Query query = pm.newQuery(MPartition.class, "table.id == tblId"); + query.declareParameters("long tblId"); + query.setResult("sd"); + List partitionSds = new ArrayList<>(); + try { + List sds = (List) 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> loadColumnSchemas(Set cdIds) { + Map> 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> groupByColumnSchema(Map> cdColumns) { + Map, List> groups = new HashMap<>(); + for (Map.Entry> entry : cdColumns.entrySet()) { + groups.computeIfAbsent(entry.getValue(), ignored -> new ArrayList<>()) + .add(entry.getKey()); + } + return groups; + } + + private int deleteUnusedColumnDescriptors(PersistenceManager pm, Set candidateCdIds) { + int removed = 0; + for (long cdId : candidateCdIds) { + MColumnDescriptor cd = pm.getObjectById(MColumnDescriptor.class, cdId); + if (cd == null || hasRemainingCDReference(pm, cd)) { + continue; + } + removeConstraintsForCd(pm, cd); + pm.retrieve(cd); + pm.deletePersistent(cd); + removed++; + } + return removed; + } + + /** Same constraint cleanup as {@code TableStoreImpl.removeUnusedColumnDescriptor}. */ + private void removeConstraintsForCd(PersistenceManager pm, MColumnDescriptor cd) { + Query query = pm.newQuery(MConstraint.class, "parentColumn == inCD || childColumn == inCD"); + query.declareParameters("MColumnDescriptor inCD"); + try { + List constraints = (List) 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 a65e0b280b1d..dc5786f629b3 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; @@ -960,4 +961,65 @@ public List run(List 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 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 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 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 28814335280a..2ab17c67edcf 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 29af84c407f6..3a466b37bca0 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,23 @@ class HiveMetaToolCommandLine { ) .create("diffExtTblLocs"); + @SuppressWarnings("static-access") + private static final Option DEDUP_COLUMNS = OptionBuilder + .withArgName("catalog> " + " " + " 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(); + + final AtomicReference progress = new AtomicReference<>(); + AtomicBoolean stopped = new AtomicBoolean(false); + Thread daemon = null; + 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(); + } + MetaToolObjectStore.DedupColumnsResult result; + try { + result = getObjectStore().dedupColumns(catalogFilter, dbFilter, tableFilter, progress, isDryRun, isVerbose); + printSummary(result, isDryRun, isVerbose); + } finally { + if (daemon != null) { + stopped.set(true); + daemon.interrupt(); + } + } + + if (result.getException() != null) { + 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 008ee3b5a67a..20fc99b640c7 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,58 @@ 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 progress = new AtomicReference<>(); + MetaToolObjectStore metaToolStore = new MetaToolObjectStore(); + metaToolStore.setConf(conf); + 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 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()))); + } + 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 9297194196e7..983bce02617d 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,39 @@ 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); + } + + @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 +172,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"}); }