public void run(HiveConf conf, Table t, StorageDescriptor sd, ValidWriteIdList writeIds,
           CompactionInfo ci, AcidDirectory dir) throws IOException {

    String jobName = ci.workerId + "-compactor-" + ci.getFullPartitionName();
    JobConf job = createBaseJobConf(conf, jobName, t, sd, writeIds, ci);

    List<AcidUtils.ParsedDelta> parsedDeltas = dir.getCurrentDirectories();
    int maxDeltasToHandle = conf.getIntVar(HiveConf.ConfVars.COMPACTOR_MAX_NUM_DELTA);
    if (parsedDeltas.size() > maxDeltasToHandle) {
      /*
       * if here, that means we have very high number of delta files.  This may be sign of a temporary
       * glitch or a real issue.  For example, if transaction batch size or transaction size is set too
       * low for the event flow rate in Streaming API, it may generate lots of delta files very
       * quickly.  Another possibility is that Compaction is repeatedly failing and not actually compacting.
       * Thus, force N minor compactions first to reduce number of deltas and then follow up with
       * the compaction actually requested in {@link ci} which now needs to compact a lot fewer deltas
       */
      if (LOG.isWarnEnabled()) {
        LOG.warn("{} delta files found for {} located at {}! This is likely a sign of misconfiguration, " +
                "especially if this message repeats.  Check that compaction is running properly.  Check for any " +
                "runaway/mis-configured process writing to ACID tables, especially using Streaming Ingest API.",
            parsedDeltas.size(), ci.getFullPartitionName(), sd.getLocation());
      }
      int numMinorCompactions = parsedDeltas.size() / maxDeltasToHandle;
      parsedDeltas.sort(AcidUtils.ParsedDeltaLight::compareTo);

      int start = 0;
      int end = maxDeltasToHandle;

      for (int jobSubId = 0; jobSubId < numMinorCompactions; jobSubId++) {
        while (end > 0 && end < parsedDeltas.size() &&
          parsedDeltas.get(end).getMinWriteId() == parsedDeltas.get(end - 1).getMinWriteId() &&
          parsedDeltas.get(end).getMaxWriteId() == parsedDeltas.get(end - 1).getMaxWriteId()) {
          end--;
        }
        List<AcidUtils.ParsedDelta> split = parsedDeltas.subList(start, end);
        start = end;
        end = start + maxDeltasToHandle;

        JobConf jobMinorCompact = createBaseJobConf(conf, jobName + "_" + jobSubId, t, sd, writeIds, ci);
        launchCompactionJob(jobMinorCompact,
          null, CompactionType.MINOR, null,
            split, split.size(), -1, conf, msc, ci.id, jobName);
      }
      //now recompute state since we've done minor compactions and have different 'best' set of deltas
      dir = AcidUtils.getAcidState(null, new Path(sd.getLocation()), conf, writeIds, Ref.from(false), false);
    }

    StringableList dirsToSearch = new StringableList();
    Path baseDir = null;
    if (CompactionType.MAJOR.equals(ci.type)) {
      // There may not be a base dir if the partition was empty before inserts or if this
      // partition is just now being converted to ACID.
      baseDir = dir.getBaseDirectory();
      if (baseDir == null) {
        List<HdfsFileStatusWithId> originalFiles = dir.getOriginalFiles();
        if (originalFiles != null && !originalFiles.isEmpty()) {
          // There are original format files
          for (HdfsFileStatusWithId stat : originalFiles) {
            Path path = stat.getFileStatus().getPath();
            //note that originalFiles are all original files recursively not dirs
            dirsToSearch.add(path);
            if (LOG.isDebugEnabled()) {
              LOG.debug("Adding original file {} to dirs to search", path);
            }
          }
          // Set base to the location so that the input format reads the original files.
          baseDir = new Path(sd.getLocation());
        }
      } else {
        // add our base to the list of directories to search for files in.
        if (LOG.isDebugEnabled()) {
          LOG.debug("Adding base directory {} to dirs to search", baseDir);
        }
        dirsToSearch.add(baseDir);
      }
    }

    launchCompactionJob(job, baseDir, ci.type, dirsToSearch, dir.getCurrentDirectories(),
      dir.getCurrentDirectories().size(), dir.getObsolete().size(), conf, msc, ci.id, jobName);
  }
Run Compaction which may consist of several jobs on the cluster.
@param conf Hive configuration file
@param t metastore table
@param sd metastore storage descriptor
@param writeIds list of valid write ids
@param ci CompactionInfo
@throws java.io.IOException if the job fails
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MRCompactor.java