ggangadharan commented on code in PR #6898: URL: https://github.com/apache/hive/pull/6898#discussion_r4347688987
########## ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorVariance.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.hadoop.hive.ql.exec.vector.ptf; + +import org.apache.hadoop.hive.ql.exec.vector.ColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.ColumnVector.Type; +import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector; +import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; +import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.hive.ql.plan.ptf.WindowFrameDef; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFVariance.VarianceKind; +import org.apache.hadoop.hive.ql.udf.ptf.Range; +import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable; + +import com.google.common.base.Preconditions; + +/** + * This class evaluates the variance family functions for a PTF group. + * + * The group result is calculated from the non-null values of the window range. When a range + * starts where the previous one started and does not end before it (e.g. UNBOUNDED PRECEDING + * frames), the previous aggregation state is reused and only the new rows are iterated. As the + * values are still iterated in row order, the result is the same as a full recalculation. + */ +public class VectorPTFEvaluatorVariance extends VectorPTFEvaluatorAbstractVariance { + + private Range previousRange; + private long previousCount; + private double previousSum; + private double previousVariance; + + public VectorPTFEvaluatorVariance(WindowFrameDef windowFrameDef, VectorExpression inputVecExpr, + int outputColumnNum, Type inputColumnVectorType, VarianceKind varianceKind) { + super(windowFrameDef, inputVecExpr, outputColumnNum, inputColumnVectorType, varianceKind); + resetEvaluator(); + } + + @Override + public void evaluateGroupBatch(VectorizedRowBatch batch) + throws HiveException { + + // We do not filter when PTF is in reducer. + Preconditions.checkState(!batch.selectedInUse); + + final int size = batch.size; + if (size == 0) { + return; + } + ColumnVector colVector = batch.cols[inputColumnNum]; + if (colVector.isRepeating) { + if (colVector.noNulls || !colVector.isNull[0]) { + final double value = getDouble(colVector, 0); + for (int i = 0; i < size; i++) { + iterate(value); + } + } + return; + } + + switch (inputColumnVectorType) { + case LONG: + iterateLongs((LongColumnVector) colVector, size); + break; + case DOUBLE: + iterateDoubles((DoubleColumnVector) colVector, size); + break; + case DECIMAL: + iterateDecimals((DecimalColumnVector) colVector, size); + break; + default: + throw new IllegalStateException("Unexpected column vector type " + inputColumnVectorType); + } + } + + private void iterateLongs(LongColumnVector colVector, int size) { Review Comment: Thanks! benchmarked this before deciding. The details are below so the numbers can be checked or reproduced. **Result, with all 3 input types running in the same JVM (as in HS2/LLAP):** | Evaluator | NULLs | Per-type loops (current) | Single loop with `getDouble()` (suggested) | |---|---|---|---| | buffered | none | 2.17 ± 0.12 ns/row | 2.76 ± 0.41 ns/row (+27%) | | buffered | 10% | 2.56 ± 0.10 ns/row | 3.40 ± 0.33 ns/row (+33%) | | streaming | none | 5.48 ± 0.27 ns/row | 6.99 ± 0.15 ns/row (+28%) | | streaming | 10% | 5.31 ± 0.28 ns/row | 6.34 ± 0.67 ns/row (+19%) | So I'd prefer to keep one loop per column vector type, which is also the usual pattern in the vectorized code. #### What was compared Two versions of `evaluateGroupBatch`; everything else is identical: - **Per-type loops (current PR code):** one `switch` per batch, then a tight loop over `long[]`, `double[]` or `HiveDecimalWritable[]`. - **Single loop (suggested):** `iterate(getDouble(colVector, i))` for every row, so the `switch` and the cast run once per row. Both versions extend the real `VectorPTFEvaluatorAbstractVariance` from the PR's `hive-exec` jar, so `iterate()`, `getDouble()`, `isResultNull()` and `calculateResult()` are the actual PR code. Only the loop shape differs. Both evaluators are measured: the buffered one, and the streaming one, which also writes the running result into the output column for every row. The function is `stddev_samp`, so the streaming result includes the `sqrt`. #### Input data - 1024-row batches (`VectorizedRowBatch.DEFAULT_SIZE`), 100 batches per input type. They are generated once in `@Setup` with a fixed random seed (42), and both versions get the same data. - LONG: random integers in [-1,000,000, 1,000,000). DOUBLE: random values in [0, 1,000,000). DECIMAL(12,2): random values with 2 decimal places. - "10% NULLs": about 1 row in 10 is randomly marked NULL (`noNulls = false`). #### What one operation measures One JMH operation = `resetEvaluator()` followed by `evaluateGroupBatch()` over all 100 batches, for each input type in the run. In the MIXED case, that is a LONG, a DOUBLE and a DECIMAL evaluator one after another, 307,200 rows in total. The final result goes to a `Blackhole`, so the JIT can't optimize the work away. ns/row = JMH score (µs/op) × 1000 / rows per operation. For example, buffered, per-type loops, no NULLs: 667.1 µs/op × 1000 / 307,200 rows = 2.17 ns/row. The ± is JMH's 99.9% confidence interval. #### Why the "all 3 types in one JVM" case is the one that matters The JIT compiles code based on what it has seen at runtime: - If a JVM only ever runs one input type, say LONG, the `switch` inside `getDouble()` always takes the same branch. HotSpot then compiles the single loop as if it were a LONG-only loop, and both versions perform about the same (table below). - In HS2/LLAP, one long-running JVM runs queries over all column types, and all of them share one compiled `evaluateGroupBatch`. Once LONG, DOUBLE and DECIMAL inputs have all gone through the single loop, the JIT has to keep the per-row `switch` and casts inside the hot loop, and it gets slower. - With per-type loops, each loop only ever sees one type, so each stays fully specialized no matter what else the JVM runs. The MIXED benchmark reproduces this by running all three input types in the same JVM fork. #### Settings and environment - JMH 1.37, `Mode.AverageTime`, 3 forks × (5 warmup + 8 measurement iterations of 1 s) = 24 samples per result, `-Xms1g -Xmx1g` - JDK 21.0.12 (OpenJDK 64-Bit Server VM), Apple M3 Pro (12 cores), macOS; machine otherwise idle during the run <details> <summary>Single input type per JVM (2 forks × 6 iterations): both versions about the same</summary> | Evaluator | Input type | NULLs | Per-type loops (current) | Single loop with getDouble() | Difference | |---|---|---|---|---|---| | buffered | LONG | none | 0.87 ± 0.13 | 0.93 ± 0.16 | +7% | | buffered | LONG | 10% | 1.29 ± 0.24 | 1.65 ± 0.95 | +29% | | buffered | DOUBLE | none | 0.81 ± 0.03 | 0.82 ± 0.02 | +1% | | buffered | DOUBLE | 10% | 1.29 ± 0.25 | 1.57 ± 0.14 | +22% | | buffered | DECIMAL | none | 4.00 ± 1.11 | 4.45 ± 1.06 | +11% | | buffered | DECIMAL | 10% | 4.54 ± 0.14 | 4.39 ± 1.19 | -3% | | streaming | LONG | none | 2.36 ± 0.70 | 1.80 ± 0.04 | -24% | | streaming | LONG | 10% | 2.28 ± 0.10 | 2.16 ± 0.07 | -5% | | streaming | DOUBLE | none | 1.73 ± 0.03 | 1.75 ± 0.04 | +1% | | streaming | DOUBLE | 10% | 2.17 ± 0.12 | 2.17 ± 0.20 | -0% | | streaming | DECIMAL | none | 7.84 ± 1.14 | 7.27 ± 0.21 | -7% | | streaming | DECIMAL | 10% | 7.01 ± 0.32 | 6.87 ± 0.14 | -2% | No consistent winner here. The larger differences are within the error bars, e.g. streaming LONG has ± 0.70 on the per-type side. </details> <details> <summary>How to reproduce</summary> Uses the `hive-exec` jar built from this PR (`mvn install -pl ql -DskipTests`) and JMH 1.37: ```bash M=~/.m2/repository CP=$M/org/apache/hive/hive-exec/4.3.0-SNAPSHOT/hive-exec-4.3.0-SNAPSHOT.jar CP=$CP:$M/org/apache/hadoop/hadoop-common/3.4.2/hadoop-common-3.4.2.jar CP=$CP:$M/org/slf4j/slf4j-api/2.0.9/slf4j-api-2.0.9.jar CP=$CP:$M/org/openjdk/jmh/jmh-core/1.37/jmh-core-1.37.jar CP=$CP:$M/org/openjdk/jmh/jmh-generator-annprocess/1.37/jmh-generator-annprocess-1.37.jar CP=$CP:$M/net/sf/jopt-simple/jopt-simple/5.0.4/jopt-simple-5.0.4.jar CP=$CP:$M/org/apache/commons/commons-math3/3.6.1/commons-math3-3.6.1.jar mkdir -p classes javac -proc:full -cp "$CP" -d classes bench/VarianceLoopBenchmark.java # all 3 input types in one JVM (table above) java -cp "classes:$CP" org.openjdk.jmh.Main bench.VarianceLoopBenchmark -p type=MIXED -f 3 -wi 5 -i 8 # one input type per JVM (table in the section above) java -cp "classes:$CP" org.openjdk.jmh.Main bench.VarianceLoopBenchmark -p type=LONG,DOUBLE,DECIMAL ``` `bench/VarianceLoopBenchmark.java`: ```java package bench; import java.util.Random; import java.util.concurrent.TimeUnit; import org.apache.hadoop.hive.ql.exec.vector.ColumnVector; import org.apache.hadoop.hive.ql.exec.vector.ColumnVector.Type; import org.apache.hadoop.hive.ql.exec.vector.DecimalColumnVector; import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector; import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector; import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; import org.apache.hadoop.hive.ql.exec.vector.expressions.IdentityExpression; import org.apache.hadoop.hive.ql.exec.vector.ptf.VectorPTFEvaluatorAbstractVariance; import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFVariance.VarianceKind; import org.apache.hadoop.hive.serde2.io.HiveDecimalWritable; import org.openjdk.jmh.annotations.*; import org.openjdk.jmh.infra.Blackhole; /** * split = current PR code: one loop per column vector type. * unified = reviewer suggestion: one loop calling getDouble(colVector, i) per row. * One op = 100 batches x 1024 rows. */ @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.MICROSECONDS) @State(Scope.Thread) @Warmup(iterations = 4, time = 1) @Measurement(iterations = 6, time = 1) @Fork(value = 2, jvmArgsAppend = {"-Xms1g", "-Xmx1g"}) public class VarianceLoopBenchmark { static final int SIZE = VectorizedRowBatch.DEFAULT_SIZE; static final int BATCHES = 100; @Param({"LONG", "DOUBLE", "DECIMAL", "MIXED"}) String type; @Param({"false", "true"}) boolean withNulls; VectorizedRowBatch[][] batches; // [typeIndex][batch] VectorPTFEvaluatorAbstractVariance[] splitBuffered, unifiedBuffered, splitStreaming, unifiedStreaming; @Setup public void setup() { Type[] types = "MIXED".equals(type) ? new Type[] {Type.LONG, Type.DOUBLE, Type.DECIMAL} : new Type[] {Type.valueOf(type)}; Random rnd = new Random(42); batches = new VectorizedRowBatch[types.length][BATCHES]; splitBuffered = new VectorPTFEvaluatorAbstractVariance[types.length]; unifiedBuffered = new VectorPTFEvaluatorAbstractVariance[types.length]; splitStreaming = new VectorPTFEvaluatorAbstractVariance[types.length]; unifiedStreaming = new VectorPTFEvaluatorAbstractVariance[types.length]; for (int t = 0; t < types.length; t++) { for (int b = 0; b < BATCHES; b++) { batches[t][b] = newBatch(types[t], rnd); } splitBuffered[t] = new SplitBuffered(types[t]); unifiedBuffered[t] = new UnifiedBuffered(types[t]); splitStreaming[t] = new SplitStreaming(types[t]); unifiedStreaming[t] = new UnifiedStreaming(types[t]); } } private VectorizedRowBatch newBatch(Type t, Random rnd) { VectorizedRowBatch batch = new VectorizedRowBatch(2, SIZE); ColumnVector in; switch (t) { case LONG: LongColumnVector l = new LongColumnVector(SIZE); for (int i = 0; i < SIZE; i++) l.vector[i] = rnd.nextInt(2_000_000) - 1_000_000; in = l; break; case DOUBLE: DoubleColumnVector d = new DoubleColumnVector(SIZE); for (int i = 0; i < SIZE; i++) d.vector[i] = rnd.nextDouble() * 1e6; in = d; break; default: DecimalColumnVector dec = new DecimalColumnVector(SIZE, 12, 2); for (int i = 0; i < SIZE; i++) dec.vector[i] = new HiveDecimalWritable(String.format("%.2f", rnd.nextDouble() * 1e6)); in = dec; } if (withNulls) { in.noNulls = false; for (int i = 0; i < SIZE; i++) in.isNull[i] = rnd.nextInt(10) == 0; } batch.cols[0] = in; batch.cols[1] = new DoubleColumnVector(SIZE); batch.size = SIZE; return batch; } private double run(VectorPTFEvaluatorAbstractVariance[] evaluators) throws Exception { double r = 0; for (int t = 0; t < evaluators.length; t++) { VectorPTFEvaluatorAbstractVariance e = evaluators[t]; e.resetEvaluator(); for (VectorizedRowBatch batch : batches[t]) { e.evaluateGroupBatch(batch); } r += ((Probe) e).probe(); } return r; } @Benchmark public void bufferedSplit(Blackhole bh) throws Exception { bh.consume(run(splitBuffered)); } @Benchmark public void bufferedUnified(Blackhole bh) throws Exception { bh.consume(run(unifiedBuffered)); } @Benchmark public void streamingSplit(Blackhole bh) throws Exception { bh.consume(run(splitStreaming)); } @Benchmark public void streamingUnified(Blackhole bh) throws Exception { bh.consume(run(unifiedStreaming)); } interface Probe { double probe(); } // ---- buffered: split (current PR code) ---- static final class SplitBuffered extends VectorPTFEvaluatorAbstractVariance implements Probe { SplitBuffered(Type t) { super(null, new IdentityExpression(0), 1, t, VarianceKind.STANDARD_DEVIATION_SAMPLE); resetEvaluator(); } public double probe() { return isResultNull() ? 0 : calculateResult(); } @Override public boolean streamsResult() { return false; } @Override public void evaluateGroupBatch(VectorizedRowBatch batch) { final int size = batch.size; ColumnVector colVector = batch.cols[inputColumnNum]; switch (inputColumnVectorType) { case LONG: iterateLongs((LongColumnVector) colVector, size); break; case DOUBLE: iterateDoubles((DoubleColumnVector) colVector, size); break; case DECIMAL: iterateDecimals((DecimalColumnVector) colVector, size); break; default: throw new IllegalStateException(); } } private void iterateLongs(LongColumnVector colVector, int size) { final long[] vector = colVector.vector; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { iterate((double) vector[i]); } } } private void iterateDoubles(DoubleColumnVector colVector, int size) { final double[] vector = colVector.vector; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { iterate(vector[i]); } } } private void iterateDecimals(DecimalColumnVector colVector, int size) { final HiveDecimalWritable[] vector = colVector.vector; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { iterate(vector[i].doubleValue()); } } } } // ---- buffered: unified (reviewer suggestion) ---- static final class UnifiedBuffered extends VectorPTFEvaluatorAbstractVariance implements Probe { UnifiedBuffered(Type t) { super(null, new IdentityExpression(0), 1, t, VarianceKind.STANDARD_DEVIATION_SAMPLE); resetEvaluator(); } public double probe() { return isResultNull() ? 0 : calculateResult(); } @Override public boolean streamsResult() { return false; } @Override public void evaluateGroupBatch(VectorizedRowBatch batch) { final int size = batch.size; ColumnVector colVector = batch.cols[inputColumnNum]; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { iterate(getDouble(colVector, i)); } } } } // ---- streaming: split (current PR code) ---- static final class SplitStreaming extends VectorPTFEvaluatorAbstractVariance implements Probe { private boolean isCurrentResultNull; private double currentResult; SplitStreaming(Type t) { super(null, new IdentityExpression(0), 1, t, VarianceKind.STANDARD_DEVIATION_SAMPLE); resetEvaluator(); } public double probe() { return currentResult; } @Override public boolean streamsResult() { return true; } @Override public void resetEvaluator() { super.resetEvaluator(); isCurrentResultNull = true; currentResult = 0; } @Override public void evaluateGroupBatch(VectorizedRowBatch batch) { final int size = batch.size; ColumnVector colVector = batch.cols[inputColumnNum]; DoubleColumnVector out = (DoubleColumnVector) batch.cols[outputColumnNum]; switch (inputColumnVectorType) { case LONG: evaluateLongs((LongColumnVector) colVector, out, size); break; case DOUBLE: evaluateDoubles((DoubleColumnVector) colVector, out, size); break; case DECIMAL: evaluateDecimals((DecimalColumnVector) colVector, out, size); break; default: throw new IllegalStateException(); } } private void evaluateLongs(LongColumnVector colVector, DoubleColumnVector out, int size) { final long[] vector = colVector.vector; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { aggregate((double) vector[i]); } setOutput(out, i); } } private void evaluateDoubles(DoubleColumnVector colVector, DoubleColumnVector out, int size) { final double[] vector = colVector.vector; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { aggregate(vector[i]); } setOutput(out, i); } } private void evaluateDecimals(DecimalColumnVector colVector, DoubleColumnVector out, int size) { final HiveDecimalWritable[] vector = colVector.vector; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { aggregate(vector[i].doubleValue()); } setOutput(out, i); } } private void aggregate(double value) { iterate(value); isCurrentResultNull = isResultNull(); if (!isCurrentResultNull) { currentResult = calculateResult(); } } private void setOutput(DoubleColumnVector out, int i) { if (isCurrentResultNull) { out.noNulls = false; out.isNull[i] = true; } else { out.isNull[i] = false; out.vector[i] = currentResult; } } } // ---- streaming: unified (reviewer suggestion) ---- static final class UnifiedStreaming extends VectorPTFEvaluatorAbstractVariance implements Probe { private boolean isCurrentResultNull; private double currentResult; UnifiedStreaming(Type t) { super(null, new IdentityExpression(0), 1, t, VarianceKind.STANDARD_DEVIATION_SAMPLE); resetEvaluator(); } public double probe() { return currentResult; } @Override public boolean streamsResult() { return true; } @Override public void resetEvaluator() { super.resetEvaluator(); isCurrentResultNull = true; currentResult = 0; } @Override public void evaluateGroupBatch(VectorizedRowBatch batch) { final int size = batch.size; ColumnVector colVector = batch.cols[inputColumnNum]; DoubleColumnVector out = (DoubleColumnVector) batch.cols[outputColumnNum]; final boolean noNulls = colVector.noNulls; final boolean[] isNull = colVector.isNull; for (int i = 0; i < size; i++) { if (noNulls || !isNull[i]) { aggregate(getDouble(colVector, i)); } setOutput(out, i); } } private void aggregate(double value) { iterate(value); isCurrentResultNull = isResultNull(); if (!isCurrentResultNull) { currentResult = calculateResult(); } } private void setOutput(DoubleColumnVector out, int i) { if (isCurrentResultNull) { out.noNulls = false; out.isNull[i] = true; } else { out.isNull[i] = false; out.vector[i] = currentResult; } } } } ``` </details> So I'd prefer to keep one loop per column vector type, which is also the usual pattern in the vectorized code. Thanks to Gemini for helping to sharing this details. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
