HIVE-29696: Vectorization: ClassCastException in VectorPTFGroupBatches when buffering complex window function inputs

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/BufferedVectorizedRowBatch.java

@@ -24,7 +24,6 @@
  */
 public class BufferedVectorizedRowBatch extends VectorizedRowBatch {
   boolean isLastGroupBatch;
-  boolean isInputExpressionEvaluated;
 
   public BufferedVectorizedRowBatch(int numCols) {
     super(numCols);

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorBase.java

@@ -86,12 +86,6 @@ public void setNullsLast(boolean nullsLast) {
     this.nullsLast = nullsLast;
   }
 
-  // Evaluate the aggregation input argument expression.
-  public void evaluateInputExpr(VectorizedRowBatch batch) throws HiveException {
-    if (inputVecExpr != null) {
-      inputVecExpr.evaluate(batch);
-    }
-  }
 
   /*
    * Evaluate the aggregation over one of the group's batches. In this mode, evaluator is not

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorCount.java

@@ -46,7 +46,6 @@ public VectorPTFEvaluatorCount(WindowFrameDef windowFrameDef, VectorExpression i
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Count non-null column rows.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorCountDistinct.java

@@ -68,7 +68,6 @@ public VectorPTFEvaluatorCountDistinct(WindowFrameDef windowFrameDef,
   @Override
   public void evaluateGroupBatch(VectorizedRowBatch batch) throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // We do not filter when PTF is in reducer.
     Preconditions.checkState(!batch.selectedInUse);

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorCountStar.java

@@ -42,7 +42,6 @@ public VectorPTFEvaluatorCountStar(WindowFrameDef windowFrameDef, VectorExpressi
   @Override
   public void evaluateGroupBatch(VectorizedRowBatch batch) {
     // No input expression for COUNT(*).
-    // evaluateInputExpr(batch);
 
     // Count all rows.
     count += batch.size;

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDecimalAvg.java

@@ -54,7 +54,6 @@ public VectorPTFEvaluatorDecimalAvg(WindowFrameDef windowFrameDef, VectorExpress
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null decimal column values for avg; maintain isGroupResultNull; after last row of
     // last group batch compute the group avg when sum is non-null.

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDecimalFirstValue.java

@@ -50,7 +50,6 @@ public VectorPTFEvaluatorDecimalFirstValue(WindowFrameDef windowFrameDef,
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // First row determines isGroupResultNull and decimal firstValue; stream fill result as repeated.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDecimalLastValue.java

@@ -50,7 +50,6 @@ public VectorPTFEvaluatorDecimalLastValue(WindowFrameDef windowFrameDef,
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Last row of last batch determines isGroupResultNull and decimal lastValue.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDecimalMax.java

@@ -48,7 +48,6 @@ public VectorPTFEvaluatorDecimalMax(WindowFrameDef windowFrameDef, VectorExpress
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine maximum of all non-null decimal column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDecimalMin.java

@@ -46,7 +46,6 @@ public VectorPTFEvaluatorDecimalMin(WindowFrameDef windowFrameDef, VectorExpress
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine minimum of all non-null decimal column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDecimalSum.java

@@ -48,7 +48,6 @@ public VectorPTFEvaluatorDecimalSum(WindowFrameDef windowFrameDef, VectorExpress
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null decimal column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDoubleAvg.java

@@ -46,7 +46,6 @@ public VectorPTFEvaluatorDoubleAvg(WindowFrameDef windowFrameDef, VectorExpressi
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null double column values for avg; maintain isGroupResultNull; after last row of
     // last group batch compute the group avg when sum is non-null.

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDoubleFirstValue.java

@@ -49,7 +49,6 @@ public VectorPTFEvaluatorDoubleFirstValue(WindowFrameDef windowFrameDef,
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // First row determines isGroupResultNull and double firstValue; stream fill result as repeated.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDoubleLastValue.java

@@ -47,7 +47,6 @@ public VectorPTFEvaluatorDoubleLastValue(WindowFrameDef windowFrameDef,
   public void evaluateGroupBatch(VectorizedRowBatch batch)
     throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Last row of last batch determines isGroupResultNull and double lastValue.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDoubleMax.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorDoubleMax(WindowFrameDef windowFrameDef, VectorExpressi
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine maximum of all non-null double column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDoubleMin.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorDoubleMin(WindowFrameDef windowFrameDef, VectorExpressi
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine minimum of all non-null double column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorDoubleSum.java

@@ -42,7 +42,6 @@ public VectorPTFEvaluatorDoubleSum(WindowFrameDef windowFrameDef, VectorExpressi
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null double column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLag.java

@@ -38,7 +38,7 @@ public Object runOnRange(int rowNum, Range range, VectorPTFGroupBatches batches)
     if (lagRow < 0 || lagRow >= batches.size()) {
       return getDefaultValue(rowNum, batches);
     } else {
-      return batches.getValueAndEvaluateInputExpression(this, lagRow, inputColumnNum);
+      return batches.getValue(lagRow, inputColumnNum);
     }
   }
 }

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLead.java

@@ -38,7 +38,7 @@ public Object runOnRange(int rowNum, Range range, VectorPTFGroupBatches batches)
     if (leadRow < 0 || leadRow >= batches.size()) {
       return getDefaultValue(rowNum, batches);
     } else {
-      return batches.getValueAndEvaluateInputExpression(this, leadRow, inputColumnNum);
+      return batches.getValue(leadRow, inputColumnNum);
     }
   }
 }

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLongAvg.java

@@ -46,7 +46,6 @@ public VectorPTFEvaluatorLongAvg(WindowFrameDef windowFrameDef, VectorExpression
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null long column values for avg; maintain isGroupResultNull; after last row of
     // last group batch compute the group avg when sum is non-null.

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLongFirstValue.java

@@ -49,7 +49,6 @@ public VectorPTFEvaluatorLongFirstValue(WindowFrameDef windowFrameDef,
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // First row determines isGroupResultNull and long firstValue; stream fill result as repeated.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLongLastValue.java

@@ -48,7 +48,6 @@ public VectorPTFEvaluatorLongLastValue(WindowFrameDef windowFrameDef,
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Last row of last batch determines isGroupResultNull and long lastValue.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLongMax.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorLongMax(WindowFrameDef windowFrameDef, VectorExpression
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine maximum of all non-null long column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLongMin.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorLongMin(WindowFrameDef windowFrameDef, VectorExpression
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine minimum of all non-null long column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorLongSum.java

@@ -42,7 +42,6 @@ public VectorPTFEvaluatorLongSum(WindowFrameDef windowFrameDef, VectorExpression
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null long column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorRowNumber.java

@@ -44,7 +44,6 @@ public VectorPTFEvaluatorRowNumber(WindowFrameDef windowFrameDef, VectorExpressi
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     final int size = batch.size;
     LongColumnVector longColVector = (LongColumnVector) batch.cols[outputColumnNum];

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDecimalAvg.java

@@ -56,7 +56,6 @@ public VectorPTFEvaluatorStreamingDecimalAvg(WindowFrameDef windowFrameDef, Vect
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null decimal column values for avg; maintain isGroupResultNull; after last row of
     // last group batch compute the group avg when sum is non-null.

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDecimalMax.java

@@ -48,7 +48,6 @@ public VectorPTFEvaluatorStreamingDecimalMax(WindowFrameDef windowFrameDef, Vect
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine maximum of all non-null decimal column values; maintain isNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDecimalMin.java

@@ -48,7 +48,6 @@ public VectorPTFEvaluatorStreamingDecimalMin(WindowFrameDef windowFrameDef, Vect
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine minimum of all non-null decimal column values; maintain isNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDecimalSum.java

@@ -48,7 +48,6 @@ public VectorPTFEvaluatorStreamingDecimalSum(WindowFrameDef windowFrameDef, Vect
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null decimal column values; maintain isGroupResultNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDoubleAvg.java

@@ -50,7 +50,6 @@ public VectorPTFEvaluatorStreamingDoubleAvg(WindowFrameDef windowFrameDef, Vecto
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null double column values for avg; maintain isGroupResultNull; after last row of
     // last group batch compute the group avg when sum is non-null.

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDoubleMax.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorStreamingDoubleMax(WindowFrameDef windowFrameDef, Vecto
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine maximum of all non-null double column values; maintain isNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDoubleMin.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorStreamingDoubleMin(WindowFrameDef windowFrameDef, Vecto
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine minimum of all non-null double column values; maintain isNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingDoubleSum.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorStreamingDoubleSum(WindowFrameDef windowFrameDef, Vecto
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // We do not filter when PTF is in reducer.
     Preconditions.checkState(!batch.selectedInUse);

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingLongAvg.java

@@ -50,7 +50,6 @@ public VectorPTFEvaluatorStreamingLongAvg(WindowFrameDef windowFrameDef, VectorE
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null long column values for avg; maintain isGroupResultNull; after last row of
     // last group batch compute the group avg when sum is non-null.

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingLongMax.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorStreamingLongMax(WindowFrameDef windowFrameDef, VectorE
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine maximum of all non-null long column values; maintain isNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingLongMin.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorStreamingLongMin(WindowFrameDef windowFrameDef, VectorE
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Determine minimum of all non-null long column values; maintain isNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorStreamingLongSum.java

@@ -45,7 +45,6 @@ public VectorPTFEvaluatorStreamingLongSum(WindowFrameDef windowFrameDef, VectorE
   public void evaluateGroupBatch(VectorizedRowBatch batch)
       throws HiveException {
 
-    evaluateInputExpr(batch);
 
     // Sum all non-null long column values; maintain isNull.
 

ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java

@@ -336,15 +336,6 @@ public Object getValue(int row, int col) throws HiveException {
     return getValueFromBatch(bufferedBatches.get(rp.batchIndex), col, rp.rowIndexInBatch);
   }
 
-  public Object getValueAndEvaluateInputExpression(VectorPTFEvaluatorBase evaluator, int row,
-      int col) throws HiveException {
-    RowPositionInBatch rp = getPosition(row);
-    BufferedVectorizedRowBatch batch = bufferedBatches.get(rp.batchIndex);
-    if (!batch.isInputExpressionEvaluated) {
-      evaluator.evaluateInputExpr(batch);
-    }
-    return getValueFromBatch(batch, col, rp.rowIndexInBatch);
-  }
 
   public RowPositionInBatch getPosition(int i) throws HiveException {
     if (positionCache[i] != null) {
@@ -997,7 +988,6 @@ private VectorizedRowBatch createTruncatedBufferedBatch(BufferedVectorizedRowBat
 
     partialBatch.size = size;
     partialBatch.isLastGroupBatch = bufferedBatch.isLastGroupBatch;
-    partialBatch.isInputExpressionEvaluated = bufferedBatch.isInputExpressionEvaluated;
     return partialBatch;
   }
 
@@ -1159,7 +1149,6 @@ private void readBlockFromSpillToMemory(VectorSpillBlock block) throws HiveExcep
         block.readSingleRowFromBytesContainer(bufferedBatch);
         bufferedBatch.size += 1;
         bufferedBatch.isLastGroupBatch = block.isLastGroupBatch[batchIndex];
-        bufferedBatch.isInputExpressionEvaluated = block.isInputExpressionEvaluated[batchIndex];
       }
       currentBufferedBatchCount = block.spillBatchCount;
       partitionMetrics.batchesReadFromSpill += currentBufferedBatchCount;
@@ -1187,7 +1176,6 @@ private void spillAndResetCurrentBufferedBatches() throws HiveException {
       BufferedVectorizedRowBatch bufferedBatch = bufferedBatches.get(i);
       block.spillBatch(bufferedBatch);
       block.isLastGroupBatch[i] = bufferedBatch.isLastGroupBatch;
-      block.isInputExpressionEvaluated[i] = bufferedBatch.isInputExpressionEvaluated;
       block.spillRowCount += bufferedBatch.size;
       block.spillBatchCount += 1;
       if (previousBatch != null) {