public static ReduceWork createReduceWork(
      GenTezProcContext context, Operator<?> root, TezWork tezWork) {
    assert !root.getParentOperators().isEmpty();

    boolean isAutoReduceParallelism =
        context.conf.getBoolVar(HiveConf.ConfVars.TEZ_AUTO_REDUCER_PARALLELISM);

    float maxPartitionFactor =
        context.conf.getFloatVar(HiveConf.ConfVars.TEZ_MAX_PARTITION_FACTOR);
    float minPartitionFactor = context.conf.getFloatVar(HiveConf.ConfVars.TEZ_MIN_PARTITION_FACTOR);
    long bytesPerReducer = context.conf.getLongVar(HiveConf.ConfVars.BYTES_PER_REDUCER);
    int defaultTinyBufferSize = context.conf.getIntVar(HiveConf.ConfVars.TEZ_SIMPLE_CUSTOM_EDGE_TINY_BUFFER_SIZE_MB);

    ReduceWork reduceWork = new ReduceWork(Utilities.REDUCENAME + context.nextSequenceNumber());
    LOG.debug("Adding reduce work (" + reduceWork.getName() + ") for " + root);
    reduceWork.setReducer(root);
    reduceWork.setNeedsTagging(GenMapRedUtils.needsTagging(reduceWork));

    // All parents should be reduce sinks. We pick the one we just walked
    // to choose the number of reducers. In the join/union case they will
    // all be -1. In sort/order case where it matters there will be only
    // one parent.
    assert context.parentOfRoot instanceof ReduceSinkOperator;
    ReduceSinkOperator reduceSink = (ReduceSinkOperator) context.parentOfRoot;

    reduceWork.setNumReduceTasks(reduceSink.getConf().getNumReducers());
    reduceWork.setSlowStart(reduceSink.getConf().isSlowStart());
    float minSrcFraction = context.conf.getFloat(
        ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MIN_SRC_FRACTION,
        ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MIN_SRC_FRACTION_DEFAULT);
    reduceWork.setMinSrcFraction(minSrcFraction);
    float maxSrcFraction = context.conf.getFloat(
        ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MAX_SRC_FRACTION,
        ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MAX_SRC_FRACTION_DEFAULT);
    reduceWork.setMaxSrcFraction(maxSrcFraction);
    reduceWork.setUniformDistribution(reduceSink.getConf().getReducerTraits().contains(UNIFORM));

    // Disabling TEZ_AUTO_REDUCER_PARALLELISM for BucketMapJoin until TEZ-4603 is fixed.
    if (hasBucketMapJoin(reduceSink, 0)) {
      reduceSink.getConf().getReducerTraits().remove(AUTOPARALLEL);
    }

    if (isAutoReduceParallelism && reduceSink.getConf().getReducerTraits().contains(AUTOPARALLEL)) {

      // configured limit for reducers
      final int maxReducers = context.conf.getIntVar(HiveConf.ConfVars.MAX_REDUCERS);
      // estimated number of reducers
      final int nReducers = reduceSink.getConf().getNumReducers();

      // min we allow tez to pick
      int minPartition = Math.max(1, (int) (nReducers * minPartitionFactor));
      minPartition = (minPartition > maxReducers) ? maxReducers : minPartition;

      if (isRestrictReducerExtrapolation(context.parseContext.getContext())) {
        LOG.debug("Overriding maxPartitionFactor to 1.0 to prevent creation of small files after delete operation");
        maxPartitionFactor = 1f;
      }
      
      // max we allow tez to pick
      int maxPartition = Math.max(1, (int) (nReducers * maxPartitionFactor));
      maxPartition = (maxPartition > maxReducers) ? maxReducers : maxPartition;
      LOG.debug("max partition factor={}, max partition={}", maxPartitionFactor, maxPartition);

      // reduce only if the parameters are significant
      final float minThreshold = context.conf.getFloatVar(HiveConf.ConfVars.TEZ_AUTO_REDUCER_PARALLELISM_MIN_THRESHOLD);
      if (minPartition < maxPartition && nReducers * minPartitionFactor >= minThreshold) {
        reduceWork.setAutoReduceParallelism(true);

        reduceWork.setMinReduceTasks(minPartition);
        reduceWork.setMaxReduceTasks(maxPartition);
      } else if (nReducers < maxPartition) {
        // the max is good, the min is too low
        reduceWork.setNumReduceTasks(maxPartition);
      }
    }

    setupReduceSink(context, reduceWork, reduceSink);

    tezWork.add(reduceWork);

    TezEdgeProperty edgeProp;
    EdgeType edgeType = determineEdgeType(context.preceedingWork, reduceWork, reduceSink);
    if (reduceWork.isAutoReduceParallelism()) {
      edgeProp =
          new TezEdgeProperty(context.conf, edgeType, true, reduceWork.isSlowStart(),
              reduceWork.getMinReduceTasks(), reduceWork.getMaxReduceTasks(), bytesPerReducer,
              reduceWork.getMinSrcFraction(), reduceWork.getMaxSrcFraction());
    } else {
      edgeProp = new TezEdgeProperty(edgeType);
      edgeProp.setSlowStart(reduceWork.isSlowStart());
    }
    edgeProp.setBufferSize(obtainBufferSize(root, reduceSink, defaultTinyBufferSize));
    reduceWork.setEdgePropRef(edgeProp);

    tezWork.connect(
        context.preceedingWork,
        reduceWork, edgeProp);
    context.connectedReduceSinks.add(reduceSink);

    return reduceWork;
  }
GenTezUtils is a collection of shared helper methods to produce TezWork.
All the methods in this class should be static, but some aren't; this is to facilitate testing.
Methods are made non-static on as needed basis.
ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java