This is an automated email from the ASF dual-hosted git repository.
LuciferYang 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 03750ab553d9 [SPARK-56892][SQL] Bulk read optimization for Parquet
DELTA_BINARY_PACKED decoding
03750ab553d9 is described below
commit 03750ab553d9ebf446e61e077e867d45436f72b0
Author: Ismaël Mejía <[email protected]>
AuthorDate: Tue Jun 16 10:48:23 2026 +0800
[SPARK-56892][SQL] Bulk read optimization for Parquet DELTA_BINARY_PACKED
decoding
### What changes were proposed in this pull request?
Replace per-element lambda dispatch in `readIntegers`/`readLongs` with bulk
paths that compute prefix sums in-place over the unpacked delta buffer and
write via `putInts`/`putLongs` (backed by `System.arraycopy` on-heap).
Three optimizations in this PR:
1. **Bulk read for INT32/INT64**: `readBulkIntegers` and `readBulkLongs`
replace the generic `readValues()` lambda-per-value path. A single
`loadMiniBlockBulk` method handles block/mini-block loading, prefix-sum
computation, and delegates the type-specific write to a `BulkWriter` callback
(called once per mini-block, not per value).
2. **Zero-allocation unsigned long encoding**: Replace `new
BigInteger(Long.toUnsignedString(v)).toByteArray()` (3 allocations per value:
String + BigInteger + byte[]) with a loop-based `byte[]` encoder using
`Long.numberOfLeadingZeros` to compute minimal BigInteger-compatible encoding
directly. The shared utility `encodeUnsignedLongBigEndian` is extracted into
`VectorizedReaderBase` and applied to all call sites
(`VectorizedDeltaBinaryPackedReader`, `UnsignedLongUpdater`, `ParquetDic [...]
3. **Widening overrides** (`readIntegersAsLongs`, `readIntegersAsDoubles`):
Since the delta decoder already works on `long[]` internally, these overrides
skip the int narrowing step and write longs/doubles directly from the
prefix-sum buffer. Benefits `DateToTimestampNTZUpdater`,
`IntegerToLongUpdater`, and `IntegerToDoubleUpdater` via the two-pass updater
pattern.
4. **Benchmark fix**: Add `unsignedLongVec.reset()` before
`readUnsignedLongs` to prevent unbounded `arrayData()` growth across benchmark
iterations (OOM).
### Why are the changes needed?
The DELTA_BINARY_PACKED decoder was 2-5x slower than PLAIN encoding for
INT32/INT64 reads due to per-element lambda dispatch and lack of bulk vector
writes. The `readUnsignedLongs` path allocated 3 objects per value (12,288
allocations per 4096-row batch) due to `BigInteger(Long.toUnsignedString(v))`.
#### Benchmark Results (AMD EPYC 7763, JDK 17/21/25)
Results from GHA benchmark workflow runs committed to this branch. Both
baseline (upstream master) and this PR ran on AMD EPYC 7763 64-Core Processor.
**DELTA_BINARY_PACKED INT64** (primary optimization target):
| Case | JDK 17 Baseline | JDK 17 PR | Speedup | JDK 21 Baseline | JDK 21
PR | Speedup | JDK 25 Baseline | JDK 25 PR | Speedup |
|---|---|---|---|---|---|---|---|---|---|
| readLongs, constant | 191.0 | 535.8 | **2.8x** | 195.2 | 477.9 | **2.4x**
| 163.0 | 508.3 | **3.1x** |
| readLongs, monotonic | 144.9 | 534.1 | **3.7x** | 158.8 | 478.2 |
**3.0x** | 139.1 | 520.9 | **3.7x** |
| readLongs, small-delta random | 134.8 | 368.4 | **2.7x** | 139.4 | 334.4
| **2.4x** | 122.7 | 356.6 | **2.9x** |
| readLongs, wide random | 97.8 | 187.7 | **1.9x** | 102.9 | 182.3 |
**1.8x** | 94.4 | 189.4 | **2.0x** |
| skipLongs, constant | 170.1 | 612.6 | **3.6x** | 175.4 | 520.5 | **3.0x**
| 152.1 | 603.1 | **4.0x** |
| skipLongs, monotonic | 170.1 | 611.2 | **3.6x** | 175.6 | 526.5 |
**3.0x** | 152.0 | 603.7 | **4.0x** |
| skipLongs, small-delta random | 152.8 | 427.1 | **2.8x** | 152.5 | 356.2
| **2.3x** | 133.9 | 397.2 | **3.0x** |
**DELTA_BINARY_PACKED INT32:**
| Case | JDK 17 Baseline | JDK 17 PR | Speedup | JDK 21 Baseline | JDK 21
PR | Speedup | JDK 25 Baseline | JDK 25 PR | Speedup |
|---|---|---|---|---|---|---|---|---|---|
| readIntegers, constant | 482.3 | 375.1 | 0.78x | 451.8 | 482.9 | 1.07x |
487.1 | 476.4 | 0.98x |
| readIntegers, monotonic | 314.9 | 375.1 | **1.19x** | 369.7 | 478.9 |
**1.30x** | 303.1 | 476.4 | **1.57x** |
| readIntegers, small-delta random | 257.5 | 297.3 | **1.15x** | 308.6 |
333.9 | 1.08x | 241.3 | 333.6 | **1.38x** |
| readIntegers, wide random | 209.3 | 240.8 | **1.15x** | 249.2 | 270.9 |
1.09x | 201.4 | 267.5 | **1.33x** |
| skipIntegers, constant | 335.4 | 608.9 | **1.82x** | 415.1 | 519.8 |
**1.25x** | 364.8 | 606.4 | **1.66x** |
| skipIntegers, monotonic | 335.1 | 612.2 | **1.83x** | 415.6 | 529.0 |
**1.27x** | 365.3 | 603.8 | **1.65x** |
**DELTA_BYTE_ARRAY** (indirect benefit from INT64 bulk path):
| Case | JDK 17 Baseline | JDK 17 PR | Speedup | JDK 21 Baseline | JDK 21
PR | Speedup | JDK 25 Baseline | JDK 25 PR | Speedup |
|---|---|---|---|---|---|---|---|---|---|
| readBinary, no overlap, len=16 | 29.2 | 38.5 | **1.32x** | 29.6 | 38.7 |
**1.31x** | 28.2 | 38.5 | **1.37x** |
| readBinary, half overlap, len=16 | 25.0 | 31.0 | **1.24x** | 25.4 | 31.2
| **1.23x** | 24.3 | 31.7 | **1.30x** |
| readBinary, full overlap, len=16 | 24.8 | 30.2 | **1.22x** | 25.1 | 30.9
| **1.23x** | 24.0 | 31.3 | **1.30x** |
**DELTA_LENGTH_BYTE_ARRAY:**
| Case | JDK 17 Baseline | JDK 17 PR | Speedup | JDK 21 Baseline | JDK 21
PR | Speedup | JDK 25 Baseline | JDK 25 PR | Speedup |
|---|---|---|---|---|---|---|---|---|---|
| readBinary, payloadLen=8 | 48.1 | 63.3 | **1.32x** | 51.7 | 65.7 |
**1.27x** | 48.7 | 63.8 | **1.31x** |
| skipBinary, payloadLen=8 | 95.8 | 174.7 | **1.82x** | 106.3 | 182.0 |
**1.71x** | 97.0 | 183.2 | **1.89x** |
**Variant reads (unsigned long encoding + widening overrides):**
| Case | JDK 17 Baseline | JDK 17 PR | Speedup | JDK 21 Baseline | JDK 21
PR | Speedup | JDK 25 Baseline | JDK 25 PR | Speedup |
|---|---|---|---|---|---|---|---|---|---|
| readIntegersAsLongs (INT32 -> Long) | 121.4 | 291.1 | **2.4x** | 117.6 |
286.8 | **2.4x** | 105.8 | 283.9 | **2.7x** |
| readIntegersAsDoubles (INT32 -> Double) | 121.4 | 258.8 | **2.1x** |
117.6 | 258.1 | **2.2x** | 105.8 | 253.0 | **2.4x** |
| readUnsignedLongs (INT64 -> Decimal(20,0)) | 4.6 | 37.6 | **8.2x** | 4.5
| 36.9 | **8.2x** | 5.0 | 36.8 | **7.4x** |
> Note: Baseline for `readIntegersAsLongs`/`readIntegersAsDoubles` is
`readUnsignedIntegers` which uses the per-row default `readInteger()` virtual
dispatch path (same cost as the default widening methods without overrides).
> Note: These results are from independent GHA runs (not back-to-back on
the same runner). Some cross-run variance is present in unchanged paths. The
INT64 bulk-read improvement (1.8x-3.7x), widening overrides (2.1x-2.7x), and
the unsigned long encoding fix (7-9x) are the primary contributions of this PR.
Full committed results: [JDK
17](https://github.com/iemejia/spark/actions/runs/27426991381), [JDK
21](https://github.com/iemejia/spark/actions/runs/27426992516), [JDK
25](https://github.com/iemejia/spark/actions/runs/27431820511)
### Does this PR introduce _any_ user-facing change?
No. This is a performance improvement to internal Parquet decoding. No API
or behavior changes.
### How was this patch tested?
- Existing unit tests: `ParquetDeltaEncodingInteger` (13 tests),
`ParquetDeltaEncodingLong` (13 tests), `ParquetDeltaByteArrayEncodingSuite`,
`ParquetDeltaLengthByteArrayEncodingSuite`, `ParquetVectorizedSuite` (25
tests), `ParquetIOSuite` (unsigned Parquet logical types test) -- all pass.
- Benchmark: `VectorizedDeltaReaderBenchmark` run via GHA workflow on JDK
17, 21, and 25.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenCode (Claude claude-opus-4.6)
Closes #55919 from iemejia/SPARK-56892-delta-binary-packed-bulk-read.
Authored-by: Ismaël Mejía <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
...ectorizedDeltaReaderBenchmark-jdk21-results.txt | 98 ++++++-------
...ectorizedDeltaReaderBenchmark-jdk25-results.txt | 98 ++++++-------
.../VectorizedDeltaReaderBenchmark-results.txt | 98 ++++++-------
.../datasources/parquet/ParquetDictionary.java | 4 +-
.../parquet/ParquetVectorUpdaterFactory.java | 11 +-
.../parquet/VectorizedDeltaBinaryPackedReader.java | 151 ++++++++++++++++++++-
.../datasources/parquet/VectorizedReaderBase.java | 38 ++++++
.../parquet/VectorizedDeltaReaderBenchmark.scala | 10 +-
8 files changed, 351 insertions(+), 157 deletions(-)
diff --git
a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
index a7199750be60..9f0397dd4d8b 100644
--- a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
+++ b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
@@ -2,92 +2,94 @@
DELTA_BINARY_PACKED INT32
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BINARY_PACKED INT32: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readIntegers, constant 2 2
0 451.8 2.2 1.0X
-skipIntegers, constant 3 3
0 415.1 2.4 0.9X
-readIntegers, monotonic 3 3
0 369.7 2.7 0.8X
-skipIntegers, monotonic 3 3
0 415.6 2.4 0.9X
-readIntegers, small-delta random 3 3
0 308.6 3.2 0.7X
-skipIntegers, small-delta random 3 3
0 358.6 2.8 0.8X
-readIntegers, wide random 4 4
0 249.2 4.0 0.6X
-skipIntegers, wide random 4 4
0 281.4 3.6 0.6X
+readIntegers, constant 2 2
0 476.2 2.1 1.0X
+skipIntegers, constant 2 2
0 561.5 1.8 1.2X
+readIntegers, monotonic 2 2
0 480.9 2.1 1.0X
+skipIntegers, monotonic 2 2
0 562.4 1.8 1.2X
+readIntegers, small-delta random 3 3
0 330.9 3.0 0.7X
+skipIntegers, small-delta random 3 3
0 376.2 2.7 0.8X
+readIntegers, wide random 4 4
0 268.4 3.7 0.6X
+skipIntegers, wide random 4 4
0 293.1 3.4 0.6X
================================================================================================
DELTA_BINARY_PACKED INT64
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BINARY_PACKED INT64: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readLongs, constant 5 6
0 195.2 5.1 1.0X
-skipLongs, constant 6 6
0 175.4 5.7 0.9X
-readLongs, monotonic 7 7
0 158.8 6.3 0.8X
-skipLongs, monotonic 6 6
0 175.6 5.7 0.9X
-readLongs, small-delta random 8 8
0 139.4 7.2 0.7X
-skipLongs, small-delta random 7 7
0 152.5 6.6 0.8X
-readLongs, wide random 10 10
1 102.9 9.7 0.5X
-skipLongs, wide random 6 9
2 185.5 5.4 1.0X
+readLongs, constant 2 2
0 518.5 1.9 1.0X
+skipLongs, constant 2 2
0 545.8 1.8 1.1X
+readLongs, monotonic 2 2
0 517.6 1.9 1.0X
+skipLongs, monotonic 2 2
0 559.7 1.8 1.1X
+readLongs, small-delta random 3 3
0 357.3 2.8 0.7X
+skipLongs, small-delta random 3 3
0 377.0 2.7 0.7X
+readLongs, wide random 6 6
0 186.1 5.4 0.4X
+skipLongs, wide random 5 6
0 190.7 5.2 0.4X
================================================================================================
DELTA_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BYTE_ARRAY: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readBinary, no overlap, len=16 35 41
4 29.6 33.8 1.0X
-skipBinary, no overlap, len=16 41 45
3 25.9 38.6 0.9X
-readBinary, half overlap, len=16 41 44
3 25.4 39.3 0.9X
-skipBinary, half overlap, len=16 47 50
5 22.3 44.8 0.8X
-readBinary, full overlap, len=16 42 44
3 25.1 39.9 0.8X
-skipBinary, full overlap, len=16 48 50
3 22.0 45.6 0.7X
-readBinary, half overlap, len=64 42 42
1 25.2 39.6 0.9X
-skipBinary, half overlap, len=64 47 50
14 22.4 44.6 0.8X
+readBinary, no overlap, len=16 27 32
2 38.2 26.2 1.0X
+skipBinary, no overlap, len=16 34 36
2 30.8 32.5 0.8X
+readBinary, half overlap, len=16 33 34
0 31.5 31.7 0.8X
+skipBinary, half overlap, len=16 38 39
1 27.7 36.1 0.7X
+readBinary, full overlap, len=16 33 34
1 31.5 31.7 0.8X
+skipBinary, full overlap, len=16 38 39
2 27.4 36.5 0.7X
+readBinary, half overlap, len=64 33 34
1 31.6 31.7 0.8X
+skipBinary, half overlap, len=64 38 39
2 27.9 35.8 0.7X
================================================================================================
DELTA_LENGTH_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_LENGTH_BYTE_ARRAY: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readBinary, payloadLen=8 20 21
0 51.7 19.3 1.0X
-skipBinary, payloadLen=8 10 10
0 106.3 9.4 2.1X
-readBinary, payloadLen=32 16 18
2 63.9 15.7 1.2X
-skipBinary, payloadLen=32 6 6
0 175.9 5.7 3.4X
-readBinary, payloadLen=128 19 19
0 56.0 17.8 1.1X
-skipBinary, payloadLen=128 6 6
0 176.3 5.7 3.4X
-readBinary, payloadLen=512 41 43
3 25.6 39.0 0.5X
-skipBinary, payloadLen=512 6 6
1 176.4 5.7 3.4X
+readBinary, payloadLen=8 16 16
0 65.7 15.2 1.0X
+skipBinary, payloadLen=8 6 6
0 184.2 5.4 2.8X
+readBinary, payloadLen=32 16 16
1 65.0 15.4 1.0X
+skipBinary, payloadLen=32 6 6
0 185.6 5.4 2.8X
+readBinary, payloadLen=128 18 19
1 56.9 17.6 0.9X
+skipBinary, payloadLen=128 6 6
0 185.6 5.4 2.8X
+readBinary, payloadLen=512 39 40
1 27.1 36.9 0.4X
+skipBinary, payloadLen=512 6 6
0 185.5 5.4 2.8X
================================================================================================
Variant reads
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Variant reads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------------
-readBytes (INT32) 7 7
0 142.3 7.0 1.0X
-readShorts (INT32) 7 8
0 143.0 7.0 1.0X
-readUnsignedIntegers (INT32 -> Long) 7 7
0 147.1 6.8 1.0X
-readUnsignedLongs (INT64 -> Decimal(20,0)) 232 243
23 4.5 221.5 0.0X
-skipBytes 4 4
0 285.4 3.5 2.0X
-skipShorts 4 4
0 285.4 3.5 2.0X
-readByte (INT32 single-value) 13 13
1 80.2 12.5 0.6X
-readShort (INT32 single-value) 13 13
1 82.0 12.2 0.6X
-readInteger (INT32 single-value) 13 13
0 80.2 12.5 0.6X
-readLong (INT64 single-value) 14 14
0 74.9 13.3 0.5X
-readBinary(len) (DELTA_BYTE_ARRAY single-value) 64 67
3 16.5 60.6 0.1X
+readBytes (INT32) 4 4
0 236.3 4.2 1.0X
+readShorts (INT32) 9 9
2 117.6 8.5 0.5X
+readUnsignedIntegers (INT32 -> Long) 9 9
1 117.6 8.5 0.5X
+readIntegersAsLongs (INT32 -> Long) 4 4
0 286.8 3.5 1.2X
+readIntegersAsDoubles (INT32 -> Double) 4 4
0 258.1 3.9 1.1X
+readUnsignedLongs (INT64 -> Decimal(20,0)) 28 29
0 36.9 27.1 0.2X
+skipBytes 6 6
0 175.1 5.7 0.7X
+skipShorts 6 6
1 175.4 5.7 0.7X
+readByte (INT32 single-value) 14 14
1 76.5 13.1 0.3X
+readShort (INT32 single-value) 14 14
0 76.6 13.1 0.3X
+readInteger (INT32 single-value) 14 14
0 76.6 13.1 0.3X
+readLong (INT64 single-value) 16 16
0 67.3 14.9 0.3X
+readBinary(len) (DELTA_BYTE_ARRAY single-value) 63 65
2 16.6 60.2 0.1X
diff --git
a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
index fe7b64c3834b..ef6274f71249 100644
--- a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
+++ b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
@@ -2,92 +2,94 @@
DELTA_BINARY_PACKED INT32
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BINARY_PACKED INT32: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readIntegers, constant 2 2
0 487.1 2.1 1.0X
-skipIntegers, constant 3 3
0 364.8 2.7 0.7X
-readIntegers, monotonic 3 4
0 303.1 3.3 0.6X
-skipIntegers, monotonic 3 3
0 365.3 2.7 0.7X
-readIntegers, small-delta random 4 4
0 241.3 4.1 0.5X
-skipIntegers, small-delta random 4 4
0 278.7 3.6 0.6X
-readIntegers, wide random 5 5
0 201.4 5.0 0.4X
-skipIntegers, wide random 5 5
0 225.6 4.4 0.5X
+readIntegers, constant 2 2
0 473.1 2.1 1.0X
+skipIntegers, constant 2 2
0 604.4 1.7 1.3X
+readIntegers, monotonic 2 2
0 473.5 2.1 1.0X
+skipIntegers, monotonic 2 2
0 607.2 1.6 1.3X
+readIntegers, small-delta random 3 3
0 333.4 3.0 0.7X
+skipIntegers, small-delta random 3 3
0 391.2 2.6 0.8X
+readIntegers, wide random 4 4
0 271.5 3.7 0.6X
+skipIntegers, wide random 4 4
0 294.7 3.4 0.6X
================================================================================================
DELTA_BINARY_PACKED INT64
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BINARY_PACKED INT64: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readLongs, constant 6 7
1 163.0 6.1 1.0X
-skipLongs, constant 7 7
0 152.1 6.6 0.9X
-readLongs, monotonic 8 8
1 139.1 7.2 0.9X
-skipLongs, monotonic 7 7
0 152.0 6.6 0.9X
-readLongs, small-delta random 9 9
0 122.7 8.2 0.8X
-skipLongs, small-delta random 8 8
0 133.9 7.5 0.8X
-readLongs, wide random 11 11
0 94.4 10.6 0.6X
-skipLongs, wide random 10 10
0 100.9 9.9 0.6X
+readLongs, constant 2 2
0 497.0 2.0 1.0X
+skipLongs, constant 2 2
0 583.4 1.7 1.2X
+readLongs, monotonic 2 2
0 496.7 2.0 1.0X
+skipLongs, monotonic 2 2
0 588.6 1.7 1.2X
+readLongs, small-delta random 3 3
0 342.8 2.9 0.7X
+skipLongs, small-delta random 3 3
0 392.5 2.5 0.8X
+readLongs, wide random 6 6
0 189.2 5.3 0.4X
+skipLongs, wide random 5 5
0 197.4 5.1 0.4X
================================================================================================
DELTA_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BYTE_ARRAY: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readBinary, no overlap, len=16 37 38
2 28.2 35.5 1.0X
-skipBinary, no overlap, len=16 42 43
2 25.2 39.7 0.9X
-readBinary, half overlap, len=16 43 44
2 24.3 41.2 0.9X
-skipBinary, half overlap, len=16 48 49
2 21.8 45.9 0.8X
-readBinary, full overlap, len=16 44 47
4 24.0 41.7 0.9X
-skipBinary, full overlap, len=16 48 50
2 21.6 46.3 0.8X
-readBinary, half overlap, len=64 44 45
2 23.8 42.0 0.8X
-skipBinary, half overlap, len=64 49 53
4 21.6 46.3 0.8X
+readBinary, no overlap, len=16 27 31
3 38.3 26.1 1.0X
+skipBinary, no overlap, len=16 32 32
0 33.2 30.2 0.9X
+readBinary, half overlap, len=16 33 34
1 31.5 31.7 0.8X
+skipBinary, half overlap, len=16 38 39
1 27.6 36.2 0.7X
+readBinary, full overlap, len=16 34 34
1 31.2 32.1 0.8X
+skipBinary, full overlap, len=16 38 39
1 27.3 36.7 0.7X
+readBinary, half overlap, len=64 34 35
1 30.5 32.7 0.8X
+skipBinary, half overlap, len=64 38 39
1 27.6 36.2 0.7X
================================================================================================
DELTA_LENGTH_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_LENGTH_BYTE_ARRAY: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readBinary, payloadLen=8 22 22
0 48.7 20.5 1.0X
-skipBinary, payloadLen=8 11 11
0 97.0 10.3 2.0X
-readBinary, payloadLen=32 21 22
0 48.9 20.5 1.0X
-skipBinary, payloadLen=32 11 11
0 97.0 10.3 2.0X
-readBinary, payloadLen=128 24 24
1 43.7 22.9 0.9X
-skipBinary, payloadLen=128 11 11
1 97.1 10.3 2.0X
-readBinary, payloadLen=512 51 53
1 20.5 48.7 0.4X
-skipBinary, payloadLen=512 11 11
0 97.1 10.3 2.0X
+readBinary, payloadLen=8 16 17
1 63.7 15.7 1.0X
+skipBinary, payloadLen=8 6 6
0 182.4 5.5 2.9X
+readBinary, payloadLen=32 16 17
1 64.5 15.5 1.0X
+skipBinary, payloadLen=32 6 6
0 182.7 5.5 2.9X
+readBinary, payloadLen=128 19 19
1 55.7 18.0 0.9X
+skipBinary, payloadLen=128 6 6
0 182.4 5.5 2.9X
+readBinary, payloadLen=512 42 43
1 25.1 39.8 0.4X
+skipBinary, payloadLen=512 6 6
0 182.7 5.5 2.9X
================================================================================================
Variant reads
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Variant reads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------------
-readBytes (INT32) 9 10
1 115.6 8.6 1.0X
-readShorts (INT32) 9 10
1 116.0 8.6 1.0X
-readUnsignedIntegers (INT32 -> Long) 9 9
1 117.1 8.5 1.0X
-readUnsignedLongs (INT64 -> Decimal(20,0)) 211 223
27 5.0 201.5 0.0X
-skipBytes 9 9
0 122.2 8.2 1.1X
-skipShorts 9 9
0 122.3 8.2 1.1X
-readByte (INT32 single-value) 13 13
0 80.1 12.5 0.7X
-readShort (INT32 single-value) 14 15
1 72.7 13.7 0.6X
-readInteger (INT32 single-value) 14 15
0 72.9 13.7 0.6X
-readLong (INT64 single-value) 16 17
1 64.3 15.6 0.6X
-readBinary(len) (DELTA_BYTE_ARRAY single-value) 79 81
2 13.3 74.9 0.1X
+readBytes (INT32) 5 5
0 211.3 4.7 1.0X
+readShorts (INT32) 9 9
1 120.9 8.3 0.6X
+readUnsignedIntegers (INT32 -> Long) 10 10
0 105.8 9.5 0.5X
+readIntegersAsLongs (INT32 -> Long) 4 4
0 283.9 3.5 1.3X
+readIntegersAsDoubles (INT32 -> Double) 4 4
0 253.0 4.0 1.2X
+readUnsignedLongs (INT64 -> Decimal(20,0)) 28 29
0 36.8 27.2 0.2X
+skipBytes 6 6
0 166.5 6.0 0.8X
+skipShorts 6 6
0 166.8 6.0 0.8X
+readByte (INT32 single-value) 12 12
0 86.2 11.6 0.4X
+readShort (INT32 single-value) 13 13
1 81.6 12.3 0.4X
+readInteger (INT32 single-value) 14 14
0 74.7 13.4 0.4X
+readLong (INT64 single-value) 16 16
1 66.9 14.9 0.3X
+readBinary(len) (DELTA_BYTE_ARRAY single-value) 64 66
3 16.3 61.3 0.1X
diff --git a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
index 9ff1678901e9..b3dc35eea94e 100644
--- a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
+++ b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
@@ -2,92 +2,94 @@
DELTA_BINARY_PACKED INT32
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BINARY_PACKED INT32: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readIntegers, constant 2 2
0 482.3 2.1 1.0X
-skipIntegers, constant 3 3
0 335.4 3.0 0.7X
-readIntegers, monotonic 3 3
0 314.9 3.2 0.7X
-skipIntegers, monotonic 3 3
0 335.1 3.0 0.7X
-readIntegers, small-delta random 4 4
0 257.5 3.9 0.5X
-skipIntegers, small-delta random 4 4
0 269.5 3.7 0.6X
-readIntegers, wide random 5 5
0 209.3 4.8 0.4X
-skipIntegers, wide random 5 5
0 218.2 4.6 0.5X
+readIntegers, constant 3 3
0 374.8 2.7 1.0X
+skipIntegers, constant 2 2
0 612.8 1.6 1.6X
+readIntegers, monotonic 3 3
0 374.3 2.7 1.0X
+skipIntegers, monotonic 2 2
0 612.0 1.6 1.6X
+readIntegers, small-delta random 4 4
0 297.3 3.4 0.8X
+skipIntegers, small-delta random 2 3
0 426.7 2.3 1.1X
+readIntegers, wide random 4 4
0 240.3 4.2 0.6X
+skipIntegers, wide random 3 3
0 307.3 3.3 0.8X
================================================================================================
DELTA_BINARY_PACKED INT64
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BINARY_PACKED INT64: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readLongs, constant 5 6
1 191.0 5.2 1.0X
-skipLongs, constant 6 6
0 170.1 5.9 0.9X
-readLongs, monotonic 7 7
0 144.9 6.9 0.8X
-skipLongs, monotonic 6 6
0 170.1 5.9 0.9X
-readLongs, small-delta random 8 8
0 134.8 7.4 0.7X
-skipLongs, small-delta random 7 7
0 152.8 6.5 0.8X
-readLongs, wide random 11 11
0 97.8 10.2 0.5X
-skipLongs, wide random 10 10
0 106.4 9.4 0.6X
+readLongs, constant 2 2
0 535.0 1.9 1.0X
+skipLongs, constant 2 2
0 607.1 1.6 1.1X
+readLongs, monotonic 2 2
0 536.2 1.9 1.0X
+skipLongs, monotonic 2 2
0 605.8 1.7 1.1X
+readLongs, small-delta random 3 3
0 367.0 2.7 0.7X
+skipLongs, small-delta random 2 2
0 426.4 2.3 0.8X
+readLongs, wide random 6 6
0 187.1 5.3 0.3X
+skipLongs, wide random 5 5
0 194.8 5.1 0.4X
================================================================================================
DELTA_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_BYTE_ARRAY: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readBinary, no overlap, len=16 36 37
1 29.2 34.2 1.0X
-skipBinary, no overlap, len=16 41 41
1 25.8 38.7 0.9X
-readBinary, half overlap, len=16 42 43
1 25.0 39.9 0.9X
-skipBinary, half overlap, len=16 48 50
2 21.6 46.2 0.7X
-readBinary, full overlap, len=16 42 44
2 24.8 40.3 0.8X
-skipBinary, full overlap, len=16 48 49
1 21.8 45.9 0.7X
-readBinary, half overlap, len=64 42 44
1 24.8 40.4 0.8X
-skipBinary, half overlap, len=64 47 49
1 22.2 45.1 0.8X
+readBinary, no overlap, len=16 27 28
1 38.6 25.9 1.0X
+skipBinary, no overlap, len=16 32 33
1 32.8 30.5 0.8X
+readBinary, half overlap, len=16 33 34
1 31.4 31.8 0.8X
+skipBinary, half overlap, len=16 39 40
1 26.9 37.2 0.7X
+readBinary, full overlap, len=16 34 35
0 30.5 32.8 0.8X
+skipBinary, full overlap, len=16 40 40
0 26.4 37.9 0.7X
+readBinary, half overlap, len=64 35 35
0 30.3 33.0 0.8X
+skipBinary, half overlap, len=64 39 40
1 26.6 37.6 0.7X
================================================================================================
DELTA_LENGTH_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
DELTA_LENGTH_BYTE_ARRAY: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-readBinary, payloadLen=8 22 25
2 48.1 20.8 1.0X
-skipBinary, payloadLen=8 11 12
1 95.8 10.4 2.0X
-readBinary, payloadLen=32 22 26
3 48.0 20.9 1.0X
-skipBinary, payloadLen=32 11 12
1 97.6 10.2 2.0X
-readBinary, payloadLen=128 26 29
2 40.0 25.0 0.8X
-skipBinary, payloadLen=128 11 12
1 96.0 10.4 2.0X
-readBinary, payloadLen=512 54 62
4 19.3 51.9 0.4X
-skipBinary, payloadLen=512 11 13
2 95.7 10.4 2.0X
+readBinary, payloadLen=8 17 19
2 63.1 15.9 1.0X
+skipBinary, payloadLen=8 6 6
0 175.2 5.7 2.8X
+readBinary, payloadLen=32 17 17
1 62.3 16.0 1.0X
+skipBinary, payloadLen=32 6 6
0 175.3 5.7 2.8X
+readBinary, payloadLen=128 19 19
0 55.1 18.1 0.9X
+skipBinary, payloadLen=128 6 6
1 174.9 5.7 2.8X
+readBinary, payloadLen=512 38 40
1 27.2 36.7 0.4X
+skipBinary, payloadLen=512 6 6
0 174.8 5.7 2.8X
================================================================================================
Variant reads
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Variant reads: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------------
-readBytes (INT32) 7 7
0 145.8 6.9 1.0X
-readShorts (INT32) 7 8
1 140.7 7.1 1.0X
-readUnsignedIntegers (INT32 -> Long) 7 7
0 143.3 7.0 1.0X
-readUnsignedLongs (INT64 -> Decimal(20,0)) 227 239
25 4.6 216.4 0.0X
-skipBytes 8 8
0 136.8 7.3 0.9X
-skipShorts 8 8
0 136.6 7.3 0.9X
-readByte (INT32 single-value) 12 12
1 85.8 11.7 0.6X
-readShort (INT32 single-value) 12 12
0 85.9 11.6 0.6X
-readInteger (INT32 single-value) 12 12
0 86.1 11.6 0.6X
-readLong (INT64 single-value) 14 15
1 73.4 13.6 0.5X
-readBinary(len) (DELTA_BYTE_ARRAY single-value) 71 77
5 14.7 68.2 0.1X
+readBytes (INT32) 5 5
0 202.0 5.0 1.0X
+readShorts (INT32) 9 9
0 115.1 8.7 0.6X
+readUnsignedIntegers (INT32 -> Long) 9 9
0 121.4 8.2 0.6X
+readIntegersAsLongs (INT32 -> Long) 4 4
0 291.1 3.4 1.4X
+readIntegersAsDoubles (INT32 -> Double) 4 4
0 258.8 3.9 1.3X
+readUnsignedLongs (INT64 -> Decimal(20,0)) 28 28
0 37.6 26.6 0.2X
+skipBytes 6 7
1 165.5 6.0 0.8X
+skipShorts 7 7
0 158.4 6.3 0.8X
+readByte (INT32 single-value) 14 14
0 75.4 13.3 0.4X
+readShort (INT32 single-value) 14 14
0 75.4 13.3 0.4X
+readInteger (INT32 single-value) 14 14
0 75.4 13.3 0.4X
+readLong (INT64 single-value) 15 16
0 68.1 14.7 0.3X
+readBinary(len) (DELTA_BYTE_ARRAY single-value) 62 64
1 17.0 58.7 0.1X
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetDictionary.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetDictionary.java
index ffdb94ee3765..aba5e1996a2e 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetDictionary.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetDictionary.java
@@ -19,8 +19,6 @@ package org.apache.spark.sql.execution.datasources.parquet;
import org.apache.spark.sql.execution.vectorized.Dictionary;
-import java.math.BigInteger;
-
public final class ParquetDictionary implements Dictionary {
private org.apache.parquet.column.Dictionary dictionary;
private boolean needTransform = false;
@@ -68,7 +66,7 @@ public final class ParquetDictionary implements Dictionary {
// whenever dictionary is available.
// Here we lazily decode it to the original signed long value then
convert to decimal(20, 0).
long signed = dictionary.decodeToLong(id);
- return new BigInteger(Long.toUnsignedString(signed)).toByteArray();
+ return VectorizedReaderBase.unsignedLongToBytesBigEndian(signed);
} else {
return dictionary.decodeToBinary(id).getBytesUnsafe();
}
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
index 90e4c3e04f81..8a72d2e5e731 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java
@@ -717,6 +717,8 @@ public class ParquetVectorUpdaterFactory {
}
private static class UnsignedLongUpdater implements ParquetVectorUpdater {
+ private final byte[] unsignedLongScratch = new byte[9];
+
@Override
public void readValues(
int total,
@@ -736,8 +738,9 @@ public class ParquetVectorUpdaterFactory {
int offset,
WritableColumnVector values,
VectorizedValuesReader valuesReader) {
- byte[] bytes = new
BigInteger(Long.toUnsignedString(valuesReader.readLong())).toByteArray();
- values.putByteArray(offset, bytes);
+ int start = VectorizedReaderBase.encodeUnsignedLongBigEndian(
+ valuesReader.readLong(), unsignedLongScratch);
+ values.putByteArray(offset, unsignedLongScratch, start, 9 - start);
}
@Override
@@ -747,8 +750,8 @@ public class ParquetVectorUpdaterFactory {
WritableColumnVector dictionaryIds,
Dictionary dictionary) {
long signed = dictionary.decodeToLong(dictionaryIds.getDictId(offset));
- byte[] unsigned = new
BigInteger(Long.toUnsignedString(signed)).toByteArray();
- values.putByteArray(offset, unsigned);
+ int start = VectorizedReaderBase.encodeUnsignedLongBigEndian(signed,
unsignedLongScratch);
+ values.putByteArray(offset, unsignedLongScratch, start, 9 - start);
}
}
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
index 7b56e1ebf239..6abea3b40066 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
@@ -17,7 +17,6 @@
package org.apache.spark.sql.execution.datasources.parquet;
import java.io.IOException;
-import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Arrays;
@@ -70,6 +69,9 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
private int remainingInBlock = 0; // values in current block still to be read
private int remainingInMiniBlock = 0; // values in current mini block still
to be read
private long[] unpackedValuesBuffer;
+ // Scratch buffer for narrowing long -> int during bulk readIntegers.
+ // Allocated eagerly in initFromPage, sized to miniBlockSizeInValues.
+ private int[] intScratchBuffer;
private ByteBufferInputStream in;
@@ -95,6 +97,7 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
this.totalValueCount = BytesUtils.readUnsignedVarInt(in);
this.bitWidths = new int[miniBlockNumInABlock];
this.unpackedValuesBuffer = new long[miniBlockSizeInValues];
+ this.intScratchBuffer = new int[miniBlockSizeInValues];
// read the first value
firstValue = BytesUtils.readZigZagVarLong(in);
}
@@ -140,7 +143,32 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
@Override
public void readIntegers(int total, WritableColumnVector c, int rowId) {
- readValues(total, c, rowId, (w, r, v) -> w.putInt(r, (int) v));
+ readBulkIntegers(total, c, rowId);
+ }
+
+ @Override
+ public void readIntegersAsLongs(int total, WritableColumnVector c, int
rowId) {
+ // Delta decoder already works on long[]; skip the int narrowing and write
longs directly.
+ readBulkLongs(total, c, rowId);
+ }
+
+ @Override
+ public void readIntegersAsDoubles(int total, WritableColumnVector c, int
rowId) {
+ checkReadBounds(total);
+ int remaining = total;
+ if (valuesRead == 0) {
+ c.putDouble(rowId, (double) firstValue);
+ lastValueRead = firstValue;
+ rowId++;
+ remaining--;
+ }
+ readBulkLoop(remaining, c, rowId,
+ (col, r, buf, s, n) -> {
+ for (int i = s; i < s + n; i++) {
+ col.putDouble(r + i - s, (double) buf[i]);
+ }
+ });
+ valuesRead += total;
}
// Based on VectorizedPlainValuesReader.readIntegersWithRebase
@@ -170,13 +198,13 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
@Override
public void readUnsignedLongs(int total, WritableColumnVector c, int rowId) {
readValues(total, c, rowId, (w, r, v) -> {
- w.putByteArray(r, new
BigInteger(Long.toUnsignedString(v)).toByteArray());
+ putUnsignedLongAsBigInteger(w, r, v);
});
}
@Override
public void readLongs(int total, WritableColumnVector c, int rowId) {
- readValues(total, c, rowId, WritableColumnVector::putLong);
+ readBulkLongs(total, c, rowId);
}
@Override
@@ -215,6 +243,111 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
skipValues(total);
}
+ // ---- Bulk read helpers for readIntegers / readLongs
----------------------------
+ //
+ // The generic readValues() path dispatches a lambda per value. For the two
most
+ // common callers (readIntegers, readLongs) we can do much better: compute a
prefix
+ // sum over the unpacked deltas in-place, then bulk-copy the result into the
column
+ // vector with putInts / putLongs (backed by System.arraycopy on-heap).
+
+ /**
+ * Callback for writing a chunk of prefix-summed absolute values from
+ * {@code unpackedValuesBuffer} into a column vector. Called once per
mini-block
+ * (not per value), so lambda overhead is negligible.
+ */
+ @FunctionalInterface
+ private interface BulkWriter {
+ void write(WritableColumnVector c, int rowId, long[] values, int start,
int count);
+ }
+
+ /** Narrows long[] -> int[] scratch and bulk-writes via putInts. */
+ private void bulkWriteInts(WritableColumnVector c, int rowId,
+ long[] buf, int start, int count) {
+ for (int i = start; i < start + count; i++) {
+ intScratchBuffer[i] = (int) buf[i];
+ }
+ c.putInts(rowId, count, intScratchBuffer, start);
+ }
+
+ private void readBulkIntegers(int total, WritableColumnVector c, int rowId) {
+ checkReadBounds(total);
+ int remaining = total;
+ if (valuesRead == 0) {
+ c.putInt(rowId, (int) firstValue);
+ lastValueRead = firstValue;
+ rowId++;
+ remaining--;
+ }
+ readBulkLoop(remaining, c, rowId, this::bulkWriteInts);
+ valuesRead += total;
+ }
+
+ private void readBulkLongs(int total, WritableColumnVector c, int rowId) {
+ checkReadBounds(total);
+ int remaining = total;
+ if (valuesRead == 0) {
+ c.putLong(rowId, firstValue);
+ lastValueRead = firstValue;
+ rowId++;
+ remaining--;
+ }
+ readBulkLoop(remaining, c, rowId,
+ (col, r, buf, s, n) -> col.putLongs(r, n, buf, s));
+ valuesRead += total;
+ }
+
+ private void checkReadBounds(int total) {
+ if (valuesRead + total > totalValueCount) {
+ throw new ParquetDecodingException(
+ "No more values to read. Total values read: " + valuesRead + ",
total count: "
+ + totalValueCount + ", trying to read " + total + " more.");
+ }
+ }
+
+ /** Mini-block iteration loop shared by readBulkIntegers and readBulkLongs.
*/
+ private void readBulkLoop(int remaining, WritableColumnVector c, int rowId,
+ BulkWriter writer) {
+ while (remaining > 0) {
+ int n;
+ try {
+ n = loadMiniBlockBulk(remaining, c, rowId, writer);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Error reading mini block.", e);
+ }
+ rowId += n;
+ remaining -= n;
+ }
+ }
+
+ /**
+ * Loads the next mini-block (if needed), prefix-sums the deltas in-place,
and
+ * delegates the type-specific column write to {@code writer}.
+ */
+ private int loadMiniBlockBulk(int remaining, WritableColumnVector c, int
rowId,
+ BulkWriter writer) throws IOException {
+ if (remainingInBlock == 0) readBlockHeader();
+ if (remainingInMiniBlock == 0) unpackMiniBlock();
+
+ int start = miniBlockSizeInValues - remainingInMiniBlock;
+ int count = Math.min(remaining, remainingInMiniBlock);
+
+ // Prefix-sum: convert raw deltas -> absolute values in-place.
+ long running = lastValueRead;
+ long minDelta = minDeltaInCurrentBlock;
+ long[] buf = unpackedValuesBuffer;
+ for (int i = start; i < start + count; i++) {
+ running += minDelta + buf[i];
+ buf[i] = running;
+ }
+ lastValueRead = running;
+
+ writer.write(c, rowId, buf, start, count);
+
+ remainingInBlock -= count;
+ remainingInMiniBlock -= count;
+ return count;
+ }
+
private void readValues(int total, WritableColumnVector c, int rowId,
IntegerOutputWriter outputWriter) {
if (valuesRead + total > totalValueCount) {
@@ -240,7 +373,7 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
rowId += n;
remaining -= n;
}
- valuesRead = total - remaining;
+ valuesRead += total;
}
@@ -328,4 +461,12 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
readValues(total, null, -1, (w, r, v) -> {});
}
+ // Reusable buffer for unsigned long -> BigInteger encoding (9 bytes: sign +
8 value bytes).
+ private final byte[] unsignedLongScratch = new byte[9];
+
+ private void putUnsignedLongAsBigInteger(WritableColumnVector c, int rowId,
long v) {
+ int start = encodeUnsignedLongBigEndian(v, unsignedLongScratch);
+ c.putByteArray(rowId, unsignedLongScratch, start, 9 - start);
+ }
+
}
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedReaderBase.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedReaderBase.java
index 77bb9340e526..f5ab1a68bea6 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedReaderBase.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedReaderBase.java
@@ -27,6 +27,44 @@ import
org.apache.spark.sql.execution.vectorized.WritableColumnVector;
*/
public class VectorizedReaderBase extends ValuesReader implements
VectorizedValuesReader {
+ /**
+ * Encodes an unsigned long as a minimal big-endian two's-complement byte
array
+ * compatible with {@link java.math.BigInteger} encoding, written into
+ * {@code scratch[start .. 8]} (length = 9 - start). {@code scratch} must
have length >= 9.
+ *
+ * <p>This avoids the per-value overhead of
+ * {@code new BigInteger(Long.toUnsignedString(v)).toByteArray()} which
allocates a
+ * String, a BigInteger, and a byte[] on every call.
+ */
+ static int encodeUnsignedLongBigEndian(long v, byte[] scratch) {
+ // Minimal BigInteger-compatible length is bitLength/8 + 1, so the
+ // encoding occupies scratch[start .. 8]; the loop runs once past the
+ // significant bytes, which writes the 0x00 sign byte when one is needed.
+ int start = 8 - (64 - Long.numberOfLeadingZeros(v)) / 8;
+ long x = v;
+ for (int i = 8; i >= start; i--) {
+ scratch[i] = (byte) x;
+ x >>>= 8;
+ }
+ return start;
+ }
+
+ /**
+ * Convenience: encodes an unsigned long and returns a new byte[] with the
minimal
+ * BigInteger-compatible encoding. Use {@link
#encodeUnsignedLongBigEndian(long, byte[])}
+ * with a reusable buffer in hot paths to avoid allocation.
+ */
+ static byte[] unsignedLongToBytesBigEndian(long v) {
+ int len = (64 - Long.numberOfLeadingZeros(v)) / 8 + 1;
+ byte[] out = new byte[len];
+ long x = v;
+ for (int i = len - 1; i >= 0 && x != 0; i--) {
+ out[i] = (byte) x;
+ x >>>= 8;
+ }
+ return out;
+ }
+
@Override
public void skip() {
throw SparkUnsupportedOperationException.apply();
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaReaderBenchmark.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaReaderBenchmark.scala
index 40e08d9870c7..b97faeaf57b6 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaReaderBenchmark.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaReaderBenchmark.scala
@@ -29,7 +29,7 @@ import org.apache.parquet.io.api.Binary
import org.apache.spark.benchmark.{Benchmark, BenchmarkBase}
import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
-import org.apache.spark.sql.types.{BinaryType, ByteType, DecimalType,
IntegerType, LongType, ShortType}
+import org.apache.spark.sql.types.{BinaryType, ByteType, DecimalType,
DoubleType, IntegerType, LongType, ShortType}
/**
* Low-level benchmark for the three Parquet delta-encoding decoders:
@@ -299,6 +299,7 @@ object VectorizedDeltaReaderBenchmark extends BenchmarkBase
{
val byteVec = new OnHeapColumnVector(NUM_ROWS, ByteType)
val shortVec = new OnHeapColumnVector(NUM_ROWS, ShortType)
val longVec = new OnHeapColumnVector(NUM_ROWS, LongType)
+ val doubleVec = new OnHeapColumnVector(NUM_ROWS, DoubleType)
val unsignedLongVec = new OnHeapColumnVector(NUM_ROWS, DecimalType(20, 0))
def newIntReader(): VectorizedDeltaBinaryPackedReader = {
@@ -320,7 +321,14 @@ object VectorizedDeltaReaderBenchmark extends
BenchmarkBase {
benchmark.addCase("readUnsignedIntegers (INT32 -> Long)") { _ =>
newIntReader().readUnsignedIntegers(NUM_ROWS, longVec, 0)
}
+ benchmark.addCase("readIntegersAsLongs (INT32 -> Long)") { _ =>
+ newIntReader().readIntegersAsLongs(NUM_ROWS, longVec, 0)
+ }
+ benchmark.addCase("readIntegersAsDoubles (INT32 -> Double)") { _ =>
+ newIntReader().readIntegersAsDoubles(NUM_ROWS, doubleVec, 0)
+ }
benchmark.addCase("readUnsignedLongs (INT64 -> Decimal(20,0))") { _ =>
+ unsignedLongVec.reset()
newLongReader().readUnsignedLongs(NUM_ROWS, unsignedLongVec, 0)
}
benchmark.addCase("skipBytes") { _ => newIntReader().skipBytes(NUM_ROWS) }
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
