private void initExpressionColumns() {
    for (int i = 0; i < evaluators.length; i++) {
      VectorPTFEvaluatorBase evaluator = evaluators[i];
      /*
       * Non-streaming evaluators work on buffered batches, we need to adapt them. Before PTF
       * bounded start vectorization (HIVE-24761), VectorExpression.outputColumnNum was closed and
       * VectorExpression.inputColumnNum didn't even exist (even though the vast majority of
       * VectorExpression subclasses use at least 1 input column). Since VectorPTFOperator and
       * VectorPTFGroupBatches work on modified batches (by not storing all the columns, and having
       * ordering columns first), the expressions planned in compile-time won't work with the
       * original config (column layout). It would make sense to move this logic to compile time,
       * because here in runtime, a very simple mapping (bufferedColumnMap) is used, so it might be
       * used. However, vectorized expression compilation affects many layers of code (having
       * VectorizationContext as the common scope), and moving the calculation of bufferedColumnMap
       * and this override logic to compile-time would create much more complicated behavior there
       * (probably involving hacking most of the time, or maybe a great re-design) just because of
       * the optimized column layout of the PTF vectorization.
       */
      if (!evaluator.streamsResult()) {
        evaluator.inputColumnNum = IntStream.range(0, bufferedColumnMap.length)
            .filter(j -> bufferedColumnMap[j] == evaluator.inputColumnNum).findFirst()
            .orElseGet(() -> evaluator.inputColumnNum);

        if (evaluator.inputVecExpr != null) {
          for (int j = 0; j <  evaluator.inputVecExpr.inputColumnNum.length; j++){
            final int jj = j; // need a final in stream filter
            evaluator.inputVecExpr.inputColumnNum[jj] = IntStream.range(0, bufferedColumnMap.length)
                .filter(k -> bufferedColumnMap[k] == evaluator.inputVecExpr.inputColumnNum[jj])
                .findFirst().orElseGet(() -> evaluator.inputVecExpr.inputColumnNum[jj]);
          }
          evaluator.inputVecExpr.outputColumnNum = IntStream.range(0, bufferedColumnMap.length)
              .filter(j -> bufferedColumnMap[j] == evaluator.inputVecExpr.outputColumnNum)
              .findFirst().orElseGet(() -> evaluator.inputVecExpr.outputColumnNum);
        }
        evaluator.mapCustomColumns(bufferedColumnMap);
      }
    }
  }
We are processing a batch from reduce processor that is only for one reducer key or PTF group.

For a simple OVER (PARTITION BY column) or OVER (ORDER BY column), the reduce processor's
group key is the partition or order by key.

For an OVER (PARTITION BY column1, ORDER BY column2), the reduce-shuffle group key is
the combination of the partition column1 and the order by column2.  In this case, this method
has to watch for changes in the partition and reset the group aggregations.

The reduce processor calls setNextVectorBatchGroupStatus beforehand to tell us whether the
batch supplied to our process method is the last batch for the group key, or not.  This helps
us intelligently process the batch.
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java