@Test
  public void autoCompactOnStreamingIngestWithDynamicPartition() throws Exception {
    String dbName = "default";
    String tblName = "cws";
    String columnNamesProperty = "a,b";
    String columnTypesProperty = "string:int";
    String agentInfo = "UT_" + Thread.currentThread().getName();

    dropTables(tblName);
    executeStatementOnDriver("CREATE TABLE " + tblName + "(a STRING) " +
        " PARTITIONED BY (b INT)" + //currently ACID requires table to be bucketed
        " STORED AS ORC  TBLPROPERTIES ('transactional'='true')", driver);

    StrictDelimitedInputWriter writer1 = StrictDelimitedInputWriter.newBuilder()
        .withFieldDelimiter(',')
        .build();
    StrictDelimitedInputWriter writer2 = StrictDelimitedInputWriter.newBuilder()
        .withFieldDelimiter(',')
        .build();

    StreamingConnection connection1 = HiveStreamingConnection.newBuilder()
        .withDatabase(dbName)
        .withTable(tblName)
        .withAgentInfo(agentInfo)
        .withHiveConf(conf)
        .withRecordWriter(writer1)
        .withStreamingOptimizations(true)
        // Transaction size has to be one or exception should happen.
        .withTransactionBatchSize(1)
        .connect();

    StreamingConnection connection2 = HiveStreamingConnection.newBuilder()
        .withDatabase(dbName)
        .withTable(tblName)
        .withAgentInfo(agentInfo)
        .withHiveConf(conf)
        .withRecordWriter(writer2)
        .withStreamingOptimizations(true)
        // Transaction size has to be one or exception should happen.
        .withTransactionBatchSize(1)
        .connect();

    try {
      connection1.beginTransaction();
      connection1.write("1,1".getBytes());
      connection1.commitTransaction();

      connection1.beginTransaction();
      connection1.write("1,1".getBytes());
      connection1.commitTransaction();
      connection1.close();

      conf.setIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_DELTA_NUM_THRESHOLD, 1);
      runInitiator(conf);

      TxnStore txnHandler = TxnUtils.getTxnStore(conf);
      ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());

      List<ShowCompactResponseElement> compacts1 = rsp.getCompacts();
      Assert.assertEquals(1, compacts1.size());
      SortedSet<String> partNames1 = new TreeSet<>();
      verifyCompactions(compacts1, partNames1, tblName);
      List<String> names1 = new ArrayList<>(partNames1);
      Assert.assertEquals("b=1", names1.get(0));

      runWorker(conf);
      runCleaner(conf);

      connection2.beginTransaction();
      connection2.write("1,1".getBytes());
      connection2.commitTransaction();

      connection2.beginTransaction();
      connection2.write("1,1".getBytes());
      connection2.commitTransaction();
      connection2.close();

      runInitiator(conf);

      List<ShowCompactResponseElement> compacts2 = rsp.getCompacts();
      Assert.assertEquals(1, compacts2.size());
      SortedSet<String> partNames2 = new TreeSet<>();
      verifyCompactions(compacts2, partNames2, tblName);
      List<String> names2 = new ArrayList<>(partNames2);
      Assert.assertEquals("b=1", names2.get(0));

      runWorker(conf);
      runCleaner(conf);

      // Find the location of the table
      IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
      Table table = msClient.getTable(dbName, tblName);
      String tablePath = table.getSd().getLocation();
      String partName = "b=1";
      Path partPath = new Path(tablePath, partName);
      FileSystem fs = FileSystem.get(conf);
      FileStatus[] stat = fs.listStatus(partPath, AcidUtils.baseFileFilter);
      if (1 != stat.length) {
        Assert.fail("Expecting 1 file \"base_0000004\" and found " + stat.length + " files " + Arrays.toString(stat));
      }
      String name = stat[0].getPath().getName();
      Assert.assertEquals("base_0000004_v0000008", name);
      CompactorTestUtil
          .checkExpectedTxnsPresent(stat[0].getPath(), null, columnNamesProperty, columnTypesProperty, 0, 1L, 4L, null,
              1);
    } finally {
      if (connection1 != null) {
        connection1.close();
      }
      if (connection2 != null) {
        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