private void launchCompactionJob(JobConf job, Path baseDir, CompactionType compactionType,
                                   StringableList dirsToSearch,
                                   List<AcidUtils.ParsedDelta> parsedDeltas,
                                   int curDirNumber, int obsoleteDirNumber, HiveConf hiveConf,
                                   IMetaStoreClient msc, long id, String jobName) throws IOException {
    job.setBoolean(IS_MAJOR, compactionType == CompactionType.MAJOR);
    if(dirsToSearch == null) {
      dirsToSearch = new StringableList();
    }
    StringableList deltaDirs = new StringableList();
    // Note: if compaction creates a delta, it won't replace an existing base dir, so the txn ID
    //       of the base dir won't be a part of delta's range. If otoh compaction creates a base,
    //       we don't care about this value because bases don't have min txn ID in the name.
    //       However logically this should also take base into account if it's included.
    long minTxn = Long.MAX_VALUE;
    long maxTxn = Long.MIN_VALUE;
    for (AcidUtils.ParsedDelta delta : parsedDeltas) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Adding delta {} to directories to search", delta.getPath());
      }
      dirsToSearch.add(delta.getPath());
      deltaDirs.add(delta.getPath());
      minTxn = Math.min(minTxn, delta.getMinWriteId());
      maxTxn = Math.max(maxTxn, delta.getMaxWriteId());
    }

    if (baseDir != null) job.set(BASE_DIR, baseDir.toString());
    job.set(DELTA_DIRS, deltaDirs.toString());
    job.set(DIRS_TO_SEARCH, dirsToSearch.toString());
    job.setLong(MIN_TXN, minTxn);
    job.setLong(MAX_TXN, maxTxn);
    // HIVE-23354 enforces that MR speculative execution is disabled
    job.setBoolean(MRJobConfig.REDUCE_SPECULATIVE, false);
    job.setBoolean(MRJobConfig.MAP_SPECULATIVE, false);

    // Add tokens for all the file system in the input path.
    ArrayList<Path> dirs = new ArrayList<>();
    if (baseDir != null) {
      dirs.add(baseDir);
    }
    dirs.addAll(deltaDirs);
    dirs.addAll(dirsToSearch);
    TokenCache.obtainTokensForNamenodes(job.getCredentials(), dirs.toArray(new Path[]{}), job);

    if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVE_IN_TEST)) {
      mrJob = job;
    }
    if (LOG.isInfoEnabled()) {
      LOG.info("Submitting {} compaction job '{}' to {} queue. (current delta dirs count={}, obsolete delta dirs " +
              "count={}. TxnIdRange[{},{}}]",
          compactionType, job.getJobName(), job.getQueueName(), curDirNumber, obsoleteDirNumber, minTxn, maxTxn);
    }
    try (JobClient jc = new JobClient(job)) {
      RunningJob rj = jc.submitJob(job);
      if (LOG.isInfoEnabled()) {
        LOG.info("Submitted compaction job '{}' with jobID={} compaction ID={}", job.getJobName(), rj.getID(), id);
      }
      try {
        msc.setHadoopJobid(rj.getID().toString(), id);
      } catch (TException e) {
        LOG.warn("Error setting hadoop job, jobId=" + rj.getID().toString()
            + " compactionId=" + id, e);
      }
      rj.waitForCompletion();
      if (!rj.isSuccessful()) {
        throw new IOException((compactionType == CompactionType.MAJOR ? "Major" : "Minor") +
            " compactor job failed for " + jobName + "! Hadoop JobId: " + rj.getID());
      }
    }
  }
@param baseDir if not null, it's either table/partition root folder or base_xxxx.
               If it's base_xxxx, it's in dirsToSearch, else the actual original files
               (all leaves recursively) are in the dirsToSearch list
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MRCompactor.java