private LinkedHashSet<ReduceSinkOperator> findCorrelatedReduceSinkOperators(
        Operator<? extends OperatorDesc> child,
        List<ExprNodeDesc> childKeyCols, List<ExprNodeDesc> childPartitionCols,
        String childRSOrder,
        Operator<? extends OperatorDesc> current,
        IntraQueryCorrelation correlation) throws SemanticException {

      LOG.info("now detecting operator " + current.getIdentifier() + " " + current.getName());
      LinkedHashSet<ReduceSinkOperator> correlatedReduceSinkOperators =
          new LinkedHashSet<ReduceSinkOperator>();
      if (skipedJoinOperators.contains(current)) {
        LOG.info(current.getName() + " " + current.getIdentifier() +
            " may be converted to MapJoin by " +
            "CommonJoinResolver. Correlation optimizer will not detect correlations" +
            "involved in this operator");
        return correlatedReduceSinkOperators;
      }
      if ((current.getParentOperators() == null) || (current.getParentOperators().isEmpty())) {
        return correlatedReduceSinkOperators;
      }
      if (current instanceof PTFOperator) {
        // Currently, we do not support PTF operator.
        LOG.info("Currently, correlation optimizer does not support PTF operator.");
        return correlatedReduceSinkOperators;
      }
      if (current instanceof UnionOperator) {
        // If we get a UnionOperator, right now, we only handle it when
        // we can find correlated ReduceSinkOperators from all inputs.
        LinkedHashSet<ReduceSinkOperator> corrRSs = new LinkedHashSet<ReduceSinkOperator>();
        for (Operator<? extends OperatorDesc> parent : current.getParentOperators()) {
          LinkedHashSet<ReduceSinkOperator> tmp =
              findCorrelatedReduceSinkOperators(
                  current, childKeyCols, childPartitionCols, childRSOrder, parent, correlation);
          if (tmp != null && tmp.size() > 0) {
            corrRSs.addAll(tmp);
          } else {
            return correlatedReduceSinkOperators;
          }
        }
        correlatedReduceSinkOperators.addAll(corrRSs);
        UnionOperator union = (UnionOperator)current;
        union.getConf().setAllInputsInSameReducer(true);
      } else if (current.getColumnExprMap() == null && !(current instanceof ReduceSinkOperator)) {
        for (Operator<? extends OperatorDesc> parent : current.getParentOperators()) {
          correlatedReduceSinkOperators.addAll(
              findCorrelatedReduceSinkOperators(
                  current, childKeyCols, childPartitionCols, childRSOrder, parent, correlation));
        }
      } else if (current.getColumnExprMap() != null && !(current instanceof ReduceSinkOperator)) {
        List<ExprNodeDesc> backtrackedKeyCols =
            ExprNodeDescUtils.backtrack(childKeyCols, child, current);
        List<ExprNodeDesc> backtrackedPartitionCols =
            ExprNodeDescUtils.backtrack(childPartitionCols, child, current);

        RowSchema rowSchema = current.getSchema();
        Set<String> tableNeedToCheck = new HashSet<String>();
        for (ExprNodeDesc expr: childKeyCols) {
          if (!(expr instanceof ExprNodeColumnDesc)) {
            return correlatedReduceSinkOperators;
          }
          String colName = ((ExprNodeColumnDesc)expr).getColumn();
          ColumnInfo columnInfo = rowSchema.getColumnInfo(colName);
          if (columnInfo != null) {
            tableNeedToCheck.add(columnInfo.getTabAlias());
          }
        }
        if (current instanceof JoinOperator) {
          boolean isCorrelated = true;
          int expectedNumCorrelatedRsops = current.getParentOperators().size();
          LinkedHashSet<ReduceSinkOperator> correlatedRsops = null;
          for (Operator<? extends OperatorDesc> parent : current.getParentOperators()) {
            Set<String> tableNames = parent.getSchema().getTableNames();
            for (String tbl : tableNames) {
              if (tableNeedToCheck.contains(tbl)) {
                correlatedRsops = findCorrelatedReduceSinkOperators(current,
                    backtrackedKeyCols, backtrackedPartitionCols,
                    childRSOrder, parent, correlation);
                if (correlatedRsops.size() != expectedNumCorrelatedRsops) {
                  isCorrelated = false;
                }
              }
            }
            if (!isCorrelated) {
              break;
            }
          }
          // If current is JoinOperaotr, we will stop to traverse the tree
          // when any of parent ReduceSinkOperaotr of this JoinOperator is
          // not considered as a correlated ReduceSinkOperator.
          if (isCorrelated && correlatedRsops != null) {
            correlatedReduceSinkOperators.addAll(correlatedRsops);
          } else {
            correlatedReduceSinkOperators.clear();
          }
        } else {
          for (Operator<? extends OperatorDesc> parent : current.getParentOperators()) {
            correlatedReduceSinkOperators.addAll(findCorrelatedReduceSinkOperators(
                current, backtrackedKeyCols, backtrackedPartitionCols, childRSOrder,
                parent, correlation));
          }
        }
      } else if (current.getColumnExprMap() != null && current instanceof ReduceSinkOperator) {
        ReduceSinkOperator rsop = (ReduceSinkOperator) current;
        List<ExprNodeDesc> backtrackedKeyCols =
            ExprNodeDescUtils.backtrack(childKeyCols, child, current);
        List<ExprNodeDesc> backtrackedPartitionCols =
            ExprNodeDescUtils.backtrack(childPartitionCols, child, current);
        List<ExprNodeDesc> rsKeyCols = rsop.getConf().getKeyCols();
        List<ExprNodeDesc> rsPartitionCols = rsop.getConf().getPartitionCols();

        // Two ReduceSinkOperators are correlated means that
        // they have same sorting columns (key columns), same partitioning columns,
        // same sorting orders, and no conflict on the numbers of reducers.
        // TODO: we should relax this condition
        // TODO: we need to handle aggregation functions with distinct keyword. In this case,
        // distinct columns will be added to the key columns.
        boolean isCorrelated = sameKeys(rsKeyCols, backtrackedKeyCols) &&
            sameOrder(rsop.getConf().getOrder(), childRSOrder) &&
            sameKeys(backtrackedPartitionCols, rsPartitionCols) &&
            correlation.adjustNumReducers(rsop.getConf().getNumReducers());
        GroupByOperator cGBY =
            CorrelationUtilities.getSingleChild(rsop, GroupByOperator.class);
        if (cGBY != null) {
          if (CorrelationUtilities.hasGroupingSet(rsop) ||
              cGBY.getConf().isGroupingSetsPresent()) {
            // Do not support grouping set right now
            isCorrelated = false;
          }
        }

        if (isCorrelated) {
          LOG.info("Operator " + current.getIdentifier() + " " +
              current.getName() + " is correlated");
          Operator<? extends OperatorDesc> childOperator =
              CorrelationUtilities.getSingleChild(current, true);
          if (childOperator instanceof JoinOperator) {
            JoinOperator joinOp = (JoinOperator) childOperator;
            JoinCondDesc[] joinConds = joinOp.getConf().getConds();
            List<Operator<? extends OperatorDesc>> rsOps = joinOp.getParentOperators();
            LinkedHashSet<ReduceSinkOperator> correlatedRsOps =
                new LinkedHashSet<ReduceSinkOperator>();
            analyzeReduceSinkOperatorsOfJoinOperator(joinConds, rsOps, current, correlatedRsOps);
            correlatedReduceSinkOperators.addAll(correlatedRsOps);
          } else {
            correlatedReduceSinkOperators.add(rsop);
          }
        } else {
          LOG.info("Operator " + current.getIdentifier() + " " +
              current.getName() + " is not correlated");
          correlatedReduceSinkOperators.clear();
        }
      } else {
        LOG.error("ReduceSinkOperator " + current.getIdentifier() + " does not have ColumnExprMap");
        throw new SemanticException("CorrelationOptimizer cannot optimize this plan. " +
            "ReduceSinkOperator " + current.getIdentifier()
            + " does not have ColumnExprMap");
      }
      return correlatedReduceSinkOperators;
    }
This method is used to recursively traverse the tree to find
ReduceSinkOperators which share the same key columns and partitioning
columns. Those ReduceSinkOperators are called correlated ReduceSinkOperaotrs.

@param child The child of the current operator
@param childKeyCols The key columns from the child operator
@param childPartitionCols The partitioning columns from the child operator
@param childRSOrder The sorting order of key columns from the child operator
@param current The current operator we are visiting
@param correlation The object keeps tracking the correlation
@return
@throws SemanticException
ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/CorrelationOptimizer.java