andygrove commented on PR #2419:
URL:
https://github.com/apache/datafusion-ballista/pull/2419#issuecomment-5591677739
## Digging into q8: the 6B-row lineitem shuffle should not exist
Following up on Theme 1 with a closer look at where q8's 164s goes and why.
Summing max task duration per stage gets to ~170s, so the DAG is effectively
serialized stage by stage. One stage is 80% of the query:
| Stage | What | Max task |
| ----- | ---- | -------: |
| 1 | `SortShuffleWriter Hash([l_partkey], 256)` over the raw lineitem scan
| **131.1s** |
| 2 | part ⋈ lineitem, Partitioned | 11.4s |
| 5 | ⋈ supplier, CollectLeft | 7.4s |
| 7 | customer scan | 7.0s |
| 8 | ⋈ customer / nation | 5.4s |
| rest | | < 4s each |
So q8 is really a single-stage problem.
### Why we shuffle 6B rows to join against 1.3M
The build side of join `plan_id=0` is `FilterExec(p_type = 'ECONOMY ANODIZED
STEEL')` over `part`. Stage 0's runtime `input_rows` is 1,332,379, which is
exactly 200M / 150 distinct `p_type` values. That is one `p_partkey` i64
column, roughly **10.7 MB**, about 12x under the configured 128 MiB broadcast
threshold. This should have been a broadcast join with no lineitem shuffle at
all.
Two independent gates block it, both in `to_actual_join`
(`dynamic_join.rs:296`):
**Gate 1: the pre-execution estimate is wrong by 30x.** There are no runtime
stats when the initial plan is built, so `FilterExec` falls back to
`datafusion.execution.default_filter_selectivity` (20%). Estimate: 40M rows,
~320 MB. Over threshold, so `under_threshold = false`, which gives
`PartitionMode::Partitioned` and therefore `JoinSelectionAction::Repartition`.
That arm at `join_selection.rs:297` unconditionally wraps **both** children in
an `ExchangeExec`. The 131s is committed right there, before a single byte of
measured statistics exists.
**Gate 2: the 1M row ceiling is a hard AND, evaluated before the byte
estimate.** In `supports_collect_by_thresholds`, `dynamic_join.rs:493`:
```rust
if num_rows == 0 || num_rows >= threshold_num_rows {
return false;
}
estimate_output_byte_size(...).is_some_and(|est| est < threshold_byte_size)
```
Even with a perfect 1.33M row estimate, the row check rejects the broadcast
before the byte estimate runs. `broadcast_join_threshold_rows` defaults to
1,000,000.
@avantgardnerio your instinct about bumping `broadcast_join_threshold_rows`
to ~1.5M is right about gate 2, but I do not think it fixes q8 on its own,
because gate 1 hands the check 40M rows rather than 1.33M.
One more thing worth knowing: once `Repartition` fires there is no way back.
On re-resolution `selection_state == Repartitioned`, so `to_actual_join` only
chooses between Hash and SortMerge. AQE cannot recover the broadcast even after
stage 0 finishes in 1.4s and reveals the true size.
### Possible fixes, cheapest first
**1. Config-only experiment to confirm the diagnosis.** Set
`datafusion.execution.default_filter_selectivity=1` and
`ballista.optimizer.broadcast_join_threshold_rows=5000000`. That drops the part
estimate to 2M rows / 16 MB and clears both gates. If q8 falls from 164s to
roughly 35s the diagnosis holds. Neither is a good permanent default, but it
isolates the cause in one run. I can do this next.
**2. Make the byte estimate authoritative when it is available.** Reorder
`supports_collect_by_thresholds` so the row ceiling only applies when
`estimate_output_byte_size` returns `None`. A 1M-row ceiling on a single i64
key column is 8 MB, which is 16x under the byte threshold. The row rule reads
like a fallback for unknown widths, but today it vetoes narrow build sides the
byte rule would accept.
**3. Asymmetric exchange: resolve the cheap side first.** This is the
structural fix. Today `Repartition` shuffles both sides at once. When one
side's estimate is `Inexact` and the other is orders of magnitude larger, we
could insert the exchange on the cheap side only and leave the big side
unresolved. For q8 that means stage 0 runs for 1.4s, returns an exact 1.33M
rows / 10.7 MB, and the join flips to CollectLeft. Lineitem is then scanned
once inside the join stage against a broadcast build side, and stages 1 and 2
(142s combined) collapse into roughly 12s. Same shape should help q9 s1, q3 s3,
and q17.
The distinction worth encoding is that "estimated large but inexact" is not
the same as "known large". Right now both take the same branch.
**4. Cross-stage dynamic filters (#1375).** Biggest lever for the whole
suite rather than just q8. `enable_dynamic_filter_pushdown` is hard-disabled at
`extension.rs:775` because a filter produced in one stage never reaches a scan
in another. The scheduler is arguably the right place to bridge that, though:
stage 0 finishes holding 1.33M `p_partkey` values, and we could ship a bloom
filter of maybe 2 MB into stage 1's `DataSourceExec` before launching it. Stage
1 would then read ~40M rows instead of 6B. Even keeping the shuffle, that is a
~100x reduction. This is essentially Spark's runtime filter / DPP.
### Two secondary findings
**Stage 5 runs the whole join in a single task.** The metrics are
unambiguous: `min=median=max=7429ms`, one task, 79.9M rows.
`ShuffleReaderExec::try_new_broadcast` gives the reader
`UnknownPartitioning(1)` (`shuffle_reader.rs:169-178`), and in stage 5 that
broadcast reader sits on the **probe** side (`upstream_stage: 2`), not the
build. The build (supplier, stage 3) is read through a plain 256-partition
reader under `CoalescePartitionsExec`. So both inputs gather everything, the
stage collapses to one core, and it also serializes a 256-way sort shuffle
write. Every path I traced through `SelectJoinRule` looks like it should put
the broadcast on the build side, so the executed shape does not match what the
code appears to produce. Worth tracing which branch actually built this.
**Stage 1's time tail looks like contention rather than shuffle cost.**
Compare against q9 s1, which is the identical operator over the identical 6B
rows with **six** projected columns instead of five:
| | cols | median | max | task input min/max |
| --- | ---: | -----: | --: | ---: |
| q8 s1 | 5 | 62.5s | 131.1s | 1.36x |
| q9 s1 | 6 | 19.2s | 58.1s | 8.4x |
q8 moves less data and takes 3.3x longer per task, on *more* balanced input.
Together with the repeat run above showing q8 at 57s and 142s across two
identical iterations, that reads as variance rather than a deterministic cost.
Candidates: 34 concurrent tasks each buffering up to 256 MiB in the sort
shuffle writer (up to ~8.7 GiB per executor before spill), or S3 read
bandwidth. Separately, 34 tasks for the most expensive stage in the query seems
low for 32-core executors, so the scan's file-group count may be worth a look
on its own.
--
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]
