HIVE-29619: Hive ACID: Long-running queries on one table shouldn't block cleanup operations on other tables.

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/CompactionCleaner.java

@@ -36,6 +36,7 @@
 import org.apache.hadoop.hive.metastore.metrics.Metrics;
 import org.apache.hadoop.hive.metastore.metrics.MetricsConstants;
 import org.apache.hadoop.hive.metastore.metrics.PerfLogger;
+import org.apache.hadoop.hive.metastore.txn.TxnHandler;
 import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
 import org.apache.hadoop.hive.metastore.txn.TxnStore;
 import org.apache.hadoop.hive.metastore.txn.TxnUtils;
@@ -73,7 +74,8 @@ public CompactionCleaner(HiveConf conf, TxnStore txnHandler,
   @Override
   public List<Runnable> getTasks(HiveConf conf) throws MetaException {
     long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
-    long retentionTime = HiveConf.getBoolVar(conf, HIVE_COMPACTOR_DELAYED_CLEANUP_ENABLED)
+    long retentionTime = (HiveConf.getBoolVar(conf, HIVE_COMPACTOR_DELAYED_CLEANUP_ENABLED)
+          || TxnHandler.ConfVars.useMinHistoryWriteId())
         ? HiveConf.getTimeVar(conf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, TimeUnit.MILLISECONDS)
         : 0;
     List<CompactionInfo> readyToClean = txnHandler.findReadyToClean(minOpenTxnId, retentionTime);

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/service/AcidCompactionService.java

@@ -123,7 +123,7 @@
     }
   }
 
   public Boolean compact(Table table, CompactionInfo ci) throws Exception {
 
     try (CompactionTxn compactionTxn = new CompactionTxn()) {
 
@@ -198,6 +198,10 @@
       txnWriteIds.addTableValidWriteIdList(tblValidWriteIds);
       conf.set(ValidTxnWriteIdList.VALID_TABLES_WRITEIDS_KEY, txnWriteIds.toString());
 
+      // Register in MIN_HISTORY_WRITE_ID so the per-table cleaner admission blocks while open.
+      msc.addWriteIdsToMinHistory(compactionTxn.getTxnId(),
+          Map.of(fullTableName, txnWriteIds.getMinOpenWriteId(fullTableName)));
+
       ci.highestWriteId = tblValidWriteIds.getHighWatermark();
       //this writes TXN_COMPONENTS to ensure that if compactorTxnId fails, we keep metadata about
       //it until after any data written by it are physically removed

ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/CompactorTest.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.hive.common.ServerUtils;
 import org.apache.hadoop.hive.common.ValidCompactorWriteIdList;
 import org.apache.hadoop.hive.common.ValidTxnList;
+import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
 import org.apache.hadoop.hive.common.ValidWriteIdList;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
@@ -113,6 +114,8 @@
 import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RETENTION_TIME;
+
 /**
  * Super class for all of the compactor test modules.
  */
@@ -145,6 +148,9 @@ protected final void setup(HiveConf conf) throws Exception {
     MetastoreConf.setBoolVar(conf, ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, useMinHistoryWriteId());
     MetastoreConf.setVar(conf, ConfVars.COMPACTOR_INITIATOR_TABLE_OPTIMIZERS,
         "org.apache.hadoop.hive.ql.txn.compactor.AcidTableOptimizer");
+    if (useMinHistoryWriteId()) {
+      HiveConf.setTimeVar(conf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, 0, TimeUnit.SECONDS);
+    }
     // Set this config to true in the base class, there are extended test classes which set this config to false.
     MetastoreConf.setBoolVar(conf, ConfVars.COMPACTOR_CLEAN_ABORTS_USING_CLEANER, true);
     TestTxnDbUtil.setConfValues(conf);
@@ -388,7 +394,7 @@ protected void burnThroughTransactions(String dbName, String tblName, int num, S
         txnHandler.commitTxn(new CommitTxnRequest(tid));
       } else if (open.contains(tid) && useMinHistoryWriteId()){
         txnHandler.addWriteIdsToMinHistory(tid,
-          Collections.singletonMap(dbName + "." + tblName, minOpenWriteId));
+            Map.of(dbName + "." + tblName, minOpenWriteId));
       }
     }
   }
@@ -754,21 +760,28 @@ long compactInTxn(CompactionRequest rqst, CommitAction commitAction) throws Exce
     ci.runAs = rqst.getRunas() == null ? System.getProperty("user.name") : rqst.getRunas();
 
     long compactorTxnId = openTxn(TxnType.COMPACTION);
+    String fullTableName = ci.getFullTableName().toLowerCase();
 
     // Need to create a valid writeIdList to set the highestWriteId in ci
     ValidTxnList validTxnList = TxnCommonUtils.createValidReadTxnList(
-        txnHandler.getOpenTxns(Collections.singletonList(TxnType.READ_ONLY)), compactorTxnId);
+        txnHandler.getOpenTxns(List.of(TxnType.READ_ONLY)), compactorTxnId);
 
-    GetValidWriteIdsRequest writeIdsRequest = new GetValidWriteIdsRequest(
-        Collections.singletonList(
-            ci.getFullTableName().toLowerCase()));
+    GetValidWriteIdsRequest writeIdsRequest = new GetValidWriteIdsRequest(List.of(fullTableName));
     writeIdsRequest.setValidTxnList(validTxnList.writeToString());
 
     // with this ValidWriteIdList is capped at whatever HWM validTxnList has
     ValidCompactorWriteIdList tblValidWriteIds = TxnUtils.createValidCompactWriteIdList(
         txnHandler.getValidWriteIds(writeIdsRequest).getTblValidWriteIds()
             .getFirst());
 
+    if (useMinHistoryWriteId()) {
+      ValidTxnWriteIdList txnWriteIds = new ValidTxnWriteIdList(compactorTxnId);
+      txnWriteIds.addTableValidWriteIdList(tblValidWriteIds);
+
+      txnHandler.addWriteIdsToMinHistory(compactorTxnId,
+          Map.of(fullTableName, txnWriteIds.getMinOpenWriteId(fullTableName)));
+    }
+
     ci.highestWriteId = tblValidWriteIds.getHighWatermark();
     txnHandler.updateCompactorState(ci, compactorTxnId);
 

ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleanerWithMinHistoryWriteId.java

@@ -20,7 +20,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.metastore.api.CommitTxnRequest;
+import org.apache.hadoop.hive.metastore.api.AbortTxnsRequest;
 import org.apache.hadoop.hive.metastore.api.CompactionRequest;
 import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.ShowCompactRequest;
@@ -32,9 +32,10 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
 import java.util.List;
 
-import static org.apache.hadoop.hive.metastore.txn.TxnStore.FAILED_RESPONSE;
+import static org.apache.hadoop.hive.metastore.txn.TxnStore.CLEANING_RESPONSE;
 import static org.apache.hadoop.hive.metastore.txn.TxnStore.SUCCEEDED_RESPONSE;
 import static org.apache.hadoop.hive.metastore.txn.TxnStore.WORKING_RESPONSE;
 import static org.apache.hadoop.hive.metastore.txn.TxnStore.INITIATED_STATE;
@@ -59,7 +60,7 @@
 
   @Test
   public void cleanupAfterAbortedAndRetriedMajorCompaction() throws Exception {
-    Table t = prepareTestTable();
+    Table t = prepareTestTable("camtc");
     CompactionRequest rqst = new CompactionRequest("default", "camtc", CompactionType.MAJOR);
     long compactTxn = compactInTxn(rqst, CommitAction.ABORT);
     addBaseFile(t, null, 25L, 25, compactTxn);
@@ -83,10 +84,10 @@
 
   @Test
   public void cleanupAfterKilledAndRetriedMajorCompaction() throws Exception {
-    Table t = prepareTestTable();
+    Table t = prepareTestTable("camtc");
     CompactionRequest rqst = new CompactionRequest("default", "camtc", CompactionType.MAJOR);
-    long compactTxn = compactInTxn(rqst, CommitAction.NONE);
-    addBaseFile(t, null, 25L, 25, compactTxn);
+    long compactTxn1 = compactInTxn(rqst, CommitAction.NONE);
+    addBaseFile(t, null, 25L, 25, compactTxn1);
 
     txnHandler.revokeTimedoutWorkers(1L);
     // an open txn should prevent the retry
@@ -96,20 +97,32 @@
 
     // force retry
     revokeTimedoutWorkers(conf);
-    compactTxn = compactInTxn(rqst);
-    addBaseFile(t, null, 25L, 25, compactTxn);
+    long compactTxn2 = compactInTxn(rqst);
+    addBaseFile(t, null, 25L, 25, compactTxn2);
 
     startCleaner();
 
-    // Validate that the cleanup attempt has failed.
+    // Validate that the cleanup attempt was skipped.
     rsp = txnHandler.showCompact(new ShowCompactRequest());
     assertEquals(1, rsp.getCompactsSize());
-    assertEquals(FAILED_RESPONSE, rsp.getCompacts().getFirst().getState());
-    assertEquals("txnid:26 is open and <= hwm: 27", rsp.getCompacts().getFirst().getErrorMessage());
+    assertEquals(CLEANING_RESPONSE, rsp.getCompacts().getFirst().getState());
 
     // Check that the files are not removed
     List<Path> paths = getDirectories(conf, t, null);
     assertEquals(6, paths.size());
+
+    // Abort the open compaction txn, so that the Cleaner can proceed.
+    txnHandler.abortTxns(
+        new AbortTxnsRequest(Collections.singletonList(compactTxn1)));
+    startCleaner();
+
+    rsp = txnHandler.showCompact(new ShowCompactRequest());
+    assertEquals(1, rsp.getCompactsSize());
+    assertEquals(SUCCEEDED_RESPONSE, rsp.getCompacts().getFirst().getState());
+
+    // Check that the files are removed
+    paths = getDirectories(conf, t, null);
+    assertEquals(1, paths.size());
   }
 
   private static void revokeTimedoutWorkers(Configuration conf) throws Exception {
@@ -121,39 +134,51 @@
   }
 
   @Test
-  public void cleanupAfterMajorCompactionWithQueryWaitingToLockTheSnapshot() throws Exception {
-    Table t = prepareTestTable();
-    CompactionRequest rqst = new CompactionRequest("default", "camtc", CompactionType.MAJOR);
-    long compactTxn = compactInTxn(rqst, CommitAction.MARK_COMPACTED);
-    addBaseFile(t, null, 25L, 25, compactTxn);
-
-    // Open a query during compaction,
-    // Do not register minOpenWriteId (i.e. simulate delay locking the snapshot)
-    openTxn();
+  public void cleanupNotBlockedByOpenTxnOnAnotherTable() throws Exception {
+    // Two tables, two compactions: camtc1's compactTxn is registered in MIN_HISTORY_WRITE_ID
+    // (via compactInTxn → addWriteIdsToMinHistory). camtc2's cleanup must proceed despite
+    // camtc1's open compactTxn id being ≤ camtc2's hwm — per-table independence.
+    Table t1 = prepareTestTable("camtc1");
+    Table t2 = prepareTestTable("camtc2");
+
+    CompactionRequest rqstTbl1 = new CompactionRequest("default", "camtc1", CompactionType.MAJOR);
+    long compactTxn = compactInTxn(rqstTbl1, CommitAction.NONE);
+    addBaseFile(t1, null, 25L, 25, compactTxn);
+
+    CompactionRequest rqstTbl2 = new CompactionRequest("default", "camtc2", CompactionType.MAJOR);
+    compactTxn = compactInTxn(rqstTbl2);
+    addBaseFile(t2, null, 25L, 25, compactTxn);
+
+    // force retry: reset camtc1's queue entry from WORKING to INITIATED so a fresh compactor takes it
+    revokeTimedoutWorkers(conf);
+    compactTxn = compactInTxn(rqstTbl1);
+    addBaseFile(t1, null, 25L, 25, compactTxn);
 
-    txnHandler.commitTxn(new CommitTxnRequest(compactTxn));
     startCleaner();
 
-    // Validate that the cleanup attempt has failed.
     ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
-    assertEquals(1, rsp.getCompactsSize());
-    assertEquals(FAILED_RESPONSE, rsp.getCompacts().getFirst().getState());
-    assertEquals("txnid:27 is open and <= hwm: 27", rsp.getCompacts().getFirst().getErrorMessage());
+    assertEquals(2, rsp.getCompactsSize());
 
-    // Check that the files are not removed
-    List<Path> paths = getDirectories(conf, t, null);
-    assertEquals(5, paths.size());
+    assertEquals(SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+    assertEquals("camtc2", rsp.getCompacts().get(0).getTablename());
+    // camtc2 was cleaned: only the new base remains.
+    assertEquals(1, getDirectories(conf, t2, null).size());
+
+    assertEquals(CLEANING_RESPONSE, rsp.getCompacts().get(1).getState());
+    assertEquals("camtc1", rsp.getCompacts().get(1).getTablename());
+    // camtc1 wasn't actually cleaned (admission filter held it back).
+    assertEquals(6, getDirectories(conf, t1, null).size());
   }
 
-  private Table prepareTestTable() throws Exception {
-    Table t = newTable("default", "camtc", false);
+  private Table prepareTestTable(String tblName) throws Exception {
+    Table t = newTable("default", tblName, false);
 
     addBaseFile(t, null, 20L, 20);
     addDeltaFile(t, null, 21L, 22L, 2);
     addDeltaFile(t, null, 23L, 24L, 2);
     addDeltaFile(t, null, 25L, 25, 2);
 
-    burnThroughTransactions("default", "camtc", 25);
+    burnThroughTransactions("default", tblName, 25);
     return t;
   }
 

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

@@ -79,33 +79,28 @@
     long highWatermark = minOpenTxn - 1;
     long[] exceptions = new long[txns.getOpen_txnsSize()];
     BitSet abortedBits = BitSet.valueOf(txns.getAbortedBits());
-    int i = 0;
+    int i = 0, j = 0;
     for (long txnId : txns.getOpen_txns()) {
       if (txnId > highWatermark) {
         break;
       }
-      if (abortedBits.get(i)) {
-        exceptions[i] = txnId;
+      if (abortedBits.get(i) || isAbortCleanup) {
+        exceptions[j++] = txnId;
+      } else if (!TxnHandler.ConfVars.useMinHistoryWriteId()) {
+        throw new IllegalStateException(
+            JavaUtils.txnIdToString(txnId) + " is open and <= hwm: " + highWatermark);
       } else {
-        if (isAbortCleanup) {
-          exceptions[i] = txnId;
-        } else {
-          throw new IllegalStateException(
-              JavaUtils.txnIdToString(txnId) + " is open and <= hwm: " + highWatermark);
-        }
+        LOG.debug("Ignoring open txn {} <= hwm: {}", txnId, highWatermark);
       }
       ++i;
     }
-    exceptions = Arrays.copyOf(exceptions, i);
+    exceptions = Arrays.copyOf(exceptions, j);
+
+    BitSet bitSet = isAbortCleanup ? abortedBits : new BitSet(j);
     if (!isAbortCleanup) {
-      BitSet bitSet = new BitSet(exceptions.length);
-      bitSet.set(0, exceptions.length);
-      //add ValidCleanerTxnList? - could be problematic for all the places that read it from
-      // string as they'd have to know which object to instantiate
-      return new ValidReadTxnList(exceptions, bitSet, highWatermark, Long.MAX_VALUE);
-    } else {
-      return new ValidReadTxnList(exceptions, abortedBits, highWatermark, Long.MAX_VALUE);
+      bitSet.set(0, j);
     }
+    return new ValidReadTxnList(exceptions, bitSet, highWatermark, Long.MAX_VALUE);
   }
 
   /**