refactor: track NestedLoopJoin fallback matches in one left bitmap and drop the cancel protocol

[jayzhan211]

Which issue does this PR close?

• Part of Memory limited nest loop join #15760. Follows Share per-chunk JoinLeftData across right partitions in NLJ memory-limited fallback #22038 and fix: cancel the NestedLoopJoin coordinated fallback when a partition is dropped unfinished #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, keeps the way left chunks are scheduled, and changes how matches are tracked.

Two user-visible problems today:

• Dropping one unfinished partition of a spilling join fails the other partitions with an execution error (the behaviour fix: cancel the NestedLoopJoin coordinated fallback when a partition is dropped unfinished #25004 chose over hanging). On the in-memory path, and in HashJoinExec, the others simply finish. fix: cancel the NestedLoopJoin coordinated fallback when a partition is dropped unfinished #25004 asked reviewers to weigh in on "letting survivors finish"; nobody did, so I am picking that question up here.
• About 1,270 lines of nested_loop_join.rs (roughly 300 of production code, the rest tests and test seams) exist only to make that early drop safe: cancel_notify, registered watchers polled on every poll_next, pending_drop / coordination_started, a select! racing each chunk load against cancellation, a publish guard, and cancelled_terminally.

Both come from one coupling. On the in-memory path the shared probe counter only decides who emits the unmatched left rows. In the fallback the same per-chunk counter also decides when the next chunk may load, and the partition it elects holds the only release_chunk call while it emits. A partition that goes away can therefore strand the chunk, and the loader going away takes the shared left stream with it.

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 this change

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

• 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 fix: spill the NestedLoopJoin build side from the pass that consumed it #24677 it is executed once and spilled during that load.

3. refactor: separate match tracking from chunk scheduling

The schedule does not change. All partitions still share one left chunk at a time, as large as memory allows, and the next chunk loads once every partition has finished the current one. That is what keeps right-side replays at a minimum, and it is the part of the current design worth keeping (see the measurements below).

Match tracking moves to one bitmap over the whole spilled left side, with one final emission pass. This is the mirror image of what #24746 did for the right side (global_right_bitmaps + EmitGlobalRightUnmatched), and it is what other engines do: DuckDB's PhysicalNestedLoopJoin keeps one OuterJoinMarker over the whole build side and emits unmatched build rows in a separate final scan; Spark's BroadcastNestedLoopJoinExec ORs per-task bitsets into one and emits as a final step.

• LeftSpillData, already shared by all partitions through OnceAsync<LeftLoad>, gets the bitmap (one bit per spilled left row, addressed by the row's position in the spill file) and the probe-threads counter — the same two things JoinLeftData carries on the in-memory path.
• While probing a chunk each partition uses a private bitmap, so the probe path takes no contended lock, and merges it into the global one when it finishes the chunk.
• ProbeEnd is entered once per stream, after its last chunk. The last stream to report streams the left spill file once more and emits the final left rows batch by batch.

With no per-chunk emitter, what is left of the coordinator is a count, LeftChunkBarrier: finished == live means move on to the next chunk.

• The chunk load is a shared future, the same pattern as OnceFut for the whole left side. Whichever waiting partition is polled drives it and an error reaches all of them, so it no longer matters which partition started the load or whether it is still around. Decision, loader_in_flight, abandon_load and the select! go away.
• A stream dropped before Done departs: live -= 1 (and finished -= 1 if it had already finished the current chunk). The others carry on. Because the departed stream never reports probe completion, the final left rows are not emitted — exactly what happens on the in-memory path.
• The barrier's clone of the load keeps the chunk alive for partitions that reach it late and lets go of it when the barrier moves on, so the chunk's reservation follows the data, as asked for in Share per-chunk JoinLeftData across right partitions in NLJ memory-limited fallback #22038.

nested_loop_join.rs goes from 7,547 to 6,226 lines: production code −262, tests −1,059 (24 cancellation tests and their helpers are replaced by three drop scenarios).

What I tried first, and why this PR looks different from its first version

The first version of this PR also gave every partition its own left chunk, so that partitions never wait for each other. It was simpler still, but it costs what the shared chunk saves: memory split P ways means chunks P times smaller and P times more right-side replays. On a wide-row workload (below) main makes 7 passes over the right side and that version made 98, for about 1.6× the time and 1.5× the RSS. A variant sharing chunks through futures without a barrier reached time parity only at one hand-tuned chunk size, and never main's memory. A partition that runs ahead has to either wait or hold a second chunk; I kept the wait.

What is the testing strategy for this PR?

• The tests in the first commit pass on main and after this change.
• The multi-partition unit tests now use a multi-chunk left side for all five left-emitting join types (before, only LEFT and FULL had multi-chunk coverage).
• Three drop scenarios, each asserting that the remaining partition finishes within a timeout, without error, and that no memory stays reserved: dropped while holding a chunk (..._dropped_partition_does_not_stall_peers), dropped after finishing the current chunk while waiting for the others (..._partition_dropped_while_waiting_for_peers, which also checks the barrier's counts), and dropped right after starting a chunk load (..._partition_dropped_while_loading).
• test_nlj_memory_limited_releases_memory_{left,full}_join: nothing stays reserved after completion while the plan is still alive.
• Extended test suite (--features avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption): 11,989 passed, 0 failed.
• Performance, datafusion-cli release builds of this branch and of the main commit it is based on, 16 partitions, 24 MB limit, LEFT JOIN with a 50 MB build side (50K rows with a 1 KB string), interleaved runs on a machine that was not idle:

right side	main	this PR
right-side passes (input_rows / 60K)	7	7
60K rows × 1 KB, fair pool	0.78–0.88 s, 422–466 MB RSS	0.81–0.96 s (one 1.38 s outlier), 453–481 MB
60K rows × 1 KB, greedy pool	0.86–1.12 s, 408–505 MB	0.81–1.02 s, 462–499 MB
600K rows × 8 bytes (compute-bound)	7.7–10.3 s, 345–392 MB	8.0–11.7 s, 360–425 MB

I read that as no measurable difference in time and RSS within about 5%.

Are there any user-facing changes?

• Dropping one unfinished partition of a spilling nested loop join no longer fails the other partitions; they finish, and the final unmatched left rows are not emitted, as on the in-memory path. This is the behaviour change I would most like reviewers to weigh in on.
• Unchanged: partitions of a spilling join still have to be polled concurrently (that is inherent to sharing one chunk, and not reachable from planned queries, since the join declares no output ordering and nothing parks one partition above it); datafusion.execution.enable_nlj_coordinated_fallback and its documentation are untouched.
• No public API changes.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 82.39%. Comparing base (f7e2db3) to head (c367337).
⚠️ Report is 8 commits behind head on main.

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #25542    +/-   ##
========================================
  Coverage   82.39%   82.39%            
========================================
  Files        1138     1138            
  Lines      434592   434390   -202     
  Branches   434592   434390   -202     
========================================
- Hits       358073   357910   -163     
+ Misses      54853    54839    -14     
+ Partials    21666    21641    -25     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.