andygrove opened a new issue, #6013: URL: https://github.com/apache/datafusion-comet/issues/6013
### What is the problem the feature request solves? Spark builds the `HashedRelation` exactly once, on the driver, inside `BroadcastExchangeExec.relationFuture` via `HashedRelationBroadcastMode.transform`, and what goes over the wire is the finished hash table. Executors deserialize it at most once per JVM, because `TorrentBroadcast.readBroadcastBlock` caches the object in `BroadcastManager.cachedValues` keyed by broadcast id. Each task then calls `broadcastRelation.value.asReadOnlyCopy()`, which is O(1) — it returns a new wrapper over the *same* `BytesToBytesMap`, so the task gets its own result-row cursor and nothing else. Comet does none of that. `CometBroadcastExchangeExec` broadcasts `Array[ChunkedByteBuffer]` — serialized Arrow IPC batches, not a hash table. From there: - `CometBatchRDD.compute` decodes those bytes per partition, so once per task (`CometBroadcastExchangeExec.scala:300-304`) - each task deserializes and instantiates its own native plan, then runs `root_op.native_plan.execute(0, task_ctx)` (`jni_api.rs:981`) - the planner builds `HashJoinExec` with `PartitionMode::Partitioned` for every join except null-aware anti-join (`planner.rs:2313-2321`), and that arm does an unshared build — `OnceFut::new(collect_left_input(...))` on each `execute()` call So every task pays a full Arrow IPC decode plus a full hash-table build over the entire build side. The work scales with task count where Spark's scales with executor count. On a 1000-task stage with 8-core executors, that is roughly 1000 builds against roughly 125. One thing worth recording, because it contradicts the first half of #3692: switching to `CollectLeft` does not fix this. Its `left_fut.try_once(...)` shares the build across *partitions of a single plan instance*, and Comet gives every task its own plan instance executing partition 0, so the once-cell is trivially once. `CollectLeft` is still arguably the more honest mode to report for a broadcast join, but it buys no reuse here. #3692 was closed as completed, but only #3703 (coalescing batches before broadcast) landed and #3693 was closed unmerged, so neither the mode change nor the executor-level cache is in the tree today. This issue is the deeper version of the second half of that issue: even if the decoded Arrow batches were cached per executor, each task would still build its own hash table. ### Describe the potential solution The general shape is an executor-level cache keyed by broadcast id holding something reusable across concurrent tasks. Three levels, increasing payoff and difficulty: 1. Cache the decoded Arrow batches per executor. This is the second half of #3692. It removes the repeated IPC decode but not the hash build, and it is by far the cheapest to implement. 2. Cache the built hash table itself. This needs the native side to expose a build-side artifact that outlives one plan instance and is safe to share across concurrent tasks, plus a way to inject a prebuilt one into `HashJoinExec` — DataFusion keeps `JoinLeftData` behind a `OnceFut` owned by the exec instance, so there is no injection point today. 3. Build once on the driver and broadcast the table, mirroring Spark. Probably not worth pursuing: the native hash table is not trivially serializable, and the driver-side build would be single-threaded. There is precedent in the tree for the executor-level cache. `PlanDataInjector.basePlanCache` (`operators.scala:170`) keeps 16 parsed plan protos per executor for exactly this reason — the comment there notes that "without a cache an executor re-parses the same operator tree once per task." Lifetime and eviction need care for anything beyond level 1. Spark's broadcast cache holds soft/weak references and drops on `Broadcast.destroy`, but a native hash table is off-heap, so release has to be tied to broadcast unpersist rather than GC. ### Additional context The build side is under `autoBroadcastJoinThreshold` (10MB by default) by definition, and DataFusion's build is vectorized, so the per-task cost is small in absolute terms. This should matter most on stages with high probe-side parallelism over a non-trivial build side — TPC-DS dimension joins against a large fact table are the obvious place to measure. Worth quantifying before investing in level 2. #3762 (bypass Arrow FFI for broadcast exchange reads) attacks the decode cost from a different direction and would compose with level 1. Verified against main at 67168ca1f. -- 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]
