You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/04/20 20:36:03 UTC
(spark) branch master updated: [SPARK-56522][SQL] Batch PACKED null/non-null runs in `VectorizedRleValuesReader`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 32be66f3ca0b [SPARK-56522][SQL] Batch PACKED null/non-null runs in `VectorizedRleValuesReader`
32be66f3ca0b is described below
commit 32be66f3ca0b83db62589bb6cdaa671d71eb41ce
Author: YangJie <ya...@baidu.com>
AuthorDate: Mon Apr 20 13:35:53 2026 -0700
[SPARK-56522][SQL] Batch PACKED null/non-null runs in `VectorizedRleValuesReader`
### What changes were proposed in this pull request?
Rewrite the PACKED branches of `readBatchInternal` and `readValuesN` in `VectorizedRleValuesReader` to scan `currentBuffer` for contiguous runs and batch the value-side reads, replacing the per-row virtual dispatch.
- Non-null runs: one `updater.readValues(runLen, ...)` per run, with a `runLen == 1` fast path calling `updater.readValue(...)`.
- Null and def-level writes stay per-row (`putNull` / `putInt`). Batching them via `putNulls` / `putInts` with short counts (PACKED runs are capped at 7 by the Parquet hybrid encoder) pollutes those methods' shared JIT profile — the RLE branch calls them with `n = 4096` and loses its count-specialized code, regressing the RLE fast path by 3-4×.
- PACKED bodies are extracted into `readPackedBatch` / `readPackedBatchWithDefLevels` so the enclosing methods stay small enough for C2 to inline the RLE fast path.
### Why are the changes needed?
`VectorizedRleValuesReader` is on the critical path of every nullable Parquet column read. In PACKED mode (scattered-null columns) the original code dispatches per element through a virtual call. Batching the value side converts O(n) virtual dispatch into O(runs) batch calls while leaving the RLE fast path untouched.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- Add `VectorizedRleValuesReaderSuite`
- Pass Github Actions
**Benchmark**
`VectorizedRleValuesReaderBenchmark` on GHA (AMD EPYC 7763, JDK 17). All three JDK result files are updated in this PR; JDK 21 and JDK 25 show comparable results.
RLE fast paths stay at baseline:
| Case | master (M rows/s) | this PR (M rows/s) | Δ |
|------|-------------------:|--------------------:|------:|
| Group C null=1.0 | 8057 | 8031 | -0.3% |
| Group D null=1.0 | 11580 | 11675 | +0.8% |
PACKED (per-row ns, lower is better):
| Case | master | this PR | Δ |
|------|-------:|--------:|-----:|
| Group D null=0.1 clustered | 10.1 | 5.3 | -48% |
| Group D null=0.3 clustered | 9.6 | 5.3 | -45% |
| Group D null=0.5 clustered | 9.1 | 5.3 | -42% |
| Group D null=0.9 clustered | 8.1 | 5.1 | -37% |
| Group D null=0.1 random | 11.0 | 7.2 | -35% |
| Group D null=0.9 random | 9.3 | 6.7 | -28% |
| Group C null=0.1 clustered | 12.2 | 6.9 | -43% |
| Group C null=0.3 clustered | 11.5 | 6.4 | -44% |
| Group C null=0.5 clustered | 10.9 | 6.3 | -42% |
| Group C null=0.9 clustered | 9.6 | 5.8 | -40% |
| Group C null=0.1 random | 12.8 | 8.9 | -30% |
| Group C null=0.9 random | 10.3 | 8.0 | -22% |
Short-run random cases at mid null ratios (0.3 / 0.5) are roughly flat — short runs benefit little from batching.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.7
Closes #55386 from LuciferYang/SPARK-56522.
Authored-by: YangJie <ya...@baidu.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
...rizedRleValuesReaderBenchmark-jdk21-results.txt | 82 +++---
...rizedRleValuesReaderBenchmark-jdk25-results.txt | 84 +++---
.../VectorizedRleValuesReaderBenchmark-results.txt | 84 +++---
.../parquet/VectorizedRleValuesReader.java | 115 ++++++--
.../parquet/ParquetReadStateTestAccess.scala | 119 ++++++++
.../VectorizedRleValuesReaderBenchmark.scala | 143 +---------
.../parquet/VectorizedRleValuesReaderSuite.scala | 314 +++++++++++++++++++++
.../VectorizedRleValuesReaderTestUtils.scala | 87 ++++++
8 files changed, 753 insertions(+), 275 deletions(-)
diff --git a/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk21-results.txt b/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk21-results.txt
index 0ea6a0f4829f..acdf8d9b5173 100644
--- a/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk21-results.txt
+++ b/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk21-results.txt
@@ -7,15 +7,15 @@ AMD EPYC 7763 64-Core Processor
RLE readBooleans decode: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
cold reader, trueRatio=0.0 0 0 0 25171.1 0.0 1.0X
-reused reader, trueRatio=0.0 0 0 0 25177.7 0.0 1.0X
-cold reader, trueRatio=0.1 1 1 0 993.9 1.0 0.0X
-reused reader, trueRatio=0.1 1 1 0 754.0 1.3 0.0X
-cold reader, trueRatio=0.5 1 1 0 1103.8 0.9 0.0X
-reused reader, trueRatio=0.5 1 1 0 833.4 1.2 0.0X
-cold reader, trueRatio=0.9 1 1 0 992.8 1.0 0.0X
-reused reader, trueRatio=0.9 1 1 0 752.9 1.3 0.0X
-cold reader, trueRatio=1.0 0 0 0 4201.3 0.2 0.2X
-reused reader, trueRatio=1.0 0 0 0 25177.1 0.0 1.0X
+reused reader, trueRatio=0.0 0 0 0 25171.1 0.0 1.0X
+cold reader, trueRatio=0.1 1 1 0 755.2 1.3 0.0X
+reused reader, trueRatio=0.1 1 1 0 754.1 1.3 0.0X
+cold reader, trueRatio=0.5 1 1 0 835.6 1.2 0.0X
+reused reader, trueRatio=0.5 1 1 0 833.3 1.2 0.0X
+cold reader, trueRatio=0.9 1 1 0 753.3 1.3 0.0X
+reused reader, trueRatio=0.9 1 1 0 753.6 1.3 0.0X
+cold reader, trueRatio=1.0 0 0 0 25165.0 0.0 1.0X
+reused reader, trueRatio=1.0 0 0 0 25183.2 0.0 1.0X
================================================================================================
@@ -26,18 +26,18 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
RLE readIntegers dictionary-id decode: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-PACKED cold, bitWidth=4 2 2 0 474.9 2.1 1.0X
-PACKED reused, bitWidth=4 2 2 0 477.3 2.1 1.0X
-RLE, bitWidth=4 0 0 0 4515.4 0.2 9.5X
-PACKED cold, bitWidth=8 2 2 0 510.1 2.0 1.1X
-PACKED reused, bitWidth=8 2 2 0 500.3 2.0 1.1X
-RLE, bitWidth=8 0 0 0 4516.0 0.2 9.5X
-PACKED cold, bitWidth=12 3 3 0 415.7 2.4 0.9X
-PACKED reused, bitWidth=12 3 3 0 415.9 2.4 0.9X
-RLE, bitWidth=12 0 0 0 4515.6 0.2 9.5X
-PACKED cold, bitWidth=20 3 3 0 352.4 2.8 0.7X
-PACKED reused, bitWidth=20 3 3 0 351.7 2.8 0.7X
-RLE, bitWidth=20 0 0 0 4514.8 0.2 9.5X
+PACKED cold, bitWidth=4 2 2 0 478.0 2.1 1.0X
+PACKED reused, bitWidth=4 2 2 0 475.7 2.1 1.0X
+RLE, bitWidth=4 0 0 0 4508.2 0.2 9.4X
+PACKED cold, bitWidth=8 2 2 0 518.0 1.9 1.1X
+PACKED reused, bitWidth=8 2 2 0 515.7 1.9 1.1X
+RLE, bitWidth=8 0 0 0 4508.9 0.2 9.4X
+PACKED cold, bitWidth=12 8 8 0 136.1 7.3 0.3X
+PACKED reused, bitWidth=12 8 8 0 136.1 7.3 0.3X
+RLE, bitWidth=12 0 0 0 4503.7 0.2 9.4X
+PACKED cold, bitWidth=20 3 3 0 353.3 2.8 0.7X
+PACKED reused, bitWidth=20 3 3 0 352.1 2.8 0.7X
+RLE, bitWidth=20 0 0 0 4508.9 0.2 9.4X
================================================================================================
@@ -48,16 +48,16 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
Nullable batch with def-levels: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-nullRatio=0.0, n/a 0 0 0 6820.6 0.1 1.0X
-nullRatio=0.1, random 11 11 0 94.0 10.6 0.0X
-nullRatio=0.1, clustered 10 10 0 105.0 9.5 0.0X
-nullRatio=0.3, random 13 13 0 79.0 12.7 0.0X
-nullRatio=0.3, clustered 9 10 0 110.9 9.0 0.0X
-nullRatio=0.5, random 14 14 0 74.0 13.5 0.0X
-nullRatio=0.5, clustered 9 9 0 118.5 8.4 0.0X
-nullRatio=0.9, random 9 9 0 119.3 8.4 0.0X
-nullRatio=0.9, clustered 7 8 0 140.4 7.1 0.0X
-nullRatio=1.0, random 0 0 0 5054.2 0.2 0.7X
+nullRatio=0.0, n/a 0 0 0 6502.7 0.2 1.0X
+nullRatio=0.1, random 8 9 0 123.7 8.1 0.0X
+nullRatio=0.1, clustered 6 6 0 169.7 5.9 0.0X
+nullRatio=0.3, random 12 12 0 87.3 11.5 0.0X
+nullRatio=0.3, clustered 6 6 0 168.0 6.0 0.0X
+nullRatio=0.5, random 13 13 0 79.8 12.5 0.0X
+nullRatio=0.5, clustered 6 6 0 171.7 5.8 0.0X
+nullRatio=0.9, random 8 8 0 136.0 7.4 0.0X
+nullRatio=0.9, clustered 6 6 0 181.6 5.5 0.0X
+nullRatio=1.0, random 0 0 0 5072.8 0.2 0.8X
================================================================================================
@@ -68,15 +68,15 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
Nullable batch without def-levels: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-nullRatio=0.0, n/a 0 0 0 12353.8 0.1 1.0X
-nullRatio=0.1, random 10 10 0 100.7 9.9 0.0X
-nullRatio=0.1, clustered 10 10 0 107.0 9.3 0.0X
-nullRatio=0.3, random 12 12 0 88.9 11.3 0.0X
-nullRatio=0.3, clustered 9 9 0 112.7 8.9 0.0X
-nullRatio=0.5, random 13 13 0 82.7 12.1 0.0X
-nullRatio=0.5, clustered 9 9 0 118.8 8.4 0.0X
-nullRatio=0.9, random 9 9 0 123.2 8.1 0.0X
-nullRatio=0.9, clustered 8 8 0 134.0 7.5 0.0X
-nullRatio=1.0, random 0 0 0 6191.1 0.2 0.5X
+nullRatio=0.0, n/a 0 0 0 11512.6 0.1 1.0X
+nullRatio=0.1, random 7 7 0 144.6 6.9 0.0X
+nullRatio=0.1, clustered 5 5 0 197.2 5.1 0.0X
+nullRatio=0.3, random 11 11 0 99.6 10.0 0.0X
+nullRatio=0.3, clustered 6 6 0 189.8 5.3 0.0X
+nullRatio=0.5, random 12 12 0 89.0 11.2 0.0X
+nullRatio=0.5, clustered 5 5 0 194.7 5.1 0.0X
+nullRatio=0.9, random 7 7 0 151.9 6.6 0.0X
+nullRatio=0.9, clustered 5 5 0 200.5 5.0 0.0X
+nullRatio=1.0, random 0 0 0 11945.0 0.1 1.0X
diff --git a/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk25-results.txt b/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk25-results.txt
index c2e20dfd582e..2b245cb20630 100644
--- a/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk25-results.txt
+++ b/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-jdk25-results.txt
@@ -6,16 +6,16 @@ OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
RLE readBooleans decode: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-cold reader, trueRatio=0.0 0 0 0 6055.4 0.2 1.0X
-reused reader, trueRatio=0.0 0 0 0 58763.5 0.0 9.7X
-cold reader, trueRatio=0.1 1 2 0 707.0 1.4 0.1X
-reused reader, trueRatio=0.1 2 2 0 563.8 1.8 0.1X
-cold reader, trueRatio=0.5 1 1 0 746.8 1.3 0.1X
-reused reader, trueRatio=0.5 1 1 0 743.0 1.3 0.1X
-cold reader, trueRatio=0.9 1 2 0 707.2 1.4 0.1X
-reused reader, trueRatio=0.9 1 2 0 706.8 1.4 0.1X
-cold reader, trueRatio=1.0 0 0 0 59098.0 0.0 9.8X
-reused reader, trueRatio=1.0 0 0 0 57130.7 0.0 9.4X
+cold reader, trueRatio=0.0 0 0 0 60530.9 0.0 1.0X
+reused reader, trueRatio=0.0 0 0 0 56821.1 0.0 0.9X
+cold reader, trueRatio=0.1 1 1 0 709.0 1.4 0.0X
+reused reader, trueRatio=0.1 1 1 0 708.2 1.4 0.0X
+cold reader, trueRatio=0.5 1 1 0 747.9 1.3 0.0X
+reused reader, trueRatio=0.5 1 1 0 746.5 1.3 0.0X
+cold reader, trueRatio=0.9 1 1 0 709.0 1.4 0.0X
+reused reader, trueRatio=0.9 1 2 0 705.6 1.4 0.0X
+cold reader, trueRatio=1.0 0 0 0 59164.7 0.0 1.0X
+reused reader, trueRatio=1.0 0 0 0 58832.7 0.0 1.0X
================================================================================================
@@ -26,18 +26,18 @@ OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
RLE readIntegers dictionary-id decode: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-PACKED cold, bitWidth=4 2 2 0 516.5 1.9 1.0X
-PACKED reused, bitWidth=4 2 2 0 516.8 1.9 1.0X
-RLE, bitWidth=4 0 0 0 18371.3 0.1 35.6X
-PACKED cold, bitWidth=8 2 2 0 570.1 1.8 1.1X
-PACKED reused, bitWidth=8 2 2 0 567.8 1.8 1.1X
-RLE, bitWidth=8 0 0 0 17893.8 0.1 34.6X
-PACKED cold, bitWidth=12 2 2 0 454.1 2.2 0.9X
-PACKED reused, bitWidth=12 2 2 0 453.3 2.2 0.9X
-RLE, bitWidth=12 0 0 0 18449.2 0.1 35.7X
-PACKED cold, bitWidth=20 3 3 0 373.7 2.7 0.7X
-PACKED reused, bitWidth=20 3 3 0 369.1 2.7 0.7X
-RLE, bitWidth=20 0 0 0 18054.3 0.1 35.0X
+PACKED cold, bitWidth=4 2 2 0 517.0 1.9 1.0X
+PACKED reused, bitWidth=4 2 2 0 518.3 1.9 1.0X
+RLE, bitWidth=4 0 0 0 18145.5 0.1 35.1X
+PACKED cold, bitWidth=8 2 2 0 570.0 1.8 1.1X
+PACKED reused, bitWidth=8 2 2 0 556.9 1.8 1.1X
+RLE, bitWidth=8 0 0 0 18098.2 0.1 35.0X
+PACKED cold, bitWidth=12 2 2 0 454.5 2.2 0.9X
+PACKED reused, bitWidth=12 2 2 0 453.0 2.2 0.9X
+RLE, bitWidth=12 0 0 0 18164.4 0.1 35.1X
+PACKED cold, bitWidth=20 3 3 0 374.6 2.7 0.7X
+PACKED reused, bitWidth=20 3 3 0 374.2 2.7 0.7X
+RLE, bitWidth=20 0 0 0 18462.1 0.1 35.7X
================================================================================================
@@ -48,16 +48,16 @@ OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
Nullable batch with def-levels: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-nullRatio=0.0, n/a 0 0 0 6840.2 0.1 1.0X
-nullRatio=0.1, random 11 11 0 93.4 10.7 0.0X
-nullRatio=0.1, clustered 10 10 0 104.9 9.5 0.0X
-nullRatio=0.3, random 13 13 0 80.6 12.4 0.0X
-nullRatio=0.3, clustered 9 9 0 114.2 8.8 0.0X
-nullRatio=0.5, random 14 14 1 77.0 13.0 0.0X
-nullRatio=0.5, clustered 8 8 1 125.9 7.9 0.0X
-nullRatio=0.9, random 8 8 0 130.7 7.7 0.0X
-nullRatio=0.9, clustered 7 7 0 160.9 6.2 0.0X
-nullRatio=1.0, random 0 0 0 8318.9 0.1 1.2X
+nullRatio=0.0, n/a 0 0 0 6625.9 0.2 1.0X
+nullRatio=0.1, random 9 9 0 121.8 8.2 0.0X
+nullRatio=0.1, clustered 6 6 0 170.3 5.9 0.0X
+nullRatio=0.3, random 13 13 0 83.8 11.9 0.0X
+nullRatio=0.3, clustered 6 6 0 170.4 5.9 0.0X
+nullRatio=0.5, random 14 14 0 75.8 13.2 0.0X
+nullRatio=0.5, clustered 6 6 0 172.1 5.8 0.0X
+nullRatio=0.9, random 8 8 0 131.0 7.6 0.0X
+nullRatio=0.9, clustered 6 6 0 181.7 5.5 0.0X
+nullRatio=1.0, random 0 0 0 8305.2 0.1 1.3X
================================================================================================
@@ -68,15 +68,15 @@ OpenJDK 64-Bit Server VM 25.0.2+10-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
Nullable batch without def-levels: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-nullRatio=0.0, n/a 0 0 0 12269.8 0.1 1.0X
-nullRatio=0.1, random 8 8 0 125.7 8.0 0.0X
-nullRatio=0.1, clustered 7 8 1 140.4 7.1 0.0X
-nullRatio=0.3, random 10 10 1 104.5 9.6 0.0X
-nullRatio=0.3, clustered 7 7 1 148.8 6.7 0.0X
-nullRatio=0.5, random 11 11 0 97.6 10.2 0.0X
-nullRatio=0.5, clustered 7 7 0 159.3 6.3 0.0X
-nullRatio=0.9, random 7 7 0 161.0 6.2 0.0X
-nullRatio=0.9, clustered 6 6 0 188.3 5.3 0.0X
-nullRatio=1.0, random 0 0 0 12016.1 0.1 1.0X
+nullRatio=0.0, n/a 0 0 0 11869.8 0.1 1.0X
+nullRatio=0.1, random 7 7 0 149.1 6.7 0.0X
+nullRatio=0.1, clustered 5 5 0 208.1 4.8 0.0X
+nullRatio=0.3, random 10 10 0 101.1 9.9 0.0X
+nullRatio=0.3, clustered 5 5 0 206.6 4.8 0.0X
+nullRatio=0.5, random 12 12 0 90.7 11.0 0.0X
+nullRatio=0.5, clustered 5 5 0 206.4 4.8 0.0X
+nullRatio=0.9, random 7 7 0 160.2 6.2 0.0X
+nullRatio=0.9, clustered 5 5 0 213.5 4.7 0.0X
+nullRatio=1.0, random 0 0 0 11957.2 0.1 1.0X
diff --git a/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-results.txt b/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-results.txt
index 007df4c669a2..10def40f3e1e 100644
--- a/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-results.txt
+++ b/sql/core/benchmarks/VectorizedRleValuesReaderBenchmark-results.txt
@@ -6,16 +6,16 @@ OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
RLE readBooleans decode: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-cold reader, trueRatio=0.0 0 0 0 72032.4 0.0 1.0X
-reused reader, trueRatio=0.0 0 0 0 71341.4 0.0 1.0X
-cold reader, trueRatio=0.1 1 1 0 894.4 1.1 0.0X
-reused reader, trueRatio=0.1 1 1 0 895.8 1.1 0.0X
-cold reader, trueRatio=0.5 1 1 0 1019.1 1.0 0.0X
-reused reader, trueRatio=0.5 1 1 0 1028.5 1.0 0.0X
-cold reader, trueRatio=0.9 1 1 0 892.6 1.1 0.0X
-reused reader, trueRatio=0.9 1 1 0 891.6 1.1 0.0X
-cold reader, trueRatio=1.0 0 0 0 67001.7 0.0 0.9X
-reused reader, trueRatio=1.0 0 0 0 72126.6 0.0 1.0X
+cold reader, trueRatio=0.0 0 0 0 47464.1 0.0 1.0X
+reused reader, trueRatio=0.0 0 0 0 47485.6 0.0 1.0X
+cold reader, trueRatio=0.1 1 1 0 894.1 1.1 0.0X
+reused reader, trueRatio=0.1 1 1 0 897.4 1.1 0.0X
+cold reader, trueRatio=0.5 1 1 0 1027.8 1.0 0.0X
+reused reader, trueRatio=0.5 1 1 0 1029.3 1.0 0.0X
+cold reader, trueRatio=0.9 1 1 0 893.8 1.1 0.0X
+reused reader, trueRatio=0.9 1 1 0 896.6 1.1 0.0X
+cold reader, trueRatio=1.0 0 0 0 47421.1 0.0 1.0X
+reused reader, trueRatio=1.0 0 0 0 47485.6 0.0 1.0X
================================================================================================
@@ -26,18 +26,18 @@ OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
RLE readIntegers dictionary-id decode: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-PACKED cold, bitWidth=4 2 2 0 487.4 2.1 1.0X
-PACKED reused, bitWidth=4 2 2 0 485.7 2.1 1.0X
-RLE, bitWidth=4 0 0 0 18400.3 0.1 37.8X
-PACKED cold, bitWidth=8 2 2 0 480.0 2.1 1.0X
-PACKED reused, bitWidth=8 2 2 0 477.4 2.1 1.0X
-RLE, bitWidth=8 0 0 0 18262.1 0.1 37.5X
-PACKED cold, bitWidth=12 3 3 0 355.5 2.8 0.7X
-PACKED reused, bitWidth=12 3 3 0 355.4 2.8 0.7X
-RLE, bitWidth=12 0 0 0 18442.3 0.1 37.8X
-PACKED cold, bitWidth=20 3 4 0 303.4 3.3 0.6X
-PACKED reused, bitWidth=20 3 3 0 303.0 3.3 0.6X
-RLE, bitWidth=20 0 0 0 18435.9 0.1 37.8X
+PACKED cold, bitWidth=4 2 2 0 488.1 2.0 1.0X
+PACKED reused, bitWidth=4 2 2 0 485.9 2.1 1.0X
+RLE, bitWidth=4 0 0 0 19116.1 0.1 39.2X
+PACKED cold, bitWidth=8 2 2 0 482.1 2.1 1.0X
+PACKED reused, bitWidth=8 2 2 0 479.7 2.1 1.0X
+RLE, bitWidth=8 0 0 0 18567.1 0.1 38.0X
+PACKED cold, bitWidth=12 3 3 0 362.1 2.8 0.7X
+PACKED reused, bitWidth=12 3 3 0 361.3 2.8 0.7X
+RLE, bitWidth=12 0 0 0 19126.9 0.1 39.2X
+PACKED cold, bitWidth=20 3 3 0 308.6 3.2 0.6X
+PACKED reused, bitWidth=20 3 3 0 306.5 3.3 0.6X
+RLE, bitWidth=20 0 0 0 19074.4 0.1 39.1X
================================================================================================
@@ -48,16 +48,16 @@ OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
Nullable batch with def-levels: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-nullRatio=0.0, n/a 0 0 0 6608.6 0.2 1.0X
-nullRatio=0.1, random 13 14 0 78.2 12.8 0.0X
-nullRatio=0.1, clustered 13 13 0 82.2 12.2 0.0X
-nullRatio=0.3, random 15 15 0 70.4 14.2 0.0X
-nullRatio=0.3, clustered 12 12 0 86.9 11.5 0.0X
-nullRatio=0.5, random 15 15 0 68.6 14.6 0.0X
-nullRatio=0.5, clustered 11 12 1 91.6 10.9 0.0X
-nullRatio=0.9, random 11 11 0 96.9 10.3 0.0X
-nullRatio=0.9, clustered 10 10 0 104.1 9.6 0.0X
-nullRatio=1.0, random 0 0 0 8057.0 0.1 1.2X
+nullRatio=0.0, n/a 0 0 0 6441.9 0.2 1.0X
+nullRatio=0.1, random 9 9 0 113.0 8.9 0.0X
+nullRatio=0.1, clustered 7 7 0 144.8 6.9 0.0X
+nullRatio=0.3, random 13 13 0 78.9 12.7 0.0X
+nullRatio=0.3, clustered 7 7 0 156.6 6.4 0.0X
+nullRatio=0.5, random 15 15 0 71.9 13.9 0.0X
+nullRatio=0.5, clustered 7 7 0 159.2 6.3 0.0X
+nullRatio=0.9, random 8 8 0 124.8 8.0 0.0X
+nullRatio=0.9, clustered 6 6 0 171.3 5.8 0.0X
+nullRatio=1.0, random 0 0 0 8031.7 0.1 1.2X
================================================================================================
@@ -68,15 +68,15 @@ OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.17.0-1010-azure
AMD EPYC 7763 64-Core Processor
Nullable batch without def-levels: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-nullRatio=0.0, n/a 0 0 0 11751.8 0.1 1.0X
-nullRatio=0.1, random 11 12 0 91.2 11.0 0.0X
-nullRatio=0.1, clustered 11 11 0 99.2 10.1 0.0X
-nullRatio=0.3, random 14 14 0 76.9 13.0 0.0X
-nullRatio=0.3, clustered 10 10 0 104.2 9.6 0.0X
-nullRatio=0.5, random 14 15 0 72.4 13.8 0.0X
-nullRatio=0.5, clustered 10 10 0 109.5 9.1 0.0X
-nullRatio=0.9, random 10 10 0 107.0 9.3 0.0X
-nullRatio=0.9, clustered 9 9 0 123.0 8.1 0.0X
-nullRatio=1.0, random 0 0 0 11580.1 0.1 1.0X
+nullRatio=0.0, n/a 0 0 0 11037.9 0.1 1.0X
+nullRatio=0.1, random 8 8 0 139.2 7.2 0.0X
+nullRatio=0.1, clustered 6 6 0 190.1 5.3 0.0X
+nullRatio=0.3, random 11 11 0 96.7 10.3 0.0X
+nullRatio=0.3, clustered 6 6 0 188.5 5.3 0.0X
+nullRatio=0.5, random 12 12 0 86.9 11.5 0.0X
+nullRatio=0.5, clustered 6 6 0 188.0 5.3 0.0X
+nullRatio=0.9, random 7 7 0 149.4 6.7 0.0X
+nullRatio=0.9, clustered 5 5 0 197.9 5.1 0.0X
+nullRatio=1.0, random 0 0 0 11675.8 0.1 1.1X
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
index efb41e1753c1..5dac20209ef3 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
@@ -245,16 +245,8 @@ public final class VectorizedRleValuesReader extends ValuesReader
}
state.valueOffset += n;
}
- case PACKED -> {
- for (int i = 0; i < n; ++i) {
- int currentValue = currentBuffer[currentBufferIdx++];
- if (currentValue == state.maxDefinitionLevel) {
- updater.readValue(state.valueOffset++, values, valueReader);
- } else {
- nulls.putNull(state.valueOffset++);
- }
- }
- }
+ case PACKED ->
+ readPackedBatch(n, state, values, nulls, valueReader, updater);
}
state.levelOffset += n;
leftInBatch -= n;
@@ -269,6 +261,52 @@ public final class VectorizedRleValuesReader extends ValuesReader
state.rowId = rowId;
}
+ /**
+ * PACKED-branch decode for {@link #readBatchInternal}. Extracted into a separate method so
+ * C2 can keep {@link #readBatchInternal}'s RLE fast path small and inlineable. Scans for
+ * contiguous null/non-null runs, batches value-side reads, but keeps null writes per-row
+ * to avoid polluting {@code putNulls(offset, count)}'s JIT profile (called with large
+ * counts from the RLE branch; short PACKED counts would average out its optimization).
+ */
+ private void readPackedBatch(
+ int n,
+ ParquetReadState state,
+ WritableColumnVector values,
+ WritableColumnVector nulls,
+ VectorizedValuesReader valueReader,
+ ParquetVectorUpdater updater) {
+ final int maxDefLevel = state.maxDefinitionLevel;
+ final int bufEnd = currentBufferIdx + n;
+ int valueOff = state.valueOffset;
+ while (currentBufferIdx < bufEnd) {
+ int runStart = currentBufferIdx;
+ if (currentBuffer[currentBufferIdx] == maxDefLevel) {
+ do {
+ currentBufferIdx++;
+ } while (currentBufferIdx < bufEnd
+ && currentBuffer[currentBufferIdx] == maxDefLevel);
+ int runLen = currentBufferIdx - runStart;
+ if (runLen == 1) {
+ updater.readValue(valueOff, values, valueReader);
+ } else {
+ updater.readValues(runLen, valueOff, values, valueReader);
+ }
+ valueOff += runLen;
+ } else {
+ do {
+ currentBufferIdx++;
+ } while (currentBufferIdx < bufEnd
+ && currentBuffer[currentBufferIdx] != maxDefLevel);
+ int runLen = currentBufferIdx - runStart;
+ for (int k = 0; k < runLen; k++) {
+ nulls.putNull(valueOff + k);
+ }
+ valueOff += runLen;
+ }
+ }
+ state.valueOffset = valueOff;
+ }
+
private void readBatchInternalWithDefLevels(
ParquetReadState state,
WritableColumnVector values,
@@ -636,18 +674,57 @@ public final class VectorizedRleValuesReader extends ValuesReader
state.valueOffset += n;
defLevels.putInts(state.levelOffset, n, currentValue);
}
- case PACKED -> {
- for (int i = 0; i < n; ++i) {
- int currentValue = currentBuffer[currentBufferIdx++];
- if (currentValue == state.maxDefinitionLevel) {
- updater.readValue(state.valueOffset++, values, valueReader);
- } else {
- nulls.putNull(state.valueOffset++);
- }
- defLevels.putInt(state.levelOffset + i, currentValue);
+ case PACKED ->
+ readPackedBatchWithDefLevels(
+ n, state, defLevels, values, nulls, valueReader, updater);
+ }
+ }
+
+ /**
+ * PACKED-branch decode for {@link #readValuesN}. Extracted for the same JIT reason as
+ * {@link #readPackedBatch}: keeps the parent's RLE fast path small enough to inline.
+ * Groups by exact def-level value (required for nested columns with multiple null def-
+ * level values), batches value-side reads via {@code readValues(runLen,...)} for
+ * runs > 1, but keeps null and def-level writes per-row so the large-count RLE call
+ * sites for {@code putNulls}/{@code putInts} retain monomorphic JIT profiles.
+ */
+ private void readPackedBatchWithDefLevels(
+ int n,
+ ParquetReadState state,
+ WritableColumnVector defLevels,
+ WritableColumnVector values,
+ WritableColumnVector nulls,
+ VectorizedValuesReader valueReader,
+ ParquetVectorUpdater updater) {
+ final int maxDefLevel = state.maxDefinitionLevel;
+ final int end = currentBufferIdx + n;
+ int valueOff = state.valueOffset;
+ int levelIdx = state.levelOffset;
+ while (currentBufferIdx < end) {
+ int runStart = currentBufferIdx;
+ int runValue = currentBuffer[currentBufferIdx];
+ do {
+ currentBufferIdx++;
+ } while (currentBufferIdx < end && currentBuffer[currentBufferIdx] == runValue);
+ int runLen = currentBufferIdx - runStart;
+ if (runValue == maxDefLevel) {
+ if (runLen == 1) {
+ updater.readValue(valueOff, values, valueReader);
+ } else {
+ updater.readValues(runLen, valueOff, values, valueReader);
+ }
+ } else {
+ for (int k = 0; k < runLen; k++) {
+ nulls.putNull(valueOff + k);
}
}
+ valueOff += runLen;
+ for (int k = 0; k < runLen; k++) {
+ defLevels.putInt(levelIdx + k, runValue);
+ }
+ levelIdx += runLen;
}
+ state.valueOffset = valueOff;
}
/**
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadStateTestAccess.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadStateTestAccess.scala
new file mode 100644
index 000000000000..d718ed49d710
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadStateTestAccess.scala
@@ -0,0 +1,119 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet
+
+import java.lang.reflect.{InvocationTargetException, Method}
+import java.util.PrimitiveIterator
+
+import org.apache.parquet.column.ColumnDescriptor
+
+import org.apache.spark.sql.execution.vectorized.WritableColumnVector
+import org.apache.spark.util.SparkClassUtils
+
+/**
+ * Reflective bridge to the package-private `ParquetReadState`. Under `spark-submit --jars`,
+ * test and main classes load from different classloaders, blocking package-private access.
+ * Reflection with `setAccessible` sidesteps the check without widening production visibility.
+ */
+object ParquetReadStateTestAccess {
+
+ private val stateCls = SparkClassUtils.classForName[Any](
+ "org.apache.spark.sql.execution.datasources.parquet.ParquetReadState")
+
+ private val ctor = {
+ val c = stateCls.getDeclaredConstructor(
+ classOf[ColumnDescriptor],
+ java.lang.Boolean.TYPE,
+ classOf[PrimitiveIterator.OfLong])
+ c.setAccessible(true)
+ c
+ }
+
+ private val resetForNewBatchMethod = {
+ val m = stateCls.getDeclaredMethod("resetForNewBatch", Integer.TYPE)
+ m.setAccessible(true)
+ m
+ }
+
+ private val resetForNewPageMethod = {
+ val m = stateCls.getDeclaredMethod(
+ "resetForNewPage", Integer.TYPE, java.lang.Long.TYPE)
+ m.setAccessible(true)
+ m
+ }
+
+ private val readBatchMethod: Method =
+ classOf[VectorizedRleValuesReader].getMethods
+ .find(m =>
+ m.getName == "readBatch"
+ && m.getParameterCount == 5
+ && m.getParameterTypes()(0) == stateCls)
+ .getOrElse(throw new NoSuchMethodException(
+ "VectorizedRleValuesReader.readBatch/5"))
+
+ def newState(
+ descriptor: ColumnDescriptor,
+ isRequired: Boolean,
+ rowIndexes: PrimitiveIterator.OfLong = null): AnyRef = {
+ try {
+ ctor.newInstance(
+ descriptor,
+ Boolean.box(isRequired),
+ rowIndexes).asInstanceOf[AnyRef]
+ } catch {
+ case e: ReflectiveOperationException => throw rethrow(e)
+ }
+ }
+
+ def resetForNewBatch(state: AnyRef, batchSize: Int): Unit =
+ try { resetForNewBatchMethod.invoke(state, Int.box(batchSize)) }
+ catch { case e: ReflectiveOperationException => throw rethrow(e) }
+
+ def resetForNewPage(
+ state: AnyRef,
+ totalValuesInPage: Int,
+ pageFirstRowIndex: Long): Unit =
+ try {
+ resetForNewPageMethod.invoke(
+ state, Int.box(totalValuesInPage), Long.box(pageFirstRowIndex))
+ } catch { case e: ReflectiveOperationException => throw rethrow(e) }
+
+ def readBatch(
+ reader: VectorizedRleValuesReader,
+ state: AnyRef,
+ values: WritableColumnVector,
+ defLevels: WritableColumnVector,
+ valueReader: VectorizedValuesReader,
+ updater: ParquetVectorUpdater): Unit =
+ try {
+ readBatchMethod.invoke(
+ reader, state, values, defLevels, valueReader, updater)
+ } catch { case e: ReflectiveOperationException => throw rethrow(e) }
+
+ private def rethrow(e: ReflectiveOperationException): RuntimeException = {
+ val cause = e match {
+ case ite: InvocationTargetException => ite.getCause
+ case other => other
+ }
+ cause match {
+ case re: RuntimeException => throw re
+ case er: Error => throw er
+ case _ => throw new RuntimeException(cause)
+ }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderBenchmark.scala
index e7fddf184c3e..008b18cdcab1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderBenchmark.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderBenchmark.scala
@@ -17,24 +17,16 @@
package org.apache.spark.sql.execution.datasources.parquet
-import java.io.ByteArrayOutputStream
-import java.lang.reflect.{InvocationTargetException, Method}
import java.nio.ByteBuffer
-import java.util.PrimitiveIterator
import scala.util.Random
-import org.apache.parquet.bytes.{ByteBufferInputStream, HeapByteBufferAllocator}
-import org.apache.parquet.column.{ColumnDescriptor, Dictionary}
-import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder
-import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
-import org.apache.parquet.schema.Type.Repetition
-import org.apache.parquet.schema.Types
+import org.apache.parquet.bytes.ByteBufferInputStream
import org.apache.spark.benchmark.{Benchmark, BenchmarkBase}
+import org.apache.spark.sql.execution.datasources.parquet.VectorizedRleValuesReaderTestUtils._
import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector, WritableColumnVector}
import org.apache.spark.sql.types.{BooleanType, IntegerType}
-import org.apache.spark.util.SparkClassUtils
/**
* Low-level benchmark for `VectorizedRleValuesReader`. Measures both RLE and PACKED decode
@@ -70,132 +62,19 @@ object VectorizedRleValuesReaderBenchmark extends BenchmarkBase {
private val NUM_ITERS = 5
private val BATCH_SIZE = 4096
- // --------------- Encoding / descriptor / updater helpers ---------------
-
- private def encodeRle(values: Array[Int], bitWidth: Int): Array[Byte] = {
- val enc = new RunLengthBitPackingHybridEncoder(
- bitWidth, 64 * 1024, 1024 * 1024, new HeapByteBufferAllocator)
- values.foreach(enc.writeInt)
- val out = new ByteArrayOutputStream()
- enc.toBytes.writeAllTo(out)
- out.toByteArray
- }
-
private def toInputStream(bytes: Array[Byte]): ByteBufferInputStream =
ByteBufferInputStream.wrap(ByteBuffer.wrap(bytes))
- private def intColumnDescriptor(maxDef: Int): ColumnDescriptor = {
- val rep = if (maxDef == 0) Repetition.REQUIRED else Repetition.OPTIONAL
- val prim = Types.primitive(PrimitiveTypeName.INT32, rep).named("col")
- new ColumnDescriptor(Array("col"), prim, 0, maxDef)
- }
-
- private val integerUpdater: ParquetVectorUpdater = new ParquetVectorUpdater {
- override def readValues(
- total: Int, offset: Int,
- values: WritableColumnVector, reader: VectorizedValuesReader): Unit =
- reader.readIntegers(total, values, offset)
-
- override def skipValues(total: Int, reader: VectorizedValuesReader): Unit =
- reader.skipIntegers(total)
-
- override def readValue(
- offset: Int, values: WritableColumnVector, reader: VectorizedValuesReader): Unit =
- values.putInt(offset, reader.readInteger())
-
- override def decodeSingleDictionaryId(
- offset: Int,
- values: WritableColumnVector,
- dictionaryIds: WritableColumnVector,
- dictionary: Dictionary): Unit =
- values.putInt(offset, dictionary.decodeToInt(dictionaryIds.getDictId(offset)))
- }
-
- // --------------- Reflective bridge to package-private ParquetReadState ---------------
- //
- // Under spark-submit --jars, test and main classes load from different classloaders, blocking
- // package-private access across runtime packages. Reflection with setAccessible sidesteps
- // the check without widening production visibility.
-
- private val stateCls = SparkClassUtils.classForName[Any](
- "org.apache.spark.sql.execution.datasources.parquet.ParquetReadState")
-
- private val stateCtor = {
- val c = stateCls.getDeclaredConstructor(
- classOf[ColumnDescriptor],
- java.lang.Boolean.TYPE,
- classOf[PrimitiveIterator.OfLong])
- c.setAccessible(true)
- c
- }
-
- private val resetForNewBatchMethod = {
- val m = stateCls.getDeclaredMethod("resetForNewBatch", Integer.TYPE)
- m.setAccessible(true)
- m
- }
-
- private val resetForNewPageMethod = {
- val m = stateCls.getDeclaredMethod(
- "resetForNewPage", Integer.TYPE, java.lang.Long.TYPE)
- m.setAccessible(true)
- m
- }
-
- private val readBatchMethod: Method =
- classOf[VectorizedRleValuesReader].getMethods
- .find(m =>
- m.getName == "readBatch"
- && m.getParameterCount == 5
- && m.getParameterTypes()(0) == stateCls)
- .getOrElse(throw new NoSuchMethodException(
- "VectorizedRleValuesReader.readBatch/5"))
+ // --------------- ReadState helpers (delegate to shared reflection bridge) ---------------
private def newReadState(maxDef: Int, valuesInPage: Int): AnyRef = {
- val state = try {
- stateCtor.newInstance(
- intColumnDescriptor(maxDef),
- Boolean.box(maxDef == 0),
- null).asInstanceOf[AnyRef]
- } catch { case e: ReflectiveOperationException => throw rethrow(e) }
- resetForNewBatch(state, BATCH_SIZE)
- resetForNewPage(state, valuesInPage, 0L)
+ val state = ParquetReadStateTestAccess.newState(
+ intColumnDescriptor(maxDef), maxDef == 0)
+ ParquetReadStateTestAccess.resetForNewBatch(state, BATCH_SIZE)
+ ParquetReadStateTestAccess.resetForNewPage(state, valuesInPage, 0L)
state
}
- private def resetForNewBatch(state: AnyRef, batchSize: Int): Unit =
- try { resetForNewBatchMethod.invoke(state, Int.box(batchSize)) }
- catch { case e: ReflectiveOperationException => throw rethrow(e) }
-
- private def resetForNewPage(
- state: AnyRef, total: Int, firstRow: Long): Unit =
- try {
- resetForNewPageMethod.invoke(state, Int.box(total), Long.box(firstRow))
- } catch { case e: ReflectiveOperationException => throw rethrow(e) }
-
- private def invokeReadBatch(
- reader: VectorizedRleValuesReader,
- state: AnyRef,
- values: WritableColumnVector,
- defLevels: WritableColumnVector,
- valueReader: VectorizedValuesReader): Unit =
- try {
- readBatchMethod.invoke(
- reader, state, values, defLevels, valueReader, integerUpdater)
- } catch { case e: ReflectiveOperationException => throw rethrow(e) }
-
- private def rethrow(e: ReflectiveOperationException): RuntimeException = {
- val cause = e match {
- case ite: InvocationTargetException => ite.getCause
- case other => other
- }
- cause match {
- case re: RuntimeException => throw re
- case er: Error => throw er
- case _ => throw new RuntimeException(cause)
- }
- }
-
// --------------- Data generation helpers ---------------
// Generates def-level arrays that force PACKED mode by ensuring no 8+ consecutive identical
@@ -379,7 +258,8 @@ object VectorizedRleValuesReaderBenchmark extends BenchmarkBase {
benchmark.addCase(
f"nullRatio=${nullRatio}%.1f, $clusterTag") { _ =>
reader.initFromPage(NUM_ROWS, toInputStream(bytes))
- resetForNewPage(state, NUM_ROWS, 0L)
+ ParquetReadStateTestAccess.resetForNewPage(
+ state, NUM_ROWS, 0L)
runBatches(reader, state, values, defLevelsVec, factory())
}
}
@@ -397,8 +277,9 @@ object VectorizedRleValuesReaderBenchmark extends BenchmarkBase {
var produced = 0
while (produced < NUM_ROWS) {
val toRead = math.min(BATCH_SIZE, NUM_ROWS - produced)
- resetForNewBatch(state, toRead)
- invokeReadBatch(reader, state, values, defLevelsVec, valueReader)
+ ParquetReadStateTestAccess.resetForNewBatch(state, toRead)
+ ParquetReadStateTestAccess.readBatch(
+ reader, state, values, defLevelsVec, valueReader, integerUpdater)
produced += toRead
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala
new file mode 100644
index 000000000000..ae3dba15e376
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderSuite.scala
@@ -0,0 +1,314 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet
+
+import java.nio.ByteBuffer
+import java.util.PrimitiveIterator
+
+import org.apache.parquet.bytes.ByteBufferInputStream
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.execution.datasources.parquet.VectorizedRleValuesReaderTestUtils._
+import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector, WritableColumnVector}
+import org.apache.spark.sql.types.IntegerType
+
+/**
+ * Focused correctness tests for `VectorizedRleValuesReader.readBatch` PACKED-mode decoding,
+ * covering patterns the P0 optimization cares about: run boundaries, batch boundaries, and
+ * nested def-level grouping. Uses the same reflection bridge as the benchmark.
+ */
+class VectorizedRleValuesReaderSuite extends SparkFunSuite {
+
+ import VectorizedRleValuesReaderSuite._
+
+ test("PACKED: alternating null/non-null (many single-element runs)") {
+ val n = 1024
+ val defLevels = Array.tabulate(n)(i => i & 1)
+ runAndAssert(defLevels, maxDef = 1, batchSize = n, withDefLevels = false)
+ }
+
+ test("PACKED: 4-element runs aligned to 8-group boundaries") {
+ val n = 1024
+ val defLevels = Array.tabulate(n)(i => if ((i / 4) % 2 == 0) 0 else 1)
+ runAndAssert(defLevels, maxDef = 1, batchSize = n, withDefLevels = false)
+ }
+
+ test("PACKED: long null and non-null runs within PACKED blocks") {
+ // 7-long null then 1 non-null then 7-long null ... forces PACKED (each 8-group is mixed),
+ // with asymmetric run lengths typical of realistic sparse data.
+ val pattern = Array.fill(7)(0) ++ Array(1)
+ val defLevels = Array.fill(128)(pattern).flatten.take(1024)
+ runAndAssert(defLevels, maxDef = 1, batchSize = 1024, withDefLevels = false)
+ }
+
+ test("PACKED: runs span batch boundaries (state carries across readBatch calls)") {
+ // 32-long null run starting at position 100 spans multiple 64-row batches.
+ val n = 512
+ val defLevels = Array.tabulate(n) { i =>
+ if (i >= 100 && i < 132) 0 // null run
+ else if ((i / 4) % 2 == 0) 0 // PACKED-forcing background pattern
+ else 1
+ }
+ runAndAssert(defLevels, maxDef = 1, batchSize = 64, withDefLevels = false)
+ }
+
+ test("PACKED with defLevels: nested column maxDef=3 with mixed def-level values") {
+ // Simulates a nested column where def-level values 0, 1, 2 all mean null (at different
+ // nesting levels) and 3 means non-null. Tests that readValuesN groups by exact def-level
+ // value so per-level null semantics are preserved.
+ val pattern = Array(0, 1, 2, 3, 0, 1, 2, 3, 1, 2, 0, 3)
+ val defLevels = Array.fill(64)(pattern).flatten.take(768)
+ runAndAssert(defLevels, maxDef = 3, batchSize = 256, withDefLevels = true)
+ }
+
+ test("PACKED: cross-batch continuity with defLevels") {
+ val n = 256
+ val defLevels = Array.tabulate(n)(i => if ((i / 3) % 2 == 0) 0 else 1)
+ runAndAssert(defLevels, maxDef = 1, batchSize = 64, withDefLevels = true)
+ }
+
+ test("RLE fast path: single long run, no nulls") {
+ val defLevels = Array.fill(1024)(1)
+ runAndAssert(defLevels, maxDef = 1, batchSize = 1024, withDefLevels = false)
+ }
+
+ test("RLE fast path: single long run, all nulls") {
+ val defLevels = Array.fill(1024)(0)
+ runAndAssert(defLevels, maxDef = 1, batchSize = 1024, withDefLevels = false)
+ }
+
+ test("PACKED group larger than initial 16-int currentBuffer (triggers buffer grow)") {
+ // ~1024 alternating values produce one PACKED block well beyond the initial 16-int buffer,
+ // exercising `new int[currentCount]` in readNextGroup.
+ val defLevels = Array.tabulate(1024)(i => i & 1)
+ runAndAssert(defLevels, maxDef = 1, batchSize = 1024, withDefLevels = false)
+ }
+
+ test("RLE + PACKED mixed in a single page") {
+ val defLevels =
+ Array.fill(200)(1) ++ Array.tabulate(256)(i => i & 1) ++ Array.fill(200)(0)
+ runAndAssert(defLevels, maxDef = 1, batchSize = 256, withDefLevels = true)
+ }
+
+ test("required column (maxDef=0): single implicit RLE run") {
+ val defLevels = Array.fill(64)(0)
+ runAndAssert(defLevels, maxDef = 0, batchSize = 64, withDefLevels = false)
+ }
+
+ test("PACKED: row-index filtering with contiguous included range") {
+ val n = 256
+ val defLevels = Array.tabulate(n)(i => i & 1)
+ runAndAssertFiltered(defLevels, maxDef = 1, includedPositions = (50 to 200).toArray)
+ }
+
+ test("PACKED: row-index filtering with multiple disjoint ranges") {
+ val n = 256
+ val defLevels = Array.tabulate(n)(i => if ((i / 3) % 2 == 0) 0 else 1)
+ val included = ((10 to 30) ++ (80 to 120) ++ (200 to 240)).toArray
+ runAndAssertFiltered(defLevels, maxDef = 1, includedPositions = included)
+ }
+
+ test("multi-page: reader reinitialized between pages, state carried via resetForNewPage") {
+ val page1 = Array.tabulate(128)(i => i & 1)
+ val page2 = Array.fill(64)(1) ++ Array.tabulate(64)(i => i & 1)
+ runAndAssertMultiPage(Seq(page1, page2), maxDef = 1, batchSize = 64)
+ }
+}
+
+private object VectorizedRleValuesReaderSuite {
+
+ /**
+ * Runs readBatch end-to-end and asserts null-bits, non-null values, and def levels.
+ * Each batch uses a fresh output vector since `state.valueOffset` resets to 0 per batch —
+ * mirrors production where `VectorizedColumnReader` hands in a batch-sized vector.
+ */
+ // Non-trivial value formula: off-by-one mismatches won't coincidentally align.
+ private def valueAt(idx: Int): Int = idx * 100 + 7
+
+ private def runAndAssert(
+ defLevels: Array[Int],
+ maxDef: Int,
+ batchSize: Int,
+ withDefLevels: Boolean): Unit = {
+ val n = defLevels.length
+ val bitWidth = if (maxDef == 0) 0 else 32 - Integer.numberOfLeadingZeros(maxDef)
+ // When bitWidth == 0 (required column), the reader treats the page as an implicit RLE run
+ // of zeros and never consumes bytes; the encoded array is a placeholder.
+ val encoded = if (bitWidth == 0) Array.emptyByteArray else encodeRle(defLevels, bitWidth)
+ val nonNullCount = defLevels.count(_ == maxDef)
+ val plainBytes = plainIntBytes(nonNullCount)(valueAt)
+
+ val reader = new VectorizedRleValuesReader(bitWidth, false)
+ reader.initFromPage(n, ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded)))
+ val valueReader = new VectorizedPlainValuesReader
+ valueReader.initFromPage(
+ nonNullCount, ByteBufferInputStream.wrap(ByteBuffer.wrap(plainBytes)))
+ val state = ParquetReadStateTestAccess.newState(intColumnDescriptor(maxDef), maxDef == 0)
+ ParquetReadStateTestAccess.resetForNewPage(state, n, 0L)
+
+ var produced = 0
+ var expectedValueIdx = 0
+ while (produced < n) {
+ val toRead = math.min(batchSize, n - produced)
+ val values = new OnHeapColumnVector(toRead, IntegerType)
+ val defLevelsVec = new OnHeapColumnVector(toRead, IntegerType)
+ ParquetReadStateTestAccess.resetForNewBatch(state, toRead)
+ val defLevelsArg: WritableColumnVector = if (withDefLevels) defLevelsVec else null
+ ParquetReadStateTestAccess.readBatch(
+ reader, state, values, defLevelsArg, valueReader, integerUpdater)
+
+ var expectedNullsInBatch = 0
+ var i = 0
+ while (i < toRead) {
+ val absPos = produced + i
+ if (defLevels(absPos) == maxDef) {
+ assert(!values.isNullAt(i), s"pos $absPos should be non-null")
+ val expected = valueAt(expectedValueIdx)
+ assert(
+ values.getInt(i) == expected,
+ s"pos $absPos value mismatch: got ${values.getInt(i)}, expected $expected")
+ expectedValueIdx += 1
+ } else {
+ assert(values.isNullAt(i), s"pos $absPos should be null")
+ expectedNullsInBatch += 1
+ }
+ if (withDefLevels) {
+ assert(
+ defLevelsVec.getInt(i) == defLevels(absPos),
+ s"defLevel at pos $absPos: got ${defLevelsVec.getInt(i)}, " +
+ s"expected ${defLevels(absPos)}")
+ }
+ i += 1
+ }
+ assert(
+ values.numNulls() == expectedNullsInBatch,
+ s"batch starting at $produced: numNulls ${values.numNulls()}, " +
+ s"expected $expectedNullsInBatch")
+ produced += toRead
+ }
+ }
+
+ /**
+ * Variant of `runAndAssert` that passes a `rowIndexes` iterator so the reader only emits
+ * rows at the listed positions. Verifies that skipped value positions advance the value
+ * reader correctly and that included rows map to the expected values in order.
+ */
+ private def runAndAssertFiltered(
+ defLevels: Array[Int],
+ maxDef: Int,
+ includedPositions: Array[Int]): Unit = {
+ val n = defLevels.length
+ val bitWidth = if (maxDef == 0) 0 else 32 - Integer.numberOfLeadingZeros(maxDef)
+ val encoded = if (bitWidth == 0) Array.emptyByteArray else encodeRle(defLevels, bitWidth)
+ val nonNullCount = defLevels.count(_ == maxDef)
+ val plainBytes = plainIntBytes(nonNullCount)(valueAt)
+
+ val reader = new VectorizedRleValuesReader(bitWidth, false)
+ reader.initFromPage(n, ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded)))
+ val valueReader = new VectorizedPlainValuesReader
+ valueReader.initFromPage(
+ nonNullCount, ByteBufferInputStream.wrap(ByteBuffer.wrap(plainBytes)))
+ val state = ParquetReadStateTestAccess.newState(
+ intColumnDescriptor(maxDef), maxDef == 0, longIterator(includedPositions))
+ ParquetReadStateTestAccess.resetForNewPage(state, n, 0L)
+
+ val size = includedPositions.length
+ val values = new OnHeapColumnVector(size, IntegerType)
+ ParquetReadStateTestAccess.resetForNewBatch(state, size)
+ ParquetReadStateTestAccess.readBatch(reader, state, values, null, valueReader, integerUpdater)
+
+ val prefixNonNulls = defLevels.scanLeft(0) { (c, d) =>
+ c + (if (d == maxDef) 1 else 0)
+ }
+ var j = 0
+ while (j < size) {
+ val p = includedPositions(j)
+ if (defLevels(p) == maxDef) {
+ assert(!values.isNullAt(j), s"included pos $p (output $j) should be non-null")
+ val expected = valueAt(prefixNonNulls(p))
+ assert(
+ values.getInt(j) == expected,
+ s"included pos $p (output $j): got ${values.getInt(j)}, expected $expected")
+ } else {
+ assert(values.isNullAt(j), s"included pos $p (output $j) should be null")
+ }
+ j += 1
+ }
+ }
+
+ /**
+ * Simulates a column chunk with multiple pages: the same reader instance is reused, pointing
+ * to fresh encoded bytes per page and with `resetForNewPage` called between pages.
+ */
+ private def runAndAssertMultiPage(
+ pages: Seq[Array[Int]],
+ maxDef: Int,
+ batchSize: Int): Unit = {
+ val bitWidth = if (maxDef == 0) 0 else 32 - Integer.numberOfLeadingZeros(maxDef)
+ val reader = new VectorizedRleValuesReader(bitWidth, false)
+ val state =
+ ParquetReadStateTestAccess.newState(intColumnDescriptor(maxDef), maxDef == 0)
+
+ var pageFirstRow = 0L
+ pages.foreach { pageDefLevels =>
+ val pageN = pageDefLevels.length
+ val encoded = if (bitWidth == 0) Array.emptyByteArray else encodeRle(pageDefLevels, bitWidth)
+ val nonNullCount = pageDefLevels.count(_ == maxDef)
+ val plainBytes = plainIntBytes(nonNullCount)(valueAt)
+
+ reader.initFromPage(pageN, ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded)))
+ val valueReader = new VectorizedPlainValuesReader
+ valueReader.initFromPage(
+ nonNullCount, ByteBufferInputStream.wrap(ByteBuffer.wrap(plainBytes)))
+ ParquetReadStateTestAccess.resetForNewPage(state, pageN, pageFirstRow)
+
+ var produced = 0
+ var expectedValueIdx = 0
+ while (produced < pageN) {
+ val toRead = math.min(batchSize, pageN - produced)
+ val values = new OnHeapColumnVector(toRead, IntegerType)
+ ParquetReadStateTestAccess.resetForNewBatch(state, toRead)
+ ParquetReadStateTestAccess.readBatch(
+ reader, state, values, null, valueReader, integerUpdater)
+
+ var i = 0
+ while (i < toRead) {
+ val absPos = produced + i
+ if (pageDefLevels(absPos) == maxDef) {
+ assert(!values.isNullAt(i), s"page@$pageFirstRow pos $absPos should be non-null")
+ val expected = valueAt(expectedValueIdx)
+ assert(values.getInt(i) == expected)
+ expectedValueIdx += 1
+ } else {
+ assert(values.isNullAt(i), s"page@$pageFirstRow pos $absPos should be null")
+ }
+ i += 1
+ }
+ produced += toRead
+ }
+ pageFirstRow += pageN
+ }
+ }
+
+ private def longIterator(values: Array[Int]): PrimitiveIterator.OfLong =
+ new PrimitiveIterator.OfLong {
+ private var idx = 0
+ override def hasNext: Boolean = idx < values.length
+ override def nextLong(): Long = { val v = values(idx).toLong; idx += 1; v }
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderTestUtils.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderTestUtils.scala
new file mode 100644
index 000000000000..37674eb91249
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReaderTestUtils.scala
@@ -0,0 +1,87 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet
+
+import java.io.ByteArrayOutputStream
+import java.nio.{ByteBuffer, ByteOrder}
+
+import org.apache.parquet.bytes.HeapByteBufferAllocator
+import org.apache.parquet.column.{ColumnDescriptor, Dictionary}
+import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+import org.apache.parquet.schema.Type.Repetition
+import org.apache.parquet.schema.Types
+
+import org.apache.spark.sql.execution.vectorized.WritableColumnVector
+
+/**
+ * Shared helpers for tests and benchmarks that exercise
+ * `VectorizedRleValuesReader` directly.
+ */
+object VectorizedRleValuesReaderTestUtils {
+
+ /** Encodes values via parquet-mr's RLE/bit-packing hybrid encoder (no length prefix). */
+ def encodeRle(values: Array[Int], bitWidth: Int): Array[Byte] = {
+ val enc = new RunLengthBitPackingHybridEncoder(
+ bitWidth, 64 * 1024, 1024 * 1024, new HeapByteBufferAllocator)
+ values.foreach(enc.writeInt)
+ val out = new ByteArrayOutputStream()
+ enc.toBytes.writeAllTo(out)
+ out.toByteArray
+ }
+
+ /** Little-endian plain INT32 bytes. Layout matches `VectorizedPlainValuesReader.readIntegers`. */
+ def plainIntBytes(count: Int)(f: Int => Int): Array[Byte] = {
+ val buf = ByteBuffer.allocate(count * 4).order(ByteOrder.LITTLE_ENDIAN)
+ var i = 0
+ while (i < count) {
+ buf.putInt(f(i))
+ i += 1
+ }
+ buf.array()
+ }
+
+ /** Single-column INT32 descriptor with the given `maxDefinitionLevel`. */
+ def intColumnDescriptor(maxDef: Int): ColumnDescriptor = {
+ val rep = if (maxDef == 0) Repetition.REQUIRED else Repetition.OPTIONAL
+ val prim = Types.primitive(PrimitiveTypeName.INT32, rep).named("col")
+ new ColumnDescriptor(Array("col"), prim, 0, maxDef)
+ }
+
+ /** Stateless INT32 updater equivalent to the package-private `IntegerUpdater`. */
+ val integerUpdater: ParquetVectorUpdater = new ParquetVectorUpdater {
+ override def readValues(
+ total: Int, offset: Int,
+ values: WritableColumnVector, reader: VectorizedValuesReader): Unit =
+ reader.readIntegers(total, values, offset)
+
+ override def skipValues(total: Int, reader: VectorizedValuesReader): Unit =
+ reader.skipIntegers(total)
+
+ override def readValue(
+ offset: Int, values: WritableColumnVector, reader: VectorizedValuesReader): Unit =
+ values.putInt(offset, reader.readInteger())
+
+ override def decodeSingleDictionaryId(
+ offset: Int,
+ values: WritableColumnVector,
+ dictionaryIds: WritableColumnVector,
+ dictionary: Dictionary): Unit =
+ values.putInt(offset, dictionary.decodeToInt(dictionaryIds.getDictId(offset)))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org