private boolean isDataLengthWithInThreshold(ParseContext pctx, final long threshold)
        throws Exception {
      if (splitSample != null && splitSample.getTotalLength() != null) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Threshold " + splitSample.getTotalLength() + " exceeded for pseudoMR mode");
        }
        return (threshold - splitSample.getTotalLength()) > 0;
      }

      Status status = checkThresholdWithMetastoreStats(table, partsList, threshold);
      if (status.equals(Status.PASS)) {
        return true;
      } else if (status.equals(Status.FAIL)) {
        return false;
      } else {
        LOG.info("Cannot fetch stats from metastore for table: {}. Falling back to filesystem scan..",
          table.getCompleteName());
        // metastore stats is unavailable, fallback to old way
        final JobConf jobConf = new JobConf(pctx.getConf());
        Utilities.setColumnNameList(jobConf, scanOp, true);
        Utilities.setColumnTypeList(jobConf, scanOp, true);
        HiveStorageHandler handler = table.getStorageHandler();
        if (handler instanceof InputEstimator) {
          InputEstimator estimator = (InputEstimator) handler;
          TableDesc tableDesc = Utilities.getTableDesc(table);
          PlanUtils.configureInputJobPropertiesForStorageHandler(tableDesc);
          Utilities.copyTableJobPropertiesToConf(tableDesc, jobConf);
          long len = estimator.estimate(jobConf, scanOp, threshold).getTotalLength();
          LOG.debug("Threshold {} exceeded for pseudoMR mode", len);
          return (threshold - len) > 0;
        }
        if (table.isNonNative()) {
          return true; // nothing can be done
        }
        if (!table.isPartitioned()) {
          long len = getPathLength(jobConf, table.getPath(), table.getInputFormatClass(), threshold);
          LOG.debug("Threshold {} exceeded for pseudoMR mode", len);
          return (threshold - len) > 0;
        }
        final AtomicLong total = new AtomicLong(0);
        //TODO: use common thread pool later?
        int threadCount = HiveConf.getIntVar(pctx.getConf(),
          HiveConf.ConfVars.HIVE_STATS_GATHER_NUM_THREADS);
        final ExecutorService pool = (threadCount > 0) ?
          Executors.newFixedThreadPool(threadCount,
            new ThreadFactoryBuilder()
              .setDaemon(true)
              .setNameFormat("SimpleFetchOptimizer-FileLength-%d").build()) : null;
        try {
          List<Future> futures = Lists.newLinkedList();
          for (final Partition partition : partsList.getNotDeniedPartns()) {
            final Path path = partition.getDataLocation();
            if (pool != null) {
              futures.add(pool.submit(new Callable<Long>() {
                @Override
                public Long call() throws Exception {
                  long len = getPathLength(jobConf, path, partition.getInputFormatClass(), threshold);
                  LOG.trace(path + ", length=" + len);
                  return total.addAndGet(len);
                }
              }));
            } else {
              total.addAndGet(getPathLength(jobConf, path, partition.getInputFormatClass(), threshold));
            }
          }
          if (pool != null) {
            pool.shutdown();
            for (Future<Long> future : futures) {
              long totalLen = future.get();
              if ((threshold - totalLen) <= 0) {
                // early exit, as getting file lengths can be expensive in object stores.
                return false;
              }
            }
          }
          return (threshold - total.get()) >= 0;
        } finally {
          LOG.info("Data set size=" + total.get() + ", threshold=" + threshold);
          if (pool != null) {
            pool.shutdownNow();
          }
        }
      }
    }
Tries to convert simple fetch query to single fetch task, which fetches rows directly
from location of table/partition.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchOptimizer.java