public static void addStatsTask(FileSinkOperator nd, MoveTask mvTask,
      Task<?> currTask, HiveConf hconf) {

    MoveWork mvWork = mvTask.getWork();
    BasicStatsWork statsWork = null;
    Table table = null;
    boolean truncate = false;
    if (mvWork.getLoadTableWork() != null) {
      statsWork = new BasicStatsWork(mvWork.getLoadTableWork());
      truncate = mvWork.getLoadTableWork().getReplace();
      String tableName = mvWork.getLoadTableWork().getTable().getTableName();
      try {
        // For partitioned CTAS, the table has not been created, but we can retrieve it
        // from the loadTableWork. For rest of query types, we just retrieve it from
        // metastore.
        table = mvWork.getLoadTableWork().getMdTable() != null ?
            mvWork.getLoadTableWork().getMdTable() :
            Hive.get().getTable(SessionState.get().getCurrentDatabase(), tableName);
      } catch (HiveException e) {
        throw new RuntimeException("unexpected; table should be present already..: " + tableName, e);
      }
    } else if (mvWork.getLoadFileWork() != null) {
      statsWork = new BasicStatsWork(mvWork.getLoadFileWork());

      truncate = true;
      if (mvWork.getLoadFileWork().getCtasCreateTableDesc() != null) {
        try {
          table = mvWork.getLoadFileWork().getCtasCreateTableDesc().toTable(hconf);
        } catch (HiveException e) {
          LOG.debug("can't pre-create table for CTAS", e);
          table = null;
        }
      } else if (mvWork.getLoadFileWork().getCreateViewDesc() != null) {
        // CREATE MATERIALIZED VIEW ...
        try {
          table = mvWork.getLoadFileWork().getCreateViewDesc().toTable(hconf);
        } catch (HiveException e) {
          LOG.debug("can't pre-create table for MV", e);
          table = null;
        }
      } else {
        throw new RuntimeException("unexpected; this should be a CTAS or a CREATE/REBUILD MV - however no desc present");
      }
    }
    assert statsWork != null : "Error when generating StatsTask";

    if (currTask.getWork() instanceof MapredWork) {
      MapredWork mrWork = (MapredWork) currTask.getWork();
      mrWork.getMapWork().setGatheringStats(true);
      if (mrWork.getReduceWork() != null) {
        mrWork.getReduceWork().setGatheringStats(true);
      }
    } else { // must be TezWork
      TezWork work = (TezWork) currTask.getWork();
      for (BaseWork w: work.getAllWork()) {
        w.setGatheringStats(true);
      }
    }

    StatsWork columnStatsWork = new StatsWork(table, statsWork, hconf);
    columnStatsWork.collectStatsFromAggregator(nd.getConf());

    columnStatsWork.truncateExisting(truncate);

    columnStatsWork.setSourceTask(currTask);
    Task<?> statsTask = TaskFactory.get(columnStatsWork);

    // subscribe feeds from the MoveTask so that MoveTask can forward the list
    // of dynamic partition list to the StatsTask
    mvTask.addDependentTask(statsTask);
    statsTask.subscribeFeed(mvTask);
  }
Add the StatsTask as a dependent task of the MoveTask
because StatsTask will change the Table/Partition metadata. For atomicity, we
should not change it before the data is actually there done by MoveTask.

@param nd
         the FileSinkOperator whose results are taken care of by the MoveTask.
@param mvTask
         The MoveTask that moves the FileSinkOperator's results.
@param currTask
         The MapRedTask that the FileSinkOperator belongs to.
@param hconf
         HiveConf
ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java