@Override
    public void run() {

      Map<String, String> parameters = partish.getPartParameters();
      try {
        long numRows = 0;
        long rawDataSize = 0;
        long fileSize = 0;
        long numFiles = 0;
        long numErasureCodedFiles = 0;
        // Note: this code would be invalid for transactional tables of any kind.
        Utilities.FILE_OP_LOGGER.debug("Aggregating stats for {}", dir);
        List<FileStatus> fileList = null;
        if (partish.getTable() != null
            && AcidUtils.isTransactionalTable(partish.getTable())) {
          fileList = AcidUtils.getAcidFilesForStats(partish.getTable(), dir, jc, fs);
        } else {
          fileList = HiveStatsUtils.getFileStatusRecurse(dir, -1, fs);
        }
        ThreadPoolExecutor tpE = null;
        List<Future<FileStats>> futures = null;
        int numThreadsFactor = HiveConf.getIntVar(jc, HiveConf.ConfVars.BASIC_STATS_TASKS_MAX_THREADS_FACTOR);
        if (fileList.size() > 1 && numThreadsFactor > 0) {
          int numThreads = Math.min(fileList.size(), numThreadsFactor * Runtime.getRuntime().availableProcessors());
          ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("Basic-Stats-Thread-%d").build();
          tpE = new ThreadPoolExecutor(numThreads, numThreads, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(),
              threadFactory);
          tpE.allowsCoreThreadTimeOut();
          futures = new ArrayList<>();
          LOG.info("Processing Stats for {} file using {} threads", fileList.size(), numThreads);
        }

        for (FileStatus file : fileList) {
          Utilities.FILE_OP_LOGGER.debug("Computing stats for {}", file);
          if (!file.isDirectory()) {
            InputFormat<?, ?> inputFormat = ReflectionUtil.newInstance(partish.getInputFormatClass(), jc);
            InputSplit dummySplit = new FileSplit(file.getPath(), 0, -1, new String[] { partish.getLocation() });
            if (file.getLen() == 0) {
              numFiles += 1;
            } else {
              FileStatProcessor fsp = new FileStatProcessor(file, inputFormat, dummySplit, jc);
              if (tpE != null) {
                futures.add(tpE.submit(fsp));
              } else {
                // No parallel processing, just call the method normally & update the stats.
                FileStats fileStat = fsp.call();
                rawDataSize += fileStat.getRawDataSize();
                numRows += fileStat.getNumRows();
                fileSize += fileStat.getFileSize();
                numFiles += 1;
                numErasureCodedFiles += fileStat.getNumErasureCodedFiles();
              }
            }
          }
        }

        if (tpE != null) {
          try {
            for (Future<FileStats> future : futures) {
              FileStats fileStat = future.get();
              rawDataSize += fileStat.getRawDataSize();
              numRows += fileStat.getNumRows();
              fileSize += fileStat.getFileSize();
              numFiles += 1;
              numErasureCodedFiles += fileStat.getNumErasureCodedFiles();
            }
          } catch (Exception e) {
            LOG.error("Encountered exception while collecting stats for file lists as {}", fileList, e);
            // Cancel all the futures in the list & throw the caught exception post that.
            futures.forEach(x -> x.cancel(true));
            throw e;
          } finally {
            tpE.shutdown();
          }
        }
        StatsSetupConst.setBasicStatsState(parameters, StatsSetupConst.TRUE);

        parameters.put(StatsSetupConst.ROW_COUNT, String.valueOf(numRows));
        parameters.put(StatsSetupConst.RAW_DATA_SIZE, String.valueOf(rawDataSize));
        parameters.put(StatsSetupConst.TOTAL_SIZE, String.valueOf(fileSize));
        parameters.put(StatsSetupConst.NUM_FILES, String.valueOf(numFiles));
        parameters.put(StatsSetupConst.NUM_ERASURE_CODED_FILES, String.valueOf(numErasureCodedFiles));

        if (partish.getPartition() != null) {
          result = new Partition(partish.getTable(), partish.getPartition().getTPartition());
        } else {
          result = new Table(partish.getTable().getTTable());
        }

        String msg = partish.getSimpleName() + " stats: [" + toString(parameters) + ']';
        LOG.debug(msg);
        console.printInfo(msg);

      } catch (Exception e) {
        console.printInfo("[Warning] could not update stats for " + partish.getSimpleName() + ".", "Failed with exception " + e.getMessage() + "\n" + StringUtils.stringifyException(e));
      }
    }
StatsNoJobTask is used in cases where stats collection is the only task for the given query (no
parent MR or Tez job). It is used in the following cases 1) ANALYZE with noscan for
file formats that implement StatsProvidingRecordReader interface: ORC format (implements
StatsProvidingRecordReader) stores column statistics for all columns in the file footer. It's much
faster to compute the table/partition statistics by reading the footer than scanning all the
rows. This task can be used for computing basic stats like numFiles, numRows, fileSize,
rawDataSize from ORC footer.
However, this cannot be used for full ACID tables, since some files may contain updates
and deletes to existing rows, so summing up the per-file row counts is invalid.
ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java