@Override
  public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx opProcCtx,
      Object... nodeOutputs) throws SemanticException {
    UnionOperator union = (UnionOperator) nd;
    GenMRProcContext ctx = (GenMRProcContext) opProcCtx;
    ParseContext parseCtx = ctx.getParseCtx();
    UnionProcContext uCtx = parseCtx.getUCtx();

    // Map-only subqueries can be optimized in future to not write to a file in
    // future
    Map<Operator<? extends OperatorDesc>, GenMapRedCtx> mapCurrCtx = ctx.getMapCurrCtx();

    if (union.getConf().isAllInputsInSameReducer()) {
      // All inputs of this UnionOperator are in the same Reducer.
      // We do not need to break the operator tree.
      mapCurrCtx.put((Operator<? extends OperatorDesc>) nd,
        new GenMapRedCtx(ctx.getCurrTask(),ctx.getCurrAliasId()));
      return null;
    }

    UnionParseContext uPrsCtx = uCtx.getUnionParseContext(union);

    ctx.setCurrUnionOp(union);
    // The plan needs to be broken only if one of the sub-queries involve a
    // map-reduce job
    if (uPrsCtx.allMapOnlySubQ()) {
      return processMapOnlyUnion(union, stack, ctx, uCtx);
    }

    assert uPrsCtx != null;

    Task<?> currTask = ctx.getCurrTask();
    int pos = UnionProcFactory.getPositionParent(union, stack);

    Task<?> uTask = null;
    MapredWork uPlan = null;

    // union is encountered for the first time
    GenMRUnionCtx uCtxTask = ctx.getUnionTask(union);
    if (uCtxTask == null) {
      uPlan = GenMapRedUtils.getMapRedWork(parseCtx);
      uTask = TaskFactory.get(uPlan);
      uCtxTask = new GenMRUnionCtx(uTask);
      ctx.setUnionTask(union, uCtxTask);
    }
    else {
      uTask = uCtxTask.getUTask();
    }

    // Copy into the current union task plan if
    if (uPrsCtx.getMapOnlySubq(pos) && uPrsCtx.getRootTask(pos)) {
      processSubQueryUnionMerge(ctx, uCtxTask, union, stack);
      if (ctx.getRootTasks().contains(currTask)) {
        ctx.getRootTasks().remove(currTask);
      }
    }
    // If it a map-reduce job, create a temporary file
    else {
      // is the current task a root task
      if (shouldBeRootTask(currTask)
          && !ctx.getRootTasks().contains(currTask)
          && (currTask.getParentTasks() == null
              || currTask.getParentTasks().isEmpty())) {
        ctx.getRootTasks().add(currTask);
      }

      processSubQueryUnionCreateIntermediate(union.getParentOperators().get(pos), union, uTask,
          ctx, uCtxTask);
      // the currAliasId and CurrTopOp is not valid any more
      ctx.setCurrAliasId(null);
      ctx.setCurrTopOp(null);
      ctx.getOpTaskMap().put(null, uTask);
    }

    ctx.setCurrTask(uTask);

    mapCurrCtx.put((Operator<? extends OperatorDesc>) nd,
        new GenMapRedCtx(ctx.getCurrTask(), null));

    return true;
  }
Union Operator encountered . Currently, the algorithm is pretty simple: If
all the sub-queries are map-only, don't do anything. Otherwise, insert a
FileSink on top of all the sub-queries.

This can be optimized later on.

@param nd
         the file sink operator encountered
@param opProcCtx
         context
ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java