@Test
  public void testCleanAbortCompactSeveralTables() throws Exception {
    String dbName = "default";
    String tblName1 = "cws1";
    String tblName2 = "cws2";
    boolean useCleanerForAbortCleanup = MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_CLEAN_ABORTS_USING_CLEANER);

    HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
    HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);

    connection1.beginTransaction();
    connection1.write("1,1".getBytes());
    connection1.write("2,2".getBytes());
    connection1.abortTransaction();

    connection2.beginTransaction();
    connection2.write("1,1".getBytes());
    connection2.write("2,2".getBytes());
    connection2.abortTransaction();

    IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
    FileSystem fs = FileSystem.get(conf);
    Table table1 = msClient.getTable(dbName, tblName1);
    FileStatus[] stat =
        fs.listStatus(new Path(table1.getSd().getLocation()));
    if (2 != stat.length) {
      Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
    }
    Table table2 = msClient.getTable(dbName, tblName2);
    stat = fs.listStatus(new Path(table2.getSd().getLocation()));
    if (2 != stat.length) {
      Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
    }

    int count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='i'");
    // We should have two rows corresponding to the two aborted transactions
    Assert.assertEquals(TestTxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);

    runInitiator(conf);
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
    // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
    // with rows in TXN_COMPONENTS
    Assert.assertEquals(TestTxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"),
            useCleanerForAbortCleanup ? 0 : 2, count);

    runWorker(conf);
    runWorker(conf);

    runCleaner(conf);
    // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
    Assert.assertEquals(TestTxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);

    count = TestTxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
    Assert.assertEquals(TestTxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);

    RemoteIterator<LocatedFileStatus> it =
        fs.listFiles(new Path(table1.getSd().getLocation()), true);
    if (it.hasNext()) {
      Assert.fail("Expected cleaner to drop aborted delta & base directories, FileStatus[] stat " + Arrays.toString(stat));
    }

    connection1.close();
    connection2.close();
  }
After each major compaction, stats need to be updated on the table
1. create an ORC backed table (Orc is currently required by ACID)
2. populate with data
3. compute stats
4. Trigger major compaction (which should update stats)
5. check that stats have been updated

@throws Exception todo:
                  4. add a test with sorted table?
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java