code: protected Collection<Path> getGranularInputPaths(Path path) throws IOException {
  
      boolean appendDelta = this.state.getPropAsBoolean(MRCompactor.COMPACTION_RENAME_SOURCE_DIR_ENABLED,
          MRCompactor.DEFAULT_COMPACTION_RENAME_SOURCE_DIR_ENABLED);
  
      Set<Path> uncompacted = Sets.newHashSet();
      Set<Path> total = Sets.newHashSet();
  
      for (FileStatus fileStatus : FileListUtils.listFilesRecursively(fs, path)) {
        if (appendDelta) {
          // use source dir suffix to identify the delta input paths
          if (!fileStatus.getPath().getParent().toString().endsWith(MRCompactor.COMPACTION_RENAME_SOURCE_DIR_SUFFIX)) {
            uncompacted.add(fileStatus.getPath().getParent());
          }
          total.add(fileStatus.getPath().getParent());
        } else {
          uncompacted.add(fileStatus.getPath().getParent());
        }
      }
  
      if (appendDelta) {
        // When the output record count from mr counter doesn't match
        // the record count from input file names, we prefer file names because
        // it will be used to calculate the difference of count in next run.
        this.fileNameRecordCount = new InputRecordCountHelper(this.state).calculateRecordCount(total);
        log.info("{} has total input record count (based on file name) {}", path, this.fileNameRecordCount);
      }
  
      return uncompacted;
    }
docs: Converts a top level input path to a group of sub-paths according to user defined granularity.
  This may be required because if upstream application generates many sub-paths but the map-reduce
  job only keeps track of the top level path, after the job is done, we won't be able to tell if
  those new arriving sub-paths is processed by previous map-reduce job or not. Hence a better way
  is to pre-define those sub-paths as input paths before we start to run MR. The implementation of
  this method should depend on the data generation granularity controlled by upstream. Here we just
  list the deepest level of containing folder as the smallest granularity.
  
  @param path top level directory needs compaction
  @return A collection of input paths which will participate in map-reduce job