@Override
    public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx ctx,
        Object... nodeOutputs) throws SemanticException {
      CorrelationNodeProcCtx corrCtx = (CorrelationNodeProcCtx) ctx;
      ReduceSinkOperator op = (ReduceSinkOperator) nd;
      // Check if we have visited this operator
      if (corrCtx.isWalked(op)) {
        return null;
      }

      LOG.info("Walk to operator " + op.getIdentifier() + " " + op.getName());

      Operator<? extends OperatorDesc> child = CorrelationUtilities.getSingleChild(op, true);
      if (!(child instanceof JoinOperator) && !(child instanceof GroupByOperator)) {
        corrCtx.addWalked(op);
        return null;
      }

      // detect correlations
      IntraQueryCorrelation correlation = new IntraQueryCorrelation(corrCtx.minReducer());
      List<ReduceSinkOperator> topReduceSinkOperators =
          CorrelationUtilities.findSiblingReduceSinkOperators(op);
      List<ReduceSinkOperator> bottomReduceSinkOperators = new ArrayList<ReduceSinkOperator>();
      // Adjust the number of reducers of this correlation based on
      // those top layer ReduceSinkOperators.
      for (ReduceSinkOperator rsop : topReduceSinkOperators) {
        if (!correlation.adjustNumReducers(rsop.getConf().getNumReducers())) {
          // If we have a conflict on the number of reducers, we will not optimize
          // this plan from here.
          corrCtx.addWalked(op);
          return null;
        }
      }
      for (ReduceSinkOperator rsop : topReduceSinkOperators) {
        LinkedHashSet<ReduceSinkOperator> thisBottomReduceSinkOperators =
            exploitJobFlowCorrelation(rsop, corrCtx, correlation);
        if (thisBottomReduceSinkOperators.size() == 0) {
          thisBottomReduceSinkOperators.add(rsop);
        }
        bottomReduceSinkOperators.addAll(thisBottomReduceSinkOperators);
      }

      if (!topReduceSinkOperators.containsAll(bottomReduceSinkOperators)) {
        LOG.info("has job flow correlation");
        correlation.setJobFlowCorrelation(true, bottomReduceSinkOperators);
      }

      if (correlation.hasJobFlowCorrelation()) {
        corrCtx.addCorrelation(correlation);
      } else {
        // Since we cannot merge operators into a single MR job from here,
        // we should remove ReduceSinkOperators added into walked in exploitJFC
        corrCtx.removeWalkedAll(correlation.getAllReduceSinkOperators());
      }

      corrCtx.addWalked(op);
      return null;
    }
Start to exploit Job Flow Correlation from op.
Example: here is the operator tree we have ...
      JOIN2
     /    \
    RS4   RS5
   /        \
  GBY1     JOIN1
   |       /    \
  RS1     RS2   RS3
The op will be RS4. If we can execute GBY1, JOIN1, and JOIN2 in
the same reducer. This method will return [RS1, RS2, RS3].
@param op
@param correlationCtx
@param correlation
@return
@throws SemanticException
ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/CorrelationOptimizer.java