You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/03/16 21:36:54 UTC
(spark) branch master updated: [SPARK-55885][SQL] Optimize vectorized Parquet boolean reading with lookup-table expansion and batch buffer reads
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 124d0a9ad306 [SPARK-55885][SQL] Optimize vectorized Parquet boolean reading with lookup-table expansion and batch buffer reads
124d0a9ad306 is described below
commit 124d0a9ad306d8dc90720d74d0e306ff0f527185
Author: yangjie01 <ya...@baidu.com>
AuthorDate: Mon Mar 16 14:36:45 2026 -0700
[SPARK-55885][SQL] Optimize vectorized Parquet boolean reading with lookup-table expansion and batch buffer reads
### What changes were proposed in this pull request?
This PR optimizes the vectorized Parquet plain boolean reading path in two ways:
1. **Lookup-table-based bit expansion**: Replace 8 individual byte writes per packed boolean byte with a single 64-bit `Platform.putLong` write, using a precomputed 256-entry lookup table (`BOOL_BYTE_TO_LONG_TABLE`) that expands each bit into a separate byte within a `long`. Big-endian platforms are handled via `Long.reverseBytes()`.
2. **Batch buffer reads**: Replace per-byte `in.read()` calls in `VectorizedPlainValuesReader.readBooleans` with a single `getBuffer(fullBytes)` call (backed by `ByteBufferInputStream.slice`), reducing I/O overhead from ~N/8 individual read calls to one bulk acquisition per batch.
### Why are the changes needed?
In the current implementation, reading a batch of N boolean values from Parquet requires:
- N/8 individual `in.read()` calls (each going through `ByteBufferInputStream`)
- 8 individual byte writes per packed byte (either array stores or `Platform.putByte` calls)
For a typical batch size of 4096, this means ~512 `in.read()` calls and ~4096 individual byte writes. The optimized path reduces this to:
- 1 `getBuffer()` call for the entire batch
- ~512 single `Platform.putLong` writes (one per packed byte)
This is a meaningful improvement on the hot path of Parquet boolean column scanning.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- Add Some new tests in `ColumnVectorSuite`
- A micro-benchmark that simulates changes with JMH to verify the performance improvement.
<details>
<summary><b>Benchmark Code (click to expand)</b></summary>
```java
package org.apache.spark.sql.execution.datasources.parquet;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import org.apache.parquet.bytes.ByteBufferInputStream;
import org.apache.spark.sql.execution.vectorized.OffHeapColumnVector;
import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
import org.apache.spark.sql.types.DataTypes;
import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;
/**
* JMH benchmark comparing old (baseline) vs new (optimized) VectorizedPlainValuesReader.
*
* Run via:
* ./build/sbt "sql/Test/runMain \
* org.apache.spark.sql.execution.datasources.parquet \
* .VectorizedPlainValuesReaderBenchmark"
*/
BenchmarkMode(Mode.AverageTime)
OutputTimeUnit(TimeUnit.NANOSECONDS)
Warmup(iterations = 10, time = 2)
Measurement(iterations = 100, time = 2)
Fork(value = 2, jvmArgsAppend = {"-Xmx4g", "-Xms4g"})
State(Scope.Thread)
public class VectorizedPlainValuesReaderBenchmark {
private static final int BATCH_SIZE = 4096;
private Random random;
// --- Heap buffer data for each type ---
private byte[] booleanData;
// --- Column vectors ---
private WritableColumnVector booleanColumn;
Param({"OnHeap", "OffHeap"})
public String vectorType;
// --- Readers ---
private OldVectorizedPlainValuesReader oldReader;
private VectorizedPlainValuesReader newReader;
Setup(Level.Trial)
public void setup() {
random = new Random(42);
// Boolean data: packed bits, ceil(BATCH_SIZE / 8) bytes
int booleanBytes = (BATCH_SIZE + 7) / 8;
booleanData = new byte[booleanBytes];
random.nextBytes(booleanData);
// Allocate column vectors
if ("OffHeap".equalsIgnoreCase(vectorType)) {
booleanColumn = new OffHeapColumnVector(BATCH_SIZE, DataTypes.BooleanType);
} else {
booleanColumn = new OnHeapColumnVector(BATCH_SIZE, DataTypes.BooleanType);
}
}
TearDown(Level.Trial)
public void tearDown() {
booleanColumn.close();
}
Setup(Level.Invocation)
public void prepareInvocation() throws IOException {
booleanColumn.reset();
oldReader = initOld(wrapHeap(booleanData));
newReader = initNew(wrapHeap(booleanData));
}
// --- Helper methods ---
private ByteBufferInputStream wrapHeap(byte[] data) {
ByteBuffer buf = ByteBuffer.wrap(data);
return ByteBufferInputStream.wrap(buf);
}
private OldVectorizedPlainValuesReader initOld(ByteBufferInputStream stream) throws IOException {
OldVectorizedPlainValuesReader reader = new OldVectorizedPlainValuesReader();
reader.initFromPage(BATCH_SIZE, stream);
return reader;
}
private VectorizedPlainValuesReader initNew(ByteBufferInputStream stream) throws IOException {
VectorizedPlainValuesReader reader = new VectorizedPlainValuesReader();
reader.initFromPage(BATCH_SIZE, stream);
return reader;
}
// ==================== Boolean benchmarks ====================
Benchmark
public void readBooleans_old() throws IOException {
oldReader.readBooleans(BATCH_SIZE, booleanColumn, 0);
}
Benchmark
public void readBooleans_new() throws IOException {
newReader.readBooleans(BATCH_SIZE, booleanColumn, 0);
}
// ==================== Main entry point ====================
public static void main(String[] args) throws Exception {
Options opt = new OptionsBuilder()
.include(VectorizedPlainValuesReaderBenchmark.class.getSimpleName())
.build();
new Runner(opt).run();
}
}
```
</details>
Perform `build/sbt "sql/Test/runMain org.apache.spark.sql.execution.datasources.parquet.VectorizedPlainValuesReaderBenchmark"` to conduct the test
**Benchmark results:**
- Java 17.0.18
```
[info] Benchmark (vectorType) Mode Cnt Score Error Units
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_new OnHeap avgt 200 280.158 ± 10.135 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_new OffHeap avgt 200 264.486 ± 5.720 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_old OnHeap avgt 200 999.699 ± 0.719 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_old OffHeap avgt 200 1152.287 ± 0.407 ns/op
```
- Java 21.0.10
```
[info] Benchmark (vectorType) Mode Cnt Score Error Units
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_new OnHeap avgt 200 284.184 ± 0.557 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_new OffHeap avgt 200 262.629 ± 5.199 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_old OnHeap avgt 200 979.432 ± 0.708 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_old OffHeap avgt 200 991.920 ± 1.109 ns/op
```
- Java 25
```
[info] Benchmark (vectorType) Mode Cnt Score Error Units
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_new OnHeap avgt 200 300.721 ± 0.361 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_new OffHeap avgt 200 326.190 ± 1.530 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_old OnHeap avgt 200 993.619 ± 1.023 ns/op
[info] VectorizedPlainValuesReaderBenchmark.readBooleans_old OffHeap avgt 200 1153.824 ± 1.005 ns/op
```
From the perspective of micro-benchmarking, the new solution demonstrates over a 3-fold performance improvement in terms of latency.
### Was this patch authored or co-authored using generative AI tooling?
Assisted by Sonnet 4.6
Closes #54818 from LuciferYang/SPARK-55885.
Authored-by: yangjie01 <ya...@baidu.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../parquet/VectorizedPlainValuesReader.java | 20 ++++++++--
.../execution/vectorized/OffHeapColumnVector.java | 15 ++++----
.../execution/vectorized/OnHeapColumnVector.java | 15 ++++----
.../execution/vectorized/WritableColumnVector.java | 30 ++++++++++++++-
.../execution/vectorized/ColumnVectorSuite.scala | 43 ++++++++++++++++++++++
5 files changed, 103 insertions(+), 20 deletions(-)
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 852d02c2cdf1..a62b773c7f4f 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
@@ -72,9 +72,23 @@ public class VectorizedPlainValuesReader extends ValuesReader implements Vectori
c.putBooleans(rowId, i, currentByte, bitOffset);
bitOffset = (bitOffset + i) & 7;
}
- for (; i + 7 < total; i += 8) {
- updateCurrentByte();
- c.putBooleans(rowId + i, currentByte);
+ // Batch-read all full bytes in a single getBuffer call instead of per-byte in.read().
+ // getBuffer returns a slice with position=0 and remaining=fullBytes.
+ int fullBytes = (total - i) / 8;
+ if (fullBytes > 0) {
+ ByteBuffer buffer = getBuffer(fullBytes);
+ if (buffer.hasArray()) {
+ byte[] array = buffer.array();
+ int offset = buffer.arrayOffset() + buffer.position();
+ for (int j = 0; j < fullBytes; j++) {
+ c.putBooleans(rowId + i + j * 8, array[offset + j]);
+ }
+ } else {
+ for (int j = 0; j < fullBytes; j++) {
+ c.putBooleans(rowId + i + j * 8, buffer.get());
+ }
+ }
+ i += fullBytes * 8;
}
if (i < total) {
updateCurrentByte();
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
index 86a0365ead0c..c8554fb1b8fb 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
@@ -159,14 +159,13 @@ public final class OffHeapColumnVector extends WritableColumnVector {
@Override
public void putBooleans(int rowId, byte src) {
- Platform.putByte(null, data + rowId, (byte)(src & 1));
- Platform.putByte(null, data + rowId + 1, (byte)(src >>> 1 & 1));
- Platform.putByte(null, data + rowId + 2, (byte)(src >>> 2 & 1));
- Platform.putByte(null, data + rowId + 3, (byte)(src >>> 3 & 1));
- Platform.putByte(null, data + rowId + 4, (byte)(src >>> 4 & 1));
- Platform.putByte(null, data + rowId + 5, (byte)(src >>> 5 & 1));
- Platform.putByte(null, data + rowId + 6, (byte)(src >>> 6 & 1));
- Platform.putByte(null, data + rowId + 7, (byte)(src >>> 7 & 1));
+ assert rowId + 8 <= capacity :
+ "putBooleans requires 8 slots available at rowId=" + rowId + ", capacity=" + capacity;
+ long expanded = expandBoolByteToLong(src);
+ if (bigEndianPlatform) {
+ expanded = Long.reverseBytes(expanded);
+ }
+ Platform.putLong(null, data + rowId, expanded);
}
@Override
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
index 9084eec9ffc2..ea516871584f 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
@@ -154,14 +154,13 @@ public final class OnHeapColumnVector extends WritableColumnVector {
@Override
public void putBooleans(int rowId, byte src) {
- byteData[rowId] = (byte)(src & 1);
- byteData[rowId + 1] = (byte)(src >>> 1 & 1);
- byteData[rowId + 2] = (byte)(src >>> 2 & 1);
- byteData[rowId + 3] = (byte)(src >>> 3 & 1);
- byteData[rowId + 4] = (byte)(src >>> 4 & 1);
- byteData[rowId + 5] = (byte)(src >>> 5 & 1);
- byteData[rowId + 6] = (byte)(src >>> 6 & 1);
- byteData[rowId + 7] = (byte)(src >>> 7 & 1);
+ assert rowId + 8 <= capacity :
+ "putBooleans requires 8 slots available at rowId=" + rowId + ", capacity=" + capacity;
+ long expanded = expandBoolByteToLong(src);
+ if (bigEndianPlatform) {
+ expanded = Long.reverseBytes(expanded);
+ }
+ Platform.putLong(byteData, Platform.BYTE_ARRAY_OFFSET + rowId, expanded);
}
@Override
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
index 7ce4139df47b..76d9d0723911 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
@@ -53,6 +53,23 @@ import org.apache.spark.unsafe.types.UTF8String;
public abstract class WritableColumnVector extends ColumnVector {
private final byte[] byte8 = new byte[8];
+ // Lookup table that "expands" each of the 8 bits in a byte into a separate byte within a long.
+ // Bit k of the input byte becomes byte k of the output long, in little-endian memory order.
+ // For example: input 0xAA (10101010) -> output 0x0100010001000100L,
+ // which when stored on a little-endian machine writes bytes [0,1,0,1,0,1,0,1] in memory order.
+ // On big-endian platforms, callers must apply Long.reverseBytes() before writing to memory.
+ private static final long[] BOOL_BYTE_TO_LONG_TABLE = new long[256];
+
+ static {
+ for (int i = 0; i < 256; i++) {
+ long r = 0L;
+ for (int bit = 0; bit < 8; bit++) {
+ r |= ((long)((i >>> bit) & 1)) << (bit * 8);
+ }
+ BOOL_BYTE_TO_LONG_TABLE[i] = r;
+ }
+ }
+
protected abstract void releaseMemory();
/**
@@ -246,8 +263,11 @@ public abstract class WritableColumnVector extends ColumnVector {
}
/**
- * Sets bits from [src[0], src[7]] to [rowId, rowId + 7]
+ * Sets bits from [src[0], src[7]] to [rowId, rowId + 7].
* src must contain bit-packed 8 booleans in the byte.
+ *
+ * Caller must ensure rowId + 8 <= capacity, as implementations may use
+ * Unsafe operations that bypass array bounds checking.
*/
public abstract void putBooleans(int rowId, byte src);
@@ -966,6 +986,14 @@ public abstract class WritableColumnVector extends ColumnVector {
DecimalType.isByteArrayDecimalType(type);
}
+ /**
+ * Expands each bit of a bit-packed boolean byte into a separate byte within a long
+ * (little-endian layout). See {@link #BOOL_BYTE_TO_LONG_TABLE} for the byte-order convention.
+ */
+ protected long expandBoolByteToLong(byte b) {
+ return BOOL_BYTE_TO_LONG_TABLE[b & 0xFF];
+ }
+
/**
* Sets up the common state and also handles creating the child columns if this is a nested
* type.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
index 3be6c14eb4d3..2dbf14d3275d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
@@ -1010,4 +1010,47 @@ class ColumnVectorSuite extends SparkFunSuite with SQLHelper {
assert(!row.isNullAt(2))
assert(row.get(2, TimestampNTZType) === 1000L)
}
+
+ testVectors("putBooleans(byte)", 96, BooleanType) { testVector =>
+ // Test various bit patterns at aligned offsets
+ val patterns = Seq(
+ 0x00.toByte,
+ 0xFF.toByte,
+ 0xAA.toByte,
+ 0x55.toByte,
+ 0x80.toByte,
+ 0x01.toByte
+ )
+
+ patterns.zipWithIndex.foreach { case (pattern, idx) =>
+ val rowId = idx * 8
+ testVector.putBooleans(rowId, pattern)
+ (0 until 8).foreach { i =>
+ val expected = ((pattern & 0xFF) >>> i & 1) == 1
+ assert(testVector.getBoolean(rowId + i) === expected)
+ }
+ }
+
+ // Verify writes at different offsets don't corrupt adjacent data.
+ // Note: the loop above writes 6 patterns at rowId 0..40, so rowId 48+ is untouched.
+ testVector.putBooleans(48, 0xFF.toByte) // fill slots 48-55 with true
+ testVector.putBooleans(56, 0xAA.toByte) // write at offset 56
+ // Verify slots 48-55 are untouched
+ (0 until 8).foreach { i =>
+ assert(testVector.getBoolean(48 + i) === true,
+ s"slot ${48 + i} should still be true after writing at rowId=56")
+ }
+ // Verify offset-56 write is correct
+ (0 until 8).foreach { i =>
+ val expected = ((0xAA & 0xFF) >>> i & 1) == 1
+ assert(testVector.getBoolean(56 + i) === expected)
+ }
+
+ // Write at capacity boundary (last 8 slots: rowId=88, capacity=96)
+ testVector.putBooleans(88, 0x55.toByte)
+ (0 until 8).foreach { i =>
+ val expected = ((0x55 & 0xFF) >>> i & 1) == 1
+ assert(testVector.getBoolean(88 + i) === expected)
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org