You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@datafusion.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2026/09/19 04:29:20 UTC

[PR] perf: reuse prepared broadcast builds across executor tasks [datafusion-comet]

sunchao opened a new pull request, #6037:
URL: https://github.com/apache/datafusion-comet/pull/6037

## Which issue does this PR close?

Part of #6013. This first step supports eligible inner joins and shares builds while tasks are actively probing; later task waves may prepare the same broadcast again.

## Rationale for this change

Comet currently decodes the same broadcast and builds a native hash table in each probe task. An executor can avoid that repeated work by sharing the immutable build rows and table while keeping probe streams, residual predicates, dynamic filters and metrics task-local.

This is a draft pending [the companion DataFusion prepared-build API](https://github.com/apache/datafusion/pull/25491) and a compatible published release. The committed dependency pins remain public and unchanged, so the current released DataFusion dependency does not yet compile this integration.

## What changes are included in this PR?

- Add opt-in `spark.comet.broadcast.reuse.enabled` and a `spark.comet.broadcast.reuse.maxMemory` cap (default `1g`). Reuse requires CometPlugin and Spark off-heap memory.
- Carry the actual Spark broadcast ID into native execution without eagerly decoding it. A bounded executor lookup elects one loader and shares its immutable prepared build with compatible concurrent tasks.
- Keep only weak references in completed lookup entries. Active probes own the build; the last lease releases its storage charge. Executor generation changes retire lookups and isolate late releases.
- Charge preparation and retained native memory to executor storage, including DataFusion's temporary input/compaction overlap. Admission failure opens a fresh ordinary input. Driver broadcast storage and JVM decoding remain outside this cap.
- Support inner joins with direct matching-type column keys and fixed-width/plain UTF8 build columns. Unsupported joins retain task-local execution.
- Preserve Spark's invalid-UTF8 decoding on uncached replay. Inputs needing string repair decline caching before native-copy allocation.
- Expose preparation, hit and admission-fallback metrics, and document the lifecycle and limits.

The feature is disabled by default. No whole-query speedup is claimed; matching enabled/disabled benchmarks remain required before enabling it broadly.

## How are these changes tested?

Validation uses the public Apache Comet base and [the companion API port on public DataFusion branch-55](https://github.com/sunchao/arrow-datafusion/commit/0d00612896922328040c065ea0411da18ff0b191) through a temporary local Cargo patch. No private registry dependencies or local paths are committed.

- Full Spark 4.1.3 Maven reactor production/test compilation passed.
- All 8 memory-manager tests passed, including shared admission, task cleanup and executor retirement.
- Full Spotless/Scalastyle checks and Rust formatting passed.
- Native build and all 13 broadcast/cache tests passed, including cancellation, retirement, invalid-UTF8 replay, and compact-build memory admission.
- The four Spark join tests and Arrow stream suite are pending a rebuild of the matching native library with the final prerequisite changes; their results will be added before marking this ready.

Without the temporary prerequisite patch, native checking reaches exactly the three missing prepared-build API symbols. This PR is not ready to merge until the dependency is available and CI plus broader Spark/AQE validation and performance checks pass.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] perf: reuse prepared broadcast builds across executor tasks [datafusion-comet]

Posted by "sunchao (via GitHub)" <gi...@apache.org>.

sunchao commented on PR #6037:
URL: https://github.com/apache/datafusion-comet/pull/6037#issuecomment-6003686979

   Rebased and fixed the native dependency blocker in `2b94d426d107187d4ade50c3fd88d557396b3bf9`. The committed manifest now pins the existing public Arrow 59-compatible companion `c9a142d834b6b47c2ac15524171bbf84f3022f73` across the DataFusion 55.1 crate family. That companion is only two commits ahead of official 55.1, with eight relevant files changed. No Arrow or other dependency versions change, and all DataFusion types come from one source. The temporary pin can be removed after an official compatible release is adopted.
   
   The native build and all 12 focused native tests pass; all five Spark 3.5 broadcast/lifecycle runtime tests pass with the matching library. Strict full-reactor Scala compilation and formatting/style pass. Broader CI is pending, and draft status is preserved.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] perf: reuse prepared broadcast builds across executor tasks [datafusion-comet]

Posted by "sunchao (via GitHub)" <gi...@apache.org>.

sunchao commented on PR #6037:
URL: https://github.com/apache/datafusion-comet/pull/6037#issuecomment-5982693538

Rebased in `7a930fa205fe0076de859a1fd950da991c4d94ee`. Adapted to main's ArrowArrayStreamReader, kept the upstream BatchProducer shutdown and plugin lifecycle cleanup, and registered the broadcast lifecycle suite in both CI matrices.

Native core/test compilation passes with the same validation-only public DataFusion companion (`c9a142d834b6b47c2ac15524171bbf84f3022f73`), and full Spark 4.1 reactor test compilation, Rust formatting, Spotless, suite registration and whitespace checks pass. Dependency pins and the lockfile remain unchanged.

The existing draft blocker remains: Comet's released DataFusion55.1 dependency does not contain the prepared-build APIs, even though upstream #25491 is merged. Runtime tests/benchmarks were not rerun on this rebased head, and ordinary native CI cannot pass until that dependency prerequisite is available.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] perf: reuse prepared broadcast builds across executor tasks [datafusion-comet]

Posted by "comphead (via GitHub)" <gi...@apache.org>.

comphead commented on PR #6037:
URL: https://github.com/apache/datafusion-comet/pull/6037#issuecomment-6354315871

   @sunchao are you still on this task? 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] perf: reuse prepared broadcast builds across executor tasks [datafusion-comet]

Posted by "comphead (via GitHub)" <gi...@apache.org>.

comphead commented on PR #6037:
URL: https://github.com/apache/datafusion-comet/pull/6037#issuecomment-6354309007

   this might be interesting


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] perf: reuse prepared broadcast builds across executor tasks [datafusion-comet]

Posted by "sunchao (via GitHub)" <gi...@apache.org>.

sunchao commented on PR #6037:
URL: https://github.com/apache/datafusion-comet/pull/6037#issuecomment-5982835074

   Cleared the Scala 2.12 strict-warning failures in `137241bd8b5e3f8500480204f943b640ff1594c1`: renamed the broadcast RDD constructor parameter to avoid shadowing `RDD.name` and made the join fixture numeric conversion explicit. The complete Spark 3.5 reactor now passes `test-compile -Pstrict-warnings -DskipTests`, including Spotless/Scalastyle. The documented DataFusion prepared-build API dependency remains the native CI blocker for this draft.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org