@Override
    public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx ctx,
        Object... nodeOutputs) throws SemanticException {
      ReduceSinkOperator op = (ReduceSinkOperator) nd;
      ColumnPrunerProcCtx cppCtx = (ColumnPrunerProcCtx) ctx;
      ReduceSinkDesc conf = op.getConf();

      List<FieldNode> colLists = new ArrayList<>();
      List<ExprNodeDesc> keys = conf.getKeyCols();
      LOG.debug("Reduce Sink Operator " + op.getIdentifier() + " key:" + keys);
      for (ExprNodeDesc key : keys) {
        colLists = mergeFieldNodesWithDesc(colLists, key);
      }
      for (ExprNodeDesc key : conf.getPartitionCols()) {
        colLists = mergeFieldNodesWithDesc(colLists, key);
      }

      assert op.getNumChild() == 1;

      Operator<? extends OperatorDesc> child = op.getChildOperators().get(0);

      List<FieldNode> childCols = null;
      if (child instanceof CommonJoinOperator) {
        childCols = cppCtx.getJoinPrunedColLists().get(child) == null
                ? null : cppCtx.getJoinPrunedColLists().get(child)
                        .get((byte) conf.getTag());
      } else {
        childCols = cppCtx.getPrunedColList(child);
      }

      List<ExprNodeDesc> valCols = conf.getValueCols();
      List<String> valColNames = conf.getOutputValueColumnNames();

      if (childCols != null) {
        boolean[] flags = new boolean[valCols.size()];

        for (FieldNode childCol : childCols) {
          int index = valColNames.indexOf(Utilities.removeValueTag(childCol.getFieldName()));
          if (index < 0) {
            continue;
          }
          flags[index] = true;
          colLists = mergeFieldNodesWithDesc(colLists, valCols.get(index));
        }

        Collections.sort(colLists, new Comparator<FieldNode>() {
          @Override
          public int compare(FieldNode o1, FieldNode o2) {
            return o1.getFieldName().compareTo(o2.getFieldName());
          }
        });
        pruneReduceSinkOperator(flags, op, cppCtx);
        cppCtx.getPrunedColLists().put(op, colLists);
        return null;
      }

      // Reduce Sink contains the columns needed - no need to aggregate from
      // children
      for (ExprNodeDesc val : valCols) {
        colLists = mergeFieldNodesWithDesc(colLists, val);
      }

      cppCtx.getPrunedColLists().put(op, colLists);
      return null;
    }
The Node Processor for Column Pruning on Reduce Sink Operators.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java