@Override
    public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
        Object... nodeOutputs) throws SemanticException {
      ReduceSinkJoinDeDuplicateProcCtx dedupCtx = (ReduceSinkJoinDeDuplicateProcCtx) procCtx;
      ReduceSinkOperator cRS = (ReduceSinkOperator) nd;
      if (cRS.getConf().isForwarding()) {
        // Already set
        return false;
      }
      if( cRS.getConf().isOrdering()) {
        // 1-1 edge is skipped for sorted shuffle
        return false;
      }
      if (cRS.getConf().getKeyCols().isEmpty()) {
        // Not supported
        return false;
      }
      boolean onlyPartitioning = false;
      Operator<?> cRSChild = cRS.getChildOperators().get(0);
      if (cRSChild instanceof MapJoinOperator ||
              cRSChild instanceof CommonMergeJoinOperator) {
        // If it is a MapJoin or MergeJoin, we make sure that they are on
        // the reduce side, otherwise we bail out
        for (Operator<?> parent: cRSChild.getParentOperators()) {
          if (!(parent instanceof ReduceSinkOperator)) {
            // MapJoin and SMBJoin not supported
            return false;
          }
        }
        if (cRSChild instanceof MapJoinOperator) {
          onlyPartitioning = true;
        }
      }

      int maxNumReducers = cRS.getConf().getNumReducers();
      ReduceSinkOperator pRS;
      if (onlyPartitioning) {
        pRS = CorrelationUtilities.findFirstPossibleParent(
            cRS, ReduceSinkOperator.class, dedupCtx.trustScript());
      } else {
        pRS = CorrelationUtilities.findFirstPossibleParentPreserveSortOrder(
            cRS, ReduceSinkOperator.class, dedupCtx.trustScript());
      }
      if (pRS != null) {
        Operator<?> pRSChild = pRS.getChildOperators().get(0);
        if (pRSChild instanceof MapJoinOperator) {
          // Handle MapJoin specially and check for all its children
          MapJoinOperator pRSChildMJ = (MapJoinOperator) pRSChild;
          // In this case, both should be DHJ operators as pRSChildMJ can only guarantee
          // partitioned input, not sorted.
          if (!pRSChildMJ.getConf().isDynamicPartitionHashJoin() ||
                  !(cRSChild instanceof MapJoinOperator) ||
                  !((MapJoinOperator) cRSChild).getConf().isDynamicPartitionHashJoin()) {
            return false;
          }
          ImmutableList.Builder<ReduceSinkOperator> l = ImmutableList.builder();
          for (Operator<?> parent: pRSChild.getParentOperators()) {
            ReduceSinkOperator rsOp = (ReduceSinkOperator) parent;
            l.add(rsOp);
            if (rsOp.getConf().getNumReducers() > maxNumReducers) {
              maxNumReducers = rsOp.getConf().getNumReducers();
            }
          }
          if (ReduceSinkDeDuplicationUtils.strictMerge(cRS, l.build())) {
            LOG.debug("Set {} to forward data", cRS);
            cRS.getConf().setForwarding(true);
            propagateMaxNumReducers(dedupCtx, cRS, maxNumReducers);
            return true;
          }
        } else if (pRS.getChildOperators().get(0) instanceof CommonMergeJoinOperator) {
          // Handle MergeJoin specially and check for all its children
          ImmutableList.Builder<ReduceSinkOperator> l = ImmutableList.builder();
          for (Operator<?> parent: pRSChild.getParentOperators()) {
            if (!(parent instanceof ReduceSinkOperator)) {
              // SMBJoin not supported
              return false;
            }
            ReduceSinkOperator rsOp = (ReduceSinkOperator) parent;
            l.add(rsOp);
            if (rsOp.getConf().getNumReducers() > maxNumReducers) {
              maxNumReducers = rsOp.getConf().getNumReducers();
            }
          }
          if (ReduceSinkDeDuplicationUtils.strictMerge(cRS, l.build())) {
            LOG.debug("Set {} to forward data", cRS);
            cRS.getConf().setForwarding(true);
            propagateMaxNumReducers(dedupCtx, cRS, maxNumReducers);
            return true;
          }
        } else {
          // Rest of cases
          if (pRS.getConf().getNumReducers() > maxNumReducers) {
            maxNumReducers = pRS.getConf().getNumReducers();
          }
          if (ReduceSinkDeDuplicationUtils.strictMerge(cRS, pRS)) {
            LOG.debug("Set {} to forward data", cRS);
            cRS.getConf().setForwarding(true);
            propagateMaxNumReducers(dedupCtx, cRS, maxNumReducers);
            return true;
          }
        }
      }
      return false;
    }
Optimization to check whether any ReduceSink operator in the plan can be
simplified so data is not shuffled/sorted if it is already shuffled/sorted.

This optimization is executed after join algorithm selection logic has run,
and it is intended to optimize new cases that cannot be optimized when
{@link ReduceSinkDeDuplication} runs because some physical algorithms have
not been selected. Instead of removing ReduceSink operators from the plan,
they will be tagged, and then the execution plan compiler might take action,
e.g., on Tez, ReduceSink operators that just need to forward data will be
translated into a ONE-TO-ONE edge. The parallelism degree of these ReduceSink
operators might be adjusted, as a ReduceSink operator that just forwards data
cannot alter the degree of parallelism of the previous task.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/correlation/ReduceSinkJoinDeDuplication.java