@VisibleForTesting
  static void getInputSummaryWithPool(final Context ctx,
      final Set<Path> pathNeedProcess, final MapWork work, final long[] summary,
      final ExecutorService executor) throws IOException {
    Preconditions.checkNotNull(ctx);
    Preconditions.checkNotNull(pathNeedProcess);
    Preconditions.checkNotNull(executor);

    List<Future<?>> futures = new ArrayList<Future<?>>(pathNeedProcess.size());
    final AtomicLong totalLength = new AtomicLong(0L);
    final AtomicLong totalFileCount = new AtomicLong(0L);
    final AtomicLong totalDirectoryCount = new AtomicLong(0L);

    HiveInterruptCallback interrup = HiveInterruptUtils.add(new HiveInterruptCallback() {
      @Override
      public void interrupt() {
        for (Path path : pathNeedProcess) {
          try {
            path.getFileSystem(ctx.getConf()).close();
          } catch (IOException ignore) {
            LOG.debug("Failed to close filesystem", ignore);
          }
        }
        executor.shutdownNow();
      }
    });
    try {
      Configuration conf = ctx.getConf();
      JobConf jobConf = new JobConf(conf);
      for (final Path path : pathNeedProcess) {
        // All threads share the same Configuration and JobConf based on the
        // assumption that they are thread safe if only read operations are
        // executed. It is not stated in Hadoop's javadoc, the sourcce codes
        // clearly showed that they made efforts for it and we believe it is
        // thread safe. Will revisit this piece of codes if we find the assumption
        // is not correct.
        final Configuration myConf = conf;
        final JobConf myJobConf = jobConf;
        final Map<String, Operator<?>> aliasToWork = work.getAliasToWork();
        final Map<Path, List<String>> pathToAlias = work.getPathToAliases();
        final PartitionDesc partDesc = work.getPathToPartitionInfo().get(path);
        Runnable r = new Runnable() {
          @Override
          public void run() {
            try {
              Class<? extends InputFormat> inputFormatCls = partDesc
                      .getInputFileFormatClass();
              InputFormat inputFormatObj = HiveInputFormat.getInputFormatFromCache(
                      inputFormatCls, myJobConf);
              if (inputFormatObj instanceof ContentSummaryInputFormat) {
                ContentSummaryInputFormat csif = (ContentSummaryInputFormat) inputFormatObj;
                final ContentSummary cs = csif.getContentSummary(path, myJobConf);
                recordSummary(path, cs);
                return;
              }
              String metaTableStorage = null;
              if (partDesc.getTableDesc() != null &&
                      partDesc.getTableDesc().getProperties() != null) {
                metaTableStorage = partDesc.getTableDesc().getProperties()
                        .getProperty(hive_metastoreConstants.META_TABLE_STORAGE, null);
              }
              if (partDesc.getProperties() != null) {
                metaTableStorage = partDesc.getProperties()
                        .getProperty(hive_metastoreConstants.META_TABLE_STORAGE, metaTableStorage);
              }

              HiveStorageHandler handler = HiveUtils.getStorageHandler(myConf, metaTableStorage);
              if (handler instanceof InputEstimator) {
                long total = 0;
                TableDesc tableDesc = partDesc.getTableDesc();
                InputEstimator estimator = (InputEstimator) handler;
                for (String alias : HiveFileFormatUtils.doGetAliasesFromPath(pathToAlias, path)) {
                  JobConf jobConf = new JobConf(myJobConf);
                  TableScanOperator scanOp = (TableScanOperator) aliasToWork.get(alias);
                  Utilities.setColumnNameList(jobConf, scanOp, true);
                  Utilities.setColumnTypeList(jobConf, scanOp, true);
                  PlanUtils.configureInputJobPropertiesForStorageHandler(tableDesc);
                  Utilities.copyTableJobPropertiesToConf(tableDesc, jobConf);
                  total += estimator.estimate(jobConf, scanOp, -1).getTotalLength();
                }
                recordSummary(path, new ContentSummary(total, -1, -1));
              } else if (handler == null) {
                // Nullify summary for non-native tables,
                // in order not to be selected as a mapjoin target
                FileSystem fs = path.getFileSystem(myConf);
                recordSummary(path, fs.getContentSummary(path));
              }
            } catch (Exception e) {
              // We safely ignore this exception for summary data.
              // We don't update the cache to protect it from polluting other
              // usages. The worst case is that IOException will always be
              // retried for another getInputSummary(), which is fine as
              // IOException is not considered as a common case.
              LOG.info("Cannot get size of {}. Safely ignored.", path);
              LOG.debug("Cannot get size of {}. Safely ignored.", path, e);
            }
          }

          private void recordSummary(final Path p, final ContentSummary cs) {
            final long csLength = cs.getLength();
            final long csFileCount = cs.getFileCount();
            final long csDirectoryCount = cs.getDirectoryCount();

            totalLength.addAndGet(csLength);
            totalFileCount.addAndGet(csFileCount);
            totalDirectoryCount.addAndGet(csDirectoryCount);

            ctx.addCS(p.toString(), cs);

            LOG.debug(
                "Cache Content Summary for {} length: {} file count: {} "
                    + "directory count: {}",
                path, csLength, csFileCount, csDirectoryCount);
          }
        };

        futures.add(executor.submit(r));
      }

      for (Future<?> future : futures) {
        try {
          future.get();
        } catch (InterruptedException e) {
          LOG.info("Interrupted when waiting threads", e);
          Thread.currentThread().interrupt();
          break;
        } catch (ExecutionException e) {
          throw new IOException(e);
        }
      }
      executor.shutdown();

      HiveInterruptUtils.checkInterrupted();

      summary[0] += totalLength.get();
      summary[1] += totalFileCount.get();
      summary[2] += totalDirectoryCount.get();
    } finally {
      executor.shutdownNow();
      HiveInterruptUtils.remove(interrup);
    }
  }
Performs a ContentSummary lookup over a set of paths using 1 or more
threads. The 'summary' argument is directly modified.

@param ctx
@param pathNeedProcess
@param work
@param summary
@param executor
@throws IOException
ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java