andygrove opened a new issue, #7408: URL: https://github.com/apache/datafusion-comet/issues/7408
### What is the problem the feature request solves? In pipelined mode, a join whose build side turns out small still shuffles its whole probe side. TPC-H q17 at SF1000 filters `part` to 199,925 rows, about 2.8 MB, but Spark estimates the filtered `part` at 654 MiB, because without column statistics a filter doesn't shrink the estimate. So the static plan joins `part` and `lineitem` after shuffling both, and moves all 6 billion `lineitem` rows across the network to do it. #7391 fixes this for blocking mode with a rule in Ballista's adaptive planner, which holds the probe side's stage back until the build side is measured, then moves the join into the probe side's stage with the build side broadcast to it. Pipelined mode has no adaptive planning: every task is placed before anything runs. Ballista's pipelined plan for q17 doesn't broadcast `part` either, and on a 2-node cluster with 2.5 GbE both engines are bound by the two full `lineitem` shuffles, about 35 GB per node each way (#7392). Removing one of them halves the network traffic. ### Describe the potential solution Measure small build sides before planning the query, as the driver already runs scalar subqueries before submitting a plan: 1. When comet-scheduler plans a pipelined query, find the joins that #7391's rule can move (inner, built on the right, column keys, no condition, both sides read through exchanges) whose build side Spark estimates within Ballista's maximum estimate multiple of the broadcast threshold, and whose probe side Spark estimates many times larger. 2. Run each such build side's stages as their own pipelined job, and read the result into the scheduler, as the scheduler reads a pipelined query's result. 3. If the result is under the broadcast threshold, move the join into the probe side's stage with the result as a broadcast input read from memory, using the same rewrite as #7391. Otherwise plan the query as it was. The cost is running the build side's stages before the rest of the query, and once more when the build side turns out large. ### Additional context Part of #7288. Related to #7391 (option D there). -- 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]
