This is an automated email from the ASF dual-hosted git repository.
dengzhhu653 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new fc0db4e3fd1 HIVE-30100: Post-run integrity checks for metatool column
de-duplication (#6865)
fc0db4e3fd1 is described below
commit fc0db4e3fd1ea4734c0a37f1d3e0a00f33db7b2c
Author: dengzh <[email protected]>
AuthorDate: Fri Oct 16 08:32:27 2026 +0800
HIVE-30100: Post-run integrity checks for metatool column de-duplication
(#6865)
---
.../hive/metastore/tools/ColumnDeduplicator.java | 217 +++++++++++++++------
.../hive/metastore/tools/MetaToolObjectStore.java | 9 +
.../tools/metatool/HiveMetaToolCommandLine.java | 4 +-
.../tools/metatool/MetaToolTaskDedupColumns.java | 3 +
.../metastore/TestHMSColumnDescriptorReuse.java | 107 ++++++++++
5 files changed, 277 insertions(+), 63 deletions(-)
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
index b4854c80b3b..16cb5f29136 100644
---
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
@@ -20,18 +20,19 @@
package org.apache.hadoop.hive.metastore.tools;
import org.apache.commons.collections4.CollectionUtils;
+import org.apache.hadoop.hive.metastore.Batchable;
import org.apache.hadoop.hive.metastore.Deadline;
import org.apache.hadoop.hive.metastore.RawStore;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils;
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;
@@ -41,6 +42,7 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.Objects;
import java.util.stream.Collectors;
import static
org.apache.hadoop.hive.metastore.ObjectStore.appendPatternCondition;
@@ -53,14 +55,14 @@
* 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 {
+public 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,
+ public ColumnDeduplicator(RawStoreBundle bundle, AtomicReference<String>
progress,
boolean isDryRun, boolean isVerbose) {
this.store = bundle.getBaseStore();
this.pm = bundle.getPersistentManager();
@@ -77,10 +79,11 @@ MetaToolObjectStore.DedupColumnsResult run(String
catalogFilter, String dbFilter
for (int i = 0; i < tables.size() && result.getException() == null; i++) {
boolean committed = false;
TableInfo table = tables.get(i);
+ PostCommitValidation validation = null;
store.openTransaction();
try {
Deadline.checkTimeout();
- deduplicateTable(table, result);
+ validation = deduplicateTable(table, result);
committed = store.commitTransaction();
} catch (Exception ex) {
result.catchException(ex);
@@ -93,6 +96,16 @@ MetaToolObjectStore.DedupColumnsResult run(String
catalogFilter, String dbFilter
}
}
}
+
+ try {
+ if (committed && validation != null) {
+ validateTableAfterDedup(table, validation.schemaSnapshots());
+ result.incrementTablesValidated();
+ }
+ } catch (Exception e) {
+ result.catchException(new IllegalStateException("Failed to validate
the column deduplication for this table: " + table, e));
+ }
+
if (progress != null) {
progress.set(String.format(
"Finished %d tables in %d total tables, time taken: %d ms, columns
updated: %d, removed: %d",
@@ -106,24 +119,23 @@ MetaToolObjectStore.DedupColumnsResult run(String
catalogFilter, String dbFilter
return result;
}
- private void deduplicateTable(TableInfo table,
MetaToolObjectStore.DedupColumnsResult result) {
+ private PostCommitValidation deduplicateTable(TableInfo table,
MetaToolObjectStore.DedupColumnsResult result)
+ throws MetaException {
List<PartitionSdInfo> partitionSds =
loadPartitionStorageDescriptors(table.tableId);
if (partitionSds.isEmpty()) {
- return;
+ return null;
}
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<List<FieldSchema>, Set<Long>> groups = groupByColumnSchema(cdIds);
Map<Long, Long> cdRemap = new HashMap<>();
- for (List<Long> group : groups.values()) {
+ for (Set<Long> group : groups.values()) {
if (group.size() <= 1) {
continue;
}
- long canonicalCdId = pickCanonicalCdId(new HashSet<>(group),
table.tableCdId, partitionSds);
+ long canonicalCdId = pickCanonicalCdId(group, table.tableCdId,
partitionSds);
for (long cdId : group) {
if (cdId != canonicalCdId) {
cdRemap.put(cdId, canonicalCdId);
@@ -131,32 +143,29 @@ private void deduplicateTable(TableInfo table,
MetaToolObjectStore.DedupColumnsR
}
}
- if (cdRemap.isEmpty()) {
- return;
- }
-
List<Map.Entry<PartitionSdInfo, Long>> partSdUpdates =
buildPartitionUpdates(partitionSds, cdRemap);
if (partSdUpdates.isEmpty()) {
- return;
+ return null;
}
-
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));
+ result.addDetail(String.format("table %s.%s.%s: partition %d CD %d ->
%d",
+ table.catalogName, table.dbName, table.tableName, partSd.partId,
partSd.cdId, newCdId));
}
}
- if (!isDryRun) {
- applyTableChanges(partSdUpdates, result);
+ if (isDryRun) {
+ return null;
}
+ List<PartitionSchemaSnapshot> schemaSnapshots =
capturePartitionSchemas(partSdUpdates);
+ applySchemaChanges(partSdUpdates, result);
+ return new PostCommitValidation(schemaSnapshots);
}
- private void applyTableChanges(List<Map.Entry<PartitionSdInfo, Long>>
partSdUpdates,
+ private void applySchemaChanges(List<Map.Entry<PartitionSdInfo, Long>>
partSdUpdates,
MetaToolObjectStore.DedupColumnsResult result) {
Set<Long> replacedCdIds = new HashSet<>();
Map<Long, MColumnDescriptor> newCDs = new HashMap<>();
@@ -174,6 +183,9 @@ private void
applyTableChanges(List<Map.Entry<PartitionSdInfo, Long>> partSdUpda
private List<Map.Entry<PartitionSdInfo, Long>> buildPartitionUpdates(
List<PartitionSdInfo> partitionSds, Map<Long, Long> cdRemap) {
List<Map.Entry<PartitionSdInfo, Long>> updates = new ArrayList<>();
+ if (cdRemap.isEmpty()) {
+ return updates;
+ }
for (PartitionSdInfo partSd : partitionSds) {
Long newCdId = cdRemap.get(partSd.cdId);
if (newCdId != null && !newCdId.equals(partSd.cdId)) {
@@ -245,20 +257,24 @@ private List<TableInfo> findPartitionedTables(String
catalogFilter, String dbFil
return tables;
}
- private List<PartitionSdInfo> loadPartitionStorageDescriptors(long tableId) {
+ private List<PartitionSdInfo> loadPartitionStorageDescriptors(long tableId)
throws MetaException {
Query query = pm.newQuery(MPartition.class, "table.id == tblId");
query.declareParameters("long tblId");
- query.setResult("sd");
+ query.setResult("this.id, this.sd");
List<PartitionSdInfo> partitionSds = new ArrayList<>();
try {
- List<MStorageDescriptor> sds = (List<MStorageDescriptor>)
query.execute(tableId);
- if (sds == null) {
+ List<Object[]> results = (List<Object[]>) query.execute(tableId);
+ if (results == null) {
return partitionSds;
}
- pm.retrieveAll(sds);
- for (MStorageDescriptor sd : sds) {
- pm.retrieve(sd.getCD());
- partitionSds.add(new PartitionSdInfo(sd, sd.getCD().getId()));
+ for (Object[] row : results) {
+ long partId = MetastoreDirectSqlUtils.extractSqlLong(row[0]);
+ MStorageDescriptor sd = (MStorageDescriptor) row[1];
+ if (sd == null) {
+ continue;
+ }
+ pm.retrieve(sd);
+ partitionSds.add(new PartitionSdInfo(partId, sd, sd.getCD().getId()));
}
} finally {
query.closeAll();
@@ -266,24 +282,110 @@ private List<PartitionSdInfo>
loadPartitionStorageDescriptors(long tableId) {
return partitionSds;
}
- private Map<Long, List<FieldSchema>> loadColumnSchemas(Set<Long> cdIds) {
- Map<Long, List<FieldSchema>> result = new HashMap<>();
+ private List<PartitionSchemaSnapshot>
capturePartitionSchemas(List<Map.Entry<PartitionSdInfo, Long>> updates) {
+ List<PartitionSchemaSnapshot> snapshots = new ArrayList<>(updates.size());
+ for (Map.Entry<PartitionSdInfo, Long> update: updates) {
+ PartitionSdInfo psi = update.getKey();
+ snapshots.add(new PartitionSchemaSnapshot(
+ psi.partId,
+ update.getValue(),
+ convertToFieldSchemas(psi.sd.getCD().getCols())));
+ }
+ return snapshots;
+ }
+
+ public void validateTableAfterDedup(TableInfo table,
List<PartitionSchemaSnapshot> before)
+ throws MetaException {
+ List<PartitionSchemaSnapshot> partitions =
+ loadPartitionCdRefs(before.stream().map(b -> b.partitionId).toList());
+ if (partitions.size() != before.size()) {
+ throw new MetaException(String.format(
+ "Post-commit validation failed for %s: expected %d partitions, found
%d",
+ table, before.size(), partitions.size()));
+ }
+ Map<Long, PartitionSchemaSnapshot> beforeByPartitionId = new HashMap<>();
+ for (PartitionSchemaSnapshot snapshot : before) {
+ beforeByPartitionId.put(snapshot.partitionId(), snapshot);
+ }
+ for (PartitionSchemaSnapshot partition : partitions) {
+ PartitionSchemaSnapshot snapshot =
beforeByPartitionId.get(partition.partitionId());
+ if (snapshot == null) {
+ throw new MetaException(String.format(
+ "Post-commit validation failed for %s: unexpected partition id %d",
+ table, partition.partitionId()));
+ }
+ if (partition.cdId() != snapshot.cdId()) {
+ throw new MetaException(String.format(
+ "Post-commit validation failed for %s: partition %d still
references CD %d, expected canonical CD %d",
+ table, partition.partitionId(), partition.cdId(),
snapshot.cdId()));
+ }
+ List<FieldSchema> expected = snapshot.columns();
+ List<FieldSchema> actual = partition.columns();
+ if (!Objects.equals(expected, actual)) {
+ throw new MetaException(String.format(
+ "Post-commit validation failed for %s: column schema changed for
partition %d",
+ table, partition.partitionId()));
+ }
+ }
+ }
+
+ private List<PartitionSchemaSnapshot> loadPartitionCdRefs(List<Long>
partIds) throws MetaException {
+ if (partIds.isEmpty()) {
+ return new ArrayList<>();
+ }
+ int batchSize = MetastoreConf.getIntVar(store.getConf(),
+ MetastoreConf.ConfVars.RAWSTORE_PARTITION_BATCH_SIZE);
+ Batchable<Long, PartitionSchemaSnapshot> batch = new Batchable<>() {
+ @Override
+ public List<PartitionSchemaSnapshot> run(List<Long> batchPartIds) throws
Exception {
+ Query query = pm.newQuery(MPartition.class, "partIds.contains(id)");
+ query.declareParameters("java.util.Collection partIds");
+ query.setResult("this.id, this.sd");
+ addQueryAfterUse(query);
+ List<PartitionSchemaSnapshot> partitions = new ArrayList<>();
+ boolean success = false;
+ store.openTransaction();
+ try {
+ List<Object[]> results = (List<Object[]>)
query.execute(batchPartIds);
+ if (results != null) {
+ for (Object[] row : results) {
+ long partId = MetastoreDirectSqlUtils.extractSqlLong(row[0]);
+ MStorageDescriptor sd = (MStorageDescriptor) row[1];
+ if (sd == null) {
+ continue;
+ }
+ pm.retrieve(sd.getCD());
+ partitions.add(new PartitionSchemaSnapshot(partId,
sd.getCD().getId(),
+ convertToFieldSchemas(sd.getCD().getCols())));
+ }
+ }
+ success = store.commitTransaction();
+ } finally {
+ if (!success) {
+ store.rollbackTransaction();
+ }
+ }
+ return partitions;
+ }
+ };
+ try {
+ return Batchable.runBatched(batchSize, partIds, batch);
+ } finally {
+ batch.closeAllQueries();
+ }
+ }
+
+ private Map<List<FieldSchema>, Set<Long>> groupByColumnSchema(Set<Long>
cdIds) {
+ Map<List<FieldSchema>, Set<Long>> groups = 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()));
+ List<FieldSchema> schemas = convertToFieldSchemas(cd.getCols());
+ groups.computeIfAbsent(schemas, k -> new HashSet<>()).add(cdId);
+ schemas = null;
}
}
- 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;
}
@@ -316,28 +418,19 @@ private void removeConstraintsForCd(MColumnDescriptor cd)
{
}
}
- 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;
- }
-
+ public record TableInfo(long tableId, long tableCdId, String catalogName,
String dbName, String tableName) {
@Override
public String toString() {
- return catalogName + "." + dbName + "." + tableName;
- }
+ return catalogName + "." + dbName + "." + tableName;
+ }
}
- private record PartitionSdInfo(MStorageDescriptor sd, long cdId) {
+ private record PartitionSdInfo(long partId, MStorageDescriptor sd, long
cdId) {
+ }
+
+ public record PartitionSchemaSnapshot(long partitionId, long cdId,
List<FieldSchema> columns) {
+ }
+ private record PostCommitValidation(List<PartitionSchemaSnapshot>
schemaSnapshots) {
}
}
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 67a97243bd5..5c5a6124270 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
@@ -975,6 +975,7 @@ public static final class DedupColumnsResult {
private int tablesWithDuplicates;
private int storageDescriptorsUpdated;
private int columnDescriptorsRemoved;
+ private int tablesValidated;
private final List<String> details = new ArrayList<>();
private Exception exception;
DedupColumnsResult(int tablesScanned) {
@@ -1008,6 +1009,14 @@ void addColumnDescriptorsRemoved(int count) {
this.columnDescriptorsRemoved += count;
}
+ public int getTablesValidated() {
+ return tablesValidated;
+ }
+
+ void incrementTablesValidated() {
+ this.tablesValidated++;
+ }
+
public List<String> getDetails() {
return details;
}
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 a1c9db7a327..5bec7e3bd97 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
@@ -93,7 +93,9 @@ class HiveMetaToolCommandLine {
.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.")
+ + "-verbose to print per-partition details, and -timeout to set a
timeout in seconds. "
+ + "Live runs validate committed state (partition column schemas
unchanged, canonical CD IDs applied) "
+ + "and print a post-commit validation summary.")
.create("dedupColumns");
private static final Option VERBOSE = OptionBuilder
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
index 06994e5d493..f86c7b9480c 100644
---
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
@@ -110,6 +110,9 @@ private void
printSummary(MetaToolObjectStore.DedupColumnsResult result, boolean
+ result.getStorageDescriptorsUpdated());
if (!isDryRun) {
System.out.println("Column descriptors removed: " +
result.getColumnDescriptorsRemoved());
+ if (result.getTablesValidated() > 0) {
+ System.out.println("Post-commit validation passed for " +
result.getTablesValidated() + " table(s).");
+ }
}
if (isVerbose) {
for (String detail : result.getDetails()) {
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 483c15fccba..0f63bda5f80 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
@@ -34,6 +34,7 @@
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.ColumnDeduplicator;
import org.apache.hadoop.hive.metastore.tools.MetaToolObjectStore;
import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
import org.junit.After;
@@ -42,6 +43,7 @@
import org.junit.experimental.categories.Category;
import java.sql.Connection;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Arrays;
import java.util.Collections;
@@ -53,10 +55,12 @@
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.assertThrows;
import static org.junit.Assert.assertTrue;
@Category(MetastoreUnitTest.class)
public class TestHMSColumnDescriptorReuse {
+
private ObjectStore objectStore = null;
private Configuration conf;
@@ -196,6 +200,7 @@ public void testDeduplicateColumnDescriptorsTool() throws
Exception {
MetaToolObjectStore.DedupColumnsResult result =
metaToolStore.dedupColumns(null, "default", "person", progress,
false, false);
assertEquals(0, result.getTablesWithDuplicates());
+ assertEquals(0, result.getTablesValidated());
FieldSchema lname = new FieldSchema("lname",
ColumnType.STRING_TYPE_NAME, "");
Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
@@ -209,6 +214,7 @@ public void testDeduplicateColumnDescriptorsTool() throws
Exception {
result = metaToolStore.dedupColumns(null, "default", "person", progress,
false, false);
assertTrue(result.getStorageDescriptorsUpdated() > 0);
+ assertEquals(1, result.getTablesValidated());
assertEquals(2, countColumnDescriptors());
assertNotNull(progress.get());
@@ -230,6 +236,107 @@ public void testDeduplicateColumnDescriptorsTool() throws
Exception {
}
}
+ @Test
+ public void testValidateTableAfterDedupReloadsPartitionById() throws
Exception {
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.PARTITION_REUSE_COLUMN_DESCRIPTORS, false);
+ createTableWithDuplicateColumnDescriptors();
+ long partId = lookupPartitionId("country=Germany");
+ List<FieldSchema> cols = Arrays.asList(
+ new FieldSchema("id", ColumnType.STRING_TYPE_NAME, ""),
+ new FieldSchema("fname", ColumnType.STRING_TYPE_NAME, ""));
+
+ validate(Collections.singletonList(new
ColumnDeduplicator.PartitionSchemaSnapshot(
+ partId, lookupPartitionCdId(partId), cols)));
+ }
+
+ @Test
+ public void testValidateTableAfterDedupDetectsCdRemapFailure() throws
Exception {
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.PARTITION_REUSE_COLUMN_DESCRIPTORS, false);
+ createTableWithDuplicateColumnDescriptors();
+ long partId = lookupPartitionId("country=Germany");
+ List<FieldSchema> cols = Arrays.asList(
+ new FieldSchema("id", ColumnType.STRING_TYPE_NAME, ""),
+ new FieldSchema("fname", ColumnType.STRING_TYPE_NAME, ""));
+
+ List<ColumnDeduplicator.PartitionSchemaSnapshot> snapshots =
Collections.singletonList(
+ new ColumnDeduplicator.PartitionSchemaSnapshot(partId, -1L, cols));
+
+ MetaException ex = assertThrows(MetaException.class, () ->
validate(snapshots));
+ assertTrue(ex.getMessage().contains("expected canonical CD"));
+ }
+
+ @Test
+ public void testValidateTableAfterDedupDetectsSchemaChange() throws
Exception {
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.PARTITION_REUSE_COLUMN_DESCRIPTORS, false);
+ createTableWithDuplicateColumnDescriptors();
+ long partId = lookupPartitionId("country=Germany");
+
+ List<ColumnDeduplicator.PartitionSchemaSnapshot> snapshots =
Collections.singletonList(
+ new ColumnDeduplicator.PartitionSchemaSnapshot(partId,
lookupPartitionCdId(partId),
+ Collections.singletonList(new FieldSchema("unexpected",
ColumnType.STRING_TYPE_NAME, ""))));
+
+ MetaException ex = assertThrows(MetaException.class, () ->
validate(snapshots));
+ assertTrue(ex.getMessage().contains("column schema changed for partition "
+ partId));
+ }
+
+ private int createTableWithDuplicateColumnDescriptors() throws Exception {
+ FieldSchema id = new FieldSchema("id", ColumnType.STRING_TYPE_NAME, "");
+ FieldSchema fname = new FieldSchema("fname", ColumnType.STRING_TYPE_NAME,
"");
+ FieldSchema country = new FieldSchema("country",
ColumnType.STRING_TYPE_NAME, "");
+
+ Table tbl1 = newTable(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+ objectStore.addPartition(newPart(tbl1, "Greece"));
+
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+ objectStore.addPartition(newPart(tbl1, "Germany"));
+ objectStore.addPartition(newPart(tbl1, "Belgium"));
+ objectStore.addPartition(newPart(tbl2, "England"));
+
+ return countColumnDescriptors();
+ }
+
+ private long lookupPartitionId(String partName) throws Exception {
+ try (Connection c = TestTxnDbUtil.getConnection(conf);
+ PreparedStatement ps = c.prepareStatement(
+ "SELECT \"PART_ID\" FROM \"PARTITIONS\" WHERE \"PART_NAME\" = ?"))
{
+ ps.setString(1, partName);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (!rs.next()) {
+ throw new IllegalStateException("Partition not found: " + partName);
+ }
+ return rs.getLong(1);
+ }
+ }
+ }
+
+ private long lookupPartitionCdId(long partId) throws Exception {
+ try (Connection c = TestTxnDbUtil.getConnection(conf);
+ PreparedStatement ps = c.prepareStatement(
+ "SELECT s.\"CD_ID\" FROM \"PARTITIONS\" p JOIN \"SDS\" s ON
p.\"SD_ID\" = s.\"SD_ID\" "
+ + "WHERE p.\"PART_ID\" = ?")) {
+ ps.setLong(1, partId);
+ try (ResultSet rs = ps.executeQuery()) {
+ if (!rs.next()) {
+ throw new IllegalStateException("Partition CD not found: " + partId);
+ }
+ return rs.getLong(1);
+ }
+ }
+ }
+
+ private void validate(List<ColumnDeduplicator.PartitionSchemaSnapshot>
snapshots)
+ throws Exception {
+ ColumnDeduplicator deduplicator =
+ new ColumnDeduplicator(objectStore.createRawStoreBundle(), null,
false, false);
+ ColumnDeduplicator.TableInfo tableInfo = new
ColumnDeduplicator.TableInfo(0L, 0L, "hive", "default", "person");
+ deduplicator.validateTableAfterDedup(tableInfo, snapshots);
+ }
+
private int countColumnDescriptors() {
try(Connection c = TestTxnDbUtil.getConnection(conf)){
try(ResultSet rs = c.prepareStatement("SELECT COUNT(*) FROM
\"CDS\"").executeQuery()) {
