comphead commented on PR #2434:
URL:
https://github.com/apache/datafusion-ballista/pull/2434#issuecomment-5641130802
Reviewed this against the PR head (`1340cc97`). Baseline is green for me:
416/416 in `cargo test -p ballista-scheduler --lib`.
First, the part you asked people to push on — **the one-sided shuffle safety
argument holds.** I traced it rather than taking it on faith:
`find_runnable_exchanges` only pushes `ExchangeExec` nodes as stages and stops
descending at a resolved one, so in the staged state the probe subtree creates
no stage, and the join sits above the exchange rather than inside its subtree.
`Repartitioned` is reachable only via `to_partitioned()` or
`inputs_already_partitioned()`, and staging sets neither. `to_broadcast`
sharing `stage_id`/`shuffle_partitions` checks out too.
Two things I'd want changed before merge.
### 1. Staging fires for join types that can never broadcast
The `StageBuildSide` guard in `to_actual_join` checks the config,
`!null_aware`, the exchange guard and `should_stage_build_side` — but not
`collect_left_broadcast_safe(build_side_join_type)`, which is already computed
~80 lines above.
`partition_mode` is `CollectLeft` only when `under_threshold &&
collect_left_broadcast_safe(..)`. That second term is a pure function of the
join type, so no measurement can change it. For a non-broadcastable type the
second pass can't reach `CollectLeft`, hits the (now failing) exchange guard,
and falls to `Repartition` — one serialised stage boundary that never had a
chance of changing the decision.
I confirmed it with a throwaway test on the q8 shape:
```
join_type=Full collect_left_broadcast_safe=false action=StageBuildSide
join_type=Left collect_left_broadcast_safe=false action=StageBuildSide
join_type=LeftSemi collect_left_broadcast_safe=false action=StageBuildSide
join_type=LeftAnti collect_left_broadcast_safe=false action=StageBuildSide
```
`Full` is affected unconditionally (`Full.swap() == Full`);
`Left`/`LeftSemi`/`LeftAnti`/`LeftMark` whenever the build side is already the
left input.
One line fixes it:
```rust
(JoinInputState::Unknown, PartitionMode::Partitioned)
if bc.stage_build_side_enabled()
&& !self.null_aware
+ // Measuring the build side can only change the outcome if
+ // CollectLeft is reachable at all, which is a property of the
+ // (post-swap) join type alone.
+ && collect_left_broadcast_safe(build_side_join_type)
&& !self.left.is::<ExchangeExec>()
```
Applied locally: **415/415 pass, no changes needed to any test in this PR**
— they all use `Inner`.
**On the +3.1s band:** this is a candidate partial explanation, not a
diagnosis. q16 (`NOT IN`, which `not_in_subquery_rewrite` turns into a plain
anti join) and q22 (`NOT EXISTS`) are both anti joins and both sit in the band.
It doesn't explain q5/q7/q12, which are plain inner joins — those look more
like the serialisation cost you already acknowledge. Re-running the six with
`stage_build_side=false` per query would separate the two.
### 2. Nothing tests the multi-pass path
All the new coverage is single-pass: `should_stage_build_side` in isolation,
one `to_actual_join` call, `exchange_on`. The mechanism this PR is actually
about — stage → resolve with measured stats → re-decide → `CollectLeft` — isn't
exercised, and neither is the `EnsureRequirements`/`CoalescePartitionsExec`
behaviour you say you verified by hand.
The harness for exactly this already exists in `aqe/test/join_selection.rs`
(`AdaptivePlanner` + `mock_partitions_with_statistics` + `assert_plan!`). Two
`assert_plan!` tests would cover it: the q8 shape resolving to a broadcast
`CollectLeft`, and the fallthrough where the measured size is still too large
and `exchange_on` reuse yields one exchange per side rather than a nested pair.
Worth noting `test_left_join_not_collected_left` (the #1055 guard) passes
today only because its MemTable stats don't match the staging shape — nothing
currently pins "a Left join must not stage."
### Smaller
- I'd split the `broadcast_join_threshold_rows` semantics change out. It's
well argued and clearly flagged, but by your own account it isn't what fixes
q8, and it replaces a test that asserted the old behaviour deliberately.
Separate PRs means either half can be reverted alone if the SF1000 numbers move.
- The exchange guard only inspects immediate children, so a build side
shaped like `FilterExec(ExchangeExec(..))` gets a second boundary stacked on
the first. Edge case.
- `CoalescePartitionsRule`'s docs assume both join legs read upstream
shuffle output, and attach the same `CoalescePlan` to every leaf to keep
buckets aligned. In the staged state only one leg is an exchange. It's off by
default so the risk is low, but does the leaf grouping handle a one-exchange
join?
### On defaulting to `true`
Fix #1 first, then yes. As it stands, default-on ships a path that's
provably unprofitable for `Full` and left-sided semi/anti joins, which is a bad
trade for queries that get nothing back. After the guard I'd have no objection.
Benchmark methodology is good — real cluster, row counts cross-checked,
regressions disclosed rather than smoothed. Two caveats worth stating inline:
one iteration per query, and the current totals include wasted stages on any
anti/outer join matching the shape, so the post-fix number should come in a
little under 696.09s.
I did not check what happens to a resolved exchange's `shuffle_partitions`
when the executor holding that shuffle is lost. Looks pre-existing for all
exchanges rather than something this PR introduces, but staging makes one more
stage's output load-bearing.
--
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]
