Conversation
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.
Which issue does this PR close?
Part of #6013. This adds reuse for eligible inner joins while tasks on an executor are actively probing the same broadcast.
Rationale for this change
Spark broadcasts the build relation once, but Comet still decodes that relation and constructs a native hash table separately for each probe task. Sharing the broadcast bytes therefore does not share the native preparation work.
For example, consider eight concurrent tasks joining different partitions of
eventsto the same broadcastuserstable. Today, the executor can decodeuserseight times and build eight copies of its hash table. Those tasks could instead probe one immutable build. Switching to DataFusion'sCollectLeftalone does not solve this: it shares preparation within a single join plan, whereas each Comet task has its own plan.What changes are included in this PR?
This PR lets compatible tasks on an executor prepare a broadcast once and share it while they run. The first task opens and decodes the broadcast; other tasks using the same broadcast and build keys borrow the prepared rows and hash table. Each task still runs its own probe input and join condition. A cache hit can skip both decoding and hash-table construction.
The shared build belongs to the executor's storage memory pool, so finishing the task that created it does not invalidate another task's join. Active tasks keep it alive, and the last task releases its memory. The lookup retains only weak references: a later wave of tasks may build the broadcast again. If preparation cannot obtain memory, affected tasks use the ordinary join path with fresh broadcast streams.
Reuse is disabled by default and enabled with
spark.comet.broadcast.reuse.enabled=true. It requires CometPlugin and Spark off-heap memory;spark.comet.broadcast.reuse.maxMemorydefaults to1gper executor and caps the native memory used to prepare and retain shared builds. Existing Spark broadcast and JVM decoder buffers are outside this cap.The initial scope is inner joins with matching direct-column keys and fixed-width or plain UTF-8 build columns. Other joins continue through the existing path. The tuning guide explains the memory scope and metrics for evaluating reuse.
This remains a draft until DataFusion's prepared-build API is available in a compatible release. Dependency pins remain unchanged in this PR.
How are these changes tested?
Validated against a companion port on public DataFusion branch-55, applied locally without committing dependency changes. All 11 native broadcast/cache tests and 38 JVM tests passed: seven memory-manager tests, four prepared-broadcast join tests, the complete 26-test Arrow stream suite, and one JNI lifecycle regression. The concurrent-task case requires exactly one preparation and one cache hit. These tests cover shared ownership, admission fallback, cancellation, executor retirement, and replay through Spark-compatible decoding. The JNI regression fails with the original execution-path selection and verifies that lazy broadcasts stay on the Spark task thread, preserving task context and synchronous native-reader cleanup.
The full Spark 4.1.3 Maven reactor compiled successfully, with Spotless and Scalastyle passing. Integration tests used the rebuilt native library, and the packaged library hash matched. Rust formatting, focused native Clippy with warnings denied, and diff checks passed. Comet's released DataFusion dependency does not yet contain the prepared-build APIs, so native CI remains blocked until the dependency is updated.
Broader Spark/AQE validation and benchmarks comparing reuse enabled and disabled remain outstanding. No end-to-end speedup is claimed.