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: [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]
