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

[andygrove]

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):

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 (e843c8e) under the default static planner.
• Surfaced by the TPC-DS correctness gate (Add TPC-DS workflow #1845 / Add TPC-DS correctness testing #2048); accounts for the q78
  half of Distributed execution returns incorrect results for several TPC-DS queries (static planner) #2046. q4 (also in Distributed execution returns incorrect results for several TPC-DS queries (static planner) #2046) is not yet diagnosed.
• Related: Left/full outer join incorrect for CollectLeft / broadcast #1055 (broadcast join-type safety), Propagate input ordering across pass-through (non-repartitioning) shuffles #2006 (propagating ordering
  across pass-through shuffles).

[andygrove]

Root-caused and fixed in #2066. The promotion replaces the inner SortMergeJoinExec with a HashJoinExec(CollectLeft), which does not preserve input order, leaving the parent SMJ to merge unsorted input.

The fix re-establishes any hard input ordering a rewritten child no longer carries, and adds the suggested validation: every SortMergeJoinExec input must satisfy the join's required ordering, so this surfaces as an error rather than wrong results. The check reports no false positives across the full SF1 TPC-DS gate under both prefer_hash_join=true and false.

It also fixes q4, so with #2066 the gate goes 86 → 88.