comphead opened a new issue, #7472: URL: https://github.com/apache/datafusion-comet/issues/7472
### What is the problem the feature request solves? Since #7016, the tasks of an executor that read a broadcast at the same time share one decoded copy of it. The memory pools do not see that sharing. Each task's native hash join still reserves the whole build side, so N overlapping tasks reserve N times the build batch while one copy is resident. For a broadcast of one batch, which is what `coalesceBroadcastBatches` produces unless a string or binary vector would pass 1 GiB (#7192), nothing between the shared decode and the join copies it: - Each task reads a root of its own over the shared buffers ([`SharedDecode.Decoded.view`](https://github.com/apache/datafusion-comet/blob/209d8e3a3eee673db17098238c861a4c072f0bcb/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastValue.scala#L303-L314)). - `ColumnarBatchArrowReader` passes them on with retained references ([ColumnarBatchArrowReader.scala#L53-L75](https://github.com/apache/datafusion-comet/blob/209d8e3a3eee673db17098238c861a4c072f0bcb/spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ColumnarBatchArrowReader.scala#L53-L75)), and the C stream export hands native the same memory. - The native scan keeps the imported buffers unless a string column holds invalid UTF-8 ([scan.rs#L187-L193](https://github.com/apache/datafusion-comet/blob/209d8e3a3eee673db17098238c861a4c072f0bcb/native/core/src/execution/operators/scan.rs#L187-L193)). - DataFusion's `collect_left_input` concatenates the build batches ([exec.rs#L2365](https://github.com/apache/datafusion/blob/55.2.0/datafusion/physical-plan/src/joins/hash_join/exec.rs#L2365)), and arrow's `concat` of a single array is a slice ([concat.rs#L500-L503](https://github.com/apache/arrow-rs/blob/59.2.0/arrow-select/src/concat.rs#L500-L503)). Yet `collect_left_input` counts each build batch with a fresh `RecordBatchMemoryCounter` and calls `reservation.try_grow(batch_size)` on the task's own `HashJoinInput[{partition}]` consumer ([exec.rs#L2266-L2270](https://github.com/apache/datafusion/blob/55.2.0/datafusion/physical-plan/src/joins/hash_join/exec.rs#L2266-L2270), [exec.rs#L1514](https://github.com/apache/datafusion/blob/55.2.0/datafusion/physical-plan/src/joins/hash_join/exec.rs#L1514)). Comet's pools pass the reservation on to Spark ([fair_pool.rs#L279](https://github.com/apache/datafusion-comet/blob/209d8e3a3eee673db17098238c861a4c072f0bcb/native/core/src/execution/memory_pools/fair_pool.rs#L279)), and `fair_unified` also takes it out of the share of the task's spillable consumers ([fair_pool.rs#L250](https://github.com/apache/datafusion-comet/blob/209d8e3a3eee673db17098238c861a4c072f0bcb/native/core/src/execution/memory_pools/fair_pool.rs#L250)). For example, 8 tasks that probe a 500 MiB broadcast at the same time reserve about 4 GiB of off-heap execution memory for about 500 MiB resident. The memory management guide already says so: `reserved()` is a lower bound on Comet's footprint "except while tasks share a decoded broadcast, which they reserve once each" ([memory_management.md#L625-L627](https://github.com/apache/datafusion-comet/blob/209d8e3a3eee673db17098238c861a4c072f0bcb/docs/source/contributor-guide/memory_management.md?plain=1#L625-L627)). This is not a regression. Before #7016 every task held its own decoded copy, so the reservations matched. But the memory that sharing saves cannot go to anything else. The reservations still crowd out the task's other operators, which spill sooner, and the off-heap size cannot shrink to match what is resident. A broadcast of several batches is different. There each task's `concat_batches` makes a copy of its own, so the reservation roughly matches what the task holds. ### Describe the potential solution Charge a shared build side once per executor rather than once per task. #6037 does this for the joins it covers, by charging each prepared build once, to the executor's storage memory. It covers inner joins with direct-column keys and fixed-width or UTF-8 build columns, and it needs DataFusion 56. Other broadcast joins would need `collect_left_input` to accept a build side whose memory is already accounted for. DataFusion 55.2 does not offer that, and DataFusion 56's prepared builds offer it only for inner joins. ### Additional context Part of #6013. Found while checking what #7016 leaves of it. Line references are to `main` at 209d8e3a3e, DataFusion 55.2.0 and arrow-rs 59.2.0. Not measured. The 4 GiB figure is arithmetic from the code above, not a run. ### Willingness to contribute I can contribute a fix for this bug independently -- 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]
