Use ProbeCompletion in NestedLoopJoinExec #26148

Is your feature request related to a problem or challenge?

When a join shares one build side across probe partitions, the last partition to finish emits the build-side rows, for example the unmatched rows of a LEFT or FULL join. The rule is simple, but it is only correct if four things hold:

1. Each partition reports once. In fix: NestedLoopJoinExec emits spurious unmatched-left rows with multiple probe partitions #22791, NLJ re-entered its emit state and decremented the counter a second time. The counter reached zero while other partitions were still probing, so a partition emitted unmatched left rows too early, and some rows appeared twice.
2. The count covers every partition that shares the matches. In fix: refuse memory-limited NestedLoopJoin fallback for left-emission joins with a multi-partition probe side #24675 (Correctness guard: disable memory-limited NestedLoopJoin fallback for LEFT-family joins with multi-partition right side #22641), the spill path built a per-chunk JoinLeftData with a counter of 1. Every partition then thought it was last and emitted from its own partial bitmap. That returned wrong rows, for example 49 instead of 19 for a LEFT join.
3. A partition that goes away still reports. In fix: cancel the NestedLoopJoin coordinated fallback when a partition is dropped unfinished #25004, a dropped NLJ partition never reported. No partition became last, and the others waited forever.
4. The last partition reads the shared state only after its own decrement, with an ordering that makes the other partitions' writes visible. In fix: null-aware anti join could emit rows after a sibling probe partition saw NULL #25076, hash join read the shared "saw NULL" flag before decrementing a Relaxed counter, so NOT IN returned rows it must suppress.

#25076 introduced ProbeCompletion (joins/hash_join/probe_completion.rs) to make point 4 part of the type instead of a convention:

• The shared facts are only available through the return value of report_completed(), so they cannot be read before the decrement.
• The decrement is AcqRel.
• The orderings are written once and checked with loom.

NestedLoopJoinExec still provides all four guarantees by hand, in two places with different orderings:

• In memory: JoinLeftData::probe_threads_counter uses Relaxed. That is correct only because the bitmap happens to sit behind a Mutex, so every reader has to work that out again.
• Spill path: LeftSpillData::{probe_threads_counter, incomplete} and depart_unfinished add a separate flag for point 3.
• Unused counter: the spill path still creates a per-chunk JoinLeftData whose counter is always 1 and, per its comment, "is not used". That is the shape that caused fix: refuse memory-limited NestedLoopJoin fallback for left-emission joins with a multi-partition probe side #24675.

Moving NLJ onto ProbeCompletion brings:

• Less to understand: one documented type instead of two counters, a flag and an unused counter, each with its own reasoning.
• Loom coverage: the existing loom model also covers NLJ.
• One place to build on: upcoming work reuses the same type instead of adding another copy. This includes hash join spilling ([EPIC] Spilling Hash Join — run any join in a bounded memory budget #24768), NLJ semi/anti pruning (perf: Prune matched right rows in nested loop semi and anti joins #25413), and sharing join state for distributed CollectLeft (Proposal: Hook to better support CollectLeft joins in distributed execution #12454).
• Low risk: it is about 200 lines with no intended behaviour change. Going from Relaxed to AcqRel costs one fence per partition, not per row.

Describe the solution you'd like

Use ProbeCompletion in NestedLoopJoinExec too:

1. Move hash_join/probe_completion.rs to joins/probe_completion.rs.
2. Add a call for a partition that stops early, for example abandon(). It still counts down, so nothing waits forever (point 3), and it tells the last partition not to emit. Add a loom test for it.
3. Replace NLJ's two counters, the incomplete flag and the unused per-chunk counter with ProbeCompletion.

Tests to run:

• the existing NLJ and hash join unit tests, joins.slt, nested_loop_join_spill.slt and join_fuzz
• RUSTFLAGS="--cfg datafusion_loom" cargo test -p datafusion-physical-plan --lib loom_tests. CI does not run this, so please paste the result in the PR.

Describe alternatives you've considered

No response

Additional context

Possible follow-ups:

• PiecewiseMergeJoinExec, which has a third copy of this counter (SeqCst). Wait until perf(pwmj):settle never-matching streamed rows against the buffered extreme… #25840 lands.
• Pass the matched-rows bitmap to the last partition through the same type.
• Make point 1 part of the type too, with a guard that reports when it is dropped.

#25413 also changes nested_loop_join.rs.