[VL]Optimize driver side build hash table performance - #12765
Open
JkSelf wants to merge 2 commits into
Open
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
Optimizes Velox driver-side broadcast hash table build/serialization by reducing repeated table walks, shrinking broadcast payloads (excluding raw build-side batches), and improving executor-side reuse of deserialized native hash tables.
Changes:
- Increase serialization chunk size and reuse a per-thread staging buffer for off-heap ↔ stream transfers.
- Memoize Velox hash table serialized size in native code to avoid computing it twice during driver-side broadcast.
- Avoid shipping raw build-side batches in the broadcast payload; add driver-side recovery and executor-side handle memoization/cloning.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-arrow/src/main/java/org/apache/spark/sql/execution/unsafe/UnsafeByteArray.java | Larger chunk size + per-thread shared staging buffer for (de)serialization. |
| cpp/velox/operators/hashjoin/HashTableBuilder.h | Adds cached serialized-size state and reset on table replacement. |
| cpp/velox/jni/JniHashTable.cc | Uses cached serialized-size to avoid repeated full table scans. |
| backends-velox/src/main/scala/org/apache/spark/sql/execution/SerializedHashTableBroadcastRelation.scala | Makes raw build-side relation access explicitly driver-only with clearer failure mode on executors. |
| backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala | Tracks serialized hash table in cache entries; adds driver build-side relation lookup and eviction-time release coordination. |
| backends-velox/src/main/scala/org/apache/gluten/execution/SerializedBroadcastHashTable.scala | Excludes raw relation from payload, adds driverRelationId recovery, and memoizes/clones deserialized native handles per JVM. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+196
to
+200
| * `driverSerializedCache` pins the relation on the driver for the lifetime of the broadcast, so | ||
| * this returns a non-null value there even if the broadcast block itself had to be re-read from | ||
| * disk. On executors the cache is always empty and the result is null, which is fine because the | ||
| * raw batches are never consumed there. | ||
| */ |
Comment on lines
67
to
71
| // `buildSideRelation` is intentionally not written. It holds the raw build side batches, which | ||
| // are only consumed on the driver (DPP key extraction through `transform`, and fallback to | ||
| // vanilla Spark through `deserialized`). Executors read the serialized hash table and never | ||
| // touch the raw batches, so shipping both would roughly double the broadcast payload. | ||
| } |
JkSelf
force-pushed
the
hashtable-ser-analysis
branch
from
September 1, 2026 09:32
f042133 to
fb1b2cd
Compare
Contributor
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Pull request overview
Copilot reviewed 10 out of 10 changed files in this pull request and generated 4 comments.
Comment on lines
+277
to
+283
| // The serialized bytes have been fully consumed into the native table. On an executor | ||
| // nothing else refers to them, so free the off-heap copy instead of waiting for the | ||
| // broadcast object to be collected. On the driver the same object is still owned by the | ||
| // broadcast variable and by driverSerializedCache, so it must be left alone. | ||
| if (!isDriver) { | ||
| serialized.releaseSerializedData() | ||
| } |
| @@ -102,12 +120,12 @@ public void read(Kryo kryo, Input input) { | |||
| this.buffer = ArrowBufferAllocators.globalInstance().buffer((int) size); | |||
Comment on lines
+52
to
+59
| /** | ||
| * Returns a scratch buffer of at most {@link #CHUNK_SIZE} bytes, never larger than the payload | ||
| * itself so that relations holding many small batches do not each allocate a full-sized chunk. | ||
| */ | ||
| private byte[] chunkBuf(long dataSize) { | ||
| final int wanted = (int) Math.max(1, Math.min(CHUNK_SIZE, dataSize)); | ||
| if (chunkBuf == null || chunkBuf.length < wanted) { | ||
| chunkBuf = new byte[wanted]; |
|
|
||
| /** Get the size of serialized data in bytes. */ | ||
| def sizeInBytes: Long = serializedData.size() | ||
| def sizeInBytes: Long = if (serializedData == null) 0L else serializedData.size() |
JkSelf
force-pushed
the
hashtable-ser-analysis
branch
from
September 3, 2026 14:43
fb1b2cd to
d170e62
Compare
Contributor
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Pull request overview
Copilot reviewed 15 out of 15 changed files in this pull request and generated 5 comments.
Comment on lines
+100
to
+104
| val relation = serializedHashTable.buildSideRelation | ||
| if (relation == null) { | ||
| throw new IllegalStateException(driverOnlyMessage("transform")) | ||
| } | ||
| relation.transform(key) |
Comment on lines
+1223
to
+1227
| static_cast<bool>(joinHasNullKeys), | ||
| // Deserializing on one thread costs about as much as building the table from the raw build | ||
| // side does on all of them, which would leave the driver-side build with no upside at all. | ||
| // FIXME: This reuses the io executor which is supposed to only serve async IO tasks. | ||
| VeloxBackend::get()->ioExecutor()); |
|
|
||
| /** Get the size of serialized data in bytes. */ | ||
| def sizeInBytes: Long = serializedData.size() | ||
| def sizeInBytes: Long = if (serializedData == null) 0L else serializedData.size() |
Comment on lines
+237
to
+244
| val shared = getOrDeserializeShared( | ||
| serialized, | ||
| broadcastHashTableId, | ||
| deserializeHashTableTimeMetric) | ||
| // Register the shared table under this join's id as well, so that the native probe side | ||
| // can resolve it. The clone holds its own reference to the same native table. | ||
| val hashTableHandle = | ||
| HashJoinBuilder.cloneHashTable(broadcastHashTableId, shared.pointer) |
Comment on lines
+1316
to
+1319
| // Do not recompute serializedHashTableSize() here just to validate 'size': that walks every | ||
| // build row again to re-measure the variable-width columns. serializeTo() bounds-checks each | ||
| // write and asserts that it filled the buffer exactly, which covers the same mistake. | ||
| gluten::serializeHashTableTo(builder, reinterpret_cast<uint8_t*>(address), static_cast<size_t>(size)); |
Contributor
There was a problem hiding this comment.
🔵 Needs a closer look
There is a correctness risk around eagerly freeing serialized broadcast bytes on executors while using time-based cache eviction, which can make later required re-deserialization fail at runtime.
Review details
Suppressed comments (3)
Previously missed (1) — in code that hasn't changed since the last review.
backends-velox/src/main/scala/org/apache/gluten/execution/VeloxBroadcastBuildSideCache.scala:301
serialized.releaseSerializedData()permanently frees the off-heap broadcast bytes in this JVM. BecausesharedDeserializedCacheisexpireAfterAccess, it can evict while the Spark broadcast payload is still live (or configured with a short TTL), which would later require re-deserialization and now fails withIllegalStateException(bytes already released).
gluten-arrow/src/main/java/org/apache/spark/sql/execution/unsafe/UnsafeByteArray.java:55
- The Javadoc claims the scratch buffer is "never larger than the payload itself", but the implementation only grows
chunkBufand never shrinks it. If an instance is reused after handling a large payload, it can keep a buffer larger than subsequent payloads, so the comment is inaccurate.
/**
* Returns a scratch buffer of at most {@link #CHUNK_SIZE} bytes, never larger than the payload
* itself so that relations holding many small batches do not each allocate a full-sized chunk.
*/
cpp/velox/jni/VeloxJniWrapper.cc:1227
deserializeHashTableDirectruns a CPU-heavy deserialization but schedules it onioExecutor(), which this file notes is intended for async IO. This can starve IO tasks and make performance unpredictable under load; prefer the general compute executor (or a dedicated one) for parallel decode.
// Deserializing on one thread costs about as much as building the table from the raw build
// side does on all of them, which would leave the driver-side build with no upside at all.
// FIXME: This reuses the io executor which is supposed to only serve async IO tasks.
VeloxBackend::get()->ioExecutor());
- Files reviewed: 15/15 changed files
- Comments generated: 0 new
- Review effort level: Lite
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes are proposed in this pull request?
How was this patch tested?
Was this patch authored or co-authored using generative AI tooling?