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 c13302acc2a8 Revert "[SPARK-56892][SQL] Bulk read optimization for
Parquet DELTA_BINARY_PACKED decoding"
c13302acc2a8 is described below
commit c13302acc2a80003a0aac8b81a63170cded83937
Author: yangjie01 <[email protected]>
AuthorDate: Tue Jun 16 15:24:50 2026 +0800
Revert "[SPARK-56892][SQL] Bulk read optimization for Parquet
DELTA_BINARY_PACKED decoding"
This reverts commit 03750ab553d9ebf446e61e077e867d45436f72b0.
---
...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, 157 insertions(+), 351 deletions(-)
diff --git
a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
index 9f0397dd4d8b..a7199750be60 100644
--- a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
+++ b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk21-results.txt
@@ -2,94 +2,92 @@
DELTA_BINARY_PACKED INT32
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
DELTA_BINARY_PACKED INT64
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
DELTA_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
DELTA_LENGTH_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
Variant reads
================================================================================================
-OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1011-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) 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
+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
diff --git
a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
index ef6274f71249..fe7b64c3834b 100644
--- a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
+++ b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-jdk25-results.txt
@@ -2,94 +2,92 @@
DELTA_BINARY_PACKED INT32
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-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 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
+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
================================================================================================
DELTA_BINARY_PACKED INT64
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-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 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
+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
================================================================================================
DELTA_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-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 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
+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
================================================================================================
DELTA_LENGTH_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-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 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
+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
================================================================================================
Variant reads
================================================================================================
-OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1010-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) 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
+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
diff --git a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
index b3dc35eea94e..9ff1678901e9 100644
--- a/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
+++ b/sql/core/benchmarks/VectorizedDeltaReaderBenchmark-results.txt
@@ -2,94 +2,92 @@
DELTA_BINARY_PACKED INT32
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
DELTA_BINARY_PACKED INT64
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
DELTA_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
DELTA_LENGTH_BYTE_ARRAY
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-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 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
+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
================================================================================================
Variant reads
================================================================================================
-OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1011-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) 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
+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
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 aba5e1996a2e..ffdb94ee3765 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,6 +19,8 @@ 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;
@@ -66,7 +68,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 VectorizedReaderBase.unsignedLongToBytesBigEndian(signed);
+ return new BigInteger(Long.toUnsignedString(signed)).toByteArray();
} 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 8a72d2e5e731..90e4c3e04f81 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,8 +717,6 @@ public class ParquetVectorUpdaterFactory {
}
private static class UnsignedLongUpdater implements ParquetVectorUpdater {
- private final byte[] unsignedLongScratch = new byte[9];
-
@Override
public void readValues(
int total,
@@ -738,9 +736,8 @@ public class ParquetVectorUpdaterFactory {
int offset,
WritableColumnVector values,
VectorizedValuesReader valuesReader) {
- int start = VectorizedReaderBase.encodeUnsignedLongBigEndian(
- valuesReader.readLong(), unsignedLongScratch);
- values.putByteArray(offset, unsignedLongScratch, start, 9 - start);
+ byte[] bytes = new
BigInteger(Long.toUnsignedString(valuesReader.readLong())).toByteArray();
+ values.putByteArray(offset, bytes);
}
@Override
@@ -750,8 +747,8 @@ public class ParquetVectorUpdaterFactory {
WritableColumnVector dictionaryIds,
Dictionary dictionary) {
long signed = dictionary.decodeToLong(dictionaryIds.getDictId(offset));
- int start = VectorizedReaderBase.encodeUnsignedLongBigEndian(signed,
unsignedLongScratch);
- values.putByteArray(offset, unsignedLongScratch, start, 9 - start);
+ byte[] unsigned = new
BigInteger(Long.toUnsignedString(signed)).toByteArray();
+ values.putByteArray(offset, unsigned);
}
}
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 6abea3b40066..7b56e1ebf239 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,6 +17,7 @@
package org.apache.spark.sql.execution.datasources.parquet;
import java.io.IOException;
+import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Arrays;
@@ -69,9 +70,6 @@ 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;
@@ -97,7 +95,6 @@ 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);
}
@@ -143,32 +140,7 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
@Override
public void readIntegers(int total, WritableColumnVector c, int rowId) {
- 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;
+ readValues(total, c, rowId, (w, r, v) -> w.putInt(r, (int) v));
}
// Based on VectorizedPlainValuesReader.readIntegersWithRebase
@@ -198,13 +170,13 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
@Override
public void readUnsignedLongs(int total, WritableColumnVector c, int rowId) {
readValues(total, c, rowId, (w, r, v) -> {
- putUnsignedLongAsBigInteger(w, r, v);
+ w.putByteArray(r, new
BigInteger(Long.toUnsignedString(v)).toByteArray());
});
}
@Override
public void readLongs(int total, WritableColumnVector c, int rowId) {
- readBulkLongs(total, c, rowId);
+ readValues(total, c, rowId, WritableColumnVector::putLong);
}
@Override
@@ -243,111 +215,6 @@ 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) {
@@ -373,7 +240,7 @@ public class VectorizedDeltaBinaryPackedReader extends
VectorizedReaderBase {
rowId += n;
remaining -= n;
}
- valuesRead += total;
+ valuesRead = total - remaining;
}
@@ -461,12 +328,4 @@ 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 f5ab1a68bea6..77bb9340e526 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,44 +27,6 @@ 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 b97faeaf57b6..40e08d9870c7 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,
DoubleType, IntegerType, LongType, ShortType}
+import org.apache.spark.sql.types.{BinaryType, ByteType, DecimalType,
IntegerType, LongType, ShortType}
/**
* Low-level benchmark for the three Parquet delta-encoding decoders:
@@ -299,7 +299,6 @@ 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 = {
@@ -321,14 +320,7 @@ 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]
