review comments

Author: deniskuzZ

ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java

@@ -3566,6 +3566,7 @@ public void testInsertSnapshotIsolation() throws Exception {
     swapTxnManager(txnMgr);
 
     driver.run();
+    txnHandler.performWriteSetGC();
     txnHandler.cleanTxnToWriteIdTable();
     swapTxnManager(txnMgr2);
 

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

@@ -212,7 +212,7 @@ public void markCleaned(CompactionInfo info) throws MetaException {
 
   /**
    * Clean up entries from TXN_TO_WRITE_ID table less than min_uncommited_txnid as found by
-   * min(max(TXNS.txn_id), min(WRITE_SET.WS_COMMIT_ID), min(Aborted TXNS.txn_id)).
+   * min(max(TXNS.txn_id), min(WRITE_SET.WS_TXNID), min(Aborted TXNS.txn_id)).
    */
   @Override
   @RetrySemantics.SafeToRetry

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

@@ -715,7 +715,7 @@ Set<CompactionInfo> findPotentialCompactions(int abortedThreshold, long abortedT
 
   /**
    * Clean up entries from TXN_TO_WRITE_ID table less than min_uncommited_txnid as found by
-   * min(max(TXNS.txn_id), min(WRITE_SET.WS_COMMIT_ID), min(Aborted TXNS.txn_id)).
+   * min(max(TXNS.txn_id), min(WRITE_SET.WS_TXNID), min(Aborted TXNS.txn_id)).
    */
   @SqlRetry
   @Transactional(POOL_COMPACTOR)

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/CommitTxnFunction.java

@@ -40,6 +40,7 @@
 import org.apache.hadoop.hive.metastore.txn.TxnStore;
 import org.apache.hadoop.hive.metastore.txn.TxnUtils;
 import org.apache.hadoop.hive.metastore.txn.entities.TxnWriteDetails;
+import org.apache.hadoop.hive.metastore.txn.jdbc.InClauseBatchCommand;
 import org.apache.hadoop.hive.metastore.txn.jdbc.commands.DeleteReplTxnMapEntryCommand;
 import org.apache.hadoop.hive.metastore.txn.jdbc.commands.InsertCompletedTxnComponentsCommand;
 import org.apache.hadoop.hive.metastore.txn.jdbc.commands.RemoveTxnsFromMinHistoryLevelCommand;
@@ -70,6 +71,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -126,7 +128,7 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
         throw new RollbackException(null);
       }
       assert targetTxnIds.size() == 1;
-      txnid = targetTxnIds.get(0);
+      txnid = targetTxnIds.getFirst();
     }
 
     /**
@@ -155,8 +157,8 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
     }
 
     String conflictSQLSuffix = String.format("""
-        FROM "TXN_COMPONENTS" WHERE "TC_TXNID" = %d AND "TC_OPERATION_TYPE" IN (%s, %s)
-        """, txnid, OperationType.UPDATE, OperationType.DELETE);
+        FROM "TXN_COMPONENTS" WHERE "TC_TXNID" = :txnId AND "TC_OPERATION_TYPE" IN (%s, %s)
+        """, OperationType.UPDATE, OperationType.DELETE);
     long tempCommitId = TxnUtils.generateTemporaryId();
 
     if (txnType == TxnType.SOFT_DELETE || txnType == TxnType.COMPACTION) {
@@ -176,7 +178,10 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
           """;
 
       boolean isUpdateOrDelete = Boolean.TRUE.equals(jdbcResource.getJdbcTemplate().query(
-          jdbcResource.getSqlGenerator().addLimitClause(1, "\"TC_OPERATION_TYPE\" " + conflictSQLSuffix),
+          jdbcResource.getSqlGenerator()
+              .addLimitClause(1, "\"TC_OPERATION_TYPE\" " + conflictSQLSuffix),
+          new MapSqlParameterSource()
+              .addValue("txnId", txnid),
           ResultSet::next));
 
       if (isUpdateOrDelete) {
@@ -201,8 +206,8 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
                 (txnType != TxnType.REBALANCE_COMPACTION) ? "" : " AND \"TC_OPERATION_TYPE\" <> :type")),
             new MapSqlParameterSource()
                 .addValue("txnId", txnid)
-                .addValue("commitId", tempCommitId)
-                .addValue("type", OperationType.COMPACT.getSqlConst()));
+                .addValue("type", OperationType.COMPACT.getSqlConst())
+                .addValue("commitId", tempCommitId));
 
         /**
          * This S4U will mutex with other commitTxn() and openTxns().
@@ -249,7 +254,7 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
         jdbcResource.getJdbcTemplate().update(writeSetInsertSql + "FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" = :txnId",
             new MapSqlParameterSource()
                 .addValue("txnId", txnid)
-                .addValue("commitId", txnid));
+                .addValue("commitId", jdbcResource.execute(new GetHighWaterMarkHandler())));
       }
     } else {
       /*
@@ -274,8 +279,7 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
       jdbcResource.execute(new DeleteReplTxnMapEntryCommand(sourceTxnId, rqst.getReplPolicy()));
     }
     updateWSCommitIdAndCleanUpMetadata(jdbcResource, txnid, txnType, commitId, tempCommitId);
-    jdbcResource.execute(new RemoveTxnsFromMinHistoryLevelCommand(ImmutableList.of(txnid)));
-    jdbcResource.execute(new RemoveWriteIdsFromMinHistoryCommand(ImmutableList.of(txnid)));
+
     if (rqst.isSetKeyValue()) {
       updateKeyValueAssociatedWithTxn(jdbcResource, rqst);
     }
@@ -562,11 +566,8 @@ private void updateKeyValueAssociatedWithTxn(MultiDataSourceJdbcResource jdbcRes
     }
   }
 
-  /**
-   * See overridden method in CompactionTxnHandler also.
-   */
-  private void updateWSCommitIdAndCleanUpMetadata(MultiDataSourceJdbcResource jdbcResource, long txnid, TxnType txnType,
-                                                    Long commitId, long tempId) throws MetaException {
+  private void updateWSCommitIdAndCleanUpMetadata(MultiDataSourceJdbcResource jdbcResource,
+        long txnid, TxnType txnType, Long commitId, long tempId) throws MetaException {
     List<String> queryBatch = new ArrayList<>(6);
     // update write_set with real commitId
     if (commitId != null) {
@@ -587,9 +588,17 @@ private void updateWSCommitIdAndCleanUpMetadata(MultiDataSourceJdbcResource jdbc
       queryBatch.add("UPDATE \"COMPACTION_QUEUE\" SET \"CQ_NEXT_TXN_ID\" = " + commitId + ", \"CQ_COMMIT_TIME\" = " +
           getEpochFn(jdbcResource.getDatabaseProduct()) + " WHERE \"CQ_TXN_ID\" = " + txnid);
     }
-    
+
     // execute all in one batch
     jdbcResource.getJdbcTemplate().getJdbcTemplate().batchUpdate(queryBatch.toArray(new String[0]));
+
+    List<Function<List<Long>, InClauseBatchCommand<Long>>> commands = List.of(
+        RemoveTxnsFromMinHistoryLevelCommand::new,
+        RemoveWriteIdsFromMinHistoryCommand::new
+    );
+    for (Function<List<Long>, InClauseBatchCommand<Long>> cmd : commands) {
+      jdbcResource.execute(cmd.apply(ImmutableList.of(txnid)));
+    }
   }
 
   /**