@SuppressWarnings("unchecked")
  private static ConditionalTask createCondTask(HiveConf conf,
      Task<?> currTask, MoveWork mvWork, Serializable mergeWork,
      Path condInputPath, Path condOutputPath, Task<MoveWork> moveTaskToLink,
      DependencyCollectionTask dependencyTask, LineageState lineageState) {
    if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
      Utilities.FILE_OP_LOGGER.trace("Creating conditional merge task for " + condInputPath);
    }
    // Create a dummy task if no move is needed.
    Serializable moveWork = mvWork != null ? mvWork : new DependencyCollectionWork();

    // Note: this should never happen for mm tables.
    boolean shouldMergeMovePaths = (moveTaskToLink != null && dependencyTask == null
        && shouldMergeMovePaths(conf, condInputPath, condOutputPath, moveTaskToLink.getWork()));

    Serializable workForMoveOnlyTask = moveWork;
    if (shouldMergeMovePaths) {
      workForMoveOnlyTask = mergeMovePaths(condInputPath, moveTaskToLink.getWork(),
          lineageState);
    }

    // There are 3 options for this ConditionalTask:
    // 1) Merge the partitions
    // 2) Move the partitions (i.e. don't merge the partitions)
    // 3) Merge some partitions and move other partitions (i.e. merge some partitions and don't
    // merge others) in this case the merge is done first followed by the move to prevent
    // conflicts.
    // TODO: if we are not dealing with concatenate DDL, we should not create a merge+move path
    //       because it should be impossible to get incompatible outputs.
    Task<?> mergeOnlyMergeTask = TaskFactory.get(mergeWork);
    Task<?> moveOnlyMoveTask = TaskFactory.get(workForMoveOnlyTask);
    Task<?> mergeAndMoveMergeTask = TaskFactory.get(mergeWork);
    Task<?> mergeAndMoveMoveTask = TaskFactory.get(moveWork);

    // NOTE! It is necessary merge task is the parent of the move task, and not
    // the other way around, for the proper execution of the execute method of
    // ConditionalTask
    mergeAndMoveMergeTask.addDependentTask(mergeAndMoveMoveTask);

    List<Serializable> listWorks = new ArrayList<Serializable>();
    listWorks.add(workForMoveOnlyTask);
    listWorks.add(mergeWork);

    ConditionalWork cndWork = new ConditionalWork(listWorks);

    List<Task<?>> listTasks = new ArrayList<Task<?>>();
    listTasks.add(moveOnlyMoveTask);
    listTasks.add(mergeOnlyMergeTask);
    listTasks.add(mergeAndMoveMergeTask);

    ConditionalTask cndTsk = (ConditionalTask) TaskFactory.get(cndWork);
    cndTsk.setListTasks(listTasks);

    // create resolver
    cndTsk.setResolver(new ConditionalResolverMergeFiles());
    ConditionalResolverMergeFilesCtx mrCtx =
        new ConditionalResolverMergeFilesCtx(listTasks, condInputPath.toString());
    if (moveTaskToLink != null) {
      setStorageHandlerAndProperties(mrCtx, moveTaskToLink.getWork());
    }
    cndTsk.setResolverCtx(mrCtx);

    // make the conditional task as the child of the current leaf task
    currTask.addDependentTask(cndTsk);

    if (shouldMergeMovePaths) {
      // If a new MoveWork was created, then we should link all dependent tasks from the MoveWork to link.
      if (moveTaskToLink.getDependentTasks() != null) {
        for (Task dependentTask : moveTaskToLink.getDependentTasks()) {
          moveOnlyMoveTask.addDependentTask(dependentTask);
        }
      }
    } else {
      addDependentMoveTasks(moveTaskToLink, conf, moveOnlyMoveTask, dependencyTask);
    }


    addDependentMoveTasks(moveTaskToLink, conf, mergeOnlyMergeTask, dependencyTask);
    addDependentMoveTasks(moveTaskToLink, conf, mergeAndMoveMoveTask, dependencyTask);

    return cndTsk;
  }
Construct a conditional task given the current leaf task, the MoveWork and the MapredWork.

@param conf
         HiveConf
@param currTask
         current leaf task
@param dummyMoveWork
         MoveWork for the move task
@param mergeWork
         MapredWork for the merge task.
@param condInputPath
         the input directory of the merge/move task
@param condOutputPath
         the output directory of the merge/move task
@param moveTaskToLink
         a MoveTask that may be linked to the conditional sub-tasks
@param dependencyTask
         a dependency task that may be linked to the conditional sub-tasks
@param lineageState
         to track activity
@return The conditional task
ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java