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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
