public static void createMRWorkForMergingFiles(FileSinkOperator fsInput,
      Path finalName, DependencyCollectionTask dependencyTask,
      List<Task<MoveWork>> mvTasks, HiveConf conf,
      Task<?> currTask, LineageState lineageState)
      throws SemanticException {

    //
    // 1. create the operator tree
    //
    FileSinkDesc fsInputDesc = fsInput.getConf();
    if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
      Utilities.FILE_OP_LOGGER.trace("Creating merge work from " + System.identityHashCode(fsInput)
          + " with write ID " + (fsInputDesc.isMmTable() ? fsInputDesc.getTableWriteId() : null)
          + " into " + finalName);
    }

    boolean isBlockMerge = (conf.getBoolVar(ConfVars.HIVE_MERGE_RCFILE_BLOCK_LEVEL) &&
        fsInputDesc.getTableInfo().getInputFileFormatClass().equals(RCFileInputFormat.class)) ||
        (conf.getBoolVar(ConfVars.HIVE_MERGE_ORC_FILE_STRIPE_LEVEL) &&
            fsInputDesc.getTableInfo().getInputFileFormatClass().equals(OrcInputFormat.class));

    RowSchema inputRS = fsInput.getSchema();
    Long srcMmWriteId = fsInputDesc.isMmTable() ? fsInputDesc.getTableWriteId() : null;
    FileSinkDesc fsOutputDesc = null;
    TableScanOperator tsMerge = null;
    if (!isBlockMerge) {
      TableDesc ts = (TableDesc) fsInputDesc.getTableInfo().clone();
      String storageHandlerClass = ts.getProperties().getProperty(
              org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.META_TABLE_STORAGE);
      HiveStorageHandler storageHandler = null;
      try {
        storageHandler = HiveUtils.getStorageHandler(conf, storageHandlerClass);
      } catch (HiveException e) {
        throw new SemanticException(e);
      }
      boolean isCustomDelete = false;
      if (Context.Operation.DELETE.equals(fsInputDesc.getWriteOperation()) && storageHandler != null && storageHandler.supportsMergeFiles()) {
        storageHandler.setMergeTaskDeleteProperties(ts);
        isCustomDelete = true;
      }
      if (ts.getTableName() != null) {
        ts.getProperties().put(HiveCustomStorageHandlerUtils.MERGE_TASK_ENABLED +
            ts.getTableName(), Boolean.toString(Boolean.TRUE));
      }

      // Create a TableScan operator
      tsMerge = GenMapRedUtils.createTemporaryTableScanOperator(
          fsInput.getCompilationOpContext(), inputRS);

      // Create a FileSink operator
      Path mergeDest = srcMmWriteId == null ? finalName : finalName.getParent();
      fsOutputDesc = new FileSinkDesc(mergeDest, ts, conf.getBoolVar(ConfVars.COMPRESS_RESULT));
      if (isCustomDelete) {
        fsOutputDesc.setWriteOperation(Context.Operation.DELETE);
      }
      fsOutputDesc.setIsMerge(true);
      // Create and attach the filesink for the merge.
      OperatorFactory.getAndMakeChild(fsOutputDesc, inputRS, tsMerge);
    }

    // If the input FileSinkOperator is a dynamic partition enabled, the tsMerge input schema
    // needs to include the partition column, and the fsOutput should have
    // a DynamicPartitionCtx to indicate that it needs to dynamically partitioned.
    DynamicPartitionCtx dpCtx = fsInputDesc.getDynPartCtx();
    if (dpCtx != null && dpCtx.getNumDPCols() > 0) {
      // adding DP ColumnInfo to the RowSchema signature
      List<ColumnInfo> signature = inputRS.getSignature();
      String tblAlias = fsInputDesc.getTableInfo().getTableName();
      for (String dpCol : dpCtx.getDPColNames()) {
        ColumnInfo colInfo = new ColumnInfo(dpCol,
            TypeInfoFactory.stringTypeInfo, // all partition column type should be string
            tblAlias, true); // partition column is virtual column
        signature.add(colInfo);
      }
      inputRS.setSignature(signature);

      if (!isBlockMerge) {
      // create another DynamicPartitionCtx, which has a different input-to-DP column mapping
        DynamicPartitionCtx dpCtx2 = new DynamicPartitionCtx(dpCtx);
        fsOutputDesc.setDynPartCtx(dpCtx2);
      }

      // update the FileSinkOperator to include partition columns
      usePartitionColumns(fsInputDesc.getTableInfo().getProperties(), fsInputDesc.getTable(), dpCtx.getDPColNames());
    } else {
      // non-partitioned table
      fsInputDesc.getTableInfo().getProperties().remove(
        org.apache.hadoop.hive.metastore.api.hive_metastoreConstants.META_TABLE_PARTITION_COLUMNS);
    }

    //
    // 2. Constructing a conditional task consisting of a move task and a map reduce task
    //
    Path inputDirName = fsInputDesc.getMergeInputDirName();
    MapWork cplan;
    Serializable work;

    if (isBlockMerge) {
      cplan = GenMapRedUtils.createMergeTask(fsInputDesc, finalName,
          dpCtx != null && dpCtx.getNumDPCols() > 0, fsInput.getCompilationOpContext());
      if (conf.getVar(ConfVars.HIVE_EXECUTION_ENGINE).equals("tez")) {
        work = new TezWork(conf.getVar(HiveConf.ConfVars.HIVE_QUERY_ID), conf);
        cplan.setName("File Merge");
        ((TezWork) work).add(cplan);
      } else {
        work = cplan;
      }
    } else {
      cplan = createMRWorkForMergingFiles(conf, tsMerge, fsInputDesc);
      if (conf.getVar(ConfVars.HIVE_EXECUTION_ENGINE).equals("tez")) {
        work = new TezWork(conf.getVar(HiveConf.ConfVars.HIVE_QUERY_ID), conf);
        cplan.setName("File Merge");
        ((TezWork)work).add(cplan);
      } else {
        work = new MapredWork();
        ((MapredWork)work).setMapWork(cplan);
      }
    }
    // use CombineHiveInputFormat for map-only merging
    cplan.setInputformat("org.apache.hadoop.hive.ql.io.CombineHiveInputFormat");
    // NOTE: we should gather stats in MR1 rather than MR2 at merge job since we don't
    // know if merge MR2 will be triggered at execution time
    MoveWork dummyMv = null;
    if (srcMmWriteId == null && !fsInputDesc.isIcebergTable()) {
      // Only create the movework for non-MM table. No action needed for a MM table.
      dummyMv = new MoveWork(null, null, null,
          new LoadFileDesc(inputDirName, finalName, true, null, null, false), false);
    }
    // Use the original fsOp path here in case of MM - while the new FSOP merges files inside the
    // MM directory, the original MoveTask still commits based on the parent. Note that this path
    // can only be triggered for a merge that's part of insert for now; MM tables do not support
    // concatenate. Keeping the old logic for non-MM tables with temp directories and stuff.
    Path fsopPath = srcMmWriteId != null ? fsInputDesc.getFinalDirName() : finalName;

    Task<MoveWork> mvTask = GenMapRedUtils.findMoveTaskForFsopOutput(mvTasks, fsopPath, fsInputDesc.isMmTable(),
        fsInputDesc.isDirectInsert(), fsInputDesc.getMoveTaskId(), fsInputDesc.getAcidOperation());
    ConditionalTask cndTsk = GenMapRedUtils.createCondTask(conf, currTask, dummyMv, work,
        fsInputDesc.getMergeInputDirName(), finalName, mvTask, dependencyTask, lineageState);

    // keep the dynamic partition context in conditional task resolver context
    ConditionalResolverMergeFilesCtx mrCtx =
        (ConditionalResolverMergeFilesCtx) cndTsk.getResolverCtx();
    mrCtx.setDPCtx(fsInputDesc.getDynPartCtx());
    mrCtx.setLbCtx(fsInputDesc.getLbCtx());

  }
@param fsInput The FileSink operator.
@param finalName the final destination path the merge job should output.
@param dependencyTask
@param mvTasks
@param conf
@param currTask
@param lineageState
@throws SemanticException
create a Map-only merge job using CombineHiveInputFormat for all partitions with
following operators:
         MR job J0:
         ...
         |
         v
         FileSinkOperator_1 (fsInput)
         |
         v
         Merge job J1:
         |
         v
         TableScan (using CombineHiveInputFormat) (tsMerge)
         |
         v
         FileSinkOperator (fsMerge)

         Here the pathToPartitionInfo &amp; pathToAlias will remain the same, which means the paths
         do
         not contain the dynamic partitions (their parent). So after the dynamic partitions are
         created (after the first job finished before the moveTask or ConditionalTask start),
         we need to change the pathToPartitionInfo &amp; pathToAlias to include the dynamic
         partition
         directories.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java