Skip to content

fix: restore input ordering a SortMergeJoinExec requires after lowering - #2066

Closed
andygrove wants to merge 1 commit into
apache:mainfrom
andygrove:fix/2065-smj-broadcast-ordering
Closed

fix: restore input ordering a SortMergeJoinExec requires after lowering#2066
andygrove wants to merge 1 commit into
apache:mainfrom
andygrove:fix/2065-smj-broadcast-ordering

Conversation

@andygrove

Copy link
Copy Markdown
Member

Which issue does this PR close?

Closes #2065.

Also resolves the q4 / q78 half of #2046 (the set-operation half, q38 / q87, is
a separate DataFusion-side bug handled in #2064).

Rationale for this change

Lowering a plan into stages rewrites nodes, and a rewrite can drop an ordering
its parent depends on. Ballista's broadcast promotion turns a
SortMergeJoinExec into a HashJoinExec(CollectLeft), which does not preserve
input order — but nothing re-establishes that order, so a parent
SortMergeJoinExec ends up merging unsorted input.

A merge join given unsorted input does not error. It stops matching once the
merge cursors diverge, so rows are silently dropped and the query returns a
wrong answer.

In the q78 plan the outer SortMergeJoinExec had a SortExec on its right
input and nothing on its left, because the inner SMJ that used to supply that
ordering had been replaced by a broadcast hash join:

SortMergeJoinExec: join_type=Left, on=[(ss_sold_year, cs_sold_year), ...]
  ...
      HashJoinExec: mode=CollectLeft, join_type=Right, ...      <-- left input: no ordering
        UnresolvedShuffleExec: stage=8, broadcast=true, upstream_partitions: 16
        ...
  SortExec: expr=[cs_item_sk ASC, cs_customer_sk ASC], ...      <-- right input: sorted

Disabling the promotion isolated the cause — broadcast_join_threshold_bytes=0
or broadcast_sort_merge_join_enabled=false both restore the correct answer.

This is distinct from #1055, which concerns which join types are safe to
broadcast (already guarded by collect_left_broadcast_safe). Here the join type
is fine; the ordering contract is what breaks.

What changes are included in this PR?

  • restore_required_input_ordering: after lowering rewrites a node's children,
    re-establish any Hard input ordering a child no longer carries. Sorts are
    only added, never removed, so an input that already satisfies its requirement
    is untouched.
  • validate_sort_merge_join_orderings: verify every SortMergeJoinExec in the
    planned stages has inputs satisfying the join's required ordering, so this
    failure mode surfaces as an error instead of wrong results.
  • Un-skip TPC-DS q4 and q78 in the correctness gate (86 → 88 queries).

Soft requirements are deliberately ignored — they are opportunistic and an
operator works without them.

Verification

SF1, 1 scheduler / 1 executor, target_partitions=16:

Before After
q78 73 rows vs 100 verified OK (100 rows)
q4 0 rows vs 8 verified OK (8 rows)
minimal repro (q78 CTEs, no ORDER BY/LIMIT) 73 rows vs 223 223 rows

Full SF1 TPC-DS gate: 88/88 verified against single-process DataFusion,
under both prefer_hash_join=false and prefer_hash_join=true. The new
validation reports 0 false positives across all 88 queries under both
settings, which is why it is a hard error rather than a debug assertion.

cargo test --workspace, cargo clippy --all-targets --workspace -- -D warnings, and cargo fmt --all are clean.

Note on the q78 framing

q78 was originally reported under #2046 as "distributed LIMIT drops rows". That
framing is a red herring: the wrong answer does not depend on ORDER BY or
LIMIT and reproduces with neither. Wrapping the query in count(*) changes
the plan enough to hide the bug, which is what made it look limit-related.

Are there any user-facing changes?

Yes — queries where a broadcast-promoted join feeds a sort-merge join now
return correct results instead of silently dropping rows. A plan that would hit
this now fails with an explicit error rather than producing a wrong answer. No
API changes.

Lowering a plan into stages rewrites nodes, and a rewrite can drop an ordering
its parent depends on. Promoting a `SortMergeJoinExec` to a broadcast
`HashJoinExec(CollectLeft)` produces a join that does not preserve input order,
so a parent `SortMergeJoinExec` reading that output merged unsorted input. A
merge join given unsorted input does not error: it stops matching once the
cursors diverge, silently dropping rows and returning a wrong answer.

Re-establish any hard input ordering a rewritten child no longer carries.
Sorts are only added, never removed, so an input that already satisfies its
requirement is left untouched.

Also validate the planned stages: every `SortMergeJoinExec` input must satisfy
the join's required ordering. An unsorted merge-join input is always a bug, and
failing loudly is better than returning wrong results. The check reports no
false positives across the full SF1 TPC-DS gate under both
`prefer_hash_join=true` and `false`.

TPC-DS q4 and q78 now match single-process DataFusion and join the correctness
gate, taking it from 86 to 88 queries. q78 was previously filed as "distributed
LIMIT drops rows": the wrong answer does not in fact depend on `ORDER BY` or
`LIMIT` and reproduces with neither.
@andygrove

Copy link
Copy Markdown
Member Author

Moving this to draft while I think about this more. Maybe the current broadcast optimization for SMJ is using the wrong approach.

@andygrove
andygrove marked this pull request as ready for review July 16, 2026 17:40
@andygrove

Copy link
Copy Markdown
Member Author

Moving this to draft while I think about this more. Maybe the current broadcast optimization for SMJ is using the wrong approach.

Let's go ahead and review this. It fixes the correctness issue. I'll keep investigating a better long term solution.

@avantgardnerio avantgardnerio left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would an extra pass of EnforceSorting be a robust alternative?

@andygrove

Copy link
Copy Markdown
Member Author

Would an extra pass of EnforceSorting be a robust alternative?

Good idea... I will open a separate PR with this approach so we can compare

@andygrove
andygrove marked this pull request as draft July 16, 2026 23:02
@andygrove

Copy link
Copy Markdown
Member Author

On second thoughts, the SMJ "broadcast" optimization was just a bad idea. I created #2072 to remove it.

@andygrove andygrove closed this Jul 16, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Broadcast join promotion drops a sort order required by a parent SortMergeJoinExec, silently producing wrong results

2 participants