refactor

Author: deniskuzZ

ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java

@@ -2373,6 +2373,7 @@ public void testCleanerForTxnToWriteId() throws Exception {
     txnHandler.compact(new CompactionRequest("default", Table.ACIDTBL.name().toLowerCase(), CompactionType.MAJOR));
     runWorker(hiveConf);
     runCleaner(hiveConf);
+    txnHandler.performWriteSetGC();
     txnHandler.cleanTxnToWriteIdTable();
 
     // After compaction/cleanup, all entries from TXN_TO_WRITE_ID should be cleaned up as all txns are committed.
@@ -2416,6 +2417,7 @@ public void testCleanerForTxnToWriteId() throws Exception {
     // aborted txn would be removed from TXNS only after the compaction. Also, committed txn > open txn is retained.
     // As open txn doesn't allocate writeid, the 2 entries for aborted and committed should be retained.
     txnHandler.cleanEmptyAbortedAndCommittedTxns();
+    txnHandler.performWriteSetGC();
     txnHandler.cleanTxnToWriteIdTable();
     Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID" + acidTblWhereClause),
             3, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID" + acidTblWhereClause));
@@ -2428,6 +2430,7 @@ public void testCleanerForTxnToWriteId() throws Exception {
     runWorker(hiveConf);
     runCleaner(hiveConf);
     txnHandler.cleanEmptyAbortedAndCommittedTxns();
+    txnHandler.performWriteSetGC();
     txnHandler.cleanTxnToWriteIdTable();
     Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID"),
             3, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
@@ -2439,6 +2442,7 @@ public void testCleanerForTxnToWriteId() throws Exception {
     // The txn opened after the compaction commit should not effect the Cleaner
     runCleaner(hiveConf);
     txnHandler.cleanEmptyAbortedAndCommittedTxns();
+    txnHandler.performWriteSetGC();
     txnHandler.cleanTxnToWriteIdTable();
 
     Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID"),

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

@@ -189,7 +189,8 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
         Object undoWriteSetForCurrentTxn = context.createSavepoint();
         jdbcResource.getJdbcTemplate().update(
             writeSetInsertSql + (TxnHandler.ConfVars.useMinHistoryLevel() ? conflictSQLSuffix :
-            "FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\"= :txnId AND \"TC_OPERATION_TYPE\" <> :type"),
+            "FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\"= :txnId" + (
+                (txnType != TxnType.REBALANCE_COMPACTION) ? "" : " AND \"TC_OPERATION_TYPE\" <> :type")),
             new MapSqlParameterSource()
                 .addValue("txnId", txnid)
                 .addValue("type", OperationType.COMPACT.getSqlConst()));
@@ -236,10 +237,9 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
           }
         }
       } else if (!TxnHandler.ConfVars.useMinHistoryLevel()) {
-        jdbcResource.getJdbcTemplate().update(writeSetInsertSql + "FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" = :txnId AND \"TC_OPERATION_TYPE\" <> :type",
+        jdbcResource.getJdbcTemplate().update(writeSetInsertSql + "FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" = :txnId",
             new MapSqlParameterSource()
-                .addValue("txnId", txnid)
-                .addValue("type", OperationType.COMPACT.getSqlConst()));
+                .addValue("txnId", txnid));
         commitId = jdbcResource.execute(new GetHighWaterMarkHandler());
       }
     } else {

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/LatestTxnIdInConflictHandler.java

@@ -58,15 +58,15 @@ public String getParameterizedQueryString(DatabaseProduct databaseProduct) throw
         " ) \"CUR\"" +
         " ON \"COMMITTED\".\"WS_DATABASE\" = \"CUR\".\"TC_DATABASE\"" +
         "   AND \"COMMITTED\".\"WS_TABLE\" = \"CUR\".\"TC_TABLE\"" +
-        (TxnHandler.ConfVars.useMinHistoryLevel() ? "" :
-        "   AND \"COMMITTED\".\"WS_OPERATION_TYPE\" != :wsType") + 
         // For partitioned table we always track writes at partition level (never at table)
         // and for non partitioned - always at table level, thus the same table should never
         // have entries with partition key and w/o
         "   AND (\"COMMITTED\".\"WS_PARTITION\" = \"CUR\".\"TC_PARTITION\" OR" +
         "     \"CUR\".\"TC_PARTITION\" IS NULL) " +
         // txns overlap
-        " WHERE \"CUR\".\"TC_TXNID\" <= \"COMMITTED\".\"WS_COMMIT_ID\"";
+        " WHERE \"CUR\".\"TC_TXNID\" <= \"COMMITTED\".\"WS_COMMIT_ID\"" +
+        (TxnHandler.ConfVars.useMinHistoryLevel() ? "" :
+        "   AND \"COMMITTED\".\"WS_OPERATION_TYPE\" != :wsType");
   }
 
   @Override