Skip to content

HIVE-29512: Clean compaction related HMS metadata tables for iceberg #6679

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
VenuReddy2103 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

HIVE-29512: Clean compaction related HMS metadata tables for iceberg #6679

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergMetaHook.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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));
}
Expand All @@ -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<String> 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());
Expand Down
36 changes: 36 additions & 0 deletions iceberg/iceberg-handler/src/test/queries/positive/iceberg_show_compactions.q
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
211 changes: 211 additions & 0 deletions iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_show_compactions.q.out
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
3 changes: 2 additions & 1 deletion itests/src/test/resources/testconfiguration.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading