comphead opened a new issue, #7487: URL: https://github.com/apache/datafusion-comet/issues/7487
### What is the problem the feature request solves? #7016 lets the tasks of an executor that read a broadcast at the same time share one decoded copy. It saves CPU, but little task time: - A task that arrives while another task decodes waits for that decode ([CometBroadcastValue.scala#L167-L172](https://github.com/apache/datafusion-comet/blob/7bf572efb7/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastValue.scala#L167-L172)), so every task of a wave still spends the decode's latency. - The copy is freed when the last task holding it completes ([CometBroadcastValue.scala#L122-L134](https://github.com/apache/datafusion-comet/blob/7bf572efb7/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastValue.scala#L122-L134)). The tasks of a wave finish at about the same time, so the next wave decodes again. Measured with the `bhj` mode of a hash-join task-time benchmark: a 1M-row broadcast probed by 32 tasks on `local[8]`, so 4 waves, median of 7 runs, `main` at 7bf572efb7. Times are the mean per probe task, with executor CPU time in brackets: | codec | Spark | Comet | Comet, `shareDecoded.enabled=false` | |---|---:|---:|---:| | lz4 | 7.3 ms (5.6) | 19.3 ms (7.3) | 21.0 ms (20.7) | | zstd, level 5 | 7.8 ms (6.8) | 31.8 ms (8.9) | 33.6 ms (33.1) | Sharing cuts the CPU per task from 20.7 to 7.3 ms, but the task time only from 21.0 to 19.3 ms. The CPU totals fit about one decode per wave. The native metrics put the hash build at 1.5 ms and the join at 0.9 ms per task, so most of the remaining time, about 12 ms with lz4 and 23 ms with zstd, is the wait for the decode. On this Mac lz4-java runs its pure Java fallback, so lz4 decodes may be faster on Linux. zstd goes through zstd-jni on both. ### Describe the potential solution Keep an idle decode for a while after its last holder completes, so that later waves reuse it. For example an idle timeout such as `spark.comet.exec.broadcastExchange.shareDecoded.idleTimeout`, possibly with a cap on idle bytes per executor. The idle copy is memory that Comet's pools do not track (the memory management guide lists `SharedDecode`), so the timeout bounds how long it stays. Freeing it when the broadcast is removed would be more precise, but needs an executor-side hook. A rough expectation from the measured CPU, not a run: about 10 ms per task with lz4 and 14 ms with zstd in this 4-wave benchmark, closer to Spark with more waves. The wait within the first wave remains. A cheaper decode would shrink it (#2216, #3762). ### Additional context Part of #6013. The benchmark is not in the tree yet and will come as its own PR. cc @andygrove: #7016 frees the copy at once to bound memory. Would an idle timeout work for you? ### 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]
