@Override
    public void commitJob(JobContext context) throws IOException {
      JobConf conf = ShimLoader.getHadoopShims().getJobConf(context);
      Path tmpLocation = new Path(conf.get(TMP_LOCATION));//this contains base_xxx or delta_xxx_yyy
      Path finalLocation = new Path(conf.get(FINAL_LOCATION));
      FileSystem fs = tmpLocation.getFileSystem(conf);
      if (LOG.isDebugEnabled()) {
        LOG.debug("Moving contents of {} to {}", tmpLocation, finalLocation);
      }
      if(!fs.exists(tmpLocation)) {
        /*
         * No 'tmpLocation' may happen if job generated created 0 splits, which happens if all
         * input delta and/or base files were empty or had
         * only {@link org.apache.orc.impl.OrcAcidUtils#getSideFile(Path)} files.
         * So make sure the new base/delta is created.
         */
        AcidOutputFormat.Options options = new AcidOutputFormat.Options(conf)
            .writingBase(conf.getBoolean(IS_MAJOR, false))
            .isCompressed(conf.getBoolean(IS_COMPRESSED, false))
            .minimumWriteId(conf.getLong(MIN_TXN, Long.MAX_VALUE))
            .maximumWriteId(conf.getLong(MAX_TXN, Long.MIN_VALUE))
            .bucket(0)
            .statementId(-1)
            .visibilityTxnId(Compactor.getCompactorTxnId(conf));
        Path newDeltaDir = AcidUtils.createFilename(finalLocation, options).getParent();
        if (LOG.isInfoEnabled()) {
          LOG.info("{}: {} not found.  Assuming 0 splits.  Creating {}", context.getJobID(), tmpLocation, newDeltaDir);
        }
        fs.mkdirs(newDeltaDir);
        if (options.isWriteVersionFile()) {
          AcidUtils.OrcAcidVersion.writeVersionFile(newDeltaDir, fs);
        }
        return;
      }
      FileStatus[] contents = fs.listStatus(tmpLocation);
      AcidOutputFormat.Options options = new AcidOutputFormat.Options(conf);
      //minor compaction may actually have delta_x_y and delete_delta_x_y
      for (FileStatus fileStatus : contents) {
        //newPath is the base/delta dir
        Path newPath = new Path(finalLocation, fileStatus.getPath().getName());
        /*rename(A, B) has "interesting" behavior if A and B are directories. If  B doesn't exist,
        * it does the expected operation and everything that was in A is now in B.  If B exists,
        * it will make A a child of B...  thus make sure the rename() is done before creating the
        * meta files which will create base_x/ (i.e. B)...*/
        fs.rename(fileStatus.getPath(), newPath);
        if (options.isWriteVersionFile()) {
          AcidUtils.OrcAcidVersion.writeVersionFile(newPath, fs);
        }
      }
      fs.delete(tmpLocation, true);
    }
This input format returns its own input split as a value.  This is because our splits
contain information needed to properly construct the writer.  Crazy, huh?
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MRCompactor.java