saihemanth-cloudera commented on code in PR #6719: URL: https://github.com/apache/hive/pull/6719#discussion_r4020725229
########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/tools/ColumnDeduplicator.java: ########## @@ -0,0 +1,344 @@ +/* + * 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) + throws MetaException { + 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++) { + Deadline.checkTimeout(); + 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<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); Review Comment: This loads every partition SD for the table at once. Since large partitioned tables are likely the target here, so can we batch this? -- 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]
