Copilot commented on code in PR #6719:
URL: https://github.com/apache/hive/pull/6719#discussion_r3840429410
##########
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> " + "<db> " + "<table")
+ .hasArgs(3)
+ .hasOptionalArgs(3)
Review Comment:
The help/usage arg name for -dedupColumns is malformed (missing "<"/">"
around catalog and the closing ">" for table), which will render confusing CLI
help output.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/MetaToolTaskDedupColumns.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+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();
+
+ final AtomicReference<String> 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;
+ }
+ if (progress.get() != null) {
+ System.out.println(progress.get());
+ }
+ }
+ });
+ daemon.setDaemon(true);
+ daemon.start();
+ }
+ MetaToolObjectStore.DedupColumnsResult result =
+ getObjectStore().dedupColumns(catalogFilter, dbFilter, tableFilter,
progress, isDryRun, isVerbose);
+ printSummary(result, isDryRun, isVerbose);
+ 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 successfully.");
Review Comment:
User-facing output has a double period: "Dry run of -dedupColumns..". This
looks like a typo in CLI output.
##########
standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java:
##########
@@ -45,9 +48,13 @@
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.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
Review Comment:
Unused static import assertNotEquals causes a Java compilation error. It
looks like it was added but never used in this test.
##########
standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/tools/metatool/TestHiveMetaToolCommandLine.java:
##########
@@ -132,23 +132,44 @@ 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, -serdePropKey, -tablePropKey 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]);
+ }
Review Comment:
-dedupColumns is configured with optional args, but the command-line parsing
test only covers the 3-argument form. Adding an assertion for the 0-argument
form (no filters) will protect the intended "scan all" usage from regressions.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/metatool/MetaToolTaskDedupColumns.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+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();
+
+ final AtomicReference<String> 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;
+ }
+ if (progress.get() != null) {
+ System.out.println(progress.get());
+ }
+ }
+ });
+ daemon.setDaemon(true);
+ daemon.start();
+ }
+ MetaToolObjectStore.DedupColumnsResult result =
+ getObjectStore().dedupColumns(catalogFilter, dbFilter, tableFilter,
progress, isDryRun, isVerbose);
+ printSummary(result, isDryRun, isVerbose);
+ 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());
+ }
Review Comment:
This command uses a background progress thread when -verbose is enabled, but
the thread is only stopped on the normal path. If dedupColumns() (or
printSummary) throws at runtime, the daemon thread will keep running until
process exit. Wrap the main body in a try/finally so the thread is always
stopped.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/ColumnDeduplicator.java:
##########
@@ -0,0 +1,372 @@
+/*
+ * 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<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 {
+ 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) throws MetaException {
+ 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);
+ } else {
+ Set<Long> candidateCdIds = new HashSet<>();
+ for (Map.Entry<PartitionSdInfo, Long> update : partSdUpdates) {
+ candidateCdIds.add(update.getKey().cdId);
+ }
+
result.addColumnDescriptorsRemoved(countRemovableColumnDescriptors(candidateCdIds));
+ }
+ }
+
+ 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, "database.catalogName", catalogFilter,
parameterVals);
+ }
+ if (!isEmpty(dbFilter)) {
+ appendPatternCondition(filter, "database.name", dbFilter, parameterVals);
+ }
+ if (!isEmpty(tableFilter)) {
+ appendPatternCondition(filter, "tableName", tableFilter, parameterVals);
+ }
+
+ Query query = filter.length() > 0 ?
+ pm.newQuery(MTable.class, filter.toString()) :
+ pm.newQuery(MTable.class);
+ 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) {
+ if (!isPartitionedTable(mTable.getId())) {
+ continue;
+ }
Review Comment:
findPartitionedTables() runs an isPartitionedTable() query per table (N+1).
On large metastores, this can make -dedupColumns extremely slow because it
scans all tables and then performs an extra query for each table to check for
any partitions. Consider querying only partitioned tables up front (e.g., via a
query on MPartition/table.id with distinct IDs or a join) to avoid per-table
lookups.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
