public static MapWork createMergeTask(FileSinkDesc fsInputDesc, Path finalName,
      boolean hasDynamicPartitions, CompilationOpContext ctx) throws SemanticException {

    Path inputDir = fsInputDesc.getMergeInputDirName();

    TableDesc tblDesc = fsInputDesc.getTableInfo();

    List<Path> inputDirs = new ArrayList<Path>(1);
    ArrayList<String> inputDirstr = new ArrayList<String>(1);
    // this will be populated by MergeFileWork.resolveDynamicPartitionStoredAsSubDirsMerge
    // in case of dynamic partitioning and list bucketing
    if (!hasDynamicPartitions &&
        !GenMapRedUtils.isSkewedStoredAsDirs(fsInputDesc)) {
      inputDirs.add(inputDir);
    }
    inputDirstr.add(inputDir.toString());

    // internal input format class for CombineHiveInputFormat
    final Class<? extends InputFormat> internalIFClass;
    if (tblDesc.getInputFileFormatClass().equals(RCFileInputFormat.class)) {
      internalIFClass = RCFileBlockMergeInputFormat.class;
    } else if (tblDesc.getInputFileFormatClass().equals(OrcInputFormat.class)) {
      internalIFClass = OrcFileStripeMergeInputFormat.class;
    } else {
      throw new SemanticException("createMergeTask called on a table with file"
          + " format other than RCFile or ORCFile");
    }

    if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
      Utilities.FILE_OP_LOGGER.trace("creating mergefilework from " + inputDirs + " to " + finalName);
    }
    // create the merge file work
    MergeFileWork work = new MergeFileWork(inputDirs, finalName,
        hasDynamicPartitions, tblDesc.getInputFileFormatClass().getName(), tblDesc);
    Map<Path, List<String>> pathToAliases = new LinkedHashMap<>();
    pathToAliases.put(inputDir, inputDirstr);
    work.setMapperCannotSpanPartns(true);
    work.setPathToAliases(pathToAliases);
    PartitionDesc pDesc = new PartitionDesc(tblDesc, null);
    pDesc.setInputFileFormatClass(internalIFClass);
    work.addPathToPartitionInfo(inputDir, pDesc);
    work.setListBucketingCtx(fsInputDesc.getLbCtx());

    // create alias to work which contains the merge operator
    LinkedHashMap<String, Operator<? extends OperatorDesc>> aliasToWork =
        new LinkedHashMap<String, Operator<? extends OperatorDesc>>();
    Operator<? extends OperatorDesc> mergeOp = null;
    final FileMergeDesc fmd;
    if (tblDesc.getInputFileFormatClass().equals(RCFileInputFormat.class)) {
      fmd = new RCFileMergeDesc();
    } else {
      fmd = new OrcFileMergeDesc();
    }
    fmd.setIsMmTable(fsInputDesc.isMmTable());
    boolean isCompactionTable = AcidUtils.isCompactionTable(tblDesc.getProperties());
    fmd.setIsCompactionTable(isCompactionTable);
    fmd.setWriteId(fsInputDesc.getTableWriteId());
    int stmtId = fsInputDesc.getStatementId();
    fmd.setStmtId(stmtId == -1 ? 0 : stmtId);
    fmd.setDpCtx(fsInputDesc.getDynPartCtx());
    fmd.setOutputPath(finalName);
    fmd.setHasDynamicPartitions(work.hasDynamicPartitions());
    fmd.setListBucketingAlterTableConcatenate(work.isListBucketingAlterTableConcatenate());
    int lbLevel = work.getListBucketingCtx() == null ? 0 :
      work.getListBucketingCtx().calculateListBucketingLevel();
    fmd.setListBucketingDepth(lbLevel);
    mergeOp = OperatorFactory.get(ctx, fmd);
    aliasToWork.put(inputDir.toString(), mergeOp);
    work.setAliasToWork(aliasToWork);

    return work;
  }
Create a block level merge task for RCFiles or stripe level merge task for
ORCFiles

@param fsInputDesc
@param finalName
@param hasDynamicPartitions
@param ctx
@return MergeWork if table is stored as RCFile or ORCFile,
        null otherwise
ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java