public static void handleDirectInsertTableFinalPath(Path specPath, String unionSuffix, Configuration hconf,
      boolean success, int dpLevels, int lbLevels, MissingBucketsContext mbc, long writeId, int stmtId,
      Reporter reporter, boolean isMmTable, boolean isMmCtas, boolean isInsertOverwrite, boolean isDirectInsert,
      String staticSpec, AcidUtils.Operation acidOperation, FileSinkDesc conf) throws IOException, HiveException {
    FileSystem fs = specPath.getFileSystem(hconf);
    boolean isDelete = AcidUtils.Operation.DELETE.equals(acidOperation);
    Path manifestDir = getManifestDir(specPath, writeId, stmtId, unionSuffix, isInsertOverwrite, staticSpec, isDelete);
    if (!success) {
      AcidUtils.IdPathFilter filter = new AcidUtils.IdPathFilter(writeId, stmtId);
      tryDeleteAllDirectInsertFiles(fs, specPath, manifestDir, dpLevels, lbLevels,
          filter, writeId, stmtId, hconf, acidOperation);
      return;
    }

    Utilities.FILE_OP_LOGGER.debug("Looking for manifests in: {} ({})", manifestDir, writeId);
    List<Path> manifests = new ArrayList<>();
    try {
      FileStatus[] manifestFiles = fs.listStatus(manifestDir);
      manifests = selectManifestFiles(manifestFiles);
    } catch (FileNotFoundException ex) {
      Utilities.FILE_OP_LOGGER.info("No manifests found in directory {} - query produced no output", manifestDir);
      manifestDir = null;
      if (!fs.exists(specPath)) {
        // Empty insert to new partition
        fs.mkdirs(specPath);
      }
    }

    Map<String, List<Path>> dynamicPartitionSpecs = new HashMap<>();
    Set<Path> committed = Collections.newSetFromMap(new ConcurrentHashMap<>());
    Set<Path> directInsertDirectories = new HashSet<>();
    for (Path mfp : manifests) {
      Utilities.FILE_OP_LOGGER.info("Looking at manifest file: {}", mfp);
      try (FSDataInputStream mdis = fs.open(mfp)) {
        if (dpLevels > 0) {
          int partitionCount = mdis.readInt();
          for (int i = 0; i < partitionCount; ++i) {
            String nextPart = mdis.readUTF();
            Utilities.FILE_OP_LOGGER.debug("Looking at dynamic partition {}", nextPart);
            if (!dynamicPartitionSpecs.containsKey(nextPart)) {
              dynamicPartitionSpecs.put(nextPart, new ArrayList<>());
            }
          }
        }

        int fileCount = mdis.readInt();
        for (int i = 0; i < fileCount; ++i) {
          String nextFile = mdis.readUTF();
          Utilities.FILE_OP_LOGGER.debug("Looking at committed file {}", nextFile);
          Path path = fs.makeQualified(new Path(nextFile));
          if (!committed.add(path)) {
            throw new HiveException(nextFile + " was specified in multiple manifests");
          }
          dynamicPartitionSpecs.entrySet()
              .stream()
              .filter(dynpath -> path.toString().contains(dynpath.getKey()))
              .findAny()
              .ifPresent(dynPath -> dynPath.getValue().add(path));

          Path parentDirPath = path.getParent();
          while (AcidUtils.isChildOfDelta(parentDirPath, specPath)) {
            // 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).
            parentDirPath = parentDirPath.getParent();
          }
          directInsertDirectories.add(parentDirPath);
        }
      }
    }

    if (manifestDir != null) {
      Utilities.FILE_OP_LOGGER.info("Deleting manifest directory {}", manifestDir);
      tryDelete(fs, manifestDir);
      if (unionSuffix != null) {
        // Also delete the parent directory if we are the last union FSOP to execute.
        manifestDir = manifestDir.getParent();
        FileStatus[] remainingFiles = fs.listStatus(manifestDir);
        if (remainingFiles == null || remainingFiles.length == 0) {
          Utilities.FILE_OP_LOGGER.info("Deleting manifest directory {}", manifestDir);
          tryDelete(fs, manifestDir);
        }
      }
    }

    if (!directInsertDirectories.isEmpty()) {
      cleanDirectInsertDirectoriesConcurrently(directInsertDirectories, committed, fs, hconf, unionSuffix, lbLevels);
    }

    conf.setDynPartitionValues(dynamicPartitionSpecs);

    if (!committed.isEmpty()) {
      throw new HiveException("The following files were committed but not found: " + committed);
    }

    if (directInsertDirectories.isEmpty()) {
      return;
    }

    // TODO: see HIVE-14886 - removeTempOrDuplicateFiles is broken for list bucketing,
    //       so maintain parity here by not calling it at all.
    if (lbLevels != 0) {
      return;
    }

    if (!isDirectInsert) {
      // Create fake file statuses to avoid querying the file system. removeTempOrDuplicateFiles
      // doesn't need to check anything except path and directory status for MM directories.
      FileStatus[] finalResults = directInsertDirectories.stream()
          .map(PathOnlyFileStatus::new)
          .toArray(FileStatus[]::new);
      List<Path> emptyBuckets = Utilities.removeTempOrDuplicateFiles(fs, finalResults,
          unionSuffix, dpLevels, mbc == null ? 0 : mbc.numBuckets, hconf, writeId, stmtId,
            isMmTable, null, isInsertOverwrite);
      // create empty buckets if necessary
      if (!emptyBuckets.isEmpty()) {
        assert mbc != null;
        Utilities.createEmptyBuckets(hconf, emptyBuckets, mbc.isCompressed, mbc.tableInfo, reporter);
      }
    }
  }
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