private LinkedHashSet<ReduceSinkOperator> exploitJobFlowCorrelation(ReduceSinkOperator op,
        CorrelationNodeProcCtx correlationCtx, IntraQueryCorrelation correlation)
        throws SemanticException {
      correlationCtx.addWalked(op);
      correlation.addToAllReduceSinkOperators(op);
      boolean shouldDetect = true;
      LinkedHashSet<ReduceSinkOperator> reduceSinkOperators =
          new LinkedHashSet<ReduceSinkOperator>();
      List<ExprNodeDesc> keyCols = op.getConf().getKeyCols();
      List<ExprNodeDesc> partitionCols = op.getConf().getPartitionCols();
      for (ExprNodeDesc key : keyCols) {
        if (!(key instanceof ExprNodeColumnDesc)) {
          shouldDetect = false;
        }
      }
      for (ExprNodeDesc key : partitionCols) {
        if (!(key instanceof ExprNodeColumnDesc)) {
          shouldDetect = false;
        }
      }
      GroupByOperator cGBY =
          CorrelationUtilities.getSingleChild(op, GroupByOperator.class);
      if (cGBY != null) {
        if (CorrelationUtilities.hasGroupingSet(op) ||
            cGBY.getConf().isGroupingSetsPresent()) {
          // Do not support grouping set right now
          shouldDetect = false;
        }
      }

      if (shouldDetect) {
        LinkedHashSet<ReduceSinkOperator> newReduceSinkOperators =
            new LinkedHashSet<ReduceSinkOperator>();
        String sortOrder = op.getConf().getOrder();
        for (Operator<? extends OperatorDesc> parent : op.getParentOperators()) {
          LOG.info("Operator " + op.getIdentifier()
              + ": start detecting correlation from this operator");
          LinkedHashSet<ReduceSinkOperator> correlatedReduceSinkOperators =
              findCorrelatedReduceSinkOperators(op, keyCols, partitionCols,
                  sortOrder, parent, correlation);
          if (correlatedReduceSinkOperators.size() == 0) {
            newReduceSinkOperators.add(op);
          } else {
            for (ReduceSinkOperator rsop : correlatedReduceSinkOperators) {
              LinkedHashSet<ReduceSinkOperator> exploited =
                  exploitJobFlowCorrelation(rsop, correlationCtx, correlation);
              if (exploited.size() == 0) {
                newReduceSinkOperators.add(rsop);
              } else {
                newReduceSinkOperators.addAll(exploited);
              }
            }
          }
        }
        reduceSinkOperators.addAll(newReduceSinkOperators);
      }
      return reduceSinkOperators;
    }
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