private static void cleanDirectInsertDirectory(Path dir, FileSystem fs, String unionSuffix, int lbLevels, Set<Path> committed)
          throws IOException, HiveException {
    for (FileStatus child : fs.listStatus(dir)) {
      Path childPath = child.getPath();
      if (lbLevels > 0) {
        // We need to recurse into some LB directories. We don't check the directories themselves
        // for matches; if they are empty they don't matter, and we do will delete bad files.
        // This recursion is not the most efficient way to do this but LB is rarely used.
        if (child.isDirectory()) {
          Utilities.FILE_OP_LOGGER.trace(
              "Recursion into LB directory {}; levels remaining ", childPath, lbLevels - 1);
          cleanDirectInsertDirectory(childPath, fs, unionSuffix, lbLevels - 1, committed);
        } else {
          if (committed.contains(childPath)) {
            throw new HiveException("LB FSOP has commited "
                + childPath + " outside of LB directory levels " + lbLevels);
          }
          deleteUncommitedFile(childPath, fs);
        }
        continue;
      }
      // No more LB directories expected.
      if (unionSuffix == null) {
        if (committed.remove(childPath)) {
          continue; // A good file.
        }
        if (!childPath.getName().equals(AcidUtils.OrcAcidVersion.ACID_FORMAT)) {
          deleteUncommitedFile(childPath, fs);
        }
      } else if (!child.isDirectory()) {
        if (committed.contains(childPath)) {
          throw new HiveException("Union FSOP has commited "
              + childPath + " outside of union directory " + unionSuffix);
        }
        deleteUncommitedFile(childPath, fs);
      } else if (childPath.getName().equals(unionSuffix)) {
        // Found the right union directory; treat it as "our" directory.
        cleanDirectInsertDirectory(childPath, fs, null, 0, committed);
      } else {
        String childName = childPath.getName();
        if (!childName.startsWith(AbstractFileMergeOperator.UNION_SUDBIR_PREFIX)
            && !childName.startsWith(".") && !childName.startsWith("_")) {
          throw new HiveException("Union FSOP has an unknown directory "
              + childPath + " outside of union directory " + unionSuffix);
        }
        Utilities.FILE_OP_LOGGER.trace(
            "FSOP for {} is ignoring the other side of the union {}", unionSuffix, childPath);
      }
    }
  }
The name of a manifest file consists of the task ID and a .manifest extension, where
the task ID includes the attempt ID as well. It can happen that a task attempt already
wrote out the manifest file, and then fails, so Tez restarts it. If the next attempt
successfully finishes, the query won't fail but there could be multiple manifest files with
the same task ID, but different attempt IDs. In this case the manifest file which has
the highest attempt ID, and not empty, has to be considered.
The empty manifest files and the ones with the same task ID but lower attempt ID has to be ignored.
@param manifestFiles All the files listed in the manifest directory
@return The list of manifest files which have the highest attempt ID and are not empty
ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java