HIVE-29028: Iceberg: Implement auto compaction

Author: Dmitriy Fingerman

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java

@@ -62,13 +62,15 @@
 import org.apache.hadoop.hive.metastore.api.AggrStats;
 import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
 import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
+import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.InvalidObjectException;
 import org.apache.hadoop.hive.metastore.api.LockType;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
 import org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils;
 import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
 import org.apache.hadoop.hive.ql.Context.Operation;
@@ -180,6 +182,7 @@
 import org.apache.iceberg.mr.Catalogs;
 import org.apache.iceberg.mr.InputFormatConfig;
 import org.apache.iceberg.mr.hive.actions.HiveIcebergDeleteOrphanFiles;
+import org.apache.iceberg.mr.hive.compaction.evaluator.CompactionEvaluator;
 import org.apache.iceberg.mr.hive.plan.IcebergBucketFunction;
 import org.apache.iceberg.puffin.Blob;
 import org.apache.iceberg.puffin.BlobMetadata;
@@ -2381,4 +2384,15 @@ private static List<FieldSchema> schema(List<VirtualColumn> exprs) {
   private static List<FieldSchema> orderBy(VirtualColumn... exprs) {
     return schema(Arrays.asList(exprs));
   }
+
+  @Override
+  public CompactionType determineCompactionType(org.apache.hadoop.hive.metastore.api.Table table, CompactionInfo ci)
+      throws IOException {
+    org.apache.iceberg.Table icebergTable = IcebergTableUtil.getTable(conf, table);
+    CompactionEvaluator compactionEvaluator = new CompactionEvaluator(icebergTable, ci, table.getParameters());
+    if (compactionEvaluator.isEligibleForCompaction()) {
+      return compactionEvaluator.determineCompactionType();
+    }
+    return null;
+  }
 }

itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorOnTezTest.java

@@ -52,6 +52,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -69,6 +70,7 @@ public abstract class CompactorOnTezTest {
   protected IMetaStoreClient msClient;
   protected IDriver driver;
   protected boolean mmCompaction = false;
+  private final AtomicBoolean stop = new AtomicBoolean();
 
   @ClassRule
   public static TemporaryFolder folder = new TemporaryFolder();
@@ -541,4 +543,27 @@ protected void dropTable(String tblName) throws Exception {
       executeStatementOnDriver("drop table " + tblName, driver);
     }
   }
+
+  protected void startInitiator() throws Exception {
+    runOneLoopOfCompactorThread(CompactorTestUtilities.CompactorThreadType.INITIATOR);
+  }
+  
+  protected InitiatorBase getInitiator() {
+    return new Initiator();
+  }
+
+  private void runOneLoopOfCompactorThread(CompactorTestUtilities.CompactorThreadType type) throws Exception {
+    TestTxnDbUtil.setConfValues(conf);
+    CompactorThread t;
+    switch (type) {
+      case INITIATOR: t = getInitiator(); break;
+      case WORKER: t = new Worker(); break;
+      case CLEANER: t = new Cleaner(); break;
+      default: throw new RuntimeException("Huh? Unknown thread type.");
+    }
+    t.setConf(conf);
+    stop.set(true);
+    t.init(stop);
+    t.run();
+  }
 }

itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestIcebergCompactorOnTez.java

@@ -21,66 +21,144 @@
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
+import org.apache.hadoop.hive.metastore.txn.TxnUtils;
+import org.apache.hadoop.hive.metastore.txn.entities.CompactionState;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 
 import static org.apache.hadoop.hive.ql.txn.compactor.TestCompactor.executeStatementOnDriver;
 
 public class TestIcebergCompactorOnTez extends CompactorOnTezTest {
+  
+  private static final String DB_NAME = "default";
+  private static final String TABLE_NAME = "ice_orc";
+  private static final String QUALIFIED_TABLE_NAME = TxnUtils.getFullTableName(DB_NAME, TABLE_NAME);
+
+  @Override
+  @Before
+  public void setup() throws Exception {
+    super.setup();
+    executeStatementOnDriver("drop table if exists " + QUALIFIED_TABLE_NAME, driver);
+  }
 
   @Test
   public void testIcebergCompactorWithAllPartitionFieldTypes() throws Exception{
     conf.setVar(HiveConf.ConfVars.COMPACTOR_JOB_QUEUE, CUSTOM_COMPACTION_QUEUE);
     msClient = new HiveMetaStoreClient(conf);
 
-    String dbName = "default";
-    String tableName = "ice_orc";
-    String qualifiedTableName = dbName + "." + tableName;
-
-    executeStatementOnDriver("drop table if exists " + qualifiedTableName, driver);
     executeStatementOnDriver(String.format("create table %s " +
         "(id int, a string, b int, c bigint, d float, e double, f decimal(4, 2), g boolean, h date, i date, j date, k timestamp) " +
         "partitioned by spec(a, truncate(3, a), bucket(4, a), b, c, d, e, f, g, h, year(h), month(i), day(j), k, hour(k)) stored by iceberg stored as orc " +
-        "tblproperties ('compactor.threshold.min.input.files'='1')", qualifiedTableName), driver);
+        "tblproperties ('compactor.threshold.min.input.files'='1')", QUALIFIED_TABLE_NAME), driver);
 
     // 6 records, one records per file --> 3 partitions, 2 files per partition
-    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (1, 'aaa111', 1, 100, 1.0, 2.0, 4.00, true,  DATE '2024-05-01', DATE '2024-05-01', DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", qualifiedTableName), driver);
-    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (2, 'aaa111', 1, 100, 1.0, 2.0, 4.00, true,  DATE '2024-05-01', DATE '2024-05-01', DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", qualifiedTableName), driver);
-    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (3, 'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03', DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", qualifiedTableName), driver);
-    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (4, 'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03', DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", qualifiedTableName), driver);
-    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (5, null, null, null, null, null, null, null, null, null, null, null)", qualifiedTableName), driver);
-    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (6, null, null, null, null, null, null, null, null, null, null, null)", qualifiedTableName), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (1, 'aaa111', 1, 100, 1.0, 2.0, 4.00, true,  DATE '2024-05-01', DATE '2024-05-01', DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (2, 'aaa111', 1, 100, 1.0, 2.0, 4.00, true,  DATE '2024-05-01', DATE '2024-05-01', DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (3, 'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03', DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (4, 'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03', DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (5, null, null, null, null, null, null, null, null, null, null, null)", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (6, null, null, null, null, null, null, null, null, null, null, null)", QUALIFIED_TABLE_NAME), driver);
 
-    Assert.assertEquals(6, getFilesCount(qualifiedTableName));
-    List<String> recordsBefore = getAllRecords(qualifiedTableName);
+    Assert.assertEquals(6, getFilesCount());
+    List<String> recordsBefore = getAllRecords();
 
-    CompactorTestUtil.runCompaction(conf, dbName, tableName, CompactionType.MINOR, false, 
+    CompactorTestUtil.runCompaction(conf, DB_NAME, TABLE_NAME, CompactionType.MINOR, false, 
         "a=aaa111/a_trunc=aaa/a_bucket=0/b=1/c=100/d=1.0/e=2.0/f=4.00/g=true/h=2024-05-01/h_year=2024/i_month=2024-05/j_day=2024-05-01/k=2024-05-02T10%3A00%3A00/k_hour=2024-05-02-10",
         "a=bbb222/a_trunc=bbb/a_bucket=3/b=2/c=200/d=2.0/e=3.0/f=8.00/g=false/h=2024-05-03/h_year=2024/i_month=2024-05/j_day=2024-05-03/k=2024-05-04T13%3A00%3A00/k_hour=2024-05-04-13",
         "a=null/a_trunc=null/a_bucket=null/b=null/c=null/d=null/e=null/f=null/g=null/h=null/h_year=null/i_month=null/j_day=null/k=null/k_hour=null"
     );
 
-    Assert.assertEquals(3, getFilesCount(qualifiedTableName));
+    Assert.assertEquals(3, getFilesCount());
     verifySuccessfulCompaction(3);
-    List<String> recordsAfter = getAllRecords(qualifiedTableName);
+    List<String> recordsAfter = getAllRecords();
 
     Assert.assertEquals(recordsBefore, recordsAfter);
   }
-  
-  private int getFilesCount(String qualifiedTableName) throws Exception {
-    driver.run(String.format("select count(*) from %s.files", qualifiedTableName));
+
+  @Test
+  public void testIcebergAutoCompactionPartitionEvolution() throws Exception {
+    executeStatementOnDriver(String.format("create table %s " +
+        "(id int, a string) " +
+        "partitioned by spec(id) stored by iceberg stored as orc " +
+        "tblproperties ('compactor.threshold.min.input.files'='1')", QUALIFIED_TABLE_NAME), driver);
+
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (1, 'a')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (2, 'b')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (3, 'c')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (4, 'd')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (5, 'e')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (6, 'd')", QUALIFIED_TABLE_NAME), driver);
+
+    executeStatementOnDriver(String.format("alter table %s set partition spec(truncate(3, a))", QUALIFIED_TABLE_NAME), driver);
+
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (7, 'aaa111')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (8, 'aaa111')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (9, 'bbb222')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (10, 'bbb222')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (11, null)", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (12, null)", QUALIFIED_TABLE_NAME), driver);
+
+    startInitiator();
+    ShowCompactResponse rsp = msClient.showCompactions();
+    Assert.assertEquals(4, rsp.getCompactsSize());
+
+    // Compaction should be initiated for each partition from the latest spec
+    Assert.assertTrue(isCompactExist(rsp, "a_trunc_3=aaa", CompactionType.MINOR, CompactionState.INITIATED));
+    Assert.assertTrue(isCompactExist(rsp, "a_trunc_3=bbb", CompactionType.MINOR, CompactionState.INITIATED));
+    Assert.assertTrue(isCompactExist(rsp, "a_trunc_3=null", CompactionType.MINOR, CompactionState.INITIATED));
+
+    // Additional compaction should be initiated for all partitions from past partition specs
+    Assert.assertTrue(isCompactExist(rsp, null, CompactionType.MINOR, CompactionState.INITIATED));
+  }
+
+  @Test
+  public void testIcebergAutoCompactionUnpartitioned() throws Exception {
+    executeStatementOnDriver(String.format("create table %s " +
+        "(id int, a string) " +
+        "stored by iceberg stored as orc " +
+        "tblproperties ('compactor.threshold.min.input.files'='1')", QUALIFIED_TABLE_NAME), driver);
+
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (7, 'aaa111')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (8, 'aaa111')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (9, 'bbb222')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (10, 'bbb222')", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (11, null)", QUALIFIED_TABLE_NAME), driver);
+    executeStatementOnDriver(String.format("INSERT INTO %s VALUES (12, null)", QUALIFIED_TABLE_NAME), driver);
+
+    startInitiator();
+    ShowCompactResponse rsp = msClient.showCompactions();
+    Assert.assertEquals(1, rsp.getCompactsSize());
+    Assert.assertTrue(isCompactExist(rsp, null, CompactionType.MINOR, CompactionState.INITIATED));
+  }
+
+  private int getFilesCount() throws Exception {
+    driver.run(String.format("select count(*) from %s.files", QUALIFIED_TABLE_NAME));
     List<String> res = new ArrayList<>();
     driver.getFetchTask().fetch(res);
     return Integer.parseInt(res.get(0));
   }
 
-  private List<String> getAllRecords(String qualifiedTableName) throws Exception {
-    driver.run(String.format("select * from %s order by id", qualifiedTableName));
+  private List<String> getAllRecords() throws Exception {
+    driver.run(String.format("select * from %s order by id", QUALIFIED_TABLE_NAME));
     List<String> res = new ArrayList<>();
     driver.getFetchTask().fetch(res);
     return res;
   }
+  
+  private boolean isCompactExist(ShowCompactResponse rsp, String partName, CompactionType type, CompactionState state) {
+    return rsp.getCompacts().stream().anyMatch(c ->
+        c.getDbname().equals(DB_NAME) && c.getTablename().equals(TABLE_NAME) &&
+            Objects.equals(c.getPartitionname(), partName) && c.getType().equals(type) &&
+            c.getState().equals(state.name().toLowerCase()));
+  }
+
+  @Override
+  protected InitiatorBase getInitiator() {
+    return new IcebergInitiator();
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java

@@ -37,11 +37,13 @@
 import org.apache.hadoop.hive.metastore.api.AggrStats;
 import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
 import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
+import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.LockType;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
 import org.apache.hadoop.hive.ql.Context;
 import org.apache.hadoop.hive.ql.Context.Operation;
 import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
@@ -76,7 +78,6 @@
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.OutputCommitter;
 import org.apache.hadoop.mapred.OutputFormat;
-import org.apache.hadoop.mapred.TaskAttemptContext;
 
 import java.util.HashMap;
 import java.util.List;
@@ -980,4 +981,16 @@ default MergeTaskProperties getMergeTaskProperties(Properties properties) {
   default void setMergeTaskDeleteProperties(TableDesc tableDesc) {
     throw new UnsupportedOperationException("Storage handler does not support getting custom delete merge schema.");
   }
+
+/**
+ * Determines the appropriate {@link CompactionType} for a given table or partition and compaction information.
+ * @param table {@link org.apache.hadoop.hive.metastore.api.Table} metadata stored in Hive Metastore.
+ * @param ci {@link org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo} details about the compaction request.
+ * @return The determined {@link CompactionType} or null if table or partition does not need compaction.
+ * @throws IOException If an I/O error occurs during the compaction evaluation process.
+ * @throws UnsupportedOperationException if the implementing storage handler does not support this operation.
+ */
+  default CompactionType determineCompactionType(Table table, CompactionInfo ci) throws IOException {
+    throw new UnsupportedOperationException("Storage handler does not support determining compaction type");
+  }
 }