HIVE-29522: Prevent cleaner from creating COMPLETED_COMPACTIONS entries for soft-deleted ACID tables

Author: Venu Reddy

standalone-metastore/metastore-common/src/gen/thrift/gen-javabean/org/apache/hadoop/hive/metastore/api/hive_metastoreConstants.java

@@ -1344,4 +1344,9 @@ public static String getHttpPath(String httpPath) {
   public static boolean isDatabaseRemote(Database db) {
     return db != null && db.getType() == DatabaseType.REMOTE;
   }
+
+  /**
+   * Soft delete operation types.
+   */
+  public enum SoftDeleteOperation {DROP_DATABASE, DROP_TABLE}
 }

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/utils/MetaStoreUtils.java

@@ -39,6 +39,7 @@
 import org.apache.hadoop.hive.metastore.events.DropTableEvent;
 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 java.io.IOException;
 import java.util.Iterator;
@@ -91,6 +92,7 @@ public void onDropTable(DropTableEvent tableEvent) throws MetaException {
             rqst.setRunas(TxnUtils.findUserToRunAs(table.getSd().getLocation(), table, conf));
             rqst.putToProperties("location", table.getSd().getLocation());
             rqst.putToProperties("ifPurge", Boolean.toString(isMustPurge(tableEvent.getEnvironmentContext(), table)));
+            addSoftDeletePropertiesToRequest(tableEvent.getEnvironmentContext(), rqst);
             txnHandler.submitForCleanup(rqst, table.getWriteId(), currentTxn);
           } catch (InterruptedException | IOException e) {
             throwMetaException(e);
@@ -244,4 +246,10 @@ private long getTxnId(EnvironmentContext context) {
       .map(Long::parseLong)
       .orElse(0L);
   }
+
+  private void addSoftDeletePropertiesToRequest(EnvironmentContext context, CompactionRequest request) {
+    request.putToProperties(hive_metastoreConstants.SOFT_DELETE_OPERATION,
+        Optional.ofNullable(context.getProperties().get(hive_metastoreConstants.SOFT_DELETE_OPERATION))
+            .orElse(MetaStoreUtils.SoftDeleteOperation.DROP_TABLE.name()));
+  }
 }

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/AcidEventListener.java

@@ -21,6 +21,7 @@
 import org.apache.hadoop.hive.common.ValidCompactorWriteIdList;
 import org.apache.hadoop.hive.metastore.api.CompactionInfoStruct;
 import org.apache.hadoop.hive.metastore.api.CompactionType;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.OptionalCompactionInfoStruct;
 import org.apache.hadoop.hive.metastore.api.TableValidWriteIds;
@@ -368,4 +369,8 @@ public void setWriteIds(boolean hasUncompactedAborts, Set<Long> writeIds) {
   public boolean isAbortedTxnCleanup() {
     return type == CompactionType.ABORT_TXN_CLEANUP;
   }
+
+  public boolean isSoftDelete() {
+    return getProperty(hive_metastoreConstants.SOFT_DELETE_OPERATION) != null;
+  }
 }

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

@@ -50,7 +50,7 @@ public MarkCleanedFunction(CompactionInfo info) {
   public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaException {
     NamedParameterJdbcTemplate jdbcTemplate = jdbcResource.getJdbcTemplate();
     MapSqlParameterSource param;
-    if (!info.isAbortedTxnCleanup()) {
+    if (!info.isAbortedTxnCleanup() && !info.isSoftDelete()) {
       param = new MapSqlParameterSource()
           .addValue("id", info.id)
           .addValue("succeeded", Character.toString(SUCCEEDED_STATE), Types.CHAR);
@@ -85,6 +85,10 @@ public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExcepti
      */
     removeCompactionAndAbortRetryEntries(info, jdbcTemplate);
 
+    if (info.isSoftDelete()) {
+      return null;
+    }
+
     if (!info.isAbortedTxnCleanup()) {
       // Remove entries from completed_txn_components as well, so we don't start looking there
       // again but only up to the highest write ID include in this compaction job.
@@ -175,7 +179,7 @@ private void removeCompactionAndAbortRetryEntries(CompactionInfo info, NamedPara
     String deleteQuery = """
         DELETE FROM "COMPACTION_QUEUE" WHERE "CQ_ID" = :id
         """;
-    if (!info.isAbortedTxnCleanup()) {
+    if (!info.isAbortedTxnCleanup() && !info.isSoftDelete()) {
       deleteQuery += """
           OR ("CQ_DATABASE" = :db AND "CQ_TABLE" = :table
             AND (:partition is NULL OR "CQ_PARTITION" = :partition)