@Override
  public void closeOp(boolean abort) throws HiveException {
    try {
      if (abort) {
        if (!autoDelete || isMmTable) {
          fs.delete(outPath, true);
        }
        return;
      }
      // if outPath does not exist, then it means all paths within combine split are skipped as
      // they are incompatible for merge (for example: files without stripe stats).
      // Those files will be added to incompatFileSet
      if (fs.exists(outPath)) {
        FileStatus fss = fs.getFileStatus(outPath);
        if (!isMmTable) {
          if (!fs.rename(outPath, finalPath)) {
            throw new IOException("Unable to rename " + outPath + " to " + finalPath);
          }
          LOG.info("Renamed path " + outPath + " to " + finalPath
              + "(" + fss.getLen() + " bytes).");
        } else {
          assert finalPath.equals(outPath);
          // There's always just one file that we have merged.
          // The union/DP/etc. should already be account for in the path.
          Utilities.writeCommitManifest(Lists.newArrayList(outPath), tmpPath.getParent(), fs, taskId, conf.getWriteId(),
              conf.getStmtId(), null, false, hasDynamicPartitions, new HashSet<>(), null, false);
          LOG.info("Merged into " + finalPath + "(" + fss.getLen() + " bytes).");
        }
      }

      // move any incompatible files to final path
      if (incompatFileSet != null && !incompatFileSet.isEmpty()) {
        if (isMmTable) {
          // We only support query-time merge for MM tables, so don't handle this.
          throw new HiveException("Incompatible files should not happen in MM tables.");
        }
        Path destDir = finalPath.getParent();
        // move any incompatible files to final path
        if (incompatFileSet != null && !incompatFileSet.isEmpty()) {
          for (Path incompatFile : incompatFileSet) {
            // Hive expects filenames to be in specific format
            // like 000000_0, but "LOAD DATA" commands can let you add any files to any partitions/tables without
            // renaming.
            // This can cause a few issues:
            // MoveTask will remove files in some cases where MoveTask assumes the files are are generated by
            // speculatively executed tasks.
            // Example: MoveTask thinks the following files are same
            // part-m-00000_1417075294718
            // part-m-00001_1417075294718
            // Assumes 1417075294718 as taskId and retains only large file supposedly generated by speculative execution.
            // This can result in data loss in case of CONCATENATE/merging.

            // If filename is consistent with XXXXXX_N and another task with same task-id runs after this move, then
            // the same file name is used in the other task which will result in task failure and retry of task and
            // subsequent removal of this file as duplicate.
            // Example: if the file name is 000001_0 and another task runs with taskid 000001_0, it will fail to create
            // the file and next attempt will create 000001_1, both the files will be considered as output of same task
            // and only 000001_1 will be picked resulting it loss of existing file 000001_0.
            final String destFileName = Utilities.getTaskId(jc) + Utilities.COPY_KEYWORD + 1;
            try {
              Path destPath = Utilities.moveFile(fs, incompatFile, destDir, destFileName);
              LOG.info("Moved incompatible file " + incompatFile + " to " + destPath);
            } catch (HiveException e) {
              LOG.error("Unable to move " + incompatFile + " to " + destDir + ", " + destFileName);
              throw new IOException(e);
            }
          }
        }

      }
    } catch (IOException e) {
      throw new HiveException("Failed to close AbstractFileMergeOperator", e);
    }
  }
Validates that each input path belongs to the same partition since each
mapper merges the input to a single output directory

@param inputPath - input path
ql/src/java/org/apache/hadoop/hive/ql/exec/AbstractFileMergeOperator.java