private void processCurrentTask(Task<?> currTask,
        ConditionalTask conditionalTask) throws SemanticException {
      // get current mapred work and its local work
      MapredWork mapredWork = (MapredWork) currTask.getWork();
      MapredLocalWork localwork = mapredWork.getMapWork().getMapRedLocalWork();
      if (localwork != null) {
        // get the context info and set up the shared tmp URI
        Context ctx = physicalContext.getContext();
        Path tmpPath = Utilities.generateTmpPath(ctx.getLocalTmpPath(), currTask.getId());
        localwork.setTmpPath(tmpPath);
        mapredWork.getMapWork().setTmpHDFSPath(Utilities.generateTmpPath(
          ctx.getMRTmpPath(), currTask.getId()));
        // create a task for this local work; right now, this local work is shared
        // by the original MapredTask and this new generated MapredLocalTask.
        MapredLocalTask localTask = (MapredLocalTask) TaskFactory.get(localwork);

        // set the backup task from curr task
        localTask.setBackupTask(currTask.getBackupTask());
        localTask.setBackupChildrenTasks(currTask.getBackupChildrenTasks());
        currTask.setBackupChildrenTasks(null);
        currTask.setBackupTask(null);

        if (currTask.getTaskTag() == Task.CONVERTED_MAPJOIN) {
          localTask.setTaskTag(Task.CONVERTED_MAPJOIN_LOCAL);
        } else {
          localTask.setTaskTag(Task.HINTED_MAPJOIN_LOCAL);
          currTask.setTaskTag(Task.HINTED_MAPJOIN);
        }
        // replace the map join operator to local_map_join operator in the operator tree
        // and return all the dummy parent
        LocalMapJoinProcCtx localMapJoinProcCtx = adjustLocalTask(localTask);
        List<Operator<? extends OperatorDesc>> dummyOps =
         localMapJoinProcCtx.getDummyParentOp();

        // create new local work and setup the dummy ops
        MapredLocalWork newLocalWork = localwork.extractDirectWorks(
            localMapJoinProcCtx.getDirectWorks());
        newLocalWork.setDummyParentOp(dummyOps);
        mapredWork.getMapWork().setMapRedLocalWork(newLocalWork);

        if (localwork.getAliasToFetchWork().isEmpty()) {
          // no alias to stage.. no local task
          newLocalWork.setHasStagedAlias(false);
          currTask.setBackupTask(localTask.getBackupTask());
          currTask.setBackupChildrenTasks(localTask.getBackupChildrenTasks());
          return;
        }
        newLocalWork.setHasStagedAlias(true);
        // get all parent tasks
        List<Task<?>> parentTasks = currTask.getParentTasks();
        currTask.setParentTasks(null);
        if (parentTasks != null) {
          for (Task<?> tsk : parentTasks) {
            // make new generated task depends on all the parent tasks of current task.
            tsk.addDependentTask(localTask);
            // remove the current task from its original parent task's dependent task
            tsk.removeDependentTask(currTask);
          }
        } else {
          // in this case, current task is in the root tasks
          // so add this new task into root tasks and remove the current task from root tasks
          if (conditionalTask == null) {
            physicalContext.addToRootTask(localTask);
            physicalContext.removeFromRootTask(currTask);
          } else {
            // set list task
            List<Task<?>> listTask = conditionalTask.getListTasks();
            ConditionalWork conditionalWork = conditionalTask.getWork();
            int index = listTask.indexOf(currTask);
            listTask.set(index, localTask);
            // set list work
            List<Serializable> listWork = (List<Serializable>) conditionalWork.getListWorks();
            index = listWork.indexOf(mapredWork);
            listWork.set(index, localwork);
            conditionalWork.setListWorks(listWork);
            ConditionalResolver resolver = conditionalTask.getResolver();
            if (resolver instanceof ConditionalResolverSkewJoin) {
              // get bigKeysDirToTaskMap
              ConditionalResolverSkewJoinCtx context = (ConditionalResolverSkewJoinCtx) conditionalTask
                  .getResolverCtx();
              HashMap<Path, Task<?>> bigKeysDirToTaskMap = context
                  .getDirToTaskMap();
              // to avoid concurrent modify the hashmap
              HashMap<Path, Task<?>> newbigKeysDirToTaskMap = new HashMap<Path, Task<?>>();
              // reset the resolver
              for (Map.Entry<Path, Task<?>> entry : bigKeysDirToTaskMap
                  .entrySet()) {
                Task<?> task = entry.getValue();
                Path key = entry.getKey();
                if (task.equals(currTask)) {
                  newbigKeysDirToTaskMap.put(key, localTask);
                } else {
                  newbigKeysDirToTaskMap.put(key, task);
                }
              }
              context.setDirToTaskMap(newbigKeysDirToTaskMap);
              conditionalTask.setResolverCtx(context);
            } else if (resolver instanceof ConditionalResolverCommonJoin) {
              // get bigKeysDirToTaskMap
              ConditionalResolverCommonJoinCtx context = (ConditionalResolverCommonJoinCtx) conditionalTask
                  .getResolverCtx();
              HashMap<Task<?>, Set<String>> taskToAliases = context.getTaskToAliases();
              // to avoid concurrent modify the hashmap
              // Must be deterministic order map for consistent q-test output across Java versions
              HashMap<Task<?>, Set<String>> newTaskToAliases =
                  new LinkedHashMap<Task<?>, Set<String>>();
              // reset the resolver
              for (Map.Entry<Task<?>, Set<String>> entry : taskToAliases.entrySet()) {
                Task<?> task = entry.getKey();
                Set<String> key = new HashSet<String>(entry.getValue());

                if (task.equals(currTask)) {
                  newTaskToAliases.put(localTask, key);
                } else {
                  newTaskToAliases.put(task, key);
                }
              }
              context.setTaskToAliases(newTaskToAliases);
              conditionalTask.setResolverCtx(context);
            }
          }
        }
        // make current task depends on this new generated localMapJoinTask
        // now localTask is the parent task of the current task
        localTask.addDependentTask(currTask);
      }
    }
Iterator each tasks. If this task has a local work,create a new task for this local work, named
MapredLocalTask. then make this new generated task depends on current task's parent task, and
make current task depends on this new generated task
ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java