This is an automated email from the ASF dual-hosted git repository.
MaxGekk 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 8875371f10df [SPARK-56894][SQL] Add vectorized Parquet
BYTE_STREAM_SPLIT reader
8875371f10df is described below
commit 8875371f10dfe9d0bf2a6c9cb7a8c327468d635c
Author: Ismaël Mejía <[email protected]>
AuthorDate: Fri Jun 26 19:29:03 2026 +0200
[SPARK-56894][SQL] Add vectorized Parquet BYTE_STREAM_SPLIT reader
### What changes were proposed in this pull request?
This PR adds a vectorized reader for the Parquet `BYTE_STREAM_SPLIT`
encoding (`VectorizedByteStreamSplitValuesReader`), wired into
`VectorizedColumnReader.getValuesReader()`.
**BYTE_STREAM_SPLIT** de-interleaves N fixed-width values (W bytes each)
into W separate byte streams. Decoding gathers the original bytes back:
`value[i] = {stream[0][i], stream[1][i], ..., stream[W-1][i]}`. This encoding
is particularly effective for time-series and scientific data where adjacent
values share high-order bytes.
The new reader:
- Loads the entire encoded page into a `byte[]` via `initFromPage`
- Uses direct per-element `assembleInt` / `assembleLong` helpers for byte
gathering
- Implements all batch read methods (`readIntegers`, `readLongs`,
`readFloats`, `readDoubles`, `readBinary`) and skip methods
- Supports FLOAT (W=4), DOUBLE (W=8), INT32 (W=4), INT64 (W=8), and
FIXED_LEN_BYTE_ARRAY (W=type length)
The `VectorizedColumnReader` change is a single `case BYTE_STREAM_SPLIT ->`
block (12 lines) that resolves the type width from the column descriptor and
yields the new reader.
#### `readFixedLenByteArray` batch method for FLBA columns
The shared `FixedLenByteArrayUpdater.readValues` previously dispatched
through `readBinary(total, c, rowId)`, which in `VectorizedPlainValuesReader`
reads a 4-byte length prefix per value (variable-length BYTE_ARRAY contract).
FIXED_LEN_BYTE_ARRAY data has no length prefix -- each value is exactly
`arrayLen` raw bytes -- so PLAIN-encoded FLBA columns were misread (masked in
CI because dictionary encoding bypasses the updater path).
A dedicated `readFixedLenByteArray(total, len, c, rowId)` default method
was added to `VectorizedValuesReader` (per-value fallback via
`readBinary(len)`), with optimized overrides in:
- **`VectorizedPlainValuesReader`**: reads `len` bytes directly from the
buffer per value (no length prefix, no intermediate `Binary` allocation).
- **`VectorizedByteStreamSplitValuesReader`**: delegates to the existing
`readBinary(total, c, rowId)` which already handles fixed-width values
correctly.
This fixes PLAIN-encoded FLBA correctness and also improves performance.
The `ParquetVectorUpdaterBenchmark` (AMD EPYC 7763, all JDKs) confirms a
**1.4-1.6x speedup** for `FixedLenByteArrayUpdater` by eliminating per-value
`Binary` allocation and `getBytesUnsafe()` copy, with no regressions in any
other updater:
| Case | JDK | Baseline (ms / M/s) | PR (ms / M/s) | Speedup |
|---|---|---|---|---|
| FixedLenByteArrayUpdater (len=16 -> Binary) | 17 | 18 ms / 57.7 M/s | 13
ms / 80.2 M/s | **1.4x** |
| | 21 | 20 ms / 51.9 M/s | 13 ms / 80.0 M/s | **1.5x** |
| | 25 | 21 ms / 50.2 M/s | 13 ms / 78.7 M/s | **1.6x** |
| FixedLenByteArrayAsIntUpdater (len=4) | all | 6-7 ms / 153-174 M/s | 7 ms
/ 152-160 M/s | unchanged |
| FixedLenByteArrayAsLongUpdater (len=8) | all | 8 ms / 128-134 M/s | 8-9
ms / 123-133 M/s | unchanged |
| All other updater groups | all | -- | -- | no regressions |
### Why are the changes needed?
Before this PR, Spark fell back to parquet-mr's per-value
`ByteStreamSplitValuesReader` for BSS-encoded columns. The new vectorized batch
reader is **2.8-4.5x faster** on the benchmark:
```
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 9V74 80-Core Processor
BYTE_STREAM_SPLIT INT32:
Spark vectorized readIntegers 534.6 M/s 1.0X
Spark vectorized readIntegersAsLongs 449.1 M/s 0.8X
Spark vectorized readIntegersAsDoubles 421.1 M/s 0.8X
parquet-mr readInteger (per-value) 118.6 M/s 0.2X (4.5x slower)
BYTE_STREAM_SPLIT INT64:
Spark vectorized readLongs 211.1 M/s 1.0X
Spark vectorized readLongsAsInts 214.3 M/s 1.0X
parquet-mr readLong (per-value) 76.4 M/s 0.4X (2.8x slower)
BYTE_STREAM_SPLIT FLOAT:
Spark vectorized readFloats 513.9 M/s 1.0X
Spark vectorized readFloatsAsDoubles 453.0 M/s 0.9X
parquet-mr readFloat (per-value) 119.9 M/s 0.2X (4.3x slower)
BYTE_STREAM_SPLIT DOUBLE:
Spark vectorized readDoubles 211.5 M/s 1.0X
parquet-mr readDouble (per-value) 76.8 M/s 0.4X (2.8x slower)
```
The widening overrides (`readIntegersAsLongs`, `readIntegersAsDoubles`,
`readFloatsAsDoubles`, `readLongsAsInts`) match base method throughput,
avoiding the 4x slower per-row default path for type-converting updaters.
**Speedup vs parquet-mr per-value (all JDKs on AMD EPYC 9V74):**
| Type | JDK 17 | JDK 21 | JDK 25 |
|---|---|---|---|
| INT32 (readIntegers) | 4.5x (534.9 M/s) | 3.8x (509.3 M/s) | 4.2x (567.2
M/s) |
| INT64 (readLongs) | 2.8x (211.5 M/s) | 2.0x (159.9 M/s) | 1.8x (152.0
M/s) |
| FLOAT (readFloats) | 4.3x (517.8 M/s) | 3.6x (486.9 M/s) | 4.3x (569.1
M/s) |
| DOUBLE (readDoubles) | 2.7x (211.1 M/s) | 2.0x (161.4 M/s) | 1.8x (152.1
M/s) |
| readIntegersAsLongs | 3.8x (449.9 M/s) | 3.0x (406.9 M/s) | 3.5x (463.2
M/s) |
| readFloatsAsDoubles | 3.8x (452.9 M/s) | 3.2x (433.6 M/s) | 3.9x (521.5
M/s) |
| FLBA(12) readBinary | 1.7x (42.3 M/s) | 1.6x (43.6 M/s) | 1.5x (40.9 M/s)
|
Full committed results: [JDK
17](https://github.com/iemejia/spark/actions/runs/27459498495), [JDK
21](https://github.com/iemejia/spark/actions/runs/27459498856), [JDK
25](https://github.com/iemejia/spark/actions/runs/27459499331)
### Does this PR introduce _any_ user-facing change?
No. This is an internal performance optimization. BSS-encoded Parquet
columns that were already readable via the parquet-mr fallback are now decoded
faster through the vectorized path. No API, configuration, or behavioral
changes.
### How was this patch tested?
- **31 unit tests** across 5 test suites in
`ParquetByteStreamSplitEncodingSuite.scala`:
- Abstract base `ParquetByteStreamSplitEncodingSuite[T]` with 7 shared
test cases (roundtrip, nulls, skip, large batches, special values, sequential
reads, mixed skip-read)
- Concrete suites for Int, Long, Float, Double (Float/Double override
`assertEqual` for bitwise NaN-safe comparison)
- Standalone FLBA suite with 3 tests
- **End-to-end round-trip tests** in `ParquetEncodingSuite`:
- BSS encoding for float/double columns (via Spark's config passthrough)
- BSS encoding for all supported types (INT32, INT64, FLOAT, DOUBLE,
FLBA) via `ExampleParquetWriter` with per-column BSS
- **PLAIN-encoded FLBA regression test** (dictionary disabled): writes
FLBA columns of widths 4, 12, and 8 (nullable) with `dictionaryEncoding=false`,
verifies PLAIN encoding in footer metadata, and asserts vectorized reader
round-trip correctness
- **Benchmarks**:
- `VectorizedByteStreamSplitReaderBenchmark.scala` comparing against
parquet-mr per-value readers
- `ParquetVectorUpdaterBenchmark.scala` verifying no regressions in all
updater groups and confirming the `FixedLenByteArrayUpdater` improvement
- All 260 existing + new Parquet tests pass on JDK 17
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenCode (Claude claude-opus-4.6)
Closes #55921 from iemejia/SPARK-56894-byte-stream-split.
Authored-by: Ismaël Mejía <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit d1fc818e941227b270ea29ef1134d50e58a22fdd)
Signed-off-by: Max Gekk <[email protected]>
---
...ParquetVectorUpdaterBenchmark-jdk21-results.txt | 55 ++-
...ParquetVectorUpdaterBenchmark-jdk25-results.txt | 55 ++-
.../ParquetVectorUpdaterBenchmark-results.txt | 55 ++-
...yteStreamSplitReaderBenchmark-jdk21-results.txt | 65 +++
...yteStreamSplitReaderBenchmark-jdk25-results.txt | 65 +++
...rizedByteStreamSplitReaderBenchmark-results.txt | 65 +++
.../parquet/ParquetVectorUpdaterFactory.java | 4 +-
.../VectorizedByteStreamSplitValuesReader.java | 273 ++++++++++++
.../parquet/VectorizedColumnReader.java | 12 +
.../parquet/VectorizedPlainValuesReader.java | 12 +
.../parquet/VectorizedValuesReader.java | 17 +
.../ParquetByteStreamSplitEncodingSuite.scala | 489 +++++++++++++++++++++
.../datasources/parquet/ParquetEncodingSuite.scala | 297 +++++++++++++
.../VectorizedByteStreamSplitReaderBenchmark.scala | 282 ++++++++++++
14 files changed, 1656 insertions(+), 90 deletions(-)
diff --git
a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
index ed39b7f69c7c..79f08a7c511c 100644
--- a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
+++ b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk21-results.txt
@@ -6,14 +6,14 @@ OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Identity Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-BooleanUpdater 0 0
0 16982.4 0.1 1.0X
-ByteUpdater (INT32 -> Byte) 0 0
0 3744.9 0.3 0.2X
-ShortUpdater (INT32 -> Short) 1 1
0 1675.0 0.6 0.1X
-IntegerUpdater 0 0
0 10248.0 0.1 0.6X
-LongUpdater 0 0
0 5141.4 0.2 0.3X
-FloatUpdater 0 0
0 10286.2 0.1 0.6X
-DoubleUpdater 0 0
0 5139.3 0.2 0.3X
-BinaryUpdater 15 15
0 71.1 14.1 0.0X
+BooleanUpdater 0 0
0 16965.6 0.1 1.0X
+ByteUpdater (INT32 -> Byte) 0 0
0 3765.6 0.3 0.2X
+ShortUpdater (INT32 -> Short) 1 1
0 1681.7 0.6 0.1X
+IntegerUpdater 0 0
0 10176.1 0.1 0.6X
+LongUpdater 0 0
0 5079.9 0.2 0.3X
+FloatUpdater 0 0
0 10201.9 0.1 0.6X
+DoubleUpdater 0 0
0 5088.8 0.2 0.3X
+BinaryUpdater 15 15
3 70.6 14.2 0.0X
================================================================================================
@@ -24,12 +24,11 @@ OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Type-converting Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------
-IntegerToLongUpdater 0 0
0 6158.1 0.2 1.0X
-IntegerToDoubleUpdater 0 0
0 6228.1 0.2 1.0X
-FloatToDoubleUpdater 0 0
0 2525.4 0.4 0.4X
-DateToTimestampNTZUpdater 1 1
0 932.9 1.1 0.2X
-LongAsNanosUpdater (TimeType) 1 1
0 1228.5 0.8 0.2X
-DowncastLongUpdater (INT64 -> Decimal(9,2)) 0 0
0 5861.5 0.2 1.0X
+IntegerToLongUpdater 0 0
0 6221.7 0.2 1.0X
+IntegerToDoubleUpdater 0 0
0 6120.9 0.2 1.0X
+FloatToDoubleUpdater 0 0
0 2527.0 0.4 0.4X
+DateToTimestampNTZUpdater 1 1
0 935.1 1.1 0.2X
+DowncastLongUpdater (INT64 -> Decimal(9,2)) 0 0
0 5823.3 0.2 0.9X
================================================================================================
@@ -38,13 +37,11 @@ Rebase Updaters
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
-Rebase Updaters: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------------------
-IntegerWithRebaseUpdater (DATE legacy) 0
0 0 3647.5 0.3 1.0X
-LongWithRebaseUpdater (TIMESTAMP_MICROS legacy) 0
0 0 2668.9 0.4 0.7X
-LongAsMicrosUpdater (TIMESTAMP_MILLIS) 1
1 0 1228.3 0.8 0.3X
-DateToTimestampNTZWithRebaseUpdater (DATE legacy) 1
1 0 797.7 1.3 0.2X
-LongAsMicrosRebaseUpdater (TIMESTAMP_MILLIS legacy) 1
1 0 1099.3 0.9 0.3X
+Rebase Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------
+IntegerWithRebaseUpdater (DATE legacy) 0 0
0 3627.1 0.3 1.0X
+LongWithRebaseUpdater (TIMESTAMP_MICROS legacy) 0 0
0 2267.9 0.4 0.6X
+LongAsMicrosUpdater (TIMESTAMP_MILLIS) 2 3
0 420.5 2.4 0.1X
================================================================================================
@@ -55,8 +52,8 @@ OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Unsigned Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-----------------------------------------------------------------------------------------------------------------------------
-UnsignedIntegerUpdater (UINT32 -> Long) 0 0
0 5894.2 0.2 1.0X
-UnsignedLongUpdater (UINT64 -> Decimal(20,0)) 17 18
1 60.3 16.6 0.0X
+UnsignedIntegerUpdater (UINT32 -> Long) 0 0
0 5902.4 0.2 1.0X
+UnsignedLongUpdater (UINT64 -> Decimal(20,0)) 17 19
1 60.2 16.6 0.0X
================================================================================================
@@ -67,9 +64,9 @@ OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Decimal Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-IntegerToDecimalUpdater 0 0
0 10291.3 0.1 1.0X
-LongToDecimalUpdater 0 0
0 5139.6 0.2 0.5X
-FixedLenByteArrayToDecimalUpdater 21 21
0 49.6 20.2 0.0X
+IntegerToDecimalUpdater 0 0
0 10196.9 0.1 1.0X
+LongToDecimalUpdater 0 0
0 5079.4 0.2 0.5X
+FixedLenByteArrayToDecimalUpdater 21 22
1 49.8 20.1 0.0X
================================================================================================
@@ -80,8 +77,8 @@ OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
FixedLenByteArray Updaters: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------------------
-FixedLenByteArrayUpdater (len=16 -> Binary) 20
21 2 51.9 19.3 1.0X
-FixedLenByteArrayAsIntUpdater (len=4 -> Decimal(9,2)) 7
7 0 160.2 6.2 3.1X
-FixedLenByteArrayAsLongUpdater (len=8 -> Decimal(18,4)) 8
8 0 133.1 7.5 2.6X
+FixedLenByteArrayUpdater (len=16 -> Binary) 13
13 0 80.0 12.5 1.0X
+FixedLenByteArrayAsIntUpdater (len=4 -> Decimal(9,2)) 7
7 1 160.2 6.2 2.0X
+FixedLenByteArrayAsLongUpdater (len=8 -> Decimal(18,4)) 8
8 0 133.3 7.5 1.7X
diff --git
a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
index 16ff18ac2e9d..985b53c7d9fd 100644
--- a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
+++ b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-jdk25-results.txt
@@ -6,14 +6,14 @@ OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Identity Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-BooleanUpdater 0 0
0 17151.8 0.1 1.0X
-ByteUpdater (INT32 -> Byte) 0 0
0 3702.7 0.3 0.2X
-ShortUpdater (INT32 -> Short) 1 1
0 1662.6 0.6 0.1X
-IntegerUpdater 0 0
0 7747.5 0.1 0.5X
-LongUpdater 0 0
0 5099.0 0.2 0.3X
-FloatUpdater 0 0
0 7751.0 0.1 0.5X
-DoubleUpdater 0 0
0 3795.5 0.3 0.2X
-BinaryUpdater 16 16
0 66.4 15.1 0.0X
+BooleanUpdater 0 0
0 17154.9 0.1 1.0X
+ByteUpdater (INT32 -> Byte) 0 0
0 3686.0 0.3 0.2X
+ShortUpdater (INT32 -> Short) 1 1
0 1692.8 0.6 0.1X
+IntegerUpdater 0 0
0 7768.8 0.1 0.5X
+LongUpdater 0 0
0 3881.5 0.3 0.2X
+FloatUpdater 0 0
0 10317.6 0.1 0.6X
+DoubleUpdater 0 0
0 5151.9 0.2 0.3X
+BinaryUpdater 15 16
0 67.8 14.8 0.0X
================================================================================================
@@ -24,12 +24,11 @@ OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Type-converting Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------
-IntegerToLongUpdater 0 0
0 4994.1 0.2 1.0X
-IntegerToDoubleUpdater 0 0
0 6589.1 0.2 1.3X
-FloatToDoubleUpdater 0 0
0 3199.0 0.3 0.6X
-DateToTimestampNTZUpdater 1 1
0 1213.3 0.8 0.2X
-LongAsNanosUpdater (TimeType) 1 1
0 1115.4 0.9 0.2X
-DowncastLongUpdater (INT64 -> Decimal(9,2)) 0 0
0 4930.3 0.2 1.0X
+IntegerToLongUpdater 0 0
0 5113.9 0.2 1.0X
+IntegerToDoubleUpdater 0 0
0 6568.8 0.2 1.3X
+FloatToDoubleUpdater 0 0
0 3189.9 0.3 0.6X
+DateToTimestampNTZUpdater 1 1
0 884.2 1.1 0.2X
+DowncastLongUpdater (INT64 -> Decimal(9,2)) 0 0
0 5089.5 0.2 1.0X
================================================================================================
@@ -38,13 +37,11 @@ Rebase Updaters
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
-Rebase Updaters: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------------------
-IntegerWithRebaseUpdater (DATE legacy) 0
0 0 3665.1 0.3 1.0X
-LongWithRebaseUpdater (TIMESTAMP_MICROS legacy) 0
0 0 2667.6 0.4 0.7X
-LongAsMicrosUpdater (TIMESTAMP_MILLIS) 1
1 0 1228.5 0.8 0.3X
-DateToTimestampNTZWithRebaseUpdater (DATE legacy) 1
1 0 719.8 1.4 0.2X
-LongAsMicrosRebaseUpdater (TIMESTAMP_MILLIS legacy) 1
1 0 1092.7 0.9 0.3X
+Rebase Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------
+IntegerWithRebaseUpdater (DATE legacy) 0 0
0 3670.5 0.3 1.0X
+LongWithRebaseUpdater (TIMESTAMP_MICROS legacy) 0 0
0 2668.5 0.4 0.7X
+LongAsMicrosUpdater (TIMESTAMP_MILLIS) 3 3
0 371.3 2.7 0.1X
================================================================================================
@@ -55,8 +52,8 @@ OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Unsigned Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-----------------------------------------------------------------------------------------------------------------------------
-UnsignedIntegerUpdater (UINT32 -> Long) 0 0
0 4931.9 0.2 1.0X
-UnsignedLongUpdater (UINT64 -> Decimal(20,0)) 17 17
0 60.4 16.6 0.0X
+UnsignedIntegerUpdater (UINT32 -> Long) 0 0
0 6207.3 0.2 1.0X
+UnsignedLongUpdater (UINT64 -> Decimal(20,0)) 17 18
1 60.4 16.6 0.0X
================================================================================================
@@ -67,9 +64,9 @@ OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Decimal Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-IntegerToDecimalUpdater 0 0
0 7752.7 0.1 1.0X
-LongToDecimalUpdater 0 0
0 5144.8 0.2 0.7X
-FixedLenByteArrayToDecimalUpdater 21 21
3 50.7 19.7 0.0X
+IntegerToDecimalUpdater 0 0
0 10266.0 0.1 1.0X
+LongToDecimalUpdater 0 0
0 5149.4 0.2 0.5X
+FixedLenByteArrayToDecimalUpdater 21 21
0 50.8 19.7 0.0X
================================================================================================
@@ -80,8 +77,8 @@ OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
FixedLenByteArray Updaters: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------------------
-FixedLenByteArrayUpdater (len=16 -> Binary) 21
21 1 50.2 19.9 1.0X
-FixedLenByteArrayAsIntUpdater (len=4 -> Decimal(9,2)) 7
7 0 152.6 6.6 3.0X
-FixedLenByteArrayAsLongUpdater (len=8 -> Decimal(18,4)) 8
8 0 127.6 7.8 2.5X
+FixedLenByteArrayUpdater (len=16 -> Binary) 13
13 0 78.7 12.7 1.0X
+FixedLenByteArrayAsIntUpdater (len=4 -> Decimal(9,2)) 7
7 0 152.6 6.6 1.9X
+FixedLenByteArrayAsLongUpdater (len=8 -> Decimal(18,4)) 8
10 4 127.7 7.8 1.6X
diff --git a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
index 58d3ac10aa97..95c02e4dd9e9 100644
--- a/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
+++ b/sql/core/benchmarks/ParquetVectorUpdaterBenchmark-results.txt
@@ -6,14 +6,14 @@ OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Identity Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-BooleanUpdater 0 0
0 15918.4 0.1 1.0X
-ByteUpdater (INT32 -> Byte) 0 0
0 3983.1 0.3 0.3X
-ShortUpdater (INT32 -> Short) 0 1
0 2227.2 0.4 0.1X
-IntegerUpdater 0 0
0 8412.7 0.1 0.5X
-LongUpdater 0 0
0 5077.8 0.2 0.3X
-FloatUpdater 0 0
0 8391.8 0.1 0.5X
-DoubleUpdater 0 0
0 5568.9 0.2 0.3X
-BinaryUpdater 15 16
0 70.8 14.1 0.0X
+BooleanUpdater 0 0
0 14656.4 0.1 1.0X
+ByteUpdater (INT32 -> Byte) 0 0
0 3674.3 0.3 0.3X
+ShortUpdater (INT32 -> Short) 1 1
0 2053.3 0.5 0.1X
+IntegerUpdater 0 0
0 10174.2 0.1 0.7X
+LongUpdater 0 0
0 5112.2 0.2 0.3X
+FloatUpdater 0 0
0 10195.1 0.1 0.7X
+DoubleUpdater 0 0
0 5077.0 0.2 0.3X
+BinaryUpdater 15 15
0 69.2 14.5 0.0X
================================================================================================
@@ -24,12 +24,11 @@ OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Type-converting Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------
-IntegerToLongUpdater 1 1
0 1386.0 0.7 1.0X
-IntegerToDoubleUpdater 1 1
0 1554.5 0.6 1.1X
-FloatToDoubleUpdater 1 1
0 1537.7 0.7 1.1X
-DateToTimestampNTZUpdater 2 2
0 596.9 1.7 0.4X
-LongAsNanosUpdater (TimeType) 1 1
0 942.2 1.1 0.7X
-DowncastLongUpdater (INT64 -> Decimal(9,2)) 1 1
0 1394.2 0.7 1.0X
+IntegerToLongUpdater 1 1
0 1280.1 0.8 1.0X
+IntegerToDoubleUpdater 1 1
0 1556.5 0.6 1.2X
+FloatToDoubleUpdater 1 1
0 1418.2 0.7 1.1X
+DateToTimestampNTZUpdater 2 2
0 605.1 1.7 0.5X
+DowncastLongUpdater (INT64 -> Decimal(9,2)) 1 1
0 1287.2 0.8 1.0X
================================================================================================
@@ -38,13 +37,11 @@ Rebase Updaters
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
-Rebase Updaters: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------------------
-IntegerWithRebaseUpdater (DATE legacy) 0
0 0 2526.8 0.4 1.0X
-LongWithRebaseUpdater (TIMESTAMP_MICROS legacy) 1
1 0 1995.7 0.5 0.8X
-LongAsMicrosUpdater (TIMESTAMP_MILLIS) 1
1 0 1087.7 0.9 0.4X
-DateToTimestampNTZWithRebaseUpdater (DATE legacy) 2
2 0 470.9 2.1 0.2X
-LongAsMicrosRebaseUpdater (TIMESTAMP_MILLIS legacy) 1
1 0 961.7 1.0 0.4X
+Rebase Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------
+IntegerWithRebaseUpdater (DATE legacy) 0 0
0 2596.4 0.4 1.0X
+LongWithRebaseUpdater (TIMESTAMP_MICROS legacy) 1 1
0 2076.8 0.5 0.8X
+LongAsMicrosUpdater (TIMESTAMP_MILLIS) 2 2
0 454.6 2.2 0.2X
================================================================================================
@@ -55,8 +52,8 @@ OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Unsigned Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-----------------------------------------------------------------------------------------------------------------------------
-UnsignedIntegerUpdater (UINT32 -> Long) 1 1
0 1174.9 0.9 1.0X
-UnsignedLongUpdater (UINT64 -> Decimal(20,0)) 17 17
0 63.4 15.8 0.1X
+UnsignedIntegerUpdater (UINT32 -> Long) 1 1
0 1093.1 0.9 1.0X
+UnsignedLongUpdater (UINT64 -> Decimal(20,0)) 18 18
0 59.1 16.9 0.1X
================================================================================================
@@ -67,9 +64,9 @@ OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
Decimal Updaters: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-IntegerToDecimalUpdater 0 0
0 10115.2 0.1 1.0X
-LongToDecimalUpdater 0 0
0 5563.6 0.2 0.6X
-FixedLenByteArrayToDecimalUpdater 20 20
0 53.6 18.6 0.0X
+IntegerToDecimalUpdater 0 0
0 10210.0 0.1 1.0X
+LongToDecimalUpdater 0 0
0 5099.0 0.2 0.5X
+FixedLenByteArrayToDecimalUpdater 20 21
0 51.3 19.5 0.0X
================================================================================================
@@ -80,8 +77,8 @@ OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
FixedLenByteArray Updaters: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------------------
-FixedLenByteArrayUpdater (len=16 -> Binary) 18
19 1 57.7 17.3 1.0X
-FixedLenByteArrayAsIntUpdater (len=4 -> Decimal(9,2)) 6
6 0 173.7 5.8 3.0X
-FixedLenByteArrayAsLongUpdater (len=8 -> Decimal(18,4)) 8
8 0 133.5 7.5 2.3X
+FixedLenByteArrayUpdater (len=16 -> Binary) 13
13 0 80.2 12.5 1.0X
+FixedLenByteArrayAsIntUpdater (len=4 -> Decimal(9,2)) 7
7 0 160.1 6.2 2.0X
+FixedLenByteArrayAsLongUpdater (len=8 -> Decimal(18,4)) 9
9 0 123.2 8.1 1.5X
diff --git
a/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-jdk21-results.txt
b/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-jdk21-results.txt
new file mode 100644
index 000000000000..1ffe8eb1cde6
--- /dev/null
+++
b/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-jdk21-results.txt
@@ -0,0 +1,65 @@
+================================================================================================
+BYTE_STREAM_SPLIT INT32
+================================================================================================
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT INT32: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readIntegers 2 2
0 509.3 2.0 1.0X
+Spark vectorized readIntegersAsLongs 3 3
0 406.9 2.5 0.8X
+Spark vectorized readIntegersAsDoubles 2 2
0 450.2 2.2 0.9X
+parquet-mr readInteger (per-value) 8 8
0 133.8 7.5 0.3X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT INT64
+================================================================================================
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT INT64: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readLongs 7 7
0 159.9 6.3 1.0X
+Spark vectorized readLongsAsInts 7 7
0 157.6 6.3 1.0X
+parquet-mr readLong (per-value) 13 13
1 81.1 12.3 0.5X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT FLOAT
+================================================================================================
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT FLOAT: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readFloats 2 2
0 486.9 2.1 1.0X
+Spark vectorized readFloatsAsDoubles 2 3
0 433.6 2.3 0.9X
+parquet-mr readFloat (per-value) 8 8
0 133.7 7.5 0.3X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT DOUBLE
+================================================================================================
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT DOUBLE: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readDoubles 6 7
1 161.4 6.2 1.0X
+parquet-mr readDouble (per-value) 13 13
0 81.0 12.4 0.5X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT FLBA
+================================================================================================
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT FLBA: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+---------------------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readBinary (width=12) 24
25 3 43.6 22.9 1.0X
+Spark per-value readBinary (old updater path, width=12) 29
29 1 36.2 27.6 0.8X
+parquet-mr readBinary (per-value, width=12) 39
39 1 26.8 37.3 0.6X
+
+
diff --git
a/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-jdk25-results.txt
b/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-jdk25-results.txt
new file mode 100644
index 000000000000..f9f9c6f57f2d
--- /dev/null
+++
b/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-jdk25-results.txt
@@ -0,0 +1,65 @@
+================================================================================================
+BYTE_STREAM_SPLIT INT32
+================================================================================================
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT INT32: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readIntegers 2 2
0 567.2 1.8 1.0X
+Spark vectorized readIntegersAsLongs 2 2
0 463.2 2.2 0.8X
+Spark vectorized readIntegersAsDoubles 2 2
0 465.5 2.1 0.8X
+parquet-mr readInteger (per-value) 8 8
0 134.2 7.5 0.2X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT INT64
+================================================================================================
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT INT64: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readLongs 7 7
0 152.0 6.6 1.0X
+Spark vectorized readLongsAsInts 7 7
0 156.5 6.4 1.0X
+parquet-mr readLong (per-value) 13 13
0 83.6 12.0 0.6X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT FLOAT
+================================================================================================
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT FLOAT: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readFloats 2 2
0 569.1 1.8 1.0X
+Spark vectorized readFloatsAsDoubles 2 2
0 521.5 1.9 0.9X
+parquet-mr readFloat (per-value) 8 8
0 132.8 7.5 0.2X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT DOUBLE
+================================================================================================
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT DOUBLE: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readDoubles 7 7
0 152.1 6.6 1.0X
+parquet-mr readDouble (per-value) 13 13
0 83.3 12.0 0.5X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT FLBA
+================================================================================================
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT FLBA: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+---------------------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readBinary (width=12) 26
26 2 40.9 24.4 1.0X
+Spark per-value readBinary (old updater path, width=12) 27
28 1 38.6 25.9 0.9X
+parquet-mr readBinary (per-value, width=12) 39
39 1 27.1 36.9 0.7X
+
+
diff --git
a/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-results.txt
b/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-results.txt
new file mode 100644
index 000000000000..76cf03c160f5
--- /dev/null
+++ b/sql/core/benchmarks/VectorizedByteStreamSplitReaderBenchmark-results.txt
@@ -0,0 +1,65 @@
+================================================================================================
+BYTE_STREAM_SPLIT INT32
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT INT32: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readIntegers 2 2
0 534.9 1.9 1.0X
+Spark vectorized readIntegersAsLongs 2 2
0 449.9 2.2 0.8X
+Spark vectorized readIntegersAsDoubles 2 3
0 421.8 2.4 0.8X
+parquet-mr readInteger (per-value) 9 9
1 118.1 8.5 0.2X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT INT64
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT INT64: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readLongs 5 5
0 211.5 4.7 1.0X
+Spark vectorized readLongsAsInts 5 5
0 215.2 4.6 1.0X
+parquet-mr readLong (per-value) 14 14
1 76.5 13.1 0.4X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT FLOAT
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT FLOAT: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readFloats 2 2
0 517.8 1.9 1.0X
+Spark vectorized readFloatsAsDoubles 2 2
0 452.9 2.2 0.9X
+parquet-mr readFloat (per-value) 9 9
0 119.3 8.4 0.2X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT DOUBLE
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT DOUBLE: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readDoubles 5 5
0 211.1 4.7 1.0X
+parquet-mr readDouble (per-value) 14 14
0 76.8 13.0 0.4X
+
+
+================================================================================================
+BYTE_STREAM_SPLIT FLBA
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 9V74 80-Core Processor
+BYTE_STREAM_SPLIT FLBA: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+---------------------------------------------------------------------------------------------------------------------------------------
+Spark vectorized readBinary (width=12) 25
28 4 42.3 23.6 1.0X
+Spark per-value readBinary (old updater path, width=12) 26
26 1 40.9 24.4 1.0X
+parquet-mr readBinary (per-value, width=12) 42
43 1 25.0 40.0 0.6X
+
+
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 4fa4183e411c..544a60a21222 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
@@ -1504,9 +1504,7 @@ public class ParquetVectorUpdaterFactory {
int offset,
WritableColumnVector values,
VectorizedValuesReader valuesReader) {
- for (int i = 0; i < total; i++) {
- readValue(offset + i, values, valuesReader);
- }
+ valuesReader.readFixedLenByteArray(total, arrayLen, values, offset);
}
@Override
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java
new file mode 100644
index 000000000000..4ca479ace79d
--- /dev/null
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitValuesReader.java
@@ -0,0 +1,273 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.parquet;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.io.api.Binary;
+
+import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
+
+/**
+ * Vectorized reader for the Parquet BYTE_STREAM_SPLIT encoding.
+ *
+ * <p>BYTE_STREAM_SPLIT de-interleaves the bytes of N fixed-width values into W
+ * separate "streams", one per byte position. For example, N FLOAT values (W=4)
+ * are stored as:
+ * <pre>
+ * stream 0: byte 0 of value 0, byte 0 of value 1, ..., byte 0 of value N-1
+ * stream 1: byte 1 of value 0, byte 1 of value 1, ..., byte 1 of value N-1
+ * stream 2: byte 2 of value 0, byte 2 of value 1, ..., byte 2 of value N-1
+ * stream 3: byte 3 of value 0, byte 3 of value 1, ..., byte 3 of value N-1
+ * </pre>
+ *
+ * <p>This makes each stream highly compressible for time-series and scientific
+ * data (adjacent values share high-order bytes). Decoding gathers the original
+ * bytes back: {@code value[i] = {stream[0][i], stream[1][i], ...,
stream[W-1][i]}}.
+ *
+ * <p>Supports FLOAT (W=4), DOUBLE (W=8), INT32 (W=4), INT64 (W=8), and
+ * FIXED_LEN_BYTE_ARRAY (W=type length).
+ */
+public class VectorizedByteStreamSplitValuesReader
+ extends VectorizedReaderBase {
+
+ /** Width of each value in bytes (4 for FLOAT/INT32, 8 for DOUBLE/INT64). */
+ private final int typeWidth;
+
+ /** Total number of values in the current page. */
+ private int valueCount;
+
+ /** Raw encoded page data: W streams of valueCount bytes each. */
+ private byte[] pageData;
+
+ /** Current read position (number of values consumed so far). */
+ private int offset;
+
+ public VectorizedByteStreamSplitValuesReader(int typeWidth) {
+ this.typeWidth = typeWidth;
+ }
+
+ @Override
+ public void initFromPage(int valueCount, ByteBufferInputStream in) throws
IOException {
+ // For nullable columns the page data section contains only non-null
values,
+ // but valueCount includes nulls. Use the actual bytes available in the
stream
+ // to derive the real number of encoded values.
+ int totalBytes = in.available();
+ this.valueCount = totalBytes / typeWidth;
+ this.offset = 0;
+ this.pageData = new byte[totalBytes];
+ // Read the entire page into pageData. ByteBufferInputStream.slice()
returns a
+ // single ByteBuffer of exactly the requested length, copying if the data
spans
+ // multiple internal buffers.
+ ByteBuffer buf = in.slice(totalBytes);
+ buf.get(pageData, 0, totalBytes);
+ }
+
+ // --------------- helpers ---------------
+
+ /** Assembles a single 4-byte little-endian int from 4 streams at the given
index. */
+ private int assembleInt(int idx) {
+ return (pageData[idx] & 0xFF)
+ | ((pageData[valueCount + idx] & 0xFF) << 8)
+ | ((pageData[2 * valueCount + idx] & 0xFF) << 16)
+ | ((pageData[3 * valueCount + idx] & 0xFF) << 24);
+ }
+
+ /** Assembles a single 8-byte little-endian long from 8 streams at the given
index. */
+ private long assembleLong(int idx) {
+ return (pageData[idx] & 0xFFL)
+ | ((pageData[valueCount + idx] & 0xFFL) << 8)
+ | ((pageData[2 * valueCount + idx] & 0xFFL) << 16)
+ | ((pageData[3 * valueCount + idx] & 0xFFL) << 24)
+ | ((pageData[4 * valueCount + idx] & 0xFFL) << 32)
+ | ((pageData[5 * valueCount + idx] & 0xFFL) << 40)
+ | ((pageData[6 * valueCount + idx] & 0xFFL) << 48)
+ | ((pageData[7 * valueCount + idx] & 0xFFL) << 56);
+ }
+
+ // --------------- single-value reads ---------------
+
+ @Override
+ public byte readByte() {
+ return (byte) readInteger();
+ }
+
+ @Override
+ public short readShort() {
+ return (short) readInteger();
+ }
+
+ @Override
+ public int readInteger() {
+ return assembleInt(offset++);
+ }
+
+ @Override
+ public long readLong() {
+ return assembleLong(offset++);
+ }
+
+ @Override
+ public float readFloat() {
+ return Float.intBitsToFloat(assembleInt(offset++));
+ }
+
+ @Override
+ public double readDouble() {
+ return Double.longBitsToDouble(assembleLong(offset++));
+ }
+
+ @Override
+ public Binary readBinary(int len) {
+ byte[] result = new byte[len];
+ for (int b = 0; b < len; b++) {
+ result[b] = pageData[b * valueCount + offset];
+ }
+ offset++;
+ return Binary.fromConstantByteArray(result);
+ }
+
+ // --------------- batch reads ---------------
+
+ @Override
+ public void readBytes(int total, WritableColumnVector c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putByte(rowId + i, (byte) assembleInt(offset + i));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readShorts(int total, WritableColumnVector c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putShort(rowId + i, (short) assembleInt(offset + i));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readIntegers(int total, WritableColumnVector c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putInt(rowId + i, assembleInt(offset + i));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readIntegersAsLongs(int total, WritableColumnVector c, int
rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putLong(rowId + i, assembleInt(offset + i));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readIntegersAsDoubles(int total, WritableColumnVector c, int
rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putDouble(rowId + i, (double) assembleInt(offset + i));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readLongs(int total, WritableColumnVector c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putLong(rowId + i, assembleLong(offset + i));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readLongsAsInts(int total, WritableColumnVector c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putInt(rowId + i, (int) assembleLong(offset + i));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readFloats(int total, WritableColumnVector c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putFloat(rowId + i, Float.intBitsToFloat(assembleInt(offset + i)));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readFloatsAsDoubles(int total, WritableColumnVector c, int
rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putDouble(rowId + i, (double) Float.intBitsToFloat(assembleInt(offset
+ i)));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readDoubles(int total, WritableColumnVector c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putDouble(rowId + i, Double.longBitsToDouble(assembleLong(offset +
i)));
+ }
+ offset += total;
+ }
+
+ @Override
+ public void readBinary(int total, WritableColumnVector c, int rowId) {
+ // Reuse a single scratch buffer to avoid per-value byte[] + Binary
allocation.
+ byte[] scratch = new byte[typeWidth];
+ for (int i = 0; i < total; i++) {
+ for (int b = 0; b < typeWidth; b++) {
+ scratch[b] = pageData[b * valueCount + offset];
+ }
+ c.putByteArray(rowId + i, scratch, 0, typeWidth);
+ offset++;
+ }
+ }
+
+ @Override
+ public void readFixedLenByteArray(int total, int len, WritableColumnVector
c, int rowId) {
+ readBinary(total, c, rowId);
+ }
+
+ // --------------- skip methods ---------------
+ // All types share the same page layout, so skipping is just advancing the
offset.
+ // skipBooleans is not overridden: BSS never encodes booleans; the base
class throws.
+
+ @Override
+ public void skipBytes(int total) { offset += total; }
+
+ @Override
+ public void skipShorts(int total) { offset += total; }
+
+ @Override
+ public void skipIntegers(int total) { offset += total; }
+
+ @Override
+ public void skipLongs(int total) { offset += total; }
+
+ @Override
+ public void skipFloats(int total) { offset += total; }
+
+ @Override
+ public void skipDoubles(int total) { offset += total; }
+
+ @Override
+ public void skipBinary(int total) { offset += total; }
+
+ @Override
+ public void skipFixedLenByteArray(int total, int len) { offset += total; }
+}
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
index eb7f1bd4d27d..01f4573557dc 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
@@ -375,6 +375,18 @@ public class VectorizedColumnReader {
case DELTA_BYTE_ARRAY -> new VectorizedDeltaByteArrayReader();
case DELTA_LENGTH_BYTE_ARRAY -> new
VectorizedDeltaLengthByteArrayReader();
case DELTA_BINARY_PACKED -> new VectorizedDeltaBinaryPackedReader();
+ case BYTE_STREAM_SPLIT -> {
+ PrimitiveType.PrimitiveTypeName typeName =
+ this.descriptor.getPrimitiveType().getPrimitiveTypeName();
+ int typeWidth = switch (typeName) {
+ case FLOAT, INT32 -> 4;
+ case DOUBLE, INT64 -> 8;
+ case FIXED_LEN_BYTE_ARRAY ->
this.descriptor.getPrimitiveType().getTypeLength();
+ default -> throw new SparkUnsupportedOperationException(
+ "_LEGACY_ERROR_TEMP_3190", Map.of("typeName",
typeName.toString()));
+ };
+ yield new VectorizedByteStreamSplitValuesReader(typeWidth);
+ }
case RLE -> {
PrimitiveType.PrimitiveTypeName typeName =
this.descriptor.getPrimitiveType().getPrimitiveTypeName();
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
index 9249fab7915c..4376d526e662 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
@@ -551,6 +551,18 @@ public class VectorizedPlainValuesReader extends
ValuesReader implements Vectori
}
}
+ @Override
+ public final void readFixedLenByteArray(int total, int len,
WritableColumnVector v, int rowId) {
+ for (int i = 0; i < total; i++) {
+ ByteBuffer buffer = getBuffer(len);
+ if (buffer.hasArray()) {
+ v.putByteArray(rowId + i, buffer.array(), buffer.arrayOffset() +
buffer.position(), len);
+ } else {
+ v.putByteArray(rowId + i, buffer, buffer.position(), len);
+ }
+ }
+ }
+
@Override
public void skipFixedLenByteArray(int total, int len) {
in.skip(total * (long) len);
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java
index c62f7bcec8c3..462757f47f43 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java
@@ -137,6 +137,23 @@ public interface VectorizedValuesReader {
}
void readBinary(int total, WritableColumnVector c, int rowId);
+
+ /**
+ * Reads {@code total} fixed-length byte arrays of exactly {@code len} bytes
each into
+ * {@code c} starting at {@code c[rowId]}. Unlike {@link #readBinary(int,
WritableColumnVector,
+ * int)} which reads length-prefixed variable-length BYTE_ARRAY values, this
method reads
+ * FIXED_LEN_BYTE_ARRAY data where each value is exactly {@code len} raw
bytes with no
+ * length prefix in the encoded stream.
+ *
+ * <p>The default implementation falls back to a per-row loop calling
+ * {@link #readBinary(int)} for each value; subclasses may override for
better performance.
+ */
+ default void readFixedLenByteArray(int total, int len, WritableColumnVector
c, int rowId) {
+ for (int i = 0; i < total; i++) {
+ c.putByteArray(rowId + i, readBinary(len).getBytesUnsafe());
+ }
+ }
+
void readGeometry(int total, WritableColumnVector c, int rowId);
void readGeography(int total, WritableColumnVector c, int rowId);
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala
new file mode 100644
index 000000000000..da07bddc9723
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetByteStreamSplitEncodingSuite.scala
@@ -0,0 +1,489 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.nio.ByteBuffer
+import java.util.Random
+
+import scala.reflect.ClassTag
+
+import org.apache.parquet.bytes.{ByteBufferInputStream,
DirectByteBufferAllocator}
+import org.apache.parquet.column.values.ValuesWriter
+import
org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesWriter._
+import org.apache.parquet.io.api.Binary
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector,
WritableColumnVector}
+import org.apache.spark.sql.types._
+
+/**
+ * Unit tests for [[VectorizedByteStreamSplitValuesReader]].
+ *
+ * Uses parquet-mr's ByteStreamSplitValuesWriter to encode data, then reads it
+ * back with the vectorized reader and verifies correctness. An abstract base
+ * covers the shared test matrix (batch reads, single-value reads, skip, direct
+ * buffers, extreme values) for all numeric types; concrete sub-classes supply
+ * only the type-specific writer/reader/comparison methods. FLBA is tested in a
+ * standalone suite because its reader API differs (readBinary vs typed batch).
+ */
+abstract class ParquetByteStreamSplitEncodingSuite[T: ClassTag] extends
SparkFunSuite {
+
+ protected val random = new Random(42)
+
+ // --- type-specific hooks ---
+
+ protected def typeWidth: Int
+ protected def sparkType: DataType
+
+ /** Encode values with the parquet-mr ByteStreamSplitValuesWriter. */
+ protected def encode(data: Array[T]): Array[Byte]
+
+ /** Batch read from the vectorized reader into a column vector. */
+ protected def readBatch(
+ reader: VectorizedByteStreamSplitValuesReader,
+ total: Int, cv: WritableColumnVector, rowId: Int): Unit
+
+ /** Skip values in the vectorized reader. */
+ protected def skipBatch(
+ reader: VectorizedByteStreamSplitValuesReader, total: Int): Unit
+
+ /** Read a single value from the vectorized reader. */
+ protected def readSingle(reader: VectorizedByteStreamSplitValuesReader): T
+
+ /** Extract a value from the column vector at the given row. */
+ protected def getFromVector(cv: WritableColumnVector, rowId: Int): T
+
+ /** Return a new random value (called repeatedly for random-data tests). */
+ protected def nextRandom: T
+
+ /** Return a deterministic value for index i (for sequential-data tests). */
+ protected def sequentialValue(i: Int): T
+
+ /** Boundary / extreme values to exercise. */
+ protected def extremeValues: Array[T]
+
+ /** A single representative value. */
+ protected def singleTestValue: T
+
+ /** Override for types that need bitwise comparison (Float, Double). */
+ protected def assertEqual(expected: T, actual: T, msg: String): Unit = {
+ assert(expected === actual, msg)
+ }
+
+ // --- shared helpers ---
+
+ private def newReader(
+ page: Array[Byte], count: Int,
+ useDirect: Boolean = false): VectorizedByteStreamSplitValuesReader = {
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ val buf = if (useDirect) {
+ val b = ByteBuffer.allocateDirect(page.length)
+ b.put(page); b.flip(); b
+ } else {
+ ByteBuffer.wrap(page)
+ }
+ reader.initFromPage(count, ByteBufferInputStream.wrap(buf))
+ reader
+ }
+
+ private def readAndVerify(data: Array[T], useDirect: Boolean = false): Unit
= {
+ val page = encode(data)
+ val reader = newReader(page, data.length, useDirect)
+ val cv = new OnHeapColumnVector(data.length, sparkType)
+ try {
+ readBatch(reader, data.length, cv, 0)
+ for (i <- data.indices) {
+ assertEqual(data(i), getFromVector(cv, i), s"mismatch at index $i")
+ }
+ } finally {
+ cv.close()
+ }
+ }
+
+ // --- tests ---
+
+ test("batch read - sequential values") {
+ readAndVerify(Array.tabulate(1000)(i => sequentialValue(i)))
+ }
+
+ test("batch read - random values") {
+ readAndVerify(Array.fill(1000)(nextRandom))
+ }
+
+ test("batch read - extreme values") {
+ readAndVerify(extremeValues)
+ }
+
+ test("batch read - single value") {
+ readAndVerify(Array(singleTestValue))
+ }
+
+ test("batch read - direct byte buffer") {
+ readAndVerify(Array.fill(500)(nextRandom), useDirect = true)
+ }
+
+ test("single-value read") {
+ val data = Array.fill(100)(nextRandom)
+ val reader = newReader(encode(data), data.length)
+ for (i <- data.indices) {
+ assertEqual(data(i), readSingle(reader), s"mismatch at index $i")
+ }
+ }
+
+ test("skip then read") {
+ val data = Array.tabulate(200)(i => sequentialValue(i))
+ val page = encode(data)
+ val reader = newReader(page, data.length)
+ val cv = new OnHeapColumnVector(data.length, sparkType)
+ try {
+ // read 10, skip 20, read 10, skip 50, read remaining 110
+ readBatch(reader, 10, cv, 0)
+ for (i <- 0 until 10) assertEqual(data(i), getFromVector(cv, i),
s"mismatch at $i")
+ skipBatch(reader, 20)
+ readBatch(reader, 10, cv, 10)
+ for (i <- 0 until 10) {
+ assertEqual(data(30 + i), getFromVector(cv, 10 + i), s"mismatch at
${30 + i}")
+ }
+ skipBatch(reader, 50)
+ val remaining = data.length - 90
+ readBatch(reader, remaining, cv, 20)
+ for (i <- 0 until remaining) {
+ assertEqual(data(90 + i), getFromVector(cv, 20 + i), s"mismatch at
${90 + i}")
+ }
+ } finally {
+ cv.close()
+ }
+ }
+}
+
+// --- Concrete suites ---
+
+/** Helper to create a parquet-mr BSS writer, write values, and return the
encoded bytes. */
+private object BssWriterHelper {
+ def encode(writer: ValuesWriter)(writeAll: ValuesWriter => Unit):
Array[Byte] = {
+ writeAll(writer)
+ val bytes = writer.getBytes.toByteArray
+ writer.close()
+ bytes
+ }
+}
+
+class ParquetByteStreamSplitEncodingIntegerSuite
+ extends ParquetByteStreamSplitEncodingSuite[Int] {
+
+ override protected def typeWidth: Int = 4
+ override protected def sparkType: DataType = IntegerType
+
+ override protected def encode(data: Array[Int]): Array[Byte] =
+ BssWriterHelper.encode(
+ new IntegerByteStreamSplitValuesWriter(
+ data.length, data.length * 4, new DirectByteBufferAllocator())
+ )(w => data.foreach(w.writeInteger))
+
+ override protected def readBatch(
+ r: VectorizedByteStreamSplitValuesReader,
+ total: Int, cv: WritableColumnVector, rowId: Int): Unit =
+ r.readIntegers(total, cv, rowId)
+
+ override protected def skipBatch(
+ r: VectorizedByteStreamSplitValuesReader, total: Int): Unit =
+ r.skipIntegers(total)
+
+ override protected def readSingle(r: VectorizedByteStreamSplitValuesReader):
Int =
+ r.readInteger()
+
+ override protected def getFromVector(cv: WritableColumnVector, rowId: Int):
Int =
+ cv.getInt(rowId)
+
+ override protected def nextRandom: Int = random.nextInt()
+ override protected def sequentialValue(i: Int): Int = i * 7
+ override protected def extremeValues: Array[Int] =
+ Array(Int.MinValue, Int.MaxValue, 0, -1, 1)
+ override protected def singleTestValue: Int = 42
+}
+
+class ParquetByteStreamSplitEncodingLongSuite
+ extends ParquetByteStreamSplitEncodingSuite[Long] {
+
+ override protected def typeWidth: Int = 8
+ override protected def sparkType: DataType = LongType
+
+ override protected def encode(data: Array[Long]): Array[Byte] =
+ BssWriterHelper.encode(
+ new LongByteStreamSplitValuesWriter(
+ data.length, data.length * 8, new DirectByteBufferAllocator())
+ )(w => data.foreach(w.writeLong))
+
+ override protected def readBatch(
+ r: VectorizedByteStreamSplitValuesReader,
+ total: Int, cv: WritableColumnVector, rowId: Int): Unit =
+ r.readLongs(total, cv, rowId)
+
+ override protected def skipBatch(
+ r: VectorizedByteStreamSplitValuesReader, total: Int): Unit =
+ r.skipLongs(total)
+
+ override protected def readSingle(r: VectorizedByteStreamSplitValuesReader):
Long =
+ r.readLong()
+
+ override protected def getFromVector(cv: WritableColumnVector, rowId: Int):
Long =
+ cv.getLong(rowId)
+
+ override protected def nextRandom: Long = random.nextLong()
+ override protected def sequentialValue(i: Int): Long = i.toLong * 7
+ override protected def extremeValues: Array[Long] =
+ Array(Long.MinValue, Long.MaxValue, 0L, -1L, 1L)
+ override protected def singleTestValue: Long = 42L
+}
+
+class ParquetByteStreamSplitEncodingFloatSuite
+ extends ParquetByteStreamSplitEncodingSuite[Float] {
+
+ override protected def typeWidth: Int = 4
+ override protected def sparkType: DataType = FloatType
+
+ override protected def encode(data: Array[Float]): Array[Byte] =
+ BssWriterHelper.encode(
+ new FloatByteStreamSplitValuesWriter(
+ data.length, data.length * 4, new DirectByteBufferAllocator())
+ )(w => data.foreach(w.writeFloat))
+
+ override protected def readBatch(
+ r: VectorizedByteStreamSplitValuesReader,
+ total: Int, cv: WritableColumnVector, rowId: Int): Unit =
+ r.readFloats(total, cv, rowId)
+
+ override protected def skipBatch(
+ r: VectorizedByteStreamSplitValuesReader, total: Int): Unit =
+ r.skipFloats(total)
+
+ override protected def readSingle(r: VectorizedByteStreamSplitValuesReader):
Float =
+ r.readFloat()
+
+ override protected def getFromVector(cv: WritableColumnVector, rowId: Int):
Float =
+ cv.getFloat(rowId)
+
+ override protected def nextRandom: Float = random.nextFloat()
+ override protected def sequentialValue(i: Int): Float = i * 0.1f
+ override protected def extremeValues: Array[Float] = Array(
+ 0.0f, -0.0f, Float.MinValue, Float.MaxValue,
+ Float.NaN, Float.PositiveInfinity, Float.NegativeInfinity,
+ 1.0f, -1.0f, Float.MinPositiveValue)
+ override protected def singleTestValue: Float = 3.14f
+
+ override protected def assertEqual(expected: Float, actual: Float, msg:
String): Unit = {
+ assert(java.lang.Float.floatToRawIntBits(expected) ===
+ java.lang.Float.floatToRawIntBits(actual), msg)
+ }
+}
+
+class ParquetByteStreamSplitEncodingDoubleSuite
+ extends ParquetByteStreamSplitEncodingSuite[Double] {
+
+ override protected def typeWidth: Int = 8
+ override protected def sparkType: DataType = DoubleType
+
+ override protected def encode(data: Array[Double]): Array[Byte] =
+ BssWriterHelper.encode(
+ new DoubleByteStreamSplitValuesWriter(
+ data.length, data.length * 8, new DirectByteBufferAllocator())
+ )(w => data.foreach(w.writeDouble))
+
+ override protected def readBatch(
+ r: VectorizedByteStreamSplitValuesReader,
+ total: Int, cv: WritableColumnVector, rowId: Int): Unit =
+ r.readDoubles(total, cv, rowId)
+
+ override protected def skipBatch(
+ r: VectorizedByteStreamSplitValuesReader, total: Int): Unit =
+ r.skipDoubles(total)
+
+ override protected def readSingle(r: VectorizedByteStreamSplitValuesReader):
Double =
+ r.readDouble()
+
+ override protected def getFromVector(cv: WritableColumnVector, rowId: Int):
Double =
+ cv.getDouble(rowId)
+
+ override protected def nextRandom: Double = random.nextDouble()
+ override protected def sequentialValue(i: Int): Double = i * 0.1
+ override protected def extremeValues: Array[Double] = Array(
+ 0.0, -0.0, Double.MinValue, Double.MaxValue,
+ Double.NaN, Double.PositiveInfinity, Double.NegativeInfinity,
+ 1.0, -1.0, Double.MinPositiveValue)
+ override protected def singleTestValue: Double = 3.14159265358979
+
+ override protected def assertEqual(expected: Double, actual: Double, msg:
String): Unit = {
+ assert(java.lang.Double.doubleToRawLongBits(expected) ===
+ java.lang.Double.doubleToRawLongBits(actual), msg)
+ }
+}
+
+class ParquetByteStreamSplitEncodingFLBASuite extends SparkFunSuite {
+ private val random = new Random(42)
+
+ private def writeFLBA(data: Array[Array[Byte]], typeWidth: Int): Array[Byte]
=
+ BssWriterHelper.encode(
+ new FixedLenByteArrayByteStreamSplitValuesWriter(
+ typeWidth, data.length, data.length * typeWidth, new
DirectByteBufferAllocator())
+ )(w => data.foreach(b => w.writeBytes(Binary.fromConstantByteArray(b))))
+
+ private def readAndVerifyFLBA(data: Array[Array[Byte]], typeWidth: Int):
Unit = {
+ val page = writeFLBA(data, typeWidth)
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ reader.initFromPage(data.length,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ for (i <- data.indices) {
+ val actual = reader.readBinary(typeWidth)
+ assert(actual.getBytes.toSeq === data(i).toSeq, s"mismatch at index $i")
+ }
+ }
+
+ test("read FLBA - width 4") {
+ readAndVerifyFLBA(
+ Array.fill(200)(Array.fill(4)(random.nextInt(256).toByte)), typeWidth =
4)
+ }
+
+ test("read FLBA - width 16") {
+ readAndVerifyFLBA(
+ Array.fill(100)(Array.fill(16)(random.nextInt(256).toByte)), typeWidth =
16)
+ }
+
+ // Odd widths exercise the generic assembly loop (not aligned to int/long
boundaries)
+ test("read FLBA - width 2 (float16-sized)") {
+ readAndVerifyFLBA(
+ Array.fill(300)(Array.fill(2)(random.nextInt(256).toByte)), typeWidth =
2)
+ }
+
+ test("read FLBA - width 3") {
+ readAndVerifyFLBA(
+ Array.fill(200)(Array.fill(3)(random.nextInt(256).toByte)), typeWidth =
3)
+ }
+
+ test("read FLBA - width 5") {
+ readAndVerifyFLBA(
+ Array.fill(150)(Array.fill(5)(random.nextInt(256).toByte)), typeWidth =
5)
+ }
+
+ test("read FLBA - width 7") {
+ readAndVerifyFLBA(
+ Array.fill(120)(Array.fill(7)(random.nextInt(256).toByte)), typeWidth =
7)
+ }
+
+ test("read FLBA - width 12 (decimal-sized)") {
+ readAndVerifyFLBA(
+ Array.fill(100)(Array.fill(12)(random.nextInt(256).toByte)), typeWidth =
12)
+ }
+
+ test("skip FLBA") {
+ val typeWidth = 4
+ val data =
Array.fill(100)(Array.fill(typeWidth)(random.nextInt(256).toByte))
+ val page = writeFLBA(data, typeWidth)
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ reader.initFromPage(data.length,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ reader.skipFixedLenByteArray(10, typeWidth)
+ for (i <- 10 until data.length) {
+ val actual = reader.readBinary(typeWidth)
+ assert(actual.getBytes.toSeq === data(i).toSeq, s"mismatch at index $i")
+ }
+ }
+
+ test("skip + read interleaving FLBA - odd width 5") {
+ val typeWidth = 5
+ val data =
Array.fill(200)(Array.fill(typeWidth)(random.nextInt(256).toByte))
+ val page = writeFLBA(data, typeWidth)
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ reader.initFromPage(data.length,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ // read 10, skip 30, read 50, skip 60, read remaining
+ for (i <- 0 until 10) {
+ val actual = reader.readBinary(typeWidth)
+ assert(actual.getBytes.toSeq === data(i).toSeq, s"mismatch at index $i")
+ }
+ reader.skipFixedLenByteArray(30, typeWidth)
+ for (i <- 40 until 90) {
+ val actual = reader.readBinary(typeWidth)
+ assert(actual.getBytes.toSeq === data(i).toSeq, s"mismatch at index $i")
+ }
+ reader.skipFixedLenByteArray(60, typeWidth)
+ for (i <- 150 until data.length) {
+ val actual = reader.readBinary(typeWidth)
+ assert(actual.getBytes.toSeq === data(i).toSeq, s"mismatch at index $i")
+ }
+ }
+
+ test("batch readBinary into WritableColumnVector - width 4") {
+ val typeWidth = 4
+ val data =
Array.fill(200)(Array.fill(typeWidth)(random.nextInt(256).toByte))
+ val page = writeFLBA(data, typeWidth)
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ reader.initFromPage(data.length,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ val cv = new OnHeapColumnVector(data.length, BinaryType)
+ try {
+ reader.readBinary(data.length, cv, 0)
+ for (i <- data.indices) {
+ assert(cv.getBinary(i).toSeq === data(i).toSeq, s"mismatch at index
$i")
+ }
+ } finally {
+ cv.close()
+ }
+ }
+
+ test("batch readBinary into WritableColumnVector - odd width 7") {
+ val typeWidth = 7
+ val data =
Array.fill(120)(Array.fill(typeWidth)(random.nextInt(256).toByte))
+ val page = writeFLBA(data, typeWidth)
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ reader.initFromPage(data.length,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ val cv = new OnHeapColumnVector(data.length, BinaryType)
+ try {
+ reader.readBinary(data.length, cv, 0)
+ for (i <- data.indices) {
+ assert(cv.getBinary(i).toSeq === data(i).toSeq, s"mismatch at index
$i")
+ }
+ } finally {
+ cv.close()
+ }
+ }
+
+ test("batch readBinary with skip interleaving - width 12") {
+ val typeWidth = 12
+ val data =
Array.fill(100)(Array.fill(typeWidth)(random.nextInt(256).toByte))
+ val page = writeFLBA(data, typeWidth)
+ val reader = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ reader.initFromPage(data.length,
ByteBufferInputStream.wrap(ByteBuffer.wrap(page)))
+ val cv = new OnHeapColumnVector(data.length, BinaryType)
+ try {
+ // read first 20 in batch
+ reader.readBinary(20, cv, 0)
+ for (i <- 0 until 20) {
+ assert(cv.getBinary(i).toSeq === data(i).toSeq, s"mismatch at index
$i")
+ }
+ // skip 30
+ reader.skipFixedLenByteArray(30, typeWidth)
+ // read next 50 in batch
+ reader.readBinary(50, cv, 20)
+ for (i <- 0 until 50) {
+ assert(cv.getBinary(20 + i).toSeq === data(50 + i).toSeq, s"mismatch
at index ${50 + i}")
+ }
+ } finally {
+ cv.close()
+ }
+ }
+
+ test("single value FLBA - width 1") {
+ readAndVerifyFLBA(
+ Array(Array(0x42.toByte)), typeWidth = 1)
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
index a2391a3ec22f..9107be919b31 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
@@ -24,7 +24,12 @@ import scala.jdk.CollectionConverters._
import org.apache.hadoop.fs.Path
import org.apache.parquet.column.{Encoding, ParquetProperties}
+import org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0
+import org.apache.parquet.example.data.simple.SimpleGroup
import org.apache.parquet.hadoop.ParquetOutputFormat
+import org.apache.parquet.hadoop.example.ExampleParquetWriter
+import org.apache.parquet.io.api.Binary
+import org.apache.parquet.schema.MessageTypeParser
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.spark.TestUtils
@@ -260,4 +265,296 @@ class ParquetEncodingSuite extends
ParquetCompatibilityTest with SharedSparkSess
}
}
}
+
+ test("BYTE_STREAM_SPLIT encoding round-trip for float and double columns") {
+ // parquet-mr already includes the BYTE_STREAM_SPLIT encoder; Spark's
existing
+ // config passthrough forwards `parquet.enable.bytestreamsplit` to the
writer.
+ // This test verifies the full write-read round-trip through the
vectorized reader.
+ val extraOptions = Map[String, String](
+ ParquetOutputFormat.ENABLE_BYTE_STREAM_SPLIT -> "true",
+ ParquetOutputFormat.ENABLE_DICTIONARY -> "false"
+ )
+
+ val hadoopConf = spark.sessionState.newHadoopConfWithOptions(extraOptions)
+ withMemoryModes { offHeapMode =>
+ withSQLConf(
+ SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offHeapMode,
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
+ ParquetOutputFormat.JOB_SUMMARY_LEVEL -> "ALL") {
+ withTempPath { dir =>
+ val path = s"${dir.getCanonicalPath}/test.parquet"
+ val size = 8193
+ val data = (1 to size).map { i =>
+ Row(i, i.toLong, i.toFloat, i.toDouble,
+ if (i % 3 == 0) null else (i * 0.1f),
+ if (i % 5 == 0) null else (i * 0.01))
+ }
+ val schema = new org.apache.spark.sql.types.StructType()
+ .add("i", org.apache.spark.sql.types.IntegerType)
+ .add("l", org.apache.spark.sql.types.LongType)
+ .add("f", org.apache.spark.sql.types.FloatType)
+ .add("d", org.apache.spark.sql.types.DoubleType)
+ .add("f_nullable", org.apache.spark.sql.types.FloatType, nullable
= true)
+ .add("d_nullable", org.apache.spark.sql.types.DoubleType, nullable
= true)
+
+ spark.createDataFrame(spark.sparkContext.parallelize(data, 1),
schema)
+ .write.options(extraOptions).mode("overwrite").parquet(path)
+
+ val blockMetadata = readFooter(new Path(path),
hadoopConf).getBlocks.asScala.head
+ val columnChunkMetadataList = blockMetadata.getColumns.asScala
+
+ assert(columnChunkMetadataList.length === 6)
+ // INT32 and INT64 columns should NOT use BYTE_STREAM_SPLIT (the
boolean
+ // flag only enables the FLOATING_POINT mode, not the extended
INT32/INT64 mode)
+ assert(
+
!columnChunkMetadataList(0).getEncodings.contains(Encoding.BYTE_STREAM_SPLIT))
+ assert(
+
!columnChunkMetadataList(1).getEncodings.contains(Encoding.BYTE_STREAM_SPLIT))
+ // FLOAT and DOUBLE columns (including nullable ones) should use
BYTE_STREAM_SPLIT
+ assert(
+
columnChunkMetadataList(2).getEncodings.contains(Encoding.BYTE_STREAM_SPLIT))
+ assert(
+
columnChunkMetadataList(3).getEncodings.contains(Encoding.BYTE_STREAM_SPLIT))
+ assert(
+
columnChunkMetadataList(4).getEncodings.contains(Encoding.BYTE_STREAM_SPLIT))
+ assert(
+
columnChunkMetadataList(5).getEncodings.contains(Encoding.BYTE_STREAM_SPLIT))
+
+ // Verify round-trip data correctness through the vectorized reader
+ val actual = spark.read.parquet(path).collect()
+ assert(actual.length === size)
+ val sorted = actual.sortBy(_.getInt(0))
+ (1 to size).foreach { i =>
+ val row = sorted(i - 1)
+ assert(row.getInt(0) === i)
+ assert(row.getLong(1) === i.toLong)
+ assert(row.getFloat(2) === i.toFloat)
+ assert(row.getDouble(3) === i.toDouble)
+ if (i % 3 == 0) {
+ assert(row.isNullAt(4))
+ } else {
+ assert(row.getFloat(4) === i * 0.1f)
+ }
+ if (i % 5 == 0) {
+ assert(row.isNullAt(5))
+ } else {
+ assert(row.getDouble(5) === i * 0.01)
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("BYTE_STREAM_SPLIT encoding round-trip for all supported types") {
+ // The EXTENDED byte-stream-split mode supports INT32, INT64, FLOAT,
DOUBLE, and
+ // FIXED_LEN_BYTE_ARRAY. This test writes a Parquet file programmatically
using the
+ // parquet-mr ExampleParquetWriter with per-column BSS encoding (which
uses EXTENDED
+ // mode) and verifies the vectorized reader correctly decodes all column
types.
+ val schemaStr =
+ """message root {
+ | required int32 int_col;
+ | required int64 long_col;
+ | required float float_col;
+ | required double double_col;
+ | required fixed_len_byte_array(4) flba_col;
+ | optional int32 int_nullable;
+ | optional float float_nullable;
+ |}
+ """.stripMargin
+ val schema = MessageTypeParser.parseMessageType(schemaStr)
+
+ withMemoryModes { offHeapMode =>
+ withSQLConf(
+ SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offHeapMode,
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ withTempDir { dir =>
+ val path = new Path(dir.toURI.toString, "bss_all_types.parquet")
+ val size = 8193
+
+ // Write the file using ExampleParquetWriter with per-column BSS
enabled.
+ // The per-column withByteStreamSplitEncoding(col, true) sets
EXTENDED mode.
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val writer = ExampleParquetWriter.builder(path)
+ .withType(schema)
+ .withDictionaryEncoding(false)
+ .withByteStreamSplitEncoding("int_col", true)
+ .withByteStreamSplitEncoding("long_col", true)
+ .withByteStreamSplitEncoding("float_col", true)
+ .withByteStreamSplitEncoding("double_col", true)
+ .withByteStreamSplitEncoding("flba_col", true)
+ .withByteStreamSplitEncoding("int_nullable", true)
+ .withByteStreamSplitEncoding("float_nullable", true)
+ .withWriterVersion(PARQUET_1_0)
+ .withConf(hadoopConf)
+ .build()
+
+ try {
+ (1 to size).foreach { i =>
+ val record = new SimpleGroup(schema)
+ record.add("int_col", i)
+ record.add("long_col", i.toLong * 100000L)
+ record.add("float_col", i * 0.1f)
+ record.add("double_col", i * 0.001)
+ // FLBA(4): use the 4 bytes of the integer
+ val flbaBytes = Array[Byte](
+ ((i >> 24) & 0xFF).toByte,
+ ((i >> 16) & 0xFF).toByte,
+ ((i >> 8) & 0xFF).toByte,
+ (i & 0xFF).toByte)
+ record.add("flba_col", Binary.fromConstantByteArray(flbaBytes))
+ // Nullable columns: null every 3rd row for int, every 5th for
float
+ if (i % 3 != 0) record.add("int_nullable", i * 7)
+ if (i % 5 != 0) record.add("float_nullable", i * 0.5f)
+ writer.write(record)
+ }
+ } finally {
+ writer.close()
+ }
+
+ // Verify encoding metadata: all columns should use BYTE_STREAM_SPLIT
+ val footer = readAllFootersWithoutSummaryFiles(
+ path.getParent, hadoopConf).head.getParquetMetadata
+ val columnChunks = footer.getBlocks.asScala.head.getColumns.asScala
+ assert(columnChunks.length === 7)
+ columnChunks.foreach { chunk =>
+ assert(chunk.getEncodings.contains(Encoding.BYTE_STREAM_SPLIT),
+ s"Column ${chunk.getPath} should use BYTE_STREAM_SPLIT encoding")
+ }
+
+ // Read back with the vectorized reader and verify data
+ val actual = spark.read.parquet(path.toString).collect()
+ assert(actual.length === size)
+ val sorted = actual.sortBy(_.getInt(0))
+ (1 to size).foreach { i =>
+ val row = sorted(i - 1)
+ assert(row.getInt(0) === i, s"int_col mismatch at i=$i")
+ assert(row.getLong(1) === i.toLong * 100000L, s"long_col mismatch
at i=$i")
+ assert(row.getFloat(2) === i * 0.1f, s"float_col mismatch at i=$i")
+ assert(row.getDouble(3) === i * 0.001, s"double_col mismatch at
i=$i")
+ val expectedFlba = Array[Byte](
+ ((i >> 24) & 0xFF).toByte,
+ ((i >> 16) & 0xFF).toByte,
+ ((i >> 8) & 0xFF).toByte,
+ (i & 0xFF).toByte)
+ assert(row.getAs[Array[Byte]](4) === expectedFlba, s"flba_col
mismatch at i=$i")
+ if (i % 3 == 0) {
+ assert(row.isNullAt(5), s"int_nullable should be null at i=$i")
+ } else {
+ assert(row.getInt(5) === i * 7, s"int_nullable mismatch at i=$i")
+ }
+ if (i % 5 == 0) {
+ assert(row.isNullAt(6), s"float_nullable should be null at i=$i")
+ } else {
+ assert(row.getFloat(6) === i * 0.5f, s"float_nullable mismatch
at i=$i")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("PLAIN-encoded FIXED_LEN_BYTE_ARRAY round-trip (dictionary disabled)") {
+ // Regression test: the FixedLenByteArrayUpdater batch read path must not
use
+ // the length-prefixed readBinary(total, c, rowId) method which is
designed for
+ // variable-length BYTE_ARRAY. FLBA data has no length prefix; each value
is
+ // exactly `arrayLen` raw bytes. This test writes FLBA columns with PLAIN
encoding
+ // (dictionary disabled) and verifies the vectorized reader round-trips
correctly.
+ val schemaStr =
+ """message root {
+ | required fixed_len_byte_array(4) flba4;
+ | required fixed_len_byte_array(12) flba12;
+ | optional fixed_len_byte_array(8) flba8_nullable;
+ |}
+ """.stripMargin
+ val schema = MessageTypeParser.parseMessageType(schemaStr)
+
+ withMemoryModes { offHeapMode =>
+ withSQLConf(
+ SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offHeapMode,
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true") {
+ withTempDir { dir =>
+ val path = new Path(dir.toURI.toString, "plain_flba.parquet")
+ // Use enough rows to span multiple pages and force the batch read
path.
+ val size = 8193
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val writer = ExampleParquetWriter.builder(path)
+ .withType(schema)
+ .withDictionaryEncoding(false)
+ .withWriterVersion(PARQUET_1_0)
+ .withConf(hadoopConf)
+ .build()
+
+ try {
+ (1 to size).foreach { i =>
+ val record = new SimpleGroup(schema)
+ // FLBA(4): big-endian encoding of i
+ val flba4 = Array[Byte](
+ ((i >> 24) & 0xFF).toByte,
+ ((i >> 16) & 0xFF).toByte,
+ ((i >> 8) & 0xFF).toByte,
+ (i & 0xFF).toByte)
+ record.add("flba4", Binary.fromConstantByteArray(flba4))
+ // FLBA(12): repeat the index across 12 bytes (simulates
decimal-sized FLBA)
+ val flba12 = Array.fill(12)(((i % 256)).toByte)
+ flba12(0) = ((i >> 8) & 0xFF).toByte
+ record.add("flba12", Binary.fromConstantByteArray(flba12))
+ // Nullable: null every 4th row
+ if (i % 4 != 0) {
+ val flba8 = Array.tabulate(8)(b => ((i + b) & 0xFF).toByte)
+ record.add("flba8_nullable",
Binary.fromConstantByteArray(flba8))
+ }
+ writer.write(record)
+ }
+ } finally {
+ writer.close()
+ }
+
+ // Verify encoding metadata: columns should use PLAIN (not
dictionary)
+ val footer = readAllFootersWithoutSummaryFiles(
+ path.getParent, hadoopConf).head.getParquetMetadata
+ val columnChunks = footer.getBlocks.asScala.head.getColumns.asScala
+ assert(columnChunks.length === 3)
+ columnChunks.foreach { chunk =>
+ assert(chunk.getEncodings.contains(Encoding.PLAIN),
+ s"Column ${chunk.getPath} should use PLAIN encoding")
+ assert(!chunk.getEncodings.contains(Encoding.PLAIN_DICTIONARY) &&
+ !chunk.getEncodings.contains(Encoding.RLE_DICTIONARY),
+ s"Column ${chunk.getPath} should NOT use dictionary encoding")
+ }
+
+ // Read back with the vectorized reader and verify data
+ val actual = spark.read.parquet(path.toString).collect()
+ assert(actual.length === size)
+ val sorted = actual.sortBy(r =>
java.util.Arrays.hashCode(r.getAs[Array[Byte]](0)))
+ .sortBy(r => {
+ val b = r.getAs[Array[Byte]](0)
+ ((b(0) & 0xFF) << 24) | ((b(1) & 0xFF) << 16) |
+ ((b(2) & 0xFF) << 8) | (b(3) & 0xFF)
+ })
+ (1 to size).foreach { i =>
+ val row = sorted(i - 1)
+ val expectedFlba4 = Array[Byte](
+ ((i >> 24) & 0xFF).toByte,
+ ((i >> 16) & 0xFF).toByte,
+ ((i >> 8) & 0xFF).toByte,
+ (i & 0xFF).toByte)
+ assert(row.getAs[Array[Byte]](0) === expectedFlba4,
+ s"flba4 mismatch at i=$i")
+ val expectedFlba12 = Array.fill(12)(((i % 256)).toByte)
+ expectedFlba12(0) = ((i >> 8) & 0xFF).toByte
+ assert(row.getAs[Array[Byte]](1) === expectedFlba12,
+ s"flba12 mismatch at i=$i")
+ if (i % 4 == 0) {
+ assert(row.isNullAt(2), s"flba8_nullable should be null at i=$i")
+ } else {
+ val expectedFlba8 = Array.tabulate(8)(b => ((i + b) &
0xFF).toByte)
+ assert(row.getAs[Array[Byte]](2) === expectedFlba8,
+ s"flba8_nullable mismatch at i=$i")
+ }
+ }
+ }
+ }
+ }
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitReaderBenchmark.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitReaderBenchmark.scala
new file mode 100644
index 000000000000..d3326efd16aa
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/VectorizedByteStreamSplitReaderBenchmark.scala
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.nio.ByteBuffer
+
+import org.apache.parquet.bytes.{ByteBufferInputStream,
DirectByteBufferAllocator}
+import org.apache.parquet.column.values.ValuesReader
+import org.apache.parquet.column.values.ValuesWriter
+import
org.apache.parquet.column.values.bytestreamsplit.{ByteStreamSplitValuesReaderForDouble,
ByteStreamSplitValuesReaderForFLBA, ByteStreamSplitValuesReaderForFloat,
ByteStreamSplitValuesReaderForInteger, ByteStreamSplitValuesReaderForLong}
+import
org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesWriter._
+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._
+
+/**
+ * Benchmark for the vectorized BYTE_STREAM_SPLIT reader
+ * (`VectorizedByteStreamSplitValuesReader`).
+ *
+ * Compares Spark's vectorized reader (which eagerly loads all page bytes and
+ * assembles values from interleaved streams with direct per-element column
+ * vector stores) against parquet-mr's per-value reader as the non-vectorized
+ * baseline.
+ *
+ * To run this benchmark:
+ * {{{
+ * 1. build/sbt "sql/Test/runMain <this class>"
+ * 2. generate result:
+ * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/Test/runMain <this
class>"
+ * Results in
"benchmarks/VectorizedByteStreamSplitReaderBenchmark-results.txt".
+ * }}}
+ */
+object VectorizedByteStreamSplitReaderBenchmark extends BenchmarkBase {
+
+ private val NUM_ROWS = 1024 * 1024
+ private val NUM_ITERS = 5
+
+ // -------- BSS-encoded data producers --------
+
+ private def bssIntBytes(count: Int)(f: Int => Int): Array[Byte] = {
+ val w: ValuesWriter = new IntegerByteStreamSplitValuesWriter(
+ count, count * 4, new DirectByteBufferAllocator())
+ var i = 0
+ while (i < count) { w.writeInteger(f(i)); i += 1 }
+ val bytes = w.getBytes.toByteArray
+ w.close()
+ bytes
+ }
+
+ private def bssLongBytes(count: Int)(f: Int => Long): Array[Byte] = {
+ val w: ValuesWriter = new LongByteStreamSplitValuesWriter(
+ count, count * 8, new DirectByteBufferAllocator())
+ var i = 0
+ while (i < count) { w.writeLong(f(i)); i += 1 }
+ val bytes = w.getBytes.toByteArray
+ w.close()
+ bytes
+ }
+
+ private def bssFloatBytes(count: Int)(f: Int => Float): Array[Byte] = {
+ val w: ValuesWriter = new FloatByteStreamSplitValuesWriter(
+ count, count * 4, new DirectByteBufferAllocator())
+ var i = 0
+ while (i < count) { w.writeFloat(f(i)); i += 1 }
+ val bytes = w.getBytes.toByteArray
+ w.close()
+ bytes
+ }
+
+ private def bssDoubleBytes(count: Int)(f: Int => Double): Array[Byte] = {
+ val w: ValuesWriter = new DoubleByteStreamSplitValuesWriter(
+ count, count * 8, new DirectByteBufferAllocator())
+ var i = 0
+ while (i < count) { w.writeDouble(f(i)); i += 1 }
+ val bytes = w.getBytes.toByteArray
+ w.close()
+ bytes
+ }
+
+ private def newBssReader(
+ bytes: Array[Byte], typeWidth: Int):
VectorizedByteStreamSplitValuesReader = {
+ val r = new VectorizedByteStreamSplitValuesReader(typeWidth)
+ r.initFromPage(NUM_ROWS,
ByteBufferInputStream.wrap(ByteBuffer.wrap(bytes)))
+ r
+ }
+
+ private def newParquetMrReader(
+ bytes: Array[Byte], reader: ValuesReader): ValuesReader = {
+ reader.initFromPage(NUM_ROWS,
ByteBufferInputStream.wrap(ByteBuffer.wrap(bytes)))
+ reader
+ }
+
+ /** Adds a case that runs `body` after pre-warming the body once. */
+ private def addCase(benchmark: Benchmark, label: String)(body: () => Unit):
Unit = {
+ body()
+ benchmark.addCase(label) { _ => body() }
+ }
+
+ // -------- INT32 --------
+
+ private def runIntegerBenchmark(): Unit = {
+ val benchmark = new Benchmark(
+ "BYTE_STREAM_SPLIT INT32", NUM_ROWS.toLong, NUM_ITERS, output = output)
+
+ val intVec = new OnHeapColumnVector(NUM_ROWS, IntegerType)
+ val longVec = new OnHeapColumnVector(NUM_ROWS, LongType)
+ val doubleVec = new OnHeapColumnVector(NUM_ROWS, DoubleType)
+ val bytes = bssIntBytes(NUM_ROWS)(i => i * 7 + 42)
+
+ addCase(benchmark, "Spark vectorized readIntegers") { () =>
+ newBssReader(bytes, 4).readIntegers(NUM_ROWS, intVec, 0)
+ }
+
+ addCase(benchmark, "Spark vectorized readIntegersAsLongs") { () =>
+ newBssReader(bytes, 4).readIntegersAsLongs(NUM_ROWS, longVec, 0)
+ }
+
+ addCase(benchmark, "Spark vectorized readIntegersAsDoubles") { () =>
+ newBssReader(bytes, 4).readIntegersAsDoubles(NUM_ROWS, doubleVec, 0)
+ }
+
+ addCase(benchmark, "parquet-mr readInteger (per-value)") { () =>
+ val r = newParquetMrReader(bytes, new
ByteStreamSplitValuesReaderForInteger())
+ var i = 0
+ while (i < NUM_ROWS) { intVec.putInt(i, r.readInteger()); i += 1 }
+ }
+
+ benchmark.run()
+ }
+
+ // -------- INT64 --------
+
+ private def runLongBenchmark(): Unit = {
+ val benchmark = new Benchmark(
+ "BYTE_STREAM_SPLIT INT64", NUM_ROWS.toLong, NUM_ITERS, output = output)
+
+ val longVec = new OnHeapColumnVector(NUM_ROWS, LongType)
+ val intVec = new OnHeapColumnVector(NUM_ROWS, IntegerType)
+ val bytes = bssLongBytes(NUM_ROWS)(i => i.toLong * 7 + 42)
+
+ addCase(benchmark, "Spark vectorized readLongs") { () =>
+ newBssReader(bytes, 8).readLongs(NUM_ROWS, longVec, 0)
+ }
+
+ addCase(benchmark, "Spark vectorized readLongsAsInts") { () =>
+ newBssReader(bytes, 8).readLongsAsInts(NUM_ROWS, intVec, 0)
+ }
+
+ addCase(benchmark, "parquet-mr readLong (per-value)") { () =>
+ val r = newParquetMrReader(bytes, new
ByteStreamSplitValuesReaderForLong())
+ var i = 0
+ while (i < NUM_ROWS) { longVec.putLong(i, r.readLong()); i += 1 }
+ }
+
+ benchmark.run()
+ }
+
+ // -------- FLOAT --------
+
+ private def runFloatBenchmark(): Unit = {
+ val benchmark = new Benchmark(
+ "BYTE_STREAM_SPLIT FLOAT", NUM_ROWS.toLong, NUM_ITERS, output = output)
+
+ val floatVec = new OnHeapColumnVector(NUM_ROWS, FloatType)
+ val doubleVec = new OnHeapColumnVector(NUM_ROWS, DoubleType)
+ val bytes = bssFloatBytes(NUM_ROWS)(i => i * 0.1f + 3.14f)
+
+ addCase(benchmark, "Spark vectorized readFloats") { () =>
+ newBssReader(bytes, 4).readFloats(NUM_ROWS, floatVec, 0)
+ }
+
+ addCase(benchmark, "Spark vectorized readFloatsAsDoubles") { () =>
+ newBssReader(bytes, 4).readFloatsAsDoubles(NUM_ROWS, doubleVec, 0)
+ }
+
+ addCase(benchmark, "parquet-mr readFloat (per-value)") { () =>
+ val r = newParquetMrReader(bytes, new
ByteStreamSplitValuesReaderForFloat())
+ var i = 0
+ while (i < NUM_ROWS) { floatVec.putFloat(i, r.readFloat()); i += 1 }
+ }
+
+ benchmark.run()
+ }
+
+ // -------- DOUBLE --------
+
+ private def runDoubleBenchmark(): Unit = {
+ val benchmark = new Benchmark(
+ "BYTE_STREAM_SPLIT DOUBLE", NUM_ROWS.toLong, NUM_ITERS, output = output)
+
+ val doubleVec = new OnHeapColumnVector(NUM_ROWS, DoubleType)
+ val bytes = bssDoubleBytes(NUM_ROWS)(i => i * 0.1 + 3.14)
+
+ addCase(benchmark, "Spark vectorized readDoubles") { () =>
+ newBssReader(bytes, 8).readDoubles(NUM_ROWS, doubleVec, 0)
+ }
+
+ addCase(benchmark, "parquet-mr readDouble (per-value)") { () =>
+ val r = newParquetMrReader(bytes, new
ByteStreamSplitValuesReaderForDouble())
+ var i = 0
+ while (i < NUM_ROWS) { doubleVec.putDouble(i, r.readDouble()); i += 1 }
+ }
+
+ benchmark.run()
+ }
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ runBenchmark("BYTE_STREAM_SPLIT INT32") { runIntegerBenchmark() }
+ runBenchmark("BYTE_STREAM_SPLIT INT64") { runLongBenchmark() }
+ runBenchmark("BYTE_STREAM_SPLIT FLOAT") { runFloatBenchmark() }
+ runBenchmark("BYTE_STREAM_SPLIT DOUBLE") { runDoubleBenchmark() }
+ runBenchmark("BYTE_STREAM_SPLIT FLBA") { runFlbaBenchmark() }
+ }
+
+ // -------- FIXED_LEN_BYTE_ARRAY --------
+
+ private def bssFlbaBytes(count: Int, typeWidth: Int)(f: Int => Array[Byte]):
Array[Byte] = {
+ val w: ValuesWriter = new FixedLenByteArrayByteStreamSplitValuesWriter(
+ typeWidth, count, count * typeWidth, new DirectByteBufferAllocator())
+ var i = 0
+ while (i < count) { w.writeBytes(Binary.fromConstantByteArray(f(i))); i +=
1 }
+ val bytes = w.getBytes.toByteArray
+ w.close()
+ bytes
+ }
+
+ private def runFlbaBenchmark(): Unit = {
+ val typeWidth = 12 // decimal-sized FLBA
+ val benchmark = new Benchmark(
+ "BYTE_STREAM_SPLIT FLBA", NUM_ROWS.toLong, NUM_ITERS, output = output)
+
+ val binaryVec = new OnHeapColumnVector(NUM_ROWS, BinaryType)
+ val rng = new java.util.Random(42)
+ val bytes = bssFlbaBytes(NUM_ROWS, typeWidth) { _ =>
+ val b = new Array[Byte](typeWidth); rng.nextBytes(b); b
+ }
+
+ addCase(benchmark, s"Spark vectorized readBinary (width=$typeWidth)") { ()
=>
+ binaryVec.reset()
+ newBssReader(bytes, typeWidth).readBinary(NUM_ROWS, binaryVec, 0)
+ }
+
+ addCase(benchmark, s"Spark per-value readBinary (old updater path,
width=$typeWidth)") { () =>
+ binaryVec.reset()
+ val r = newBssReader(bytes, typeWidth)
+ var i = 0
+ while (i < NUM_ROWS) {
+ binaryVec.putByteArray(i, r.readBinary(typeWidth).getBytesUnsafe())
+ i += 1
+ }
+ }
+
+ addCase(benchmark, s"parquet-mr readBinary (per-value, width=$typeWidth)")
{ () =>
+ binaryVec.reset()
+ val r = newParquetMrReader(bytes, new
ByteStreamSplitValuesReaderForFLBA(typeWidth))
+ var i = 0
+ while (i < NUM_ROWS) {
+ val v = r.readBytes()
+ binaryVec.putByteArray(i, v.getBytes)
+ i += 1
+ }
+ }
+
+ benchmark.run()
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
