andygrove opened a new pull request, #26330:
URL: https://github.com/apache/datafusion/pull/26330
> [!WARNING]
> **Work in progress / experimental.** This draft is for discussion. It is
not ready for review or merge.
>
> **Companion Ballista draft PR:** BALLISTA_PR_LINK
## Which issue does this PR close?
- No dedicated issue yet, because this is an experimental draft.
- Related: #23194 (adaptive query execution), #25798 (`StageBoundary`),
apache/datafusion-ballista#1900, apache/datafusion-ballista#2315.
## Rationale for this change
Engines such as Ballista and datafusion-distributed run each output
partition of a physical plan as an isolated task: a separate plan instance,
often in another process. In that setting, DataFusion's default join selection
can return wrong results.
`JoinSelection` collects (broadcasts) a small build side for every join
type, and nested-loop joins always collect their left side. For `Left`,
`LeftSemi`, `LeftAnti`, `LeftMark` and `Full` joins, the build-side rows those
joins emit depend on match state that all output partitions share. The last
probe partition to finish emits them. When partitions run as separate plan
instances, every task emits those rows, or none does:
- Ballista saw duplicated rows: TPC-DS q77 came back 16x inflated
(apache/datafusion-ballista#2315).
- A new test in this PR reproduces the same root cause inside DataFusion:
`small LEFT JOIN big` loses its unmatched `small` rows.
Two engines have independently re-implemented the same fix downstream, and
their copies already differ:
- **Ballista:** `DemoteUnsafeBroadcastJoinRule` and
`collect_left_broadcast_safe` (apache/datafusion-ballista#1900).
- **datafusion-distributed:** `normalize_collect_joins.rs` and
`is_left_broadcast_safe`.
This PR lets an engine declare its execution model with one option, so the
fix lives in one place. That follows the proposed physical optimizer contract
(#26107): use configuration rather than editing the default rule list. It also
matches the existing `datafusion.execution.enable_file_stream_work_stealing`,
which was added for the same execution model.
## What changes are included in this PR?
- **New option** `datafusion.execution.isolated_partitions` (default
`false`). When it is set, `JoinSelection` never collects the build side of a
join that emits build-side rows:
- **Hash joins:** a `CollectLeft` candidate is only used if its join type
is safe *after any swap*. Otherwise the join is `Partitioned`. If both inputs
are under the thresholds and only the swapped orientation is safe, the inputs
are swapped. That swap only happens when `join_reordering` is enabled.
- **Nested loop joins:** `Left`, `LeftSemi`, `LeftAnti` and `LeftMark` are
swapped to the probe-emitting join type. There is no partitioned form to fall
back to.
- **Never unsafe:** a safe join is never swapped into an unsafe one, so
re-running the rule is stable. Ballista's adaptive planner re-runs it after
every stage.
- **Exceptions:** null-aware joins (`NOT IN`, planned as `LeftAnti` or
`LeftMark`), `Full` nested-loop joins and piecewise merge joins are unchanged.
The option's documentation says engines must still handle them, for example by
running them in a single task.
- **Not covered:** execution-time coordination between partitions of a
plan instance. Dynamic join filters and the memory-limited nested-loop join
fallback still expect every partition of the instance to run, and the option's
documentation says so.
- **New public method** `JoinType::emits_build_side_rows()`. It is the
classification core already used privately (`need_produce_result_in_final`,
which now delegates to it).
With the option off, nothing changes: no existing plan, EXPLAIN output or
test expectation is modified.
## What is the testing strategy for this PR?
- **`join_isolated_partitions.slt`:** EXPLAIN tests and result checks for
every case:
- Left, Full, LeftSemi, LeftAnti and LeftMark hash joins
- the safe swap
- a residual join filter and projection
- null-aware `NOT IN`, both as an anti join and inside an `OR` (a mark
join)
- Left and Full nested-loop joins
- the interaction with `repartition_joins = false`
- **`core/tests/physical_optimizer/join_selection.rs`:**
- Rule-level plan snapshots for each decision, including `join_reordering
= false` and re-running the rule.
- A test that plans a query once per output partition and runs each
partition on its own plan instance, the way isolated-task engines do. With the
option off, the results differ from normal execution; with it on, they match.
The test fails without the `JoinSelection` change.
- **`JoinType::emits_build_side_rows`:** a unit test covering every variant.
**Downstream check, not part of this PR** (companion Ballista draft PR:
BALLISTA_PR_LINK):
- **Setup:** Ballista, with this change backported onto DataFusion 55.2.0
([`andygrove/[email protected]`](https://github.com/andygrove/datafusion/tree/isolated-partitions-55.2))
and the option enabled. `DemoteUnsafeBroadcastJoinRule` was deleted, and
`collect_left_broadcast_safe` was replaced by
`JoinType::emits_build_side_rows`. That removes 173 lines of Ballista source
and adds 17.
- **Results:** Ballista's test suite passes (1,599 passed, 0 failed), and
its 22 approved TPC-H distributed plans are unchanged.
## Are there any user-facing changes?
Yes, both additive (one caveat below):
- The new option `datafusion.execution.isolated_partitions`, off by default,
documented in `configs.md`.
- The new public method `JoinType::emits_build_side_rows()`.
There are no breaking behaviour changes. One API caveat: `config_namespace!`
structs are not `#[non_exhaustive]`, so code that builds `ExecutionOptions`
with an exhaustive struct literal must add the new field.
--
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]
