jayzhan211 opened a new pull request, #25542: URL: https://github.com/apache/datafusion/pull/25542
## Which issue does this PR close? - Part of #15760. Follows #22038 and #25004. ## Rationale for this change Thanks @viirya for #22038 and #25004, which made `LEFT` / `FULL` / `LEFT SEMI` / `LEFT ANTI` / `LEFT MARK` nested loop joins spill with a multi-partition right side instead of failing. This PR keeps that capability and changes how it is achieved. The visited-left bitmap must be complete across all right partitions before the final left rows are emitted. `FallbackCoordinator` gets there by sharing each left chunk: one partition loads chunk N, every partition probes it, the last one to finish emits, and only then can chunk N+1 load. For users that lock-step means: - Draining the join's partitions one after another hangs on a multi-chunk left side (the tests note this and collect concurrently). `enable_nlj_coordinated_fallback = false` avoids it only for left-emitting joins; `INNER` / `RIGHT` share chunks the same way. - Dropping one unfinished partition fails the others with an execution error (the behaviour #25004 chose over hanging), unlike the in-memory path and `HashJoinExec`, where the others finish. - Every chunk is a barrier, so the slowest partition gates the rest. - The cancelled state lives on the plan, so it outlives one execution. ## What changes are included in this PR? Three commits, meant to be reviewed in order. It will be squashed on merge; the split is only there to make the last one easier to read. ### 1. `test:` coverage that does not depend on the redesign These pass on `main` as well, so they show the same behaviour before and after. - `test_nested_loop_join_spill_fuzz`: all 10 join types, right side spread over 4 partitions, comparing a run under a 4 KB limit with the same join without a limit, and asserting the limited run spilled. Until now no fuzz test reached the NLJ fallback. - `nested_loop_join_spill.slt`: multi-partition `LEFT MARK` and `LIMIT` cases. - `memory_limit/nlj_spill_unmatched.rs`: assert `spill_count > 0` exactly when a memory limit is set. These three regressions would otherwise keep passing if the join stopped spilling. ### 2. `chore:` small cleanups, no behaviour change Split out so they do not add noise to the next commit. - `execute()` computed the "can this join spill" condition twice, with the same long comment; compute it once. - Remove `left_buffered_in_one_pass`, which is written in three places and never read. - `process_left_unmatched_range` copied the visited bitmap bit by bit under an `assert!` whose message is `"DBG: .."`; use `BooleanBuffer::collect_bool`. - The module docs still said each output partition re-executes the left child; since #24677 it is executed once and spilled during that load. ### 3. `refactor:` the redesign The right side already needs no coordination: per-pass bitmaps are OR-ed into `global_right_bitmaps` and emission is deferred to `EmitGlobalRightUnmatched` (#24746). This does the mirror image for the left side. - `LeftSpillData`, already shared by all partitions through `OnceAsync<LeftLoad>`, gets one visited bitmap with one bit per spilled left row plus the probe-threads counter — the same two things `JoinLeftData` carries on the in-memory path. - Each partition reads its own chunks from the shared left spill file, with a private per-chunk bitmap, so probing takes no contended lock. When it finishes a chunk it merges the bitmap into the global one at the chunk's row offset. Offsets are positions in the spill file, so they agree across partitions even though chunk boundaries do not. - `ProbeEnd` is entered once per stream, after its last chunk, and reports completion on `LeftSpillData`. The last stream to report streams the spill file once more and emits the final left rows batch by batch. Partitions never wait on each other, so there is nothing to elect, release or cancel. `FallbackCoordinator`, `FallbackCoordinatorInner`, `CurrentChunk`, `Decision`, `LoadOutcome`, the cancel watcher, `cancelled_terminally` and the stream's `Drop` impl go away, as does the plan-level field and its rebuild in `replace_children`. `nested_loop_join.rs` goes from 7,547 to 5,997 lines: production code shrinks by about 500 lines and the test module by about 1,050. That includes the code I tidied in #25533, which this replaces. #### Keeping memory in check Chunks are no longer shared, so up to one chunk per partition is resident. My first version failed the existing 4-partition cases in `nested_loop_join_spill.slt`: with a 50K greedy pool each partition force-loaded its own copy of the single 40 KB left batch, and the `AggregateStream` above the join could no longer allocate. Two changes fix that, and both are needed: 1. `spill_left_input` writes each left batch as `right_partition_count` slices, so the batches every partition must hold to make progress add up to one input batch, the same bound as before. 2. Each partition's chunk is capped at an even share of what the in-memory load had reserved when it gave up, so the first partition to load cannot take everything under a first-come pool. `try_grow` stays the real limit; the cap only stops one partition from taking the others' share. With `FairSpillPool` the cap is not needed, and a single partition is not capped. (1) alone is not enough: with the cap disabled the slt still fails on every run, with one chunk holding 40.7 KB and the other partitions forcing a 10 KB slice each into the 50K pool. Limitations I know of: - If not even the first left batch fit, the share is zero and every chunk is one slice, so right-side passes multiply by the partition count. Results are unaffected; the cost is per-pass overhead. The slt cases are in this regime, because their left side is a single batch. - The share is a snapshot from the moment the load failed. If memory frees up later the chunks stay small; if it gets tighter only `try_grow` protects the pool. - The load is measured on input batches while chunks are built from batches read back from the spill file. `get_array_memory_size` over-reports sliced input, which can make the cap too generous. That weakens the cap but cannot breach the pool. #### Questions for reviewers - Is "reserved bytes when the load gave up / partition count" an acceptable chunk budget, or would you rather derive it from `MemoryPool::memory_limit()` when it is finite? - Is there a constraint behind sharing chunks that I am missing, for instance very large left batches, where one chunk per partition is a real cost? - `enable_nlj_coordinated_fallback` keeps its meaning — an engine with one plan instance per partition would never drive the shared counter to zero, and would now lose the final left rows rather than stall, as on the in-memory path — but its name describes a mechanism that no longer exists. Leave it, or rename before 56 ships? ## What is the testing strategy for this PR? - The tests in the first commit pass on `main` and after the redesign. - The multi-partition unit tests now use a multi-chunk left side and run for all five left-emitting join types, both draining the partitions one after another (which hung before) and concurrently. Before, only `LEFT` and `FULL` had multi-chunk coverage. - `test_nlj_memory_limited_dropped_partition_does_not_fail_peers`: dropping a partition mid-fallback leaves the others unaffected and no memory reserved. - `test_nlj_memory_limited_partitions_share_the_chunk_budget`: the load fits 3 of 8 left batches (36,864 bytes), each of 4 partitions gets 9,216, and with all four chunks resident the pool stays within its limit. With the cap disabled the same test sees 67,800 bytes reserved against a 43,008 byte pool. - About 1,070 lines of tests that pinned the coordinator's cancellation protocol are removed with it. - Extended test suite: 11,993 passed, 0 failed. - `dfbench nlj -n 4` with 40K–150K limits, where both versions spill (`spill_count=5`): no measurable difference, for example Q7 about 620 ms against 640–665 ms and Q10 about 447 ms against 453 ms. The built-in NLJ queries are small, so this says the fallback did not get slower; I have not measured a large build side or peak RSS. ## Are there any user-facing changes? - A spilling nested loop join no longer requires its partitions to be polled concurrently, and dropping one partition no longer fails the others. - `datafusion.execution.enable_nlj_coordinated_fallback` keeps its behaviour; its description is reworded in `config.rs`, `configs.md`, `information_schema.slt` and the 56.0.0 upgrade guide. - No public API changes. -- 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]
