@Test
  public void testCleanerForTxnToWriteId() throws Exception {
    int[][] tableData1 = {{1, 2}};
    int[][] tableData2 = {{2, 3}};
    int[][] tableData3 = {{3, 4}};
    runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData1));
    runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData2));
    runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData3));
    runStatementOnDriver("insert into " + Table.ACIDTBLPART + " partition(p=1) (a,b) " + makeValuesClause(tableData1));
    runStatementOnDriver("insert into " + Table.ACIDTBLPART + " partition(p=2) (a,b) " + makeValuesClause(tableData2));

    // All inserts are committed and hence would expect in TXN_TO_WRITE_ID, 3 entries for acidTbl
    // and 2 entries for acidTblPart as each insert would have allocated a writeid.
    String acidTblWhereClause = " where t2w_database = " + quoteString("default")
            + " and t2w_table = " + quoteString(Table.ACIDTBL.name().toLowerCase());
    String acidTblPartWhereClause = " where t2w_database = " + quoteString("default")
            + " and t2w_table = " + quoteString(Table.ACIDTBLPART.name().toLowerCase());
    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID" + acidTblWhereClause),
            3, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID" + acidTblWhereClause));
    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID" + acidTblPartWhereClause),
            2, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID" + acidTblPartWhereClause));

    txnHandler.compact(new CompactionRequest("default", Table.ACIDTBL.name().toLowerCase(), CompactionType.MAJOR));
    runWorker(hiveConf);
    runCleaner(hiveConf);
    txnHandler.cleanTxnToWriteIdTable();

    // After compaction/cleanup, all entries from TXN_TO_WRITE_ID should be cleaned up as all txns are committed.
    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID"),
            0, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));

    // Following sequence of commit-abort-open-abort-commit.
    int[][] tableData4 = {{4, 5}};
    int[][] tableData5 = {{5, 6}};
    runStatementOnDriver("insert into " + Table.ACIDTBLPART + " partition(p=3) (a,b) " + makeValuesClause(tableData3));

    hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_ROLLBACK_TXN, true);
    runStatementOnDriver("insert into " +  Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData4));
    hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_ROLLBACK_TXN, false);

    // Keep an open txn which refers to the aborted txn.
    Context ctx = new Context(hiveConf);
    HiveTxnManager txnMgr = TxnManagerFactory.getTxnManagerFactory().getTxnManager(hiveConf);
    // Txn is not considered committed or aborted until TXN_OPENTXN_TIMEOUT expires
    // See MinOpenTxnIdWaterMarkFunction, OpenTxnTimeoutLowBoundaryTxnIdHandler
    // TODO: revisit wait logic
    waitUntilAllTxnFinished();
    txnMgr.openTxn(ctx, "u1");
    txnMgr.getValidTxns();

    // Start an INSERT statement transaction and roll back this transaction.
    hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_ROLLBACK_TXN, true);
    runStatementOnDriver("insert into " +  Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData5));
    hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_ROLLBACK_TXN, false);

    runStatementOnDriver("insert into " +  Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData5));

    // We would expect 4 entries in TXN_TO_WRITE_ID as each insert would have allocated a writeid
    // including aborted one.
    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID" + acidTblWhereClause),
            3, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID" + acidTblWhereClause));
    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID" + acidTblPartWhereClause),
            1, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID" + acidTblPartWhereClause));

    // The entry relevant to aborted txns shouldn't be removed from TXN_TO_WRITE_ID as
    // 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.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));
    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID" + acidTblPartWhereClause),
            0, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID" + acidTblPartWhereClause));

    // The cleaner after the compaction will not run, since the open txnId < compaction commit txnId
    // Since aborted txns data/metadata was not removed, all data in TXN_TO_WRITE_ID should remain
    txnHandler.compact(new CompactionRequest("default", Table.ACIDTBL.name().toLowerCase(), CompactionType.MAJOR));
    runWorker(hiveConf);
    runCleaner(hiveConf);
    txnHandler.cleanEmptyAbortedAndCommittedTxns();
    txnHandler.cleanTxnToWriteIdTable();
    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID"),
            3, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));

    // Commit the open txn, which lets the cleanup on TXN_TO_WRITE_ID.
    txnMgr.commitTxn();
    txnMgr.openTxn(ctx, "u1");
    txnMgr.getValidTxns();
    // The txn opened after the compaction commit should not effect the Cleaner
    runCleaner(hiveConf);
    txnHandler.cleanEmptyAbortedAndCommittedTxns();
    txnHandler.cleanTxnToWriteIdTable();

    Assert.assertEquals(TestTxnDbUtil.queryToString(hiveConf, "select * from TXN_TO_WRITE_ID"),
            0, TestTxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
  }
Test cleaner for TXN_TO_WRITE_ID table.
@throws Exception
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java