@Test
  public void testNoDataLossWhenMaxNumDeltaIsUsed() throws Exception {
    String dbName = "default";
    String tblName = "cws";
    dropTables(tblName);

    executeStatementOnDriver("CREATE TABLE " + tblName + "(a INT, b STRING) " +
      " STORED AS ORC  TBLPROPERTIES ('transactional'='true')", driver);
    executeStatementOnDriver("insert into " + tblName + " values (1, 'a')", driver);
    executeStatementOnDriver("insert into " + tblName + " values (3, 'b')", driver);

    runMajorCompaction(dbName, tblName);
    runCleaner(conf);

    for (int i = 0; i < 3; i++) {
      executeStatementOnDriver("MERGE INTO " + tblName + " AS T USING (" +
        "select * from " + tblName + " union all select a+1, b from " + tblName + ") AS S " +
        "ON T.a=s.a " +
        "WHEN MATCHED THEN DELETE " +
        "WHEN not MATCHED THEN INSERT values (s.a, s.b)", driver);
    }

    driver.run("select a from " + tblName);
    List<String> res = new ArrayList<>();
    driver.getFetchTask().fetch(res);
    Assert.assertEquals(res, Arrays.asList("4", "6"));

    conf.setIntVar(HiveConf.ConfVars.COMPACTOR_MAX_NUM_DELTA, 5);
    runMajorCompaction(dbName, tblName);

    List<String> matchesNotFound = new ArrayList<>(5);
    matchesNotFound.add(AcidUtils.deleteDeltaSubdir(3, 4) + VISIBILITY_PATTERN);
    matchesNotFound.add(AcidUtils.deltaSubdir(3, 4) + VISIBILITY_PATTERN);
    matchesNotFound.add(AcidUtils.deleteDeltaSubdir(5, 5, 0));
    matchesNotFound.add(AcidUtils.deltaSubdir(5, 5, 1));
    matchesNotFound.add(AcidUtils.baseDir(5) + VISIBILITY_PATTERN);

    IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
    Table table = msClient.getTable(dbName, tblName);
    msClient.close();

    FileSystem fs = FileSystem.get(conf);
    FileStatus[] stat = fs.listStatus(new Path(table.getSd().getLocation()));

    for (FileStatus f : stat) {
      for (int j = 0; j < matchesNotFound.size(); j++) {
        if (f.getPath().getName().matches(matchesNotFound.get(j))) {
          matchesNotFound.remove(j);
          break;
        }
      }
    }
    Assert.assertEquals("Matches Not Found: " + matchesNotFound, 0, matchesNotFound.size());
    runCleaner(conf);

    driver.run("select a from " + tblName);
    res = new ArrayList<>();
    driver.getFetchTask().fetch(res);
    Assert.assertEquals(res, Arrays.asList("4", "6"));
  }
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