fixes

Author: deniskuzZ

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

@@ -23,6 +23,7 @@
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.conf.HiveConfForTest;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.txn.TxnHandler;
 import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
 import org.apache.hadoop.hive.metastore.txn.TxnStore;
 import org.apache.hadoop.hive.metastore.txn.TxnUtils;
@@ -78,7 +79,9 @@ public void setUp() throws Exception {
     HiveConf.setIntVar(conf, HiveConf.ConfVars.HIVE_LOCKS_PARTITION_THRESHOLD, -1);
     HiveConf.setBoolVar(conf, HiveConf.ConfVars.HIVE_ACID_LOCKLESS_READS_ENABLED, false);
     HiveConf.setBoolVar(conf, HiveConf.ConfVars.TXN_WRITE_X_LOCK, false);
-    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_LEVEL, true);
+
+    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, true);
+    TxnHandler.ConfVars.setUseMinHistoryWriteId(true);
 
     driver = new Driver(new QueryState.Builder().withHiveConf(conf).nonIsolated().build());
     driver2 = new Driver(new QueryState.Builder().withHiveConf(conf).build());

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

@@ -37,6 +37,7 @@
 import org.apache.hadoop.hive.metastore.api.TxnType;
 import org.apache.hadoop.hive.metastore.api.CommitTxnRequest;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.txn.TxnHandler;
 import org.apache.hadoop.hive.metastore.txn.service.CompactionHouseKeeperService;
 import org.apache.hadoop.hive.metastore.txn.service.AcidHouseKeeperService;
 import org.apache.hadoop.hive.ql.Context;
@@ -1235,6 +1236,9 @@ public void testWriteSetTracking4() throws Exception {
    */
   @Test
   public void testWriteSetTracking5() throws Exception {
+    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, false);
+    TxnHandler.ConfVars.setUseMinHistoryWriteId(false);
+
     dropTable(new String[] {"TAB_PART"});
     Assert.assertEquals(0, TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"WRITE_SET\""));
     driver.run("create table if not exists TAB_PART (a int, b int) " +
@@ -2109,6 +2113,9 @@ public void testMergeUnpartitionedConflictSharedWrite() throws Exception {
    * @param causeConflict true to make 2 operations such that they update the same entity
    */
   private void testMergeUnpartitioned(boolean causeConflict, boolean sharedWrite) throws Exception {
+    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, false);
+    TxnHandler.ConfVars.setUseMinHistoryWriteId(false);
+
     dropTable(new String[] {"target","source"});
     conf.setBoolVar(HiveConf.ConfVars.TXN_WRITE_X_LOCK, !sharedWrite);
 
@@ -2873,6 +2880,9 @@ public void testMergePartitionedConflictSharedWrite() throws Exception {
    * @param causeConflict - true to make the operations cause a Write conflict
    */
   private void testMergePartitioned(boolean causeConflict, boolean sharedWrite) throws Exception {
+    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, false);
+    TxnHandler.ConfVars.setUseMinHistoryWriteId(false);
+
     dropTable(new String[] {"target","source"});
     conf.setBoolVar(HiveConf.ConfVars.TXN_WRITE_X_LOCK, !sharedWrite);
 
@@ -3537,7 +3547,8 @@ public void testSkipAcquireLocksForExplain() throws Exception {
 
   @Test
   public void testInsertSnapshotIsolationMinHistoryDisabled() throws Exception {
-    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_LEVEL, false);
+    TxnHandler.ConfVars.setUseMinHistoryWriteId(false);
+    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, false);
     testInsertSnapshotIsolation();
   }
 
@@ -3566,7 +3577,8 @@ public void testInsertSnapshotIsolation() throws Exception {
 
   @Test
   public void testUpdateSnapshotIsolationMinHistoryDisabled() throws Exception {
-    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_LEVEL, false);
+    TxnHandler.ConfVars.setUseMinHistoryWriteId(false);
+    MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, false);
     testUpdateSnapshotIsolation();
   }
 

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

@@ -86,6 +86,7 @@ public class TestCompactionMetrics  extends CompactorTest {
   public void setUp() throws Exception {
     MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.METRICS_ENABLED, true);
     MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.TXN_USE_MIN_HISTORY_LEVEL, true);
+    TxnHandler.ConfVars.setUseMinHistoryLevel(true);
     MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, true);
     MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, true);
     // re-initialize metrics

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java

@@ -1676,7 +1676,7 @@ public enum ConfVars {
     TXN_OPENTXN_TIMEOUT("metastore.txn.opentxn.timeout", "hive.txn.opentxn.timeout", 1000, TimeUnit.MILLISECONDS,
         "Time before an open transaction operation should persist, otherwise it is considered invalid and rolled back"),
     @Deprecated
-    TXN_USE_MIN_HISTORY_LEVEL("metastore.txn.use.minhistorylevel", "hive.txn.use.minhistorylevel", false,
+    TXN_USE_MIN_HISTORY_LEVEL("metastore.txn.use.minhistorylevel", "hive.txn.use.minhistorylevel", true,
         "Set this to false, for the TxnHandler and Cleaner to not use MIN_HISTORY_LEVEL table and take advantage of openTxn optimisation.\n"
             + "If the table is dropped HMS will switch this flag to false, any other value changes need a restart to take effect."),
     TXN_USE_MIN_HISTORY_WRITE_ID("metastore.txn.use.minhistorywriteid", "hive.txn.use.minhistorywriteid", true,

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

@@ -374,7 +374,7 @@ private void updateStatus(CompactionInfo ci) throws MetaException {
     String strState = CompactionState.fromSqlConst(ci.state).toString();
 
     LOG.debug("Marking as {}: CompactionInfo: {}", strState, ci);
-    CompactionInfo ciActual = jdbcResource.execute(new GetCompactionInfoHandler(ci.id, false)); 
+    CompactionInfo ciActual = jdbcResource.execute(new GetCompactionInfoHandler(ci.id, false));
 
     long endTime = getDbTime().getTime();
     if (ciActual != null) {
@@ -505,7 +505,7 @@ public long findMinOpenTxnIdForCleaner() throws MetaException {
   @RetrySemantics.Idempotent
   @Deprecated
   public long findMinTxnIdSeenOpen() {
-    if (!ConfVars.useMinHistoryLevel() || ConfVars.useMinHistoryWriteId()) {
+    if (!ConfVars.useMinHistoryLevel()) {
       return Long.MAX_VALUE;
     }
     try {

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

@@ -205,7 +205,7 @@ private ConfVars() {}
     private boolean useMinHistoryWriteId;
 
     public boolean useMinHistoryLevel() {
-      return useMinHistoryLevel;
+      return useMinHistoryLevel && !useMinHistoryWriteId;
     }
 
     public void setUseMinHistoryLevel(boolean useMinHistoryLevel) {

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

@@ -19,9 +19,9 @@
 
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.txn.TxnHandler;
-import org.apache.hadoop.hive.metastore.txn.entities.TxnStatus;
 import org.apache.hadoop.hive.metastore.txn.jdbc.MultiDataSourceJdbcResource;
 import org.apache.hadoop.hive.metastore.txn.jdbc.TransactionalFunction;
+import org.apache.hadoop.hive.metastore.txn.jdbc.queries.MinUncommittedTxnIdHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
@@ -33,21 +33,6 @@ public class CleanTxnToWriteIdTableFunction implements TransactionalFunction<Voi
 
   private static final Logger LOG = LoggerFactory.getLogger(CleanTxnToWriteIdTableFunction.class);
 
-  //language=SQL
-  private static String minHistoryLevelSql = "SELECT MIN(\"RES\".\"ID\") AS \"ID\" FROM (" +
-      " SELECT MAX(\"TXN_ID\") + 1 AS \"ID\" FROM \"TXNS\"" +
-      "   UNION" +
-      " SELECT MIN(\"TXN_ID\") AS \"ID\" FROM \"TXNS\" WHERE \"TXN_STATE\" = :abortedState) \"RES\"";
-  //language=SQL
-  private static String noMinHistoryLevelSql = "SELECT MIN(\"RES\".\"ID\") AS \"ID\" FROM (" +
-      " SELECT MAX(\"TXN_ID\") + 1 AS \"ID\" FROM \"TXNS\"" +
-      "   UNION" +
-      " SELECT MIN(\"WS_TXNID\") AS \"ID\" FROM \"WRITE_SET\"" +
-      "   UNION" +
-      " SELECT MIN(\"TXN_ID\") AS \"ID\" FROM \"TXNS\" WHERE \"TXN_STATE\" = " + TxnStatus.ABORTED +
-      "   OR \"TXN_STATE\" = " + TxnStatus.OPEN +
-      " ) \"RES\"";
-
   private final long minTxnIdSeenOpen;
 
   public CleanTxnToWriteIdTableFunction(long minTxnIdSeenOpen) {
@@ -56,32 +41,18 @@ public CleanTxnToWriteIdTableFunction(long minTxnIdSeenOpen) {
 
   @Override
   public Void execute(MultiDataSourceJdbcResource jdbcResource) throws MetaException {
-    NamedParameterJdbcTemplate jdbcTemplate = jdbcResource.getJdbcTemplate();
-    String sql = TxnHandler.ConfVars.useMinHistoryLevel() ? minHistoryLevelSql : noMinHistoryLevelSql;
-    MapSqlParameterSource params = new MapSqlParameterSource()
-        .addValue("abortedState", TxnStatus.ABORTED.getSqlConst(), Types.CHAR);
-    if (!TxnHandler.ConfVars.useMinHistoryLevel()) {
-      params.addValue("openState", TxnStatus.OPEN.getSqlConst(), Types.CHAR);
-    }
-
     // First need to find the min_uncommitted_txnid which is currently seen by any open transactions.
     // If there are no txns which are currently open or aborted in the system, then current value of
     // max(TXNS.txn_id) could be min_uncommitted_txnid.
-    Long minTxnId = jdbcTemplate.query(sql, params, rs -> {
-      if (rs.next()) {
-        return rs.getLong(1);
-      } else {
-        return null;
-      }
-    });
-
+    Long minTxnId = jdbcResource.execute(
+        new MinUncommittedTxnIdHandler(TxnHandler.ConfVars.useMinHistoryLevel()));
     if (minTxnId == null) {
       throw new MetaException("Transaction tables not properly initialized, no record found in TXNS");
     }
     long minUncommitedTxnid = Math.min(minTxnId, minTxnIdSeenOpen);
-
     // As all txns below min_uncommitted_txnid are either committed or empty_aborted, we are allowed
     // to clean up the entries less than min_uncommitted_txnid from the TXN_TO_WRITE_ID table.
+    NamedParameterJdbcTemplate jdbcTemplate = jdbcResource.getJdbcTemplate();
     int rc = jdbcTemplate.update("DELETE FROM \"TXN_TO_WRITE_ID\" WHERE \"T2W_TXNID\" < :txnId",
         new MapSqlParameterSource("txnId", minUncommitedTxnid));
     LOG.info("Removed {} rows from TXN_TO_WRITE_ID with Txn Low-Water-Mark: {}", rc, minUncommitedTxnid);

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

@@ -256,7 +256,6 @@ public TxnType execute(MultiDataSourceJdbcResource jdbcResource) throws MetaExce
       assert true;
     }
 
-
     if (txnType != TxnType.READ_ONLY && !isReplayedReplTxn && !MetaStoreServerUtils.isCompactionTxn(txnType)) {
       moveTxnComponentsToCompleted(jdbcResource, txnid, isUpdateDelete);
     } else if (isReplayedReplTxn) {
@@ -575,7 +574,7 @@ private void updateWSCommitIdAndCleanUpMetadata(MultiDataSourceJdbcResource jdbc
     if (txnType == TxnType.MATER_VIEW_REBUILD) {
       queryBatch.add("DELETE FROM \"MATERIALIZATION_REBUILD_LOCKS\" WHERE \"MRL_TXN_ID\" = " + txnid);
     }
-    if (txnType == TxnType.SOFT_DELETE || txnType == TxnType.COMPACTION) {
+    if (txnType == TxnType.SOFT_DELETE || MetaStoreServerUtils.isCompactionTxn(txnType)) {
       queryBatch.add("UPDATE \"COMPACTION_QUEUE\" SET \"CQ_NEXT_TXN_ID\" = " + commitId + ", \"CQ_COMMIT_TIME\" = " +
           getEpochFn(jdbcResource.getDatabaseProduct()) + " WHERE \"CQ_TXN_ID\" = " + txnid);
     }