@SuppressWarnings("nls")
  /**
   * Split two tasks by creating a temporary file between them.
   *
   * @param op reduce sink operator being processed
   * @param parentTask the parent task
   * @param childTask the child task
   * @param opProcCtx context
   **/
  private static void splitTasks(ReduceSinkOperator op,
      Task<?> parentTask, Task<?> childTask,
      GenMRProcContext opProcCtx) throws SemanticException {
    if (op.getNumParent() != 1) {
      throw new IllegalStateException("Expecting operator " + op + " to have one parent. " +
          "But found multiple parents : " + op.getParentOperators());
    }

    ParseContext parseCtx = opProcCtx.getParseCtx();
    parentTask.addDependentTask(childTask);

    // Root Task cannot depend on any other task, therefore childTask cannot be
    // a root Task
    List<Task<?>> rootTasks = opProcCtx.getRootTasks();
    if (rootTasks.contains(childTask)) {
      rootTasks.remove(childTask);
    }

    // Generate the temporary file name
    Context baseCtx = parseCtx.getContext();
    Path taskTmpDir = baseCtx.getMRTmpPath();

    Operator<? extends OperatorDesc> parent = op.getParentOperators().get(0);
    TableDesc tt_desc = PlanUtils.getIntermediateFileTableDesc(PlanUtils
        .getFieldSchemasFromRowSchema(parent.getSchema(), "temporarycol"));

    // Create the temporary file, its corresponding FileSinkOperaotr, and
    // its corresponding TableScanOperator.
    TableScanOperator tableScanOp =
        createTemporaryFile(parent, op, taskTmpDir, tt_desc, parseCtx);

    Map<Operator<? extends OperatorDesc>, GenMapRedCtx> mapCurrCtx =
        opProcCtx.getMapCurrCtx();
    mapCurrCtx.put(tableScanOp, new GenMapRedCtx(childTask, null));

    String streamDesc = taskTmpDir.toUri().toString();
    MapredWork cplan = (MapredWork) childTask.getWork();

    if (needsTagging(cplan.getReduceWork())) {
      Operator<? extends OperatorDesc> reducerOp = cplan.getReduceWork().getReducer();
      String id = null;
      if (reducerOp instanceof JoinOperator) {
        if (parseCtx.getJoinOps().contains(reducerOp)) {
          id = ((JoinOperator)reducerOp).getConf().getId();
        }
      } else if (reducerOp instanceof MapJoinOperator) {
        if (parseCtx.getMapJoinOps().contains(reducerOp)) {
          id = ((MapJoinOperator)reducerOp).getConf().getId();
        }
      } else if (reducerOp instanceof SMBMapJoinOperator) {
        if (parseCtx.getSmbMapJoinOps().contains(reducerOp)) {
          id = ((SMBMapJoinOperator)reducerOp).getConf().getId();
        }
      }

      if (id != null) {
        streamDesc = id + ":$INTNAME";
      } else {
        streamDesc = "$INTNAME";
      }

      String origStreamDesc = streamDesc;
      int pos = 0;
      while (cplan.getMapWork().getAliasToWork().get(streamDesc) != null) {
        streamDesc = origStreamDesc.concat(String.valueOf(++pos));
      }

      // TODO: Allocate work to remove the temporary files and make that
      // dependent on the redTask
      cplan.getReduceWork().setNeedsTagging(true);
    }

    // Add the path to alias mapping
    setTaskPlan(taskTmpDir, streamDesc, tableScanOp, cplan.getMapWork(), false, tt_desc);
    opProcCtx.setCurrTopOp(null);
    opProcCtx.setCurrAliasId(null);
    opProcCtx.setCurrTask(childTask);
    opProcCtx.addRootIfPossible(parentTask);
  }
Break the pipeline between parent and child, and then
output data generated by parent to a temporary file stored in taskTmpDir.
A FileSinkOperator is added after parent to output the data.
Before child, we add a TableScanOperator to load data stored in the temporary
file back.
@param parent
@param child
@param taskTmpDir
@param tt_desc
@param parseCtx
@return The TableScanOperator inserted before child.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java