void init(JobConf jconf, Operator<?> reducer, boolean vectorized, TableDesc keyTableDesc,
      TableDesc valueTableDesc, Reader reader, boolean handleGroupKey, byte tag,
      VectorizedRowBatchCtx batchContext, long vectorizedVertexNum,
      int vectorizedTestingReducerBatchSize)
      throws Exception {

    this.vectorizedVertexNum = vectorizedVertexNum;
    if (vectorizedTestingReducerBatchSize > VectorizedRowBatch.DEFAULT_SIZE) {

      // For now, we don't go higher than the default batch size unless we do more work
      // to verify every vectorized operator downstream can handle a larger batch size.
      vectorizedTestingReducerBatchSize = VectorizedRowBatch.DEFAULT_SIZE;
    }
    this.vectorizedTestingReducerBatchSize = vectorizedTestingReducerBatchSize;
    ObjectInspector keyObjectInspector;

    this.reducer = reducer;
    this.vectorized = vectorized;
    this.keyTableDesc = keyTableDesc;
    if (reader instanceof KeyValueReader) {
      this.reader = new KeyValuesFromKeyValue((KeyValueReader) reader);
    } else {
      this.reader = new KeyValuesFromKeyValues((KeyValuesReader) reader);
    }
    this.handleGroupKey = handleGroupKey;
    this.tag = tag;

    try {
      inputKeySerDe = ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
      inputKeySerDe.initialize(null, keyTableDesc.getProperties(), null);

      keyObjectInspector = inputKeySerDe.getObjectInspector();

      if(vectorized) {
        keyStructInspector = (StructObjectInspector) keyObjectInspector;
        firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
      }

      // We should initialize the SerDe with the TypeInfo when available.
      this.valueTableDesc = valueTableDesc;
      inputValueSerDe = (AbstractSerDe) ReflectionUtils.newInstance(valueTableDesc.getSerDeClass(), null);
      inputValueSerDe.initialize(null, valueTableDesc.getProperties(), null);
      valueObjectInspector = inputValueSerDe.getObjectInspector();

      ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();

      if(vectorized) {
        /* vectorization only works with struct object inspectors */
        valueStructInspectors = (StructObjectInspector) valueObjectInspector;

        final int totalColumns = firstValueColumnOffset +
            valueStructInspectors.getAllStructFieldRefs().size();

        rowObjectInspector = Utilities.constructVectorizedReduceRowOI(keyStructInspector,
            valueStructInspectors);
        this.batchContext = batchContext;
        batch = batchContext.createVectorizedRowBatch();

        // Setup vectorized deserialization for the key and value.
        BinarySortableSerDe binarySortableSerDe = (BinarySortableSerDe) inputKeySerDe;

        keyBinarySortableDeserializeToRow =
            new VectorDeserializeRow<BinarySortableDeserializeRead>(
                new BinarySortableDeserializeRead(
                    VectorizedBatchUtil.typeInfosFromStructObjectInspector(
                        keyStructInspector),
                    (batchContext.getRowdataTypePhysicalVariations().length > firstValueColumnOffset)
                        ? Arrays.copyOfRange(batchContext.getRowdataTypePhysicalVariations(), 0,
                            firstValueColumnOffset)
                        : batchContext.getRowdataTypePhysicalVariations(),
                    /* useExternalBuffer */ true,
                    binarySortableSerDe.getSortOrders(),
                    binarySortableSerDe.getNullMarkers(),
                    binarySortableSerDe.getNotNullMarkers()));
        keyBinarySortableDeserializeToRow.init(0);

        final int valuesSize = valueStructInspectors.getAllStructFieldRefs().size();
        if (valuesSize > 0) {
          valueLazyBinaryDeserializeToRow =
              new VectorDeserializeRow<LazyBinaryDeserializeRead>(
                  new LazyBinaryDeserializeRead(
                      VectorizedBatchUtil.typeInfosFromStructObjectInspector(
                          valueStructInspectors),
                      (batchContext.getRowdataTypePhysicalVariations().length >= totalColumns)
                          ? Arrays.copyOfRange(batchContext.getRowdataTypePhysicalVariations(),
                              firstValueColumnOffset, totalColumns)
                          : null,
                      /* useExternalBuffer */ true));
          valueLazyBinaryDeserializeToRow.init(firstValueColumnOffset);

          // Create data buffers for value bytes column vectors.
          for (int i = firstValueColumnOffset; i < batch.numCols; i++) {
            ColumnVector colVector = batch.cols[i];
            if (colVector instanceof BytesColumnVector) {
              BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
              bytesColumnVector.initBuffer();
            }
          }
        }
      } else {
        ois.add(keyObjectInspector);
        ois.add(valueObjectInspector);
        rowObjectInspector =
            ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
                ois);
      }
    } catch (Throwable e) {
      abort = true;
      if (e instanceof OutOfMemoryError) {
        // Don't create a new object if we are already out of memory
        throw (OutOfMemoryError) e;
      } else {
        throw new RuntimeException("Reduce 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/ReduceRecordSource.java