@Override
  void init(MRTaskReporter mrReporter, Map<String, LogicalInput> inputs, Map<String, LogicalOutput> outputs)
      throws Exception {
    perfLogger.perfLogBegin(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
    super.init(mrReporter, inputs, outputs);

    MapredContext.init(false, new JobConf(jconf));
    List<LogicalInput> shuffleInputs = getShuffleInputs(inputs);
    // TODO HIVE-14042. Move to using a loop and a timed wait once TEZ-3302 is fixed.
    checkAbortCondition();
    if (shuffleInputs != null) {
      LOG.info("Waiting for ShuffleInputs to become ready");
      processorContext.waitForAllInputsReady(new ArrayList<Input>(shuffleInputs));
    }

    connectOps.clear();
    ReduceWork redWork = reduceWork;
    LOG.info("Main work is " + reduceWork.getName());
    List<HashTableDummyOperator> workOps = reduceWork.getDummyOps();
    Set<HashTableDummyOperator> dummyOps = workOps == null ? new HashSet<>() : new HashSet<>(workOps);
    tagToReducerMap.put(redWork.getTag(), redWork);
    if (mergeWorkList != null) {
      for (BaseWork mergeWork : mergeWorkList) {
        LOG.debug("Additional work {}", mergeWork.getName());
        workOps = mergeWork.getDummyOps();
        if (workOps != null) {
          dummyOps.addAll(workOps);
        }
        ReduceWork mergeReduceWork = (ReduceWork) mergeWork;
        reducer = mergeReduceWork.getReducer();
        // Check immediately after reducer is assigned, in cae the abort came in during
        checkAbortCondition();
        DummyStoreOperator dummyStoreOp = getJoinParentOp(reducer);
        connectOps.put(mergeReduceWork.getTag(), dummyStoreOp);
        tagToReducerMap.put(mergeReduceWork.getTag(), mergeReduceWork);
      }

      ((TezContext) MapredContext.get()).setDummyOpsMap(connectOps);
    }
    checkAbortCondition();

    bigTablePosition = (byte) reduceWork.getTag();

    ObjectInspector[] mainWorkOIs = null;
    ((TezContext) MapredContext.get()).setInputs(inputs);
    ((TezContext) MapredContext.get()).setTezProcessorContext(processorContext);
    int numTags = reduceWork.getTagToValueDesc().size();
    reducer = reduceWork.getReducer();
    // Check immediately after reducer is assigned, in cae the abort came in during
    checkAbortCondition();
    // set memory available for operators
    long memoryAvailableToTask = processorContext.getTotalMemoryAvailableToTask();
    if (reducer.getConf() != null) {
      reducer.getConf().setMaxMemoryAvailable(memoryAvailableToTask);
      LOG.info("Memory available for operators set to {}", LlapUtil.humanReadableByteCount(memoryAvailableToTask));
    }
    OperatorUtils.setMemoryAvailable(reducer.getChildOperators(), memoryAvailableToTask);

    // Setup values registry
    String valueRegistryKey = DynamicValue.DYNAMIC_VALUE_REGISTRY_CACHE_KEY;
    DynamicValueRegistryTez registryTez =
        dynamicValueCache.retrieve(valueRegistryKey, () -> new DynamicValueRegistryTez());
    dynamicValueCacheKeys.add(valueRegistryKey);
    RegistryConfTez registryConf = new RegistryConfTez(jconf, reduceWork, processorContext, inputs);
    registryTez.init(registryConf);
    checkAbortCondition();

    if (numTags > 1) {
      sources = new ReduceRecordSource[numTags];
      mainWorkOIs = new ObjectInspector[numTags];
      initializeMultipleSources(reduceWork, numTags, mainWorkOIs, sources);
      ((TezContext) MapredContext.get()).setRecordSources(sources);
      reducer.initialize(jconf, mainWorkOIs);
    } else {
      numTags = tagToReducerMap.keySet().size();
      sources = new ReduceRecordSource[numTags];
      mainWorkOIs = new ObjectInspector[numTags];
      for (int i : tagToReducerMap.keySet()) {
        redWork = tagToReducerMap.get(i);
        reducer = redWork.getReducer();
        // Check immediately after reducer is assigned, in cae the abort came in during
        checkAbortCondition();
        initializeSourceForTag(redWork, i, mainWorkOIs, sources, redWork.getTagToValueDesc().get(0),
            redWork.getTagToInput().get(0));
        reducer.initializeLocalWork(jconf);
      }
      reducer = reduceWork.getReducer();
      // Check immediately after reducer is assigned, in cae the abort came in during
      checkAbortCondition();
      ((TezContext) MapredContext.get()).setRecordSources(sources);
      reducer.initialize(jconf, new ObjectInspector[] {mainWorkOIs[bigTablePosition]});
      for (int i : tagToReducerMap.keySet()) {
        if (i == bigTablePosition) {
          continue;
        }
        redWork = tagToReducerMap.get(i);
        reducer = redWork.getReducer();
        // Check immediately after reducer is assigned, in cae the abort came in during
        checkAbortCondition();
        reducer.initialize(jconf, new ObjectInspector[] {mainWorkOIs[i]});
      }
    }
    checkAbortCondition();

    reducer = reduceWork.getReducer();

    // initialize reduce operator tree
    try {
      LOG.info(reducer.dump(0));

      // Initialization isn't finished until all parents of all operators
      // are initialized. For broadcast joins that means initializing the
      // dummy parent operators as well.
      for (HashTableDummyOperator dummyOp : dummyOps) {
        // TODO HIVE-14042. Propagating abort to dummyOps.
        dummyOp.initialize(jconf, null);
        checkAbortCondition();
      }

      // set output collector for any reduce sink operators in the pipeline.
      List<Operator<?>> children = new ArrayList<>();
      children.add(reducer);
      children.addAll(dummyOps);
      createOutputMap();
      OperatorUtils.setChildrenCollector(children, outMap);

      checkAbortCondition();
      reducer.setReporter(reporter);
      MapredContext.get().setReporter(reporter);

    } catch (Throwable e) {
      super.setAborted(true);
      if (e instanceof OutOfMemoryError) {
        // Don't create a new object if we are already out of memory
        throw (OutOfMemoryError) e;
      } else if (e instanceof InterruptedException) {
        LOG.info("Hit an interrupt while initializing ReduceRecordProcessor. Message={}", e.getMessage());
        throw (InterruptedException) e;
      } else {
        throw new RuntimeException(redWork.getName() + " operator initialization failed", e);
      }
    }

    perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
  }
Process input from tez LogicalInput and write output - for a map plan
Just pump the records through the query plan.
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java