This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 729c57e6382b [SPARK-56892][SQL] Bulk read optimization for Parquet
DELTA_BINARY_PACKED decoding
729c57e6382b is described below
commit 729c57e6382b797c7b4dd3311efff49e4db2fa6e
Author: Ismaël Mejía <[email protected]>
AuthorDate: Wed Jun 17 16:55:11 2026 +0800
[SPARK-56892][SQL] Bulk read optimization for Parquet DELTA_BINARY_PACKED
decoding
### What changes were proposed in this pull request?
Re-apply the bulk read optimization for `VectorizedDeltaBinaryPackedReader`
(reverted in c13302acc2a) with a fix for the INT32 widening bug that caused the
CI failure.
**Commit 1** — Reapply the original optimization (revert of the revert):
- Bulk `readIntegers`/`readLongs` via prefix-sum + `putInts`/`putLongs`
- Zero-allocation unsigned long encoding (`encodeUnsignedLongBigEndian`)
- `readIntegersAsLongs` and `readIntegersAsDoubles` overrides
**Commit 2** — Fix the INT32 widening bug:
- The Parquet INT32 delta encoder
(`DeltaBinaryPackingValuesWriterForInteger`) computes deltas using Java int
arithmetic with modular overflow. The bulk widened readers
(`readIntegersAsLongs`, `readIntegersAsDoubles`) were performing the prefix sum
in long space and writing raw long results without truncating back to int. When
delta overflow occurs (e.g. a sequence containing `Int.MinValue`), the
reconstructed long has the wrong sign.
- Fix: truncate each prefix-sum result to int before widening to long/double
- Add focused low-level tests for the overflow case (single-batch and split
reads)
- Add benchmark cases for the overflow pattern
This is the same content as #55919, which was merged and reverted due to
this bug.
### Why are the changes needed?
The bulk read path eliminates per-value lambda dispatch overhead and
enables the JIT to better vectorize the inner unpacking loop. See #55919 for
full benchmark results.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `ParquetTypeWideningSuite`: IntegerType -> LongType, IntegerType ->
DoubleType
- `ParquetDeltaEncodingInteger`: new focused tests for modular delta
overflow
- `ParquetDeltaEncodingInteger`/`Long`: full suites (30 tests)
- `ParquetIOSuite`: UINT_64 tests
- `VectorizedDeltaReaderBenchmark`: full suite including new overflow cases
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Assisted-by: GitHub Copilot:claude-opus-4.6
Closes #56543 from iemejia/SPARK-56892-delta-binary-packed-bulk-read-v2.
Authored-by: Ismaël Mejía <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
(cherry picked from commit 107111001a0b5f5258a463481deb1a039902f511)
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 | 175 ++++++++++++++++++++-
.../datasources/parquet/VectorizedReaderBase.java | 38 +++++
.../parquet/ParquetDeltaEncodingSuite.scala | 52 +++++-
.../parquet/VectorizedDeltaReaderBenchmark.scala | 30 +++-
9 files changed, 446 insertions(+), 158 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..0912ae534b9b 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,41 @@ 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) {
+ // The INT32 delta encoder uses int arithmetic (with modular overflow) for
deltas and
+ // stores minDelta as a zigzag VarInt. The prefix sum in loadMiniBlockBulk
operates in
+ // long space, so we must truncate each result to int (matching the
encoder's 32-bit
+ // modular semantics) then sign-extend back to long for the output.
+ checkReadBounds(total);
+ int remaining = total;
+ if (valuesRead == 0) {
+ c.putLong(rowId, (int) firstValue);
+ lastValueRead = firstValue;
+ rowId++;
+ remaining--;
+ }
+ readBulkLoop(remaining, c, rowId, this::bulkWriteIntsAsLongs);
+ valuesRead += total;
+ }
+
+ @Override
+ public void readIntegersAsDoubles(int total, WritableColumnVector c, int
rowId) {
+ // Same as readIntegersAsLongs: truncate the long prefix-sum result to int
(matching the
+ // INT32 encoder's modular arithmetic) then widen to double.
+ checkReadBounds(total);
+ int remaining = total;
+ if (valuesRead == 0) {
+ c.putDouble(rowId, (double) (int) firstValue);
+ lastValueRead = firstValue;
+ rowId++;
+ remaining--;
+ }
+ readBulkLoop(remaining, c, rowId, this::bulkWriteIntsAsDoubles);
+ valuesRead += total;
}
// Based on VectorizedPlainValuesReader.readIntegersWithRebase
@@ -170,13 +207,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 +252,126 @@ 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 bulkWriteIntsAsLongs(WritableColumnVector c, int rowId,
+ long[] buf, int start, int count) {
+ for (int i = start; i < start + count; i++) {
+ buf[i] = (int) buf[i];
+ }
+ c.putLongs(rowId, count, buf, start);
+ }
+
+ private void bulkWriteIntsAsDoubles(WritableColumnVector c, int rowId,
+ long[] buf, int start, int count) {
+ for (int i = start; i < start + count; i++) {
+ c.putDouble(rowId + i - start, (double) (int) buf[i]);
+ }
+ }
+
+ 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 +397,7 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
rowId += n;
remaining -= n;
}
- valuesRead = total - remaining;
+ valuesRead += total;
}
@@ -328,4 +485,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/ParquetDeltaEncodingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala
index 1c7b7b14da38..b72b55344507 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaEncodingSuite.scala
@@ -27,7 +27,7 @@ import org.apache.parquet.io.ParquetDecodingException
import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector,
WritableColumnVector}
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{IntegerType, IntegralType, LongType}
+import org.apache.spark.sql.types.{DoubleType, IntegerType, IntegralType,
LongType}
/**
* Read tests for vectorized Delta binary packed reader.
@@ -261,6 +261,26 @@ abstract class ParquetDeltaEncodingSuite[T] extends
ParquetCompatibilityTest
class ParquetDeltaEncodingInteger extends ParquetDeltaEncodingSuite[Int] {
+ test("read INT32 as long with modular delta overflow") {
+ val data = Array(1, 2, Int.MinValue, 3)
+ shouldReadIntegersAsLongs(data, reads = Seq(data.length))
+ }
+
+ test("read INT32 as long with modular delta overflow across split reads") {
+ val data = Array(1, 2, Int.MinValue, 3)
+ shouldReadIntegersAsLongs(data, reads = Seq(2, 2))
+ }
+
+ test("read INT32 as double with modular delta overflow") {
+ val data = Array(1, 2, Int.MinValue, 3)
+ shouldReadIntegersAsDoubles(data, reads = Seq(data.length))
+ }
+
+ test("read INT32 as double with modular delta overflow across split reads") {
+ val data = Array(1, 2, Int.MinValue, 3)
+ shouldReadIntegersAsDoubles(data, reads = Seq(2, 2))
+ }
+
override protected def getSparkSqlType: IntegralType = IntegerType
override protected def writeData(data: Array[Int]): Unit = writeData(data,
data.length)
@@ -309,6 +329,36 @@ class ParquetDeltaEncodingInteger extends
ParquetDeltaEncodingSuite[Int] {
override protected def compareValues(expected: Int, actual: Int) : Boolean =
expected == actual
+ private def shouldReadIntegersAsLongs(data: Array[Int], reads: Seq[Int]):
Unit = {
+ writeData(data)
+ reader = new VectorizedDeltaBinaryPackedReader
+ reader.initFromPage(data.length, writer.getBytes.toInputStream)
+ writableColumnVector = new OnHeapColumnVector(data.length, LongType)
+ var rowId = 0
+ reads.foreach { total =>
+ reader.readIntegersAsLongs(total, writableColumnVector, rowId)
+ rowId += total
+ }
+ data.indices.foreach { i =>
+ assert(writableColumnVector.getLong(i) == data(i).toLong)
+ }
+ }
+
+ private def shouldReadIntegersAsDoubles(data: Array[Int], reads: Seq[Int]):
Unit = {
+ writeData(data)
+ reader = new VectorizedDeltaBinaryPackedReader
+ reader.initFromPage(data.length, writer.getBytes.toInputStream)
+ writableColumnVector = new OnHeapColumnVector(data.length, DoubleType)
+ var rowId = 0
+ reads.foreach { total =>
+ reader.readIntegersAsDoubles(total, writableColumnVector, rowId)
+ rowId += total
+ }
+ data.indices.foreach { i =>
+ assert(writableColumnVector.getDouble(i) == data(i).toDouble)
+ }
+ }
+
}
class ParquetDeltaEncodingLong extends ParquetDeltaEncodingSuite[Long] {
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..e0426393396a 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:
@@ -292,13 +292,23 @@ object VectorizedDeltaReaderBenchmark extends
BenchmarkBase {
val rng = new Random(42)
val intValues = Array.tabulate(NUM_ROWS)(_ => rng.nextInt())
+ val overflowIntValues = Array.tabulate(NUM_ROWS) { i =>
+ i % 4 match {
+ case 0 => 1
+ case 1 => 2
+ case 2 => Int.MinValue
+ case _ => 3
+ }
+ }
val longValues = Array.tabulate(NUM_ROWS)(_ => rng.nextLong())
val intBytes = encodeDeltaInts(intValues)
+ val overflowIntBytes = encodeDeltaInts(overflowIntValues)
val longBytes = encodeDeltaLongs(longValues)
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 = {
@@ -306,6 +316,11 @@ object VectorizedDeltaReaderBenchmark extends
BenchmarkBase {
r.initFromPage(NUM_ROWS,
ByteBufferInputStream.wrap(ByteBuffer.wrap(intBytes)))
r
}
+ def newOverflowIntReader(): VectorizedDeltaBinaryPackedReader = {
+ val r = new VectorizedDeltaBinaryPackedReader
+ r.initFromPage(NUM_ROWS,
ByteBufferInputStream.wrap(ByteBuffer.wrap(overflowIntBytes)))
+ r
+ }
def newLongReader(): VectorizedDeltaBinaryPackedReader = {
val r = new VectorizedDeltaBinaryPackedReader
r.initFromPage(NUM_ROWS,
ByteBufferInputStream.wrap(ByteBuffer.wrap(longBytes)))
@@ -320,7 +335,20 @@ 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("readIntegersAsLongs (INT32 -> Long, overflow pattern)")
{ _ =>
+ newOverflowIntReader().readIntegersAsLongs(NUM_ROWS, longVec, 0)
+ }
+ benchmark.addCase("readIntegersAsDoubles (INT32 -> Double)") { _ =>
+ newIntReader().readIntegersAsDoubles(NUM_ROWS, doubleVec, 0)
+ }
+ benchmark.addCase("readIntegersAsDoubles (INT32 -> Double, overflow
pattern)") { _ =>
+ newOverflowIntReader().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]
