[ewoodbury]

Which issue does this PR close?

Closes #2847.

Related: #1055 (the same bug for broadcast hash joins) and #2849 (a similar gap for null-aware anti joins in the static planner).

Rationale for this change

A nested-loop join reads its left (build) side in full and streams its right (probe) side. For LEFT, FULL, semi, anti and mark joins, it also outputs left-side rows based on whether they matched any right-side row. For example, a LEFT JOIN outputs (left row, NULL) for each left row that matched nothing. DataFusion keeps track of those matches in memory shared by all of the probe side's partitions, and outputs these rows once the last partition is done.

Ballista can split the probe side across several tasks, often on different executors. Each task then sees only its share of the right side and decides on its own whether a left row matched. The query returns wrong results, with no error:

• A left row with no match anywhere is output once per task instead of once.
• A left row that matched in one task is still output as unmatched by the other tasks.
• EXISTS returns a row once for each task it matched in, and NOT EXISTS returns rows that did have a match.

For example, with orders read from four files and one partition per task:

SELECT a.o_id, b.o_id FROM orders a LEFT JOIN orders b
ON b.o_id < a.o_id AND b.o_id < 2 WHERE a.o_id < 3 ORDER BY 1, 2

DataFusion returns 4 rows and Ballista returns 13, with both the adaptive and the static planner.

The bug is easy to miss locally. With the default ballista.scheduler.max_partitions_per_task (unbounded), one task usually takes the whole stage. On a cluster, a stage is split whenever it has more partitions than one executor has free slots.

Hash joins are already protected: a broadcast hash join of these types is turned back into a partitioned join, and a null-aware anti join runs its probe side in one task. Nested-loop joins had no such check.

What changes are included in this PR?

Nested-loop joins of these types (the ones collect_left_broadcast_safe rejects) now run as a single task:

• Adaptive planner: DistributedExchangeRule already adds a coalesce under the probe side of a null-aware anti join. It now does the same for these nested-loop joins. The stage that produces the probe side stays parallel; only the join runs as one task.
• Static planner (also used by pipelined mode): the probe side gets its own one-partition stage. Adding a CoalescePartitionsExec under the join is not enough here. When an aggregate sits above the join, the planner runs EnsureRequirements over the join's stage again, and that removes the coalesce because the join does not ask for one.

Tests:

• nested_loop_join.rs runs LEFT JOIN, FULL JOIN, EXISTS, NOT EXISTS, a mark join (... OR EXISTS) and a GROUP BY over a LEFT JOIN under both planners with one partition per task, and compares the results with DataFusion. Every case fails on main.
• A unit test checks that the adaptive rule coalesces the probe side of a LeftAnti join and leaves an Inner join alone.

Are there any user-facing changes?

No API or config changes. These joins now return correct results, at the cost of running the join itself in one task. INNER joins and the right-side join types are unchanged.

TPC-H plan-stability goldens are unchanged: its only nested-loop joins (q11, q22) are INNER.