apache / apache/datafusion-ballista
Broadcast join promotion drops a sort order required by a parent SortMergeJoinExec, silently producing wrong results
- Dominant language
- Rust
- Stars
- 2.1k
- Forks
- 320
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 66
Description
## Describe the bug
Ballista's broadcast-join promotion rewrites a `SortMergeJoinExec` into a
`HashJoinExec(CollectLeft)`. A `HashJoinExec` does not preserve its input's
sort order, but nothing re-establishes that order afterwards. When the promoted
join feeds a **parent `SortMergeJoinExec`**, the parent then runs on unsorted
input.
A merge join on unsorted input does not error — it silently stops matching once
the merge cursors diverge, so **rows are dropped and the query returns a wrong
answer**.
This is distinct from #1055 (which is about which *join types* are safe to
broadcast, and is already guarded by `collect_left_broadcast_safe`). Here the
join type is fine; the *ordering contract* is what breaks.
## Evidence
In the distributed plan below (TPC-DS q78, SF1, `prefer_hash_join=false`), the
outer `SortMergeJoinExec` has a `SortExec` on its right input but **nothing on
its left** — the left arrives straight from a `HashJoinExec(CollectLeft)` that
replaced the inner SMJ:
```
SortMergeJoinExec: join_type=Left, on=[(ss_sold_year, cs_sold_year), (ss_item_sk, cs_item_sk), (ss_customer_sk, cs_customer_sk)]
ProjectionExec: ...
ProjectionExec: ...
HashJoinExec: mode=CollectLeft, join_type=Right, on=[(ws_sold_year, ss_sold_year), ...] <-- left input: no ordering
UnresolvedShuffleExec: stage=8, broadcast=true, upstream_partitions: 16
ProjectionExec: ...
AggregateExec: mode=FinalPartitioned, gby=[d_year, ss_item_sk, ss_customer_sk], ...
UnresolvedShuffleExec: stage=15, partitioning: Hash([d_year, ss_item_sk, ss_customer_sk], 16)
SortExec: expr=[cs_item_sk ASC, cs_customer_sk ASC], preserve_partitioning=[true] <-- right input: sorted
...
```
Before the rewrite the ordering came from a `SortExec` feeding the inner SMJ's
`ss` side; promoting that SMJ to a broadcast hash join removed the need for
that sort, and the parent SMJ's requirement went unmet.
## To Reproduce
SF1 TPC-DS, scheduler + executor, `target_partitions=16`. Using the q78 CTEs
(`ss`, `ws`, `cs`):
```sql
select ss_sold_year, ss_item_sk, ss_customer_sk, ss_qty from ss
left join ws on (ws_sold_year=ss_sold_year and ws_item_sk=ss_item_sk and ws_customer_sk=ss_customer_sk)
left join cs on (cs_sold_year=ss_sold_year and cs_item_sk=ss_item_sk and cs_customer_sk=ss_customer_sk)
where (coalesce(ws_qty,0)>0 or coalesce(cs_qty,0)>0) and ss_sold_year=2000;
-- Ballista: 73 rows. DataFusion: 223 rows.
```
Ballista's own breakdown of that set is `total=223, ws_pos=73, cs_pos=150`: the
150 rows that match only via `cs` are exactly the ones lost, i.e. the outer
`ss ⋈ cs` join stops matching.
Disabling the promotion restores the correct answer, which isolates the cause:
| Config | Result |
|---|---|
| default (broadcast on) | 73 rows — **wrong** |
| `-c ballista.optimizer.broadcast_sort_merge_join_enabled=false` | 223 rows — correct |
| `-c ballista.optimizer.broadcast_join_threshold_bytes=0` | 223 rows — correct |
| full q78 + `broadcast_join_threshold_bytes=0` | 100 rows — **verifies OK** vs DataFusion |
Note the wrong answer does **not** depend on `ORDER BY` or `LIMIT`; it
reproduces with neither. (q78 was originally filed under #2046 as "distributed
LIMIT drops rows" — that framing is a red herring. Wrapping the query in
`count(*)` changes the plan enough to hide the bug, which is what made it look
limit-related.)
## Expected behavior
Promoting a join must not silently drop an ordering that a parent operator
requires. Either re-establish the ordering (insert the `SortExec` the parent
needs) after the rewrite, or decline to promote when the join's output ordering
is required upstream.
It would also be worth asserting, over the planned stages, that every
`SortMergeJoinExec` input satisfies the join's `required_input_ordering()` — an
unsorted SMJ input is always a bug, and today it fails silently and produces
wrong results rather than erroring.
## Additional context
- Reproduced at SF1 on a 1-scheduler / 1-executor native cluster (8 slots, 16
partitions), on `upstream/main` (e843c8e8) under the default static planner.
- Surfaced by the TPC-DS correctness gate (#1845 / #2048); accounts for the q78
half of #2046. q4 (also in #2046) is not yet diagnosed.
- Related: #1055 (broadcast join-*type* safety), #2006 (propagating ordering
across pass-through shuffles).
Contributor guide
Research direction
Search the Rust optimizer for broadcast-join promotion and trace how SortMergeJoinExec, HashJoinExec, and required_input_ordering() are handled. Reproduce TPC-DS q78 with the stated scheduler and executor settings, then verify the result is 223 rows and that every SortMergeJoinExec input satisfies its required ordering.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- databases, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100