@Test
  public void testMetastoreTablesCleanup() throws Exception {
    dropTable(new String[] {"temp.T10", "temp.T11", "temp.T12p", "temp.T13p"});
    driver.run("create database if not exists temp");

    // Create some ACID tables: T10, T11 - unpartitioned table, T12p, T13p - partitioned table
    driver.run("create table temp.T10 (a int, b int) " +
        "clustered by(b) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true')");
    driver.run("create table temp.T11 (a int, b int) " +
        "clustered by(b) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true')");
    driver.run("create table temp.T12p (a int, b int) partitioned by (ds string, hour string) " +
        "clustered by(b) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true')");
    driver.run("create table temp.T13p (a int, b int) partitioned by (ds string, hour string) " +
        "clustered by(b) into 2 buckets stored as orc TBLPROPERTIES ('transactional'='true')");

    // Successfully insert some data into ACID tables, so that we have records in COMPLETED_TXN_COMPONENTS
    driver.run("insert into temp.T10 values (1, 1)");
    driver.run("insert into temp.T10 values (2, 2)");
    driver.run("insert into temp.T11 values (3, 3)");
    driver.run("insert into temp.T11 values (4, 4)");
    driver.run("insert into temp.T12p partition (ds='today', hour='1') values (5, 5)");
    driver.run("insert into temp.T12p partition (ds='tomorrow', hour='2') values (6, 6)");
    driver.run("insert into temp.T12p partition (ds='tomorrow', hour='2') values (13, 13)");
    driver.run("insert into temp.T13p partition (ds='today', hour='1') values (7, 7)");
    driver.run("insert into temp.T13p partition (ds='tomorrow', hour='2') values (8, 8)");
    int count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\" in ('t10', 't11')");
    Assert.assertEquals(4, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\" in ('t12p', 't13p')");
    Assert.assertEquals(5, count);

    // Fail some inserts, so that we have records in TXN_COMPONENTS
    conf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_ROLLBACK_TXN, true);
    driver.run("insert into temp.T10 values (9, 9)");
    driver.run("insert into temp.T11 values (10, 10)");
    driver.run("insert into temp.T12p partition (ds='today', hour='1') values (11, 11)");
    driver.run("insert into temp.T13p partition (ds='today', hour='1') values (12, 12)");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"TXN_COMPONENTS\" " +
        "where \"TC_DATABASE\"='temp' and \"TC_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(4, count);
    conf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_ROLLBACK_TXN, false);

    // Drop a table/partition; corresponding records in TXN_COMPONENTS and COMPLETED_TXN_COMPONENTS should disappear
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"TXN_COMPONENTS\" " +
        "where \"TC_DATABASE\"='temp' and \"TC_TABLE\"='t10'");
    Assert.assertEquals(1, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\"='t10'");
    Assert.assertEquals(2, count);
    driver.run("drop table temp.T10");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"TXN_COMPONENTS\" " +
        "where \"TC_DATABASE\"='temp' and \"TC_TABLE\"='t10'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\"='t10'");
    Assert.assertEquals(0, count);

    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"TXN_COMPONENTS\" " +
        "where \"TC_DATABASE\"='temp' and \"TC_TABLE\"='t12p' and \"TC_PARTITION\"='ds=today/hour=1'");
    Assert.assertEquals(1, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\"='t12p' and \"CTC_PARTITION\"='ds=today/hour=1'");
    Assert.assertEquals(1, count);
    driver.run("alter table temp.T12p drop partition (ds='today', hour='1')");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"TXN_COMPONENTS\" " +
        "where \"TC_DATABASE\"='temp' and \"TC_TABLE\"='t12p' and \"TC_PARTITION\"='ds=today/hour=1'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\"='t12p' and \"CTC_PARTITION\"='ds=today/hour=1'");
    Assert.assertEquals(0, count);

    // Successfully perform compaction on a table/partition, so that we have successful records in COMPLETED_COMPACTIONS
    driver.run("alter table temp.T11 compact 'minor'");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t11' and \"CQ_STATE\"='i' and \"CQ_TYPE\"='i'");
    Assert.assertEquals(1, count);
    runWorker(conf);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t11' and \"CQ_STATE\"='r' and \"CQ_TYPE\"='i'");
    Assert.assertEquals(1, count);
    runCleaner(conf);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t11'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\"='t11' and \"CC_STATE\"='s' and \"CC_TYPE\"='i'");
    Assert.assertEquals(1, count);

    driver.run("alter table temp.T12p partition (ds='tomorrow', hour='2') compact 'minor'");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t12p' and \"CQ_PARTITION\"='ds=tomorrow/hour=2' " +
        "and \"CQ_STATE\"='i' and \"CQ_TYPE\"='i'");
    Assert.assertEquals(1, count);
    runWorker(conf);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t12p' and \"CQ_PARTITION\"='ds=tomorrow/hour=2' " +
        "and \"CQ_STATE\"='r' and \"CQ_TYPE\"='i'");
    Assert.assertEquals(1, count);
    runCleaner(conf);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t12p'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\"='t12p' and \"CC_STATE\"='s' and \"CC_TYPE\"='i'");
    Assert.assertEquals(1, count);

    // Fail compaction, so that we have failed records in COMPLETED_COMPACTIONS.
    // Tables need at least 2 delta files to compact, and minor compaction was just run, so insert
    driver.run("insert into temp.T11 values (14, 14)");
    driver.run("insert into temp.T12p partition (ds='tomorrow', hour='2') values (15, 15)");
    conf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_FAIL_COMPACTION, true);
    driver.run("alter table temp.T11 compact 'major'");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t11' and \"CQ_STATE\"='i' and \"CQ_TYPE\"='a'");
    Assert.assertEquals(1, count);
    runWorker(conf); // will fail
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t11' and \"CQ_STATE\"='i' and \"CQ_TYPE\"='a'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\"='t11' and \"CC_STATE\"='f' and \"CC_TYPE\"='a'");
    Assert.assertEquals(1, count);

    driver.run("alter table temp.T12p partition (ds='tomorrow', hour='2') compact 'major'");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t12p' and \"CQ_PARTITION\"='ds=tomorrow/hour=2' " +
        "and \"CQ_STATE\"='i' and \"CQ_TYPE\"='a'");
    Assert.assertEquals(1, count);
    runWorker(conf); // will fail
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t12p' and \"CQ_PARTITION\"='ds=tomorrow/hour=2' " +
        "and \"CQ_STATE\"='i' and \"CQ_TYPE\"='a'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\"='t12p' and \"CC_STATE\"='f' and \"CC_TYPE\"='a'");
    Assert.assertEquals(1, count);
    conf.setBoolVar(HiveConf.ConfVars.HIVE_TEST_MODE_FAIL_COMPACTION, false);

    // Put 2 records into COMPACTION_QUEUE and do nothing
    driver.run("alter table temp.T11 compact 'major'");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t11' and \"CQ_STATE\"='i' and \"CQ_TYPE\"='a'");
    Assert.assertEquals(1, count);
    driver.run("alter table temp.T12p partition (ds='tomorrow', hour='2') compact 'major'");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t12p' and \"CQ_PARTITION\"='ds=tomorrow/hour=2' " +
        "and \"CQ_STATE\"='i' and \"CQ_TYPE\"='a'");
    Assert.assertEquals(1, count);

    // Drop a table/partition, corresponding records in COMPACTION_QUEUE and COMPLETED_COMPACTIONS should disappear
    driver.run("drop table temp.T11");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t11'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\"='t11'");
    Assert.assertEquals(0, count);

    driver.run("alter table temp.T12p drop partition (ds='tomorrow', hour='2')");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t12p'");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\"='t12p'");
    Assert.assertEquals(0, count);

    // Put 1 record into COMPACTION_QUEUE and do nothing
    driver.run("alter table temp.T13p partition (ds='today', hour='1') compact 'major'");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\"='t13p' and \"CQ_STATE\"='i' and \"CQ_TYPE\"='a'");
    Assert.assertEquals(1, count);

    // Drop database, everything in all 4 meta tables should disappear
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"TXN_COMPONENTS\" " +
        "where \"TC_DATABASE\"='temp' and \"TC_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(1, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(2, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(1, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(0, count);
    driver.run("drop database if exists temp cascade");
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"TXN_COMPONENTS\" " +
        "where \"TC_DATABASE\"='temp' and \"TC_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_TXN_COMPONENTS\" " +
        "where \"CTC_DATABASE\"='temp' and \"CTC_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPACTION_QUEUE\" " +
        "where \"CQ_DATABASE\"='temp' and \"CQ_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(0, count);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from \"COMPLETED_COMPACTIONS\" " +
        "where \"CC_DATABASE\"='temp' and \"CC_TABLE\" in ('t10', 't11', 't12p', 't13p')");
    Assert.assertEquals(0, count);
  }
Normally the compaction process will clean up records in TXN_COMPONENTS, COMPLETED_TXN_COMPONENTS,
COMPACTION_QUEUE and COMPLETED_COMPACTIONS. But if a table/partition has been dropped before
compaction and there are still relevant records in those metastore tables, the Initiator will
complain about not being able to find the table/partition. This method is to test and make sure
we clean up relevant records as soon as a table/partition is dropped.

Note, here we don't need to worry about cleaning up TXNS table, since it's handled separately.
ql/src/test/org/apache/hadoop/hive/ql/lockmgr/TestDbTxnManager2.java