private void processVectorGroup(BytesWritable keyWritable,
          Iterable<Object> values, byte tag) throws HiveException, IOException {

    if (reducer.batchNeedsClone()) {
      batch = batchContext.createVectorizedRowBatch();
    }
    Preconditions.checkState(batch.size == 0);

    // Deserialize key into vector row columns.
    //
    byte[] keyBytes = keyWritable.getBytes();
    int keyLength = keyWritable.getLength();

    // l4j.info("ReduceRecordSource processVectorGroup keyBytes " + keyLength + " " +
    //     VectorizedBatchUtil.displayBytes(keyBytes, 0, keyLength));

    keyBinarySortableDeserializeToRow.setBytes(keyBytes, 0, keyLength);
    try {
      keyBinarySortableDeserializeToRow.deserialize(batch, 0);
    } catch (Exception e) {
      throw new HiveException(
          "\nDeserializeRead details: " +
              keyBinarySortableDeserializeToRow.getDetailedReadPositionString(),
          e);
    }
    for(int i = 0; i < firstValueColumnOffset; i++) {
      VectorizedBatchUtil.setRepeatingColumn(batch, i);
    }

    final int maxSize =
        (vectorizedTestingReducerBatchSize > 0 ?
            Math.min(vectorizedTestingReducerBatchSize, batch.getMaxSize()) :
            batch.getMaxSize());
    Preconditions.checkState(maxSize > 0);
    int rowIdx = 0;
    int batchBytes = keyBytes.length;
    try {
      for (Object value : values) {
        if (rowIdx >= maxSize ||
            (rowIdx > 0 && batchBytes >= BATCH_BYTES)) {

          // Batch is full AND we have at least 1 more row...
          batch.size = rowIdx;
          if (handleGroupKey) {
            reducer.setNextVectorBatchGroupStatus(/* isLastGroupBatch */ false);
          }
          reducer.process(batch, tag);

          // Do the non-column batch reset logic.
          batch.selectedInUse = false;
          batch.size = 0;
          batch.endOfFile = false;

          // Reset just the value columns and value buffer.
          for (int i = firstValueColumnOffset; i < batch.numCols; i++) {
            // Note that reset also resets the data buffer for bytes column vectors.
            batch.cols[i].reset();
          }
          rowIdx = 0;
          batchBytes = keyBytes.length;
        }
        if (valueLazyBinaryDeserializeToRow != null) {
          // Deserialize value into vector row columns.
          BytesWritable valueWritable = (BytesWritable) value;
          byte[] valueBytes = valueWritable.getBytes();
          int valueLength = valueWritable.getLength();
          batchBytes += valueLength;

          valueLazyBinaryDeserializeToRow.setBytes(valueBytes, 0, valueLength);
          valueLazyBinaryDeserializeToRow.deserialize(batch, rowIdx);
        }
        rowIdx++;
      }
      if (rowIdx > 0) {
        // Flush final partial batch.
        batch.size = rowIdx;
        if (handleGroupKey) {
          reducer.setNextVectorBatchGroupStatus(/* isLastGroupBatch */ true);
        }
        reducer.process(batch, tag);
      }
      // reset only when we reuse the batch
      if (!reducer.batchNeedsClone()) {
        batch.reset();
      }
    } catch (Exception e) {
      String rowString = null;
      try {
        rowString = batch.toString();
      } catch (Exception e2) {
        rowString = "[Error getting row data with exception "
            + StringUtils.stringifyException(e2) + " ]";
      }
      l4j.error("Hive Runtime Error while processing vector batch (tag=" + tag
              + ") (vectorizedVertexNum " + vectorizedVertexNum + ") " + rowString, e);
      throw new HiveException("Hive Runtime Error while processing vector batch (tag="
          + tag + ") (vectorizedVertexNum " + vectorizedVertexNum + ")", e);
    }
  }
@param keyWritable
@param values
@param tag
@throws HiveException
@throws IOException
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java