ewoodbury opened a new pull request, #2848: URL: https://github.com/apache/datafusion-ballista/pull/2848
# Which issue does this PR close? Closes #2847. Related to #1055, the same problem for broadcast hash joins. # Rationale for this change A nested-loop join collects its build (left) side and keeps the record of which build rows matched in memory shared by its probe partitions. The last probe partition to finish then emits the build-side rows: unmatched rows for `Left` and `Full`, and semi/anti/mark rows for `LeftSemi`, `LeftAnti` and `LeftMark`. Ballista runs probe partitions as separate tasks, so each task decides on its own slice of the probe side. A build row unmatched in that slice is emitted as unmatched even if it matched in another task, and a semi row is emitted once per task it matched in, and a mark row once per task. This is on `main` with both the adaptive and the static planner whenever a stage is split into more than one task. In a local standalone cluster with `ballista.scheduler.max_partitions_per_task = 1`, `orders a LEFT JOIN orders b ON b.o_id < a.o_id` returns `(0, NULL)` four times, plus `(1, NULL)` and `(2, NULL)` rows that do not exist, and a `NOT EXISTS` with a non-equi correlation returns rows the subquery matched. There is no error. With the default unbounded `max_partitions_per_task` the problem is hidden locally, because one task usually takes the whole stage. Hash joins already have this guard: unsafe `CollectLeft` joins are demoted, and null-aware anti joins run their probe side in one task. Nested-loop joins had no equivalent. # What changes are included in this PR? - Adaptive planner: `DistributedExchangeRule` now applies its null-aware probe-side coalesce to nested-loop joins whose type is not `collect_left_broadcast_safe`. The probe stage stays parallel and the join runs as one task. - Static planner (also used by pipelined mode): such a join's probe side gets a one-partition stage of its own. An inline `CoalescePartitionsExec` does not hold there: with an aggregate above the join, `EnsureRequirements` removes it because the join does not require it. - Tests: an integration test compares Left, Full, semi, mark and anti non-equi self-joins, plus an aggregate over a Left join, against DataFusion under both planners with one partition per task. Every case fails on `main`. A unit test pins that the adaptive rule coalesces a `LeftAnti` probe and leaves an `Inner` one alone. # Are there any user-facing changes? No API changes. Nested-loop joins of these types now run the join in a single task, so they are correct but less parallel. Inner, Right, RightSemi, RightAnti and RightMark nested-loop joins are unchanged. TPC-H plan-stability goldens are unchanged. Its only nested-loop joins (q11, q22) are Inner. -- 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]
