@SuppressWarnings("unchecked")
  @Override
  public Object process(Node nd, Stack<Node> stack,
                        NodeProcessorCtx procContext, Object... nodeOutputs)
      throws SemanticException {

    OptimizeTezProcContext context = (OptimizeTezProcContext) procContext;

    ReduceSinkOperator sink = (ReduceSinkOperator) nd;
    ReduceSinkDesc desc = sink.getConf();

    long bytesPerReducer = context.conf.getLongVar(HiveConf.ConfVars.BYTES_PER_REDUCER);
    int maxReducers = context.conf.getIntVar(HiveConf.ConfVars.MAX_REDUCERS);
    int constantReducers = context.conf.getIntVar(HiveConf.ConfVars.HADOOP_NUM_REDUCERS);

    if (context.visitedReduceSinks.contains(sink)) {
      // skip walking the children
      LOG.debug("Already processed reduce sink: " + sink.getName());
      return true;
    }

    context.visitedReduceSinks.add(sink);

    if (desc.getNumReducers() <= 0) {
      if (constantReducers > 0) {
        LOG.info("Parallelism for reduce sink "+sink+" set by user to "+constantReducers);
        desc.setNumReducers(constantReducers);
      } else {
        long numberOfBytes = 0;

        // we need to add up all the estimates from the siblings of this reduce sink
        for (Operator<? extends OperatorDesc> sibling:
          sink.getChildOperators().get(0).getParentOperators()) {
          if (sibling.getStatistics() != null) {
            numberOfBytes = StatsUtils.safeAdd(
                numberOfBytes, sibling.getStatistics().getDataSize());
          } else {
            LOG.warn("No stats available from: "+sibling);
          }
        }

        int numReducers = Utilities.estimateReducers(numberOfBytes, bytesPerReducer,
            maxReducers, false);
        LOG.info("Set parallelism for reduce sink "+sink+" to: "+numReducers);
        desc.setNumReducers(numReducers);

        final Collection<ExprNodeDescEqualityWrapper> keyCols = ExprNodeDescEqualityWrapper.transform(desc.getKeyCols());
        final Collection<ExprNodeDescEqualityWrapper> partCols = ExprNodeDescEqualityWrapper.transform(desc.getPartitionCols());
        if (keyCols != null && keyCols.equals(partCols)) {
          desc.setReducerTraits(EnumSet.of(UNIFORM, AUTOPARALLEL));
        } else {
          desc.setReducerTraits(EnumSet.of(AUTOPARALLEL));
        }
      }
    } else {
      LOG.info("Number of reducers determined to be: "+desc.getNumReducers());
      desc.setReducerTraits(EnumSet.of(FIXED)); // usually controlled by bucketing
    }

    return false;
  }
SetReducerParallelism determines how many reducers should
be run for a given reduce sink.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SetReducerParallelism.java