HIVE-28644: Iceberg: Add support for SMART OPTIMIZE feature

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

@@ -18,6 +18,7 @@
 
 package org.apache.iceberg.mr.hive.compaction;
 
+import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.hadoop.hive.metastore.txn.TxnUtils;
 import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
@@ -40,9 +41,8 @@ public IcebergCompactionService() {
 
   public Boolean compact(Table table, CompactionInfo ci) throws Exception {
 
-    if (!ci.isMajorCompaction() && !ci.isMinorCompaction()) {
-      ci.errorMessage = String.format(
-          "Iceberg tables do not support %s compaction type, supported types are ['MINOR', 'MAJOR']", ci.type.name());
+    if (!ci.isMajorCompaction() && !ci.isMinorCompaction() && !ci.isSmartOptimize()) {
+      ci.errorMessage = String.format("Iceberg tables do not support %s compaction type", ci.type.name());
       LOG.error(ci.errorMessage + " Compaction info: {}", ci);
       try {
         msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
@@ -63,6 +63,10 @@ public Boolean compact(Table table, CompactionInfo ci) throws Exception {
       return false;
     }
 
+    if (ci.type == CompactionType.SMART) {
+      ci.type = compactionEvaluator.determineCompactionType();
+    }
+
     if (ci.runAs == null) {
       ci.runAs = TxnUtils.findUserToRunAs(table.getSd().getLocation(), table, conf);
     }

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/compaction/evaluator/CompactionEvaluator.java

@@ -23,6 +23,7 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
+import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
 import org.apache.hadoop.hive.ql.txn.compactor.CompactorContext;
 import org.apache.iceberg.DataFile;
@@ -89,11 +90,27 @@ public boolean isEligibleForCompaction() {
         return isMinorNecessary();
       case MAJOR:
         return isMajorNecessary();
+      case SMART:
+        return isMinorNecessary() || isMajorNecessary();
       default:
         return false;
     }
   }
 
+  public CompactionType determineCompactionType() {
+    if (ci.type == CompactionType.SMART) {
+      if (isMajorNecessary()) {
+        return CompactionType.MAJOR;
+      } else if (isMinorNecessary()) {
+        return CompactionType.MINOR;
+      } else {
+        throw new RuntimeException("Unknown compaction type");
+      }
+    } else {
+      return ci.type;
+    }
+  }
+
   private static TableRuntime createTableRuntime(Table icebergTable, Map<String, String> parameters) {
     OptimizingConfig optimizingConfig = OptimizingConfig.parse(Collections.emptyMap());
     optimizingConfig.setTargetSize(getTargetSizeBytes(parameters));

iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_major_compaction_partition_evolution_ordered.q.out

@@ -215,7 +215,7 @@ STAGE DEPENDENCIES:
 STAGE PLANS:
   Stage: Stage-0
     Compact
-      compaction type: major
+      compaction type: smart
       table name: default.ice_orc
       numberOfBuckets: 0
       order by: order by first_name desc

iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_major_compaction_unpartitioned.q.out

@@ -248,7 +248,7 @@ STAGE DEPENDENCIES:
 STAGE PLANS:
   Stage: Stage-0
     Compact
-      compaction type: major
+      compaction type: smart
       table name: default.ice_orc
       numberOfBuckets: 0
       pool: iceberg

iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_major_compaction_unpartitioned_ordered.q.out

@@ -145,7 +145,7 @@ STAGE DEPENDENCIES:
 STAGE PLANS:
   Stage: Stage-0
     Compact
-      compaction type: major
+      compaction type: smart
       table name: default.ice_orc
       numberOfBuckets: 0
       order by: order by last_name desc

parser/src/java/org/apache/hadoop/hive/ql/parse/AlterClauseParser.g

@@ -115,7 +115,7 @@ optimizeTblRewriteDataSuffix
 @init { gParent.msgs.push("compaction request"); }
 @after { gParent.msgs.pop(); }
     : KW_REWRITE KW_DATA compactPool? whereClause? orderByClause?
-    -> ^(TOK_ALTERTABLE_COMPACT Identifier["'MAJOR'"] TOK_BLOCKING compactPool? whereClause? orderByClause?)
+    -> ^(TOK_ALTERTABLE_COMPACT Identifier["'SMART'"] TOK_BLOCKING compactPool? whereClause? orderByClause?)
     ;
 
 alterStatementPartitionKeyType

parser/src/test/org/apache/hadoop/hive/ql/parse/TestParseOptimizeTable.java

@@ -34,7 +34,7 @@ public void testOptimizeTableWithWhere() throws Exception {
         "      TOK_TABNAME\n" +
         "         tbl0\n" +
         "      TOK_ALTERTABLE_COMPACT\n" +
-        "         'MAJOR'\n" +
+        "         'SMART'\n" +
         "         TOK_BLOCKING\n" +
         "         TOK_WHERE\n" +
         "            and\n" +
@@ -64,7 +64,7 @@ public void testOptimizeTableWithOrderBy() throws Exception {
         "      TOK_TABNAME\n" +
         "         tbl0\n" +
         "      TOK_ALTERTABLE_COMPACT\n" +
-        "         'MAJOR'\n" +
+        "         'SMART'\n" +
         "         TOK_BLOCKING\n" +
         "         TOK_ORDERBY\n" +
         "            TOK_TABSORTCOLNAMEDESC\n" +
@@ -87,7 +87,7 @@ public void testOptimizeTableWithPool() throws Exception {
         "      TOK_TABNAME\n" +
         "         tbl0\n" +
         "      TOK_ALTERTABLE_COMPACT\n" +
-        "         'MAJOR'\n" +
+        "         'SMART'\n" +
         "         TOK_BLOCKING\n" +
         "         TOK_COMPACT_POOL\n" +
         "            'iceberg'\n" +

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/entities/CompactionInfo.java

@@ -149,6 +149,10 @@ public boolean isMinorCompaction() {
     return CompactionType.MINOR == type;
   }
 
+  public boolean isSmartOptimize() {
+    return CompactionType.SMART == type;
+  }
+
   public boolean isRebalanceCompaction() {
     return CompactionType.REBALANCE == type;
   }