public static Map<Path, HdfsDirSnapshot> getHdfsDirSnapshots(final FileSystem fs, final Path path)
      throws IOException {
    Map<Path, HdfsDirSnapshot> dirToSnapshots = new HashMap<>();
    Deque<RemoteIterator<FileStatus>> stack = new ArrayDeque<>();
    stack.push(FileUtils.listStatusIterator(fs, path, acidHiddenFileFilter));
    while (!stack.isEmpty()) {
      RemoteIterator<FileStatus> itr = stack.pop();
      while (itr.hasNext()) {
        FileStatus fStatus = itr.next();
        Path fPath = fStatus.getPath();
        if (fStatus.isDirectory()) {
          stack.push(FileUtils.listStatusIterator(fs, fPath, acidHiddenFileFilter));
        } else {
          Path parentDirPath = fPath.getParent();
          if (acidTempDirFilter.accept(parentDirPath)) {
            while (isChildOfDelta(parentDirPath, path)) {
              // Some cases there are other directory layers between the delta and the datafiles
              // (export-import mm table, insert with union all to mm table, skewed tables).
              // But it does not matter for the AcidState, we just need the deltas and the data files
              // So build the snapshot with the files inside the delta directory
              parentDirPath = parentDirPath.getParent();
            }
            HdfsDirSnapshot dirSnapshot = addToSnapshot(dirToSnapshots, parentDirPath);
            // We're not filtering out the metadata file and acid format file,
            // as they represent parts of a valid snapshot
            // We're not using the cached values downstream, but we can potentially optimize more in a follow-up task
            if (fStatus.getPath().toString().contains(MetaDataFile.METADATA_FILE)) {
              dirSnapshot.addMetadataFile(fStatus);
            } else if (fStatus.getPath().toString().contains(OrcAcidVersion.ACID_FORMAT)) {
              dirSnapshot.addOrcAcidFormatFile(fStatus);
            } else {
              dirSnapshot.addFile(fStatus);
            }
          }
        }
      }
    }
    return dirToSnapshots;
  }
In case of the cleaner, we don't need to go into file level, it is enough to collect base/delta/deletedelta directories.

@param fs the filesystem used for the directory lookup
@param path the path of the table or partition needs to be cleaned
@return The listed directory snapshot needs to be checked for cleaning
@throws IOException on filesystem errors
ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java