andygrove opened a new issue, #2065:
URL: https://github.com/apache/datafusion-ballista/issues/2065
## 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).
--
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]
