From 149dd14cab456a04155a114357f0216ef670b006 Mon Sep 17 00:00:00 2001 From: Venu Reddy Date: Wed, 5 Aug 2026 16:13:53 +0530 Subject: [PATCH] HIVE-29512: Clean compaction related HMS metadata tables for iceberg --- .../iceberg/mr/hive/HiveIcebergMetaHook.java | 14 ++ .../positive/iceberg_show_compactions.q | 36 +++ .../llap/iceberg_show_compactions.q.out | 211 ++++++++++++++++++ .../resources/testconfiguration.properties | 3 +- .../hive/metastore/AcidEventListener.java | 64 +++--- .../hadoop/hive/metastore/txn/TxnHandler.java | 22 +- .../hadoop/hive/metastore/txn/TxnStore.java | 24 +- .../functions/CleanupRecordsFunction.java | 81 ++++--- .../metastore/txn/ThrowingTxnHandler.java | 9 +- 9 files changed, 383 insertions(+), 81 deletions(-) create mode 100644 iceberg/iceberg-handler/src/test/queries/positive/iceberg_show_compactions.q create mode 100644 iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_show_compactions.q.out diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java index 92070d64528d..d53b914d5d9b 100644 --- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java +++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java @@ -51,6 +51,8 @@ import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants; import org.apache.hadoop.hive.metastore.client.ThriftHiveMetaStoreClient; import org.apache.hadoop.hive.metastore.partition.spec.PartitionSpecProxy; +import org.apache.hadoop.hive.metastore.txn.TxnStore; +import org.apache.hadoop.hive.metastore.txn.TxnUtils; import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils; import org.apache.hadoop.hive.ql.QueryState; import org.apache.hadoop.hive.ql.ddl.misc.sortoder.SortFieldDesc; @@ -989,6 +991,8 @@ public void preDropPartitions(org.apache.hadoop.hive.metastore.api.Table hmsTabl deleteFiles.deleteFromRowFilter(partitionSetFilter); deleteFiles.commit(); + cleanupCompactionRecords(hmsTable, + partitionList.stream().map(pSpec::partitionToPath).distinct().toList()); } catch (IOException e) { throw new MetaException(String.format("Error while fetching the partitions due to: %s", e)); } @@ -1008,10 +1012,20 @@ public void preDropPartitions(org.apache.hadoop.hive.metastore.api.Table hmsTabl preDropPartitions(hmsTable, context, partExprs); } else if (partsSpec.isSetNames()) { preTruncateTable(hmsTable, context, partsSpec.getNames()); + cleanupCompactionRecords(hmsTable, partsSpec.getNames()); } context.putToProperties(ThriftHiveMetaStoreClient.SKIP_DROP_PARTITION, "true"); } + private void cleanupCompactionRecords(org.apache.hadoop.hive.metastore.api.Table hmsTable, + List partitionNames) throws MetaException { + if (CollectionUtils.isEmpty(partitionNames)) { + return; + } + TxnStore txnHandler = TxnUtils.getTxnStore(conf); + txnHandler.cleanupCompactionRecords(hmsTable, partitionNames); + } + private static void validatePartitionSpec(SearchArgument sarg, PartitionSpec partitionSpec) { for (PredicateLeaf leaf : sarg.getLeaves()) { TransformSpec transformSpec = TransformSpec.fromStringWithColumnName(leaf.getColumnName()); diff --git a/iceberg/iceberg-handler/src/test/queries/positive/iceberg_show_compactions.q b/iceberg/iceberg-handler/src/test/queries/positive/iceberg_show_compactions.q new file mode 100644 index 000000000000..0c5510ee54a9 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/queries/positive/iceberg_show_compactions.q @@ -0,0 +1,36 @@ +-- Mask show compactions fields that change across runs +--! qt:replace:/^[0-9]/#Masked#/ +--! qt:replace:/(MAJOR\s+succeeded\s+)[a-zA-Z0-9\-\.\s+]+(\s+manual)/$1#Masked#$2/ + +-- Test compaction entry is removed upon drop non-partitioned table +create table ice_t1 (i int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500'); +insert into ice_t1 values(1),(2); +insert into ice_t1 values(3),(4); +alter table ice_t1 compact 'major' and wait; +show compactions ice_t1; +drop table ice_t1; +show compactions ice_t1; + +-- Test compaction entry is updated upon rename non-partitioned table +create table ice_t2 (i int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500'); +insert into ice_t2 values(1),(2); +insert into ice_t2 values(3),(4); +alter table ice_t2 compact 'major' and wait; +show compactions ice_t2; +alter table ice_t2 RENAME to ice_t2_new; +show compactions ice_t2_new; +drop table ice_t2_new; + +-- Test compaction entries are removed upon drop partition and drop table for a partitioned table +create table ice_part (i int) partitioned by (j int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500'); +insert into ice_part values (1,1); +insert into ice_part values (2,1); +alter table ice_part partition (j=1) compact 'major' and wait; +insert into ice_part values (1,2); +insert into ice_part values (2,2); +alter table ice_part partition (j=2) compact 'major' and wait; +show compactions ice_part; +alter table ice_part drop partition (j=1); +show compactions ice_part; +drop table ice_part; +show compactions ice_part; \ No newline at end of file diff --git a/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_show_compactions.q.out b/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_show_compactions.q.out new file mode 100644 index 000000000000..848a820823a9 --- /dev/null +++ b/iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_show_compactions.q.out @@ -0,0 +1,211 @@ +PREHOOK: query: create table ice_t1 (i int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_t1 +POSTHOOK: query: create table ice_t1 (i int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_t1 +PREHOOK: query: insert into ice_t1 values(1),(2) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_t1 +POSTHOOK: query: insert into ice_t1 values(1),(2) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_t1 +PREHOOK: query: insert into ice_t1 values(3),(4) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_t1 +POSTHOOK: query: insert into ice_t1 values(3),(4) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_t1 +PREHOOK: query: alter table ice_t1 compact 'major' and wait +PREHOOK: type: ALTERTABLE_COMPACT +PREHOOK: Input: default@ice_t1 +PREHOOK: Output: default@ice_t1 +POSTHOOK: query: alter table ice_t1 compact 'major' and wait +POSTHOOK: type: ALTERTABLE_COMPACT +POSTHOOK: Input: default@ice_t1 +POSTHOOK: Output: default@ice_t1 +PREHOOK: query: show compactions ice_t1 +PREHOOK: type: SHOW COMPACTIONS +POSTHOOK: query: show compactions ice_t1 +POSTHOOK: type: SHOW COMPACTIONS +CompactionId Database Table Partition Type State Worker host Worker Enqueue Time Start Time Duration(ms) HadoopJobId Error message Initiator host Initiator Pool name TxnId Next TxnId Commit Time Highest WriteId +#Masked# default ice_t1 --- MAJOR succeeded #Masked# manual default 0 0 0 --- +PREHOOK: query: drop table ice_t1 +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@ice_t1 +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_t1 +POSTHOOK: query: drop table ice_t1 +POSTHOOK: type: DROPTABLE +POSTHOOK: Input: default@ice_t1 +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_t1 +PREHOOK: query: show compactions ice_t1 +PREHOOK: type: SHOW COMPACTIONS +POSTHOOK: query: show compactions ice_t1 +POSTHOOK: type: SHOW COMPACTIONS +CompactionId Database Table Partition Type State Worker host Worker Enqueue Time Start Time Duration(ms) HadoopJobId Error message Initiator host Initiator Pool name TxnId Next TxnId Commit Time Highest WriteId +PREHOOK: query: create table ice_t2 (i int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_t2 +POSTHOOK: query: create table ice_t2 (i int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_t2 +PREHOOK: query: insert into ice_t2 values(1),(2) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_t2 +POSTHOOK: query: insert into ice_t2 values(1),(2) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_t2 +PREHOOK: query: insert into ice_t2 values(3),(4) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_t2 +POSTHOOK: query: insert into ice_t2 values(3),(4) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_t2 +PREHOOK: query: alter table ice_t2 compact 'major' and wait +PREHOOK: type: ALTERTABLE_COMPACT +PREHOOK: Input: default@ice_t2 +PREHOOK: Output: default@ice_t2 +POSTHOOK: query: alter table ice_t2 compact 'major' and wait +POSTHOOK: type: ALTERTABLE_COMPACT +POSTHOOK: Input: default@ice_t2 +POSTHOOK: Output: default@ice_t2 +PREHOOK: query: show compactions ice_t2 +PREHOOK: type: SHOW COMPACTIONS +POSTHOOK: query: show compactions ice_t2 +POSTHOOK: type: SHOW COMPACTIONS +CompactionId Database Table Partition Type State Worker host Worker Enqueue Time Start Time Duration(ms) HadoopJobId Error message Initiator host Initiator Pool name TxnId Next TxnId Commit Time Highest WriteId +#Masked# default ice_t2 --- MAJOR succeeded #Masked# manual default 0 0 0 --- +PREHOOK: query: alter table ice_t2 RENAME to ice_t2_new +PREHOOK: type: ALTERTABLE_RENAME +PREHOOK: Input: default@ice_t2 +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_t2 +PREHOOK: Output: default@ice_t2_new +POSTHOOK: query: alter table ice_t2 RENAME to ice_t2_new +POSTHOOK: type: ALTERTABLE_RENAME +POSTHOOK: Input: default@ice_t2 +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_t2 +POSTHOOK: Output: default@ice_t2_new +PREHOOK: query: show compactions ice_t2_new +PREHOOK: type: SHOW COMPACTIONS +POSTHOOK: query: show compactions ice_t2_new +POSTHOOK: type: SHOW COMPACTIONS +CompactionId Database Table Partition Type State Worker host Worker Enqueue Time Start Time Duration(ms) HadoopJobId Error message Initiator host Initiator Pool name TxnId Next TxnId Commit Time Highest WriteId +#Masked# default ice_t2_new --- MAJOR succeeded #Masked# manual default 0 0 0 --- +PREHOOK: query: drop table ice_t2_new +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@ice_t2_new +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_t2_new +POSTHOOK: query: drop table ice_t2_new +POSTHOOK: type: DROPTABLE +POSTHOOK: Input: default@ice_t2_new +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_t2_new +PREHOOK: query: create table ice_part (i int) partitioned by (j int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500') +PREHOOK: type: CREATETABLE +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_part +POSTHOOK: query: create table ice_part (i int) partitioned by (j int) stored by iceberg tblproperties ('compactor.threshold.target.size'='1500') +POSTHOOK: type: CREATETABLE +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_part +PREHOOK: query: insert into ice_part values (1,1) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_part +POSTHOOK: query: insert into ice_part values (1,1) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_part +PREHOOK: query: insert into ice_part values (2,1) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_part +POSTHOOK: query: insert into ice_part values (2,1) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_part +PREHOOK: query: alter table ice_part partition (j=1) compact 'major' and wait +PREHOOK: type: ALTERTABLE_COMPACT +PREHOOK: Input: default@ice_part +PREHOOK: Output: default@ice_part@j=1 +POSTHOOK: query: alter table ice_part partition (j=1) compact 'major' and wait +POSTHOOK: type: ALTERTABLE_COMPACT +POSTHOOK: Input: default@ice_part +POSTHOOK: Output: default@ice_part@j=1 +PREHOOK: query: insert into ice_part values (1,2) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_part +POSTHOOK: query: insert into ice_part values (1,2) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_part +PREHOOK: query: insert into ice_part values (2,2) +PREHOOK: type: QUERY +PREHOOK: Input: _dummy_database@_dummy_table +PREHOOK: Output: default@ice_part +POSTHOOK: query: insert into ice_part values (2,2) +POSTHOOK: type: QUERY +POSTHOOK: Input: _dummy_database@_dummy_table +POSTHOOK: Output: default@ice_part +PREHOOK: query: alter table ice_part partition (j=2) compact 'major' and wait +PREHOOK: type: ALTERTABLE_COMPACT +PREHOOK: Input: default@ice_part +PREHOOK: Output: default@ice_part@j=2 +POSTHOOK: query: alter table ice_part partition (j=2) compact 'major' and wait +POSTHOOK: type: ALTERTABLE_COMPACT +POSTHOOK: Input: default@ice_part +POSTHOOK: Output: default@ice_part@j=2 +PREHOOK: query: show compactions ice_part +PREHOOK: type: SHOW COMPACTIONS +POSTHOOK: query: show compactions ice_part +POSTHOOK: type: SHOW COMPACTIONS +CompactionId Database Table Partition Type State Worker host Worker Enqueue Time Start Time Duration(ms) HadoopJobId Error message Initiator host Initiator Pool name TxnId Next TxnId Commit Time Highest WriteId +#Masked# default ice_part j=1 MAJOR succeeded #Masked# manual default 0 0 0 --- +#Masked# default ice_part j=2 MAJOR succeeded #Masked# manual default 0 0 0 --- +PREHOOK: query: alter table ice_part drop partition (j=1) +PREHOOK: type: ALTERTABLE_DROPPARTS +PREHOOK: Input: default@ice_part +PREHOOK: Output: default@ice_part@j=1 +POSTHOOK: query: alter table ice_part drop partition (j=1) +POSTHOOK: type: ALTERTABLE_DROPPARTS +POSTHOOK: Input: default@ice_part +POSTHOOK: Output: default@ice_part@j=1 +PREHOOK: query: show compactions ice_part +PREHOOK: type: SHOW COMPACTIONS +POSTHOOK: query: show compactions ice_part +POSTHOOK: type: SHOW COMPACTIONS +CompactionId Database Table Partition Type State Worker host Worker Enqueue Time Start Time Duration(ms) HadoopJobId Error message Initiator host Initiator Pool name TxnId Next TxnId Commit Time Highest WriteId +#Masked# default ice_part j=2 MAJOR succeeded #Masked# manual default 0 0 0 --- +PREHOOK: query: drop table ice_part +PREHOOK: type: DROPTABLE +PREHOOK: Input: default@ice_part +PREHOOK: Output: database:default +PREHOOK: Output: default@ice_part +POSTHOOK: query: drop table ice_part +POSTHOOK: type: DROPTABLE +POSTHOOK: Input: default@ice_part +POSTHOOK: Output: database:default +POSTHOOK: Output: default@ice_part +PREHOOK: query: show compactions ice_part +PREHOOK: type: SHOW COMPACTIONS +POSTHOOK: query: show compactions ice_part +POSTHOOK: type: SHOW COMPACTIONS +CompactionId Database Table Partition Type State Worker host Worker Enqueue Time Start Time Duration(ms) HadoopJobId Error message Initiator host Initiator Pool name TxnId Next TxnId Commit Time Highest WriteId diff --git a/itests/src/test/resources/testconfiguration.properties b/itests/src/test/resources/testconfiguration.properties index efdbebcf9b54..8b54a43c5174 100644 --- a/itests/src/test/resources/testconfiguration.properties +++ b/itests/src/test/resources/testconfiguration.properties @@ -436,7 +436,8 @@ iceberg.llap.query.compactor.files=\ iceberg_minor_compaction_bucket.q,\ iceberg_minor_compaction_partition_evolution.q,\ iceberg_minor_compaction_unpartitioned.q,\ - iceberg_row_lineage_compactions.q + iceberg_row_lineage_compactions.q,\ + iceberg_show_compactions.q iceberg.llap.query.rest.hms.files=\ iceberg_rest_catalog_hms.q diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/AcidEventListener.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/AcidEventListener.java index d8f51ad3ce50..c5495b790437 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/AcidEventListener.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/AcidEventListener.java @@ -42,6 +42,7 @@ import org.apache.hadoop.hive.metastore.txn.TxnUtils; import java.io.IOException; +import java.util.ArrayList; import java.util.Iterator; import java.util.List; import java.util.Optional; @@ -49,11 +50,13 @@ import static org.apache.hadoop.hive.metastore.HiveMetaStoreClient.RENAME_PARTITION_MAKE_COPY; import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.getWriteId; import static org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils.isMustPurge; +import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.isIcebergTable; import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.throwMetaException; /** - * It handles cleanup of dropped partition/table/database in ACID related metastore tables + * It handles cleanup of dropped partition/table/database in ACID related metastore tables, + * and compaction related metastore tables cleanup for Iceberg tables. */ public class AcidEventListener extends TransactionalMetaStoreEventListener { @@ -103,43 +106,51 @@ public void onDropTable(DropTableEvent tableEvent) throws MetaException { } } } + } else if (isIcebergTable(table.getParameters())) { + txnHandler = getTxnHandler(); + txnHandler.cleanupRecords(HiveObjectType.TABLE, null, table, null); } } @Override public void onDropPartition(DropPartitionEvent partitionEvent) throws MetaException { + if (!TxnUtils.isTransactionalTable(partitionEvent.getTable())) { + return; + } Table table = partitionEvent.getTable(); EnvironmentContext context = partitionEvent.getEnvironmentContext(); - if (TxnUtils.isTransactionalTable(table)) { - txnHandler = getTxnHandler(); - txnHandler.cleanupRecords(HiveObjectType.PARTITION, null, table, partitionEvent.getPartitionIterator()); + txnHandler = getTxnHandler(); + List partCols = table.getPartitionKeys(); + List partNames = new ArrayList<>(); + Iterator partitionIterator = partitionEvent.getPartitionIterator(); + while (partitionIterator.hasNext()) { + Partition partition = partitionIterator.next(); + partNames.add(Warehouse.makePartName(partCols, partition.getValues())); + } + txnHandler.cleanupRecords(HiveObjectType.PARTITION, null, table, partNames.iterator()); - if (!partitionEvent.getDeleteData()) { - long currentTxn = getTxnId(context); - - if (currentTxn > 0) { - long writeId = getWriteId(context); - try { - CompactionRequest rqst = new CompactionRequest( - table.getDbName(), table.getTableName(), CompactionType.MAJOR); - rqst.setRunas(TxnUtils.findUserToRunAs(table.getSd().getLocation(), table, conf)); - rqst.putToProperties("ifPurge", Boolean.toString(isMustPurge(context, table))); + if (!partitionEvent.getDeleteData()) { + long currentTxn = getTxnId(context); - Iterator partitionIterator = partitionEvent.getPartitionIterator(); - while (partitionIterator.hasNext()) { - Partition p = partitionIterator.next(); + if (currentTxn > 0) { + long writeId = getWriteId(context); + try { + CompactionRequest rqst = new CompactionRequest( + table.getDbName(), table.getTableName(), CompactionType.MAJOR); + rqst.setRunas(TxnUtils.findUserToRunAs(table.getSd().getLocation(), table, conf)); + rqst.putToProperties("ifPurge", Boolean.toString(isMustPurge(context, table))); - List partCols = partitionEvent.getTable().getPartitionKeys(); // partition columns - List partVals = p.getValues(); - rqst.setPartitionname(Warehouse.makePartName(partCols, partVals)); - rqst.putToProperties("location", p.getSd().getLocation()); + partitionIterator = partitionEvent.getPartitionIterator(); + while (partitionIterator.hasNext()) { + Partition p = partitionIterator.next(); + rqst.setPartitionname(Warehouse.makePartName(partCols, p.getValues())); + rqst.putToProperties("location", p.getSd().getLocation()); - txnHandler.submitForCleanup(rqst, writeId, currentTxn); - } - } catch (InterruptedException | IOException e) { - throwMetaException(e); + txnHandler.submitForCleanup(rqst, writeId, currentTxn); } + } catch (InterruptedException | IOException e) { + throwMetaException(e); } } } @@ -147,7 +158,8 @@ public void onDropPartition(DropPartitionEvent partitionEvent) throws MetaExcep @Override public void onAlterTable(AlterTableEvent tableEvent) throws MetaException { - if (!TxnUtils.isTransactionalTable(tableEvent.getNewTable())) { + if (!TxnUtils.isTransactionalTable(tableEvent.getNewTable()) && + !isIcebergTable(tableEvent.getNewTable().getParameters())) { return; } Table oldTable = tableEvent.getOldTable(); diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java index a847e01aca4d..04aa6b69c034 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java @@ -64,7 +64,6 @@ import org.apache.hadoop.hive.metastore.api.NoSuchTxnException; import org.apache.hadoop.hive.metastore.api.OpenTxnRequest; import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse; -import org.apache.hadoop.hive.metastore.api.Partition; import org.apache.hadoop.hive.metastore.api.ReplTblWriteIdStateRequest; import org.apache.hadoop.hive.metastore.api.ReplayedTxnsForPolicyResult; import org.apache.hadoop.hive.metastore.api.SeedTableWriteIdsRequest; @@ -985,18 +984,27 @@ public void addDynamicPartitions(AddDynamicPartitions rqst) throws NoSuchTxnExce */ @Override public void cleanupRecords(HiveObjectType type, Database db, Table table, - Iterator partitionIterator, boolean keepTxnToWriteIdMetaData) throws MetaException { - new CleanupRecordsFunction(type, db, table, partitionIterator, getDefaultCatalog(conf), keepTxnToWriteIdMetaData, null) - .execute(jdbcResource); + Iterator partNamesIterator, boolean keepTxnToWriteIdMetaData) throws MetaException { + new CleanupRecordsFunction(type, db, table, partNamesIterator, getDefaultCatalog(conf), + keepTxnToWriteIdMetaData, null).execute(jdbcResource); } @Override public void cleanupRecords(HiveObjectType type, Database db, Table table, - Iterator partitionIterator, long txnId) throws MetaException { - new CleanupRecordsFunction(type, db, table, partitionIterator, getDefaultCatalog(conf), false, txnId) + Iterator partNamesIterator, long txnId) throws MetaException { + new CleanupRecordsFunction(type, db, table, partNamesIterator, getDefaultCatalog(conf), false, txnId) .execute(jdbcResource); } - + + @Override + public void cleanupCompactionRecords(Table table, List partitionNames) throws MetaException { + if (CollectionUtils.isEmpty(partitionNames)) { + return; + } + new CleanupRecordsFunction(HiveObjectType.PARTITION, null, table, partitionNames.iterator(), + getDefaultCatalog(conf), false, null).execute(jdbcResource); + } + /** * Catalog hasn't been added to transactional tables yet, so it's passed in but not used. */ diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java index 22e6c279fc84..ff63a16f52dd 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnStore.java @@ -59,7 +59,6 @@ import org.apache.hadoop.hive.metastore.api.NoSuchTxnException; import org.apache.hadoop.hive.metastore.api.OpenTxnRequest; import org.apache.hadoop.hive.metastore.api.OpenTxnsResponse; -import org.apache.hadoop.hive.metastore.api.Partition; import org.apache.hadoop.hive.metastore.api.ReplTblWriteIdStateRequest; import org.apache.hadoop.hive.metastore.api.ReplayedTxnsForPolicyResult; import org.apache.hadoop.hive.metastore.api.SeedTableWriteIdsRequest; @@ -529,28 +528,37 @@ void addDynamicPartitions(AddDynamicPartitions rqst) * @param type Hive object type * @param db database object * @param table table object - * @param partitionIterator partition iterator + * @param partNamesIterator partition name iterator * @throws MetaException */ @SqlRetry @Transactional(POOL_TX) @RetrySemantics.Idempotent - default void cleanupRecords(HiveObjectType type, Database db, Table table, - Iterator partitionIterator) throws MetaException { - cleanupRecords(type, db, table, partitionIterator, false); + default void cleanupRecords(HiveObjectType type, Database db, Table table, + Iterator partNamesIterator) throws MetaException { + cleanupRecords(type, db, table, partNamesIterator, false); } @SqlRetry @Transactional(POOL_TX) @RetrySemantics.Idempotent - void cleanupRecords(HiveObjectType type, Database db, Table table, - Iterator partitionIterator, boolean keepTxnToWriteIdMetaData) throws MetaException; + void cleanupRecords(HiveObjectType type, Database db, Table table, + Iterator partNamesIterator, boolean keepTxnToWriteIdMetaData) throws MetaException; @SqlRetry @Transactional(POOL_TX) @RetrySemantics.Idempotent void cleanupRecords(HiveObjectType type, Database db, Table table, - Iterator partitionIterator, long txnId) throws MetaException; + Iterator partNamesIterator, long txnId) throws MetaException; + + /** + * Clean compaction related records for the given table partitions. + * Used for non-transactional tables (e.g. Iceberg) where partition lifecycle is managed outside HMS. + */ + @SqlRetry + @Transactional(POOL_TX) + @RetrySemantics.Idempotent + void cleanupCompactionRecords(Table table, List partitionNames) throws MetaException; @SqlRetry @Transactional(POOL_TX) diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/CleanupRecordsFunction.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/CleanupRecordsFunction.java index 5d0d6f584f1a..69d2560ed397 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/CleanupRecordsFunction.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/CleanupRecordsFunction.java @@ -18,14 +18,12 @@ package org.apache.hadoop.hive.metastore.txn.jdbc.functions; import org.apache.commons.lang3.StringUtils; -import org.apache.hadoop.hive.metastore.Warehouse; import org.apache.hadoop.hive.metastore.api.Database; -import org.apache.hadoop.hive.metastore.api.FieldSchema; import org.apache.hadoop.hive.metastore.api.HiveObjectType; import org.apache.hadoop.hive.metastore.api.MetaException; -import org.apache.hadoop.hive.metastore.api.Partition; import org.apache.hadoop.hive.metastore.api.Table; import org.apache.hadoop.hive.metastore.txn.TxnStore; +import org.apache.hadoop.hive.metastore.txn.TxnUtils; import org.apache.hadoop.hive.metastore.txn.jdbc.MultiDataSourceJdbcResource; import org.apache.hadoop.hive.metastore.txn.jdbc.TransactionalFunction; import org.slf4j.Logger; @@ -35,6 +33,7 @@ import java.sql.Types; import java.util.ArrayList; +import java.util.Arrays; import java.util.EnumSet; import java.util.Iterator; import java.util.LinkedHashMap; @@ -45,12 +44,12 @@ public class CleanupRecordsFunction implements TransactionalFunction { private static final Logger LOG = LoggerFactory.getLogger(CleanupRecordsFunction.class); - private static final EnumSet HIVE_OBJECT_TYPES = + private static final EnumSet HIVE_OBJECT_TYPES = EnumSet.of(HiveObjectType.DATABASE, HiveObjectType.TABLE, HiveObjectType.PARTITION); @SuppressWarnings("squid:S3599") //language=SQL - private static final Map, String> DELETE_COMMANDS = + private static final Map, String> DELETE_TXN_COMMANDS = new LinkedHashMap, String>() {{ put((hiveObjectType, keepTxnToWriteIdMetaData) -> HIVE_OBJECT_TYPES.contains(hiveObjectType), "DELETE FROM \"TXN_COMPONENTS\" WHERE " + @@ -62,18 +61,6 @@ public class CleanupRecordsFunction implements TransactionalFunction { "(\"CTC_DATABASE\" = :dbName) AND " + "(\"CTC_TABLE\" = :tableName OR :tableName IS NULL) AND " + "(\"CTC_PARTITION\" = :partName OR :partName IS NULL)"); - put((hiveObjectType, keepTxnToWriteIdMetaData) -> HIVE_OBJECT_TYPES.contains(hiveObjectType), - "DELETE FROM \"COMPACTION_QUEUE\" WHERE " + - "\"CQ_DATABASE\" = :dbName AND " + - "(\"CQ_TABLE\" = :tableName OR :tableName IS NULL) AND " + - "(\"CQ_PARTITION\" = :partName OR :partName IS NULL) AND " + - "(\"CQ_TXN_ID\" != :txnId OR :txnId IS NULL) AND " + - "(\"CQ_TYPE\" != :cType)"); - put((hiveObjectType, keepTxnToWriteIdMetaData) -> HIVE_OBJECT_TYPES.contains(hiveObjectType), - "DELETE FROM \"COMPLETED_COMPACTIONS\" WHERE " + - "\"CC_DATABASE\" = :dbName AND " + - "(\"CC_TABLE\" = :tableName OR :tableName IS NULL) AND " + - "(\"CC_PARTITION\" = :partName OR :partName IS NULL)"); put((hiveObjectType, keepTxnToWriteIdMetaData) -> HiveObjectType.DATABASE.equals(hiveObjectType) || (HiveObjectType.TABLE.equals(hiveObjectType) && !keepTxnToWriteIdMetaData), "DELETE FROM \"TXN_TO_WRITE_ID\" WHERE " + @@ -84,27 +71,45 @@ public class CleanupRecordsFunction implements TransactionalFunction { "DELETE FROM \"NEXT_WRITE_ID\" WHERE " + "\"NWI_DATABASE\" = :dbName AND " + "(\"NWI_TABLE\" = :tableName OR :tableName IS NULL)"); - put((hiveObjectType, keepTxnToWriteIdMetaData) -> HIVE_OBJECT_TYPES.contains(hiveObjectType), - "DELETE FROM \"COMPACTION_METRICS_CACHE\" WHERE " + - "\"CMC_DATABASE\" = :dbName AND " + - "(\"CMC_TABLE\" = :tableName OR :tableName IS NULL) AND " + - "(\"CMC_PARTITION\" = :partName OR :partName IS NULL)"); }}; + //language=SQL + private static final String DELETE_COMPACTION_QUEUE_ALL = + "DELETE FROM \"COMPACTION_QUEUE\" WHERE " + + "\"CQ_DATABASE\" = :dbName AND " + + "(\"CQ_TABLE\" = :tableName OR :tableName IS NULL) AND " + + "(\"CQ_PARTITION\" = :partName OR :partName IS NULL)"; + + //language=SQL + private static final String DELETE_COMPACTION_QUEUE_ACID = DELETE_COMPACTION_QUEUE_ALL + + " AND (\"CQ_TXN_ID\" != :txnId OR :txnId IS NULL) AND " + + "(\"CQ_TYPE\" != :cType)"; + + //language=SQL + private static final List DELETE_COMPACT_COMMANDS = Arrays.asList( + "DELETE FROM \"COMPLETED_COMPACTIONS\" WHERE " + + "\"CC_DATABASE\" = :dbName AND " + + "(\"CC_TABLE\" = :tableName OR :tableName IS NULL) AND " + + "(\"CC_PARTITION\" = :partName OR :partName IS NULL)", + "DELETE FROM \"COMPACTION_METRICS_CACHE\" WHERE " + + "\"CMC_DATABASE\" = :dbName AND " + + "(\"CMC_TABLE\" = :tableName OR :tableName IS NULL) AND " + + "(\"CMC_PARTITION\" = :partName OR :partName IS NULL)"); + private final HiveObjectType type; private final Database db; private final Table table; - private final Iterator partitionIterator; + private final Iterator partNamesIterator; private final String defaultCatalog; private final boolean keepTxnToWriteIdMetaData; private final Long txnId; - public CleanupRecordsFunction(HiveObjectType type, Database db, Table table, Iterator partitionIterator, + public CleanupRecordsFunction(HiveObjectType type, Database db, Table table, Iterator partNamesIterator, String defaultCatalog, boolean keepTxnToWriteIdMetaData, Long txnId) { this.type = type; this.db = db; this.table = table; - this.partitionIterator = partitionIterator; + this.partNamesIterator = partNamesIterator; this.defaultCatalog = defaultCatalog; this.keepTxnToWriteIdMetaData = keepTxnToWriteIdMetaData; this.txnId = txnId; @@ -114,6 +119,7 @@ public CleanupRecordsFunction(HiveObjectType type, Database db, Table table, Ite public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaException { // cleanup should be done only for objects belonging to default catalog List paramSources = new ArrayList<>(); + boolean deleteTxnCommands = false; switch (type) { case DATABASE: { if (!defaultCatalog.equals(db.getCatalogName())) { @@ -121,6 +127,7 @@ public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExcepti + "other than default catalog: " + db.getCatalogName()); return null; } + deleteTxnCommands = true; paramSources.add(new MapSqlParameterSource() .addValue("dbName", db.getName().toLowerCase()) .addValue("tableName", null, Types.VARCHAR) @@ -135,6 +142,7 @@ public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExcepti table.getTableName(), table.getCatName()); return null; } + deleteTxnCommands = TxnUtils.isTransactionalTable(table); paramSources.add(new MapSqlParameterSource() .addValue("dbName", table.getDbName().toLowerCase()) .addValue("tableName", table.getTableName().toLowerCase(), Types.VARCHAR) @@ -149,15 +157,12 @@ public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExcepti table.getCatName()); return null; } - List partCols = table.getPartitionKeys(); // partition columns - List partVals; // partition values - while (partitionIterator.hasNext()) { - Partition partition = partitionIterator.next(); - partVals = partition.getValues(); + deleteTxnCommands = TxnUtils.isTransactionalTable(table); + while (partNamesIterator.hasNext()) { paramSources.add(new MapSqlParameterSource() .addValue("dbName", table.getDbName().toLowerCase()) .addValue("tableName", table.getTableName().toLowerCase(), Types.VARCHAR) - .addValue("partName", Warehouse.makePartName(partCols, partVals), Types.VARCHAR) + .addValue("partName", partNamesIterator.next(), Types.VARCHAR) .addValue("txnId", null, Types.BIGINT) .addValue("cType", Character.toString(TxnStore.DEFERRED_CLEANUP), Types.CHAR)); } @@ -166,10 +171,18 @@ public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExcepti try { for (MapSqlParameterSource parameterSource : paramSources) { - for (Map.Entry, String> item : DELETE_COMMANDS.entrySet()) { - if (item.getKey().apply(type, keepTxnToWriteIdMetaData)) { - jdbcResource.getJdbcTemplate().update(item.getValue(), parameterSource); + String deleteCompactionQueue = DELETE_COMPACTION_QUEUE_ALL; + if (deleteTxnCommands) { + for (Map.Entry, String> item : DELETE_TXN_COMMANDS.entrySet()) { + if (item.getKey().apply(type, keepTxnToWriteIdMetaData)) { + jdbcResource.getJdbcTemplate().update(item.getValue(), parameterSource); + } } + deleteCompactionQueue = DELETE_COMPACTION_QUEUE_ACID; + } + jdbcResource.getJdbcTemplate().update(deleteCompactionQueue, parameterSource); + for (String deleteCommand : DELETE_COMPACT_COMMANDS) { + jdbcResource.getJdbcTemplate().update(deleteCommand, parameterSource); } } } catch (DataAccessException e) { diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/txn/ThrowingTxnHandler.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/txn/ThrowingTxnHandler.java index 72fa4e171ce2..c6e32dd2d387 100644 --- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/txn/ThrowingTxnHandler.java +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/txn/ThrowingTxnHandler.java @@ -21,7 +21,6 @@ import org.apache.hadoop.hive.metastore.api.Database; import org.apache.hadoop.hive.metastore.api.HiveObjectType; import org.apache.hadoop.hive.metastore.api.MetaException; -import org.apache.hadoop.hive.metastore.api.Partition; import org.apache.hadoop.hive.metastore.api.ShowCompactRequest; import org.apache.hadoop.hive.metastore.api.ShowCompactResponse; import org.apache.hadoop.hive.metastore.api.Table; @@ -34,20 +33,20 @@ public class ThrowingTxnHandler extends CompactionTxnHandler { @Override public void cleanupRecords(HiveObjectType type, Database db, Table table, - Iterator partitionIterator, boolean keepTxnToWriteIdMetaData) throws MetaException { + Iterator partNamesIterator, boolean keepTxnToWriteIdMetaData) throws MetaException { if (doThrow) { throw new RuntimeException("during transactional cleanup"); } - super.cleanupRecords(type, db, table, partitionIterator, keepTxnToWriteIdMetaData); + super.cleanupRecords(type, db, table, partNamesIterator, keepTxnToWriteIdMetaData); } @Override public void cleanupRecords(HiveObjectType type, Database db, Table table, - Iterator partitionIterator, long txnId) throws MetaException { + Iterator partNamesIterator, long txnId) throws MetaException { if (doThrow) { throw new RuntimeException("during transactional cleanup"); } - super.cleanupRecords(type, db, table, partitionIterator, txnId); + super.cleanupRecords(type, db, table, partNamesIterator, txnId); } @Override