You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "Dandandan (via GitHub)" <gi...@apache.org> on 2026/08/17 18:29:31 UTC
[PR] perf(scheduler): read an unfiltered small scan inline instead of staging it [datafusion-ballista]
Dandandan opened a new pull request, #2338:
URL: https://github.com/apache/datafusion-ballista/pull/2338
## Which issue does this PR close?
Related to the broadcast-lowering cost noted while comparing Ballista's TPC-H plans against Spark with AQE. No issue filed yet.
## Rationale for this change
DataFusion plants a `CoalescePartitionsExec` over a scan to feed a `CollectLeft` hash join's build side. `DistributedExchangeRule` planted a stage boundary under *every* such coalesce, so a build side that is just a table scan became a stage of its own.
TPC-H q7 before (AQE on, `--partitions 16`):
```
-- Stage 0 <- exists only to feed the join below
ShuffleWriterExec: partitioning: UnknownPartitioning(16)
DataSourceExec: supplier, file_groups={16}, proj=2c
-- Stage 2
HashJoinExec: mode=CollectLeft, on=[(s_suppkey@0, l_suppkey@1)]
CoalescePartitionsExec
ShuffleReaderExec: upstream_stage: 0, partitioning: UnknownPartitioning(16)
FilterExec: l_shipdate@4 >= 1995-01-01 AND l_shipdate@4 <= 1996-12-31
DataSourceExec: lineitem, file_groups={16}
```
The boundary costs a stage, a shuffle write of the whole scan, a re-read in every consuming task, and - because a stage cannot start before its inputs finish - a barrier in front of the join. The gather happens inside a single task either way, so it buys nothing.
## What changes are included in this PR?
`DistributedExchangeRule` skips the boundary when the coalesce input is an **unfiltered** scan whose bytes fit `ballista.optimizer.broadcast_join_threshold_bytes` - the same budget that decides whether a build side is small enough to sit in every probe task. Projections are transparent; aggregates, joins and shuffle reads keep their boundary, since they carry state that must be computed once.
Filtering is what disqualifies a scan, and this is the part worth reviewing. A filter makes the staged copy far smaller than the scan, so rebuilding it in every task costs much more than reading it back. `Precision::Exact` on the scan's `total_byte_size` is the test rather than looking for a `FilterExec`, because the interesting cases have the predicate pushed *into* the scan, where there is no filter node to find:
```
DataSourceExec: part, file_groups={16}, predicate=p_name LIKE 'forest%' <- no FilterExec
```
Note this also means the rewrite only fires with `datafusion.execution.collect_statistics=true`; without collected statistics the estimate is not exact and every scan keeps its boundary as before.
q7 after - 8 stages become 6, with `supplier` and `customer` read directly in the stages that join them:
```
-- Stage 0
SortShuffleWriterExec: partitioning=Hash([l_orderkey@1], 16)
HashJoinExec: mode=CollectLeft, on=[(s_suppkey@0, l_suppkey@1)]
CoalescePartitionsExec
DataSourceExec: supplier, file_groups={16}, proj=2c
FilterExec: l_shipdate@4 >= 1995-01-01 AND l_shipdate@4 <= 1996-12-31
DataSourceExec: lineitem, file_groups={16}
```
## Are these changes tested?
Five unit tests in `distributed_exchange.rs` cover: a small scan inlined, a scan over the byte budget staged, threshold `0` staged, a filtered scan staged, and a non-scan (aggregate) pipeline staged. Full `ballista-scheduler` suite passes (368 + 25), `cargo clippy --all-targets` clean.
Verified end to end on TPC-H SF10 (1 scheduler, 2 executors x 4 vcores, AQE + `max_partitions_per_task=0`): q7 8 stages -> 6, q19's filtered `part` build side keeps its boundary, and row counts are unchanged on every query touched.
**Draft, because the performance claim is not settled yet.** An earlier revision of this rule that inlined filtered scans too measured q7 0.92x and q2 0.95x but q19 **1.14x** over 9 alternating samples per variant at SF10 - that regression is what motivated the exact-statistics guard. A fresh A/B for the final rule is running and I will post it here. At SF10 on a single machine the removed shuffle write and barrier are worth little, so the measurable win is expected to be small; the case for the change is the stage, the write and the barrier that disappear at cluster scale.
## Are there any user-facing changes?
No API or configuration change. Plans change shape: an unfiltered small scan feeding a broadcast build side no longer appears as a separate stage.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] perf(scheduler): read an unfiltered small scan inline instead of staging it [datafusion-ballista]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan merged PR #2338:
URL: https://github.com/apache/datafusion-ballista/pull/2338
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] perf(scheduler): read an unfiltered small scan inline instead of staging it [datafusion-ballista]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on PR #2338:
URL: https://github.com/apache/datafusion-ballista/pull/2338#issuecomment-5318674267
A/B for the rule as it now stands, TPC-H SF10, one scheduler + two executors x 4 vcores, AQE with `collect_statistics=true` and `max_partitions_per_task=0`. Three rounds, variants alternated per round so both share any machine drift, three iterations each — 9 samples per variant per query, medians below.
| Query | base | this PR | ratio |
| ----: | ---: | ------: | ----: |
| 2 | 0.449 | 0.420 | 0.94 |
| 7 | 2.016 | 1.921 | 0.95 |
| 8 | 0.614 | 0.623 | 1.01 |
| 12 | 0.899 | 0.843 | 0.94 |
| 19 | 0.698 | 0.651 | 0.93 |
| 20 | 1.463 | 1.455 | 0.99 |
| **sum of medians** | **6.14s** | **5.91s** | **0.963** |
Read this as "no regressions" rather than as a 4% win. q12 and q19 keep their stage boundaries under this rule, so their plans are byte-identical between the two variants — the ~5% they move is the noise floor of this setup, which means q7 and q2 are inside it too. A single machine at SF10 is the wrong instrument for the thing being removed: the shuffle write lands in the page cache and the barrier costs almost nothing when every task is milliseconds from its input.
What is not noise is the earlier revision that inlined filtered scans as well. On the same harness it measured q19 at **1.14x** — inlining a `part` scan filtered to a fraction of a percent made every consuming task rebuild what one stage had materialized. That is what the `Precision::Exact` guard is for, and q19 is now back at parity.
Plan effect, unchanged from the description: q7 goes from 8 stages to 6, and q19's filtered `part` build side keeps its boundary. Row counts match the pre-change run on every query in the table.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org