Skip to content

perf: reuse prepared broadcast builds across executor tasks - #6037

Draft
sunchao wants to merge 3 commits into
apache:mainfrom
sunchao:codex/oss-broadcast-build-reuse-6013
Draft

sunchao wants to merge 3 commits into
apache:mainfrom
sunchao:codex/oss-broadcast-build-reuse-6013

Conversation

@sunchao

@sunchao sunchao commented Sep 19, 2026

Copy link
Copy Markdown
Member

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 events to the same broadcast users table. Today, the executor can decode users eight times and build eight copies of its hash table. Those tasks could instead probe one immutable build. Switching to DataFusion's CollectLeft alone 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.maxMemory defaults to 1g per 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.

@github-actions github-actions Bot added enhancement New feature or request performance area:scan Parquet scan / data reading area:memory Memory pools, reservations, OOM handling area:joins Join operators and dynamic filter pushdown labels Sep 19, 2026
@sunchao sunchao added the run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue label Sep 19, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:joins Join operators and dynamic filter pushdown area:memory Memory pools, reservations, OOM handling area:scan Parquet scan / data reading enhancement New feature or request performance run-spark-4.1-tests Run the Spark 4.1 SQL tests on this pull request instead of waiting for the merge queue

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant