private static Path[] getDirectInsertDirectoryCandidatesRecursive(FileSystem fs,
      Path path, int skipLevels, PathFilter filter) throws IOException {
    String lastRelDir = null;
    HashSet<Path> results = new HashSet<Path>();
    String relRoot = Path.getPathWithoutSchemeAndAuthority(path).toString();
    if (!relRoot.endsWith(Path.SEPARATOR)) {
      relRoot += Path.SEPARATOR;
    }
    RemoteIterator<LocatedFileStatus> allFiles = fs.listFiles(path, true);
    while (allFiles.hasNext()) {
      LocatedFileStatus lfs = allFiles.next();
      Path lfsPath = lfs.getPath();
      Path dirPath = Path.getPathWithoutSchemeAndAuthority(lfsPath);
      String dir = dirPath.toString();
      if (!dir.startsWith(relRoot)) {
        throw new IOException("Path " + lfsPath + " is not under " + relRoot
            + " (when shortened to " + dir + ")");
      }
      String subDir = dir.substring(relRoot.length());
      Utilities.FILE_OP_LOGGER.trace("Looking at {} from {}", subDir, lfsPath);

      // If sorted, we'll skip a bunch of files.
      if (lastRelDir != null && subDir.startsWith(lastRelDir)) {
        continue;
      }
      int startIx = skipLevels > 0 ? -1 : 0;
      for (int i = 0; i < skipLevels; ++i) {
        startIx = subDir.indexOf(Path.SEPARATOR_CHAR, startIx + 1);
        if (startIx == -1) {
          Utilities.FILE_OP_LOGGER.info("Expected level of nesting ({}) is not "
              + " present in {} (from {})", skipLevels, subDir, lfsPath);
          break;
        }
      }
      if (startIx == -1) {
        continue;
      }
      int endIx = subDir.indexOf(Path.SEPARATOR_CHAR, startIx + 1);
      if (endIx == -1) {
        Utilities.FILE_OP_LOGGER.info("Expected level of nesting ({}) is not present in"
            + " {} (from {})", (skipLevels + 1), subDir, lfsPath);
        continue;
      }
      lastRelDir = subDir = subDir.substring(0, endIx);
      Path candidate = new Path(relRoot, subDir);
      if (!filter.accept(candidate)) {
        continue;
      }
      results.add(fs.makeQualified(candidate));
    }
    return results.toArray(new Path[results.size()]);
  }
Create row key and value object inspectors for reduce vectorization.
The row object inspector used by ReduceWork needs to be a **standard**
struct object inspector, not just any struct object inspector.
@param keyInspector
@param valueInspector
@return OI
@throws HiveException
ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java