@Test
  public void testStatsAfterCompactionPartTbl() throws Exception {
    //as of (8/27/2014) Hive 0.14, ACID/Orc requires HiveInputFormat
    String dbName = "default";
    String tblName = "compaction_test";
    IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
    dropTables(tblName);
    executeStatementOnDriver("CREATE TABLE " + tblName + "(a INT, b STRING) " +
      " PARTITIONED BY(bkt INT)" +
      " CLUSTERED BY(a) INTO 4 BUCKETS" + //currently ACID requires table to be bucketed
      " STORED AS ORC  TBLPROPERTIES ('transactional'='true')", driver);

    StrictDelimitedInputWriter writer = StrictDelimitedInputWriter.newBuilder()
            .withFieldDelimiter(',')
            .build();
    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
            .withDatabase(dbName)
            .withTable(tblName)
            .withStaticPartitionValues(Collections.singletonList("0"))
            .withAgentInfo("UT_" + Thread.currentThread().getName())
            .withHiveConf(conf)
            .withRecordWriter(writer)
            .connect();
    connection.beginTransaction();
    connection.write("55, 'London'".getBytes());
    connection.commitTransaction();
    connection.beginTransaction();
    connection.write("56, 'Paris'".getBytes());
    connection.commitTransaction();
    connection.close();

    executeStatementOnDriver("INSERT INTO TABLE " + tblName + " PARTITION(bkt=1)" +
      " values(57, 'Budapest')", driver);
    executeStatementOnDriver("INSERT INTO TABLE " + tblName + " PARTITION(bkt=1)" +
            " values(58, 'Milano')", driver);
    execSelectAndDumpData("select * from " + tblName, driver, "Dumping data for " +
      tblName + " after load:");

    TxnStore txnHandler = TxnUtils.getTxnStore(conf);
    Table table = msClient.getTable(dbName, tblName);

    //compute stats before compaction
    CompactionInfo ci = new CompactionInfo(dbName, tblName, "bkt=0", CompactionType.MAJOR);
    statsUpdater.gatherStats(conf, ci, emptyMap(), System.getProperty("user.name"),
            CompactorUtil.getCompactorJobQueueName(conf, ci, table), msClient);
    ci = new CompactionInfo(dbName, tblName, "bkt=1", CompactionType.MAJOR);
    statsUpdater.gatherStats(conf, ci, emptyMap(), System.getProperty("user.name"),
            CompactorUtil.getCompactorJobQueueName(conf, ci, table), msClient);

    //Check basic stats are collected
    org.apache.hadoop.hive.ql.metadata.Table hiveTable = Hive.get().getTable(tblName);
    List<org.apache.hadoop.hive.ql.metadata.Partition> partitions = Hive.get().getPartitions(hiveTable);
    Map<String, String> parameters = partitions
            .stream()
            .filter(p -> p.getName().equals("bkt=0"))
            .findFirst()
            .orElseThrow(() -> new RuntimeException("Could not get Partition"))
            .getParameters();
    Assert.assertEquals("The number of files is differing from the expected", "2", parameters.get("numFiles"));
    Assert.assertEquals("The number of rows is differing from the expected", "2", parameters.get("numRows"));

    parameters = partitions
            .stream()
            .filter(p -> p.getName().equals("bkt=1"))
            .findFirst()
            .orElseThrow(() -> new RuntimeException("Could not get Partition"))
            .getParameters();
    Assert.assertEquals("The number of files is differing from the expected", "2", parameters.get("numFiles"));
    Assert.assertEquals("The number of rows is differing from the expected", "2", parameters.get("numRows"));

    //Do a major compaction
    CompactionRequest rqst = new CompactionRequest(dbName, tblName, CompactionType.MAJOR);
    rqst.setPartitionname("bkt=0");
    txnHandler.compact(rqst);
    runWorker(conf);

    ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
    List<ShowCompactResponseElement> compacts = rsp.getCompacts();
    if (1 != compacts.size()) {
      Assert.fail("Expecting 1 file and found " + compacts.size() + " files " + compacts);
    }
    Assert.assertEquals("ready for cleaning", compacts.get(0).getState());

    //Check basic stats are updated for partition bkt=0, but not updated for partition bkt=1
    partitions = Hive.get().getPartitions(hiveTable);
    parameters = partitions
            .stream()
            .filter(p -> p.getName().equals("bkt=0"))
            .findFirst()
            .orElseThrow(() -> new RuntimeException("Could not get Partition"))
            .getParameters();
    Assert.assertEquals("The number of files is differing from the expected", "1", parameters.get("numFiles"));
    Assert.assertEquals("The number of rows is differing from the expected", "2", parameters.get("numRows"));

    parameters = partitions
            .stream()
            .filter(p -> p.getName().equals("bkt=1"))
            .findFirst()
            .orElseThrow(() -> new RuntimeException("Could not get Partition"))
            .getParameters();
    Assert.assertEquals("The number of files is differing from the expected", "2", parameters.get("numFiles"));
    Assert.assertEquals("The number of rows is differing from the expected", "2", parameters.get("numRows"));
  }
After each major compaction, stats need to be updated on the table
1. create a partitioned ORC backed table (Orc is currently required by ACID)
2. populate with data
3. compute stats
4. Trigger major compaction on one of the partitions (which should update stats)
5. check that stats have been updated for that partition only

@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