Repository navigation
Use ProbeCompletion in NestedLoopJoinExec #26148
Copy link
Copy link
Open
Labels
enhancementNew feature or requestNew feature or requesthelp wantedExtra attention is neededExtra attention is needed
Description
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:
- 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.
- 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
JoinLeftDatawith a counter of1. 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. - 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.
- 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
Relaxedcounter, soNOT INreturned 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_counterusesRelaxed. That is correct only because the bitmap happens to sit behind aMutex, so every reader has to work that out again. - Spill path:
LeftSpillData::{probe_threads_counter, incomplete}anddepart_unfinishedadd a separate flag for point 3. - Unused counter: the spill path still creates a per-chunk
JoinLeftDatawhose counter is always1and, 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 supportCollectLeftjoins in distributed execution #12454). - Low risk: it is about 200 lines with no intended behaviour change. Going from
RelaxedtoAcqRelcosts one fence per partition, not per row.
Describe the solution you'd like
Use ProbeCompletion in NestedLoopJoinExec too:
- Move
hash_join/probe_completion.rstojoins/probe_completion.rs. - 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. - Replace NLJ's two counters, the
incompleteflag and the unused per-chunk counter withProbeCompletion.
Tests to run:
- the existing NLJ and hash join unit tests,
joins.slt,nested_loop_join_spill.sltandjoin_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.
Reactions are currently unavailable
Metadata
Metadata
Assignees
Labels
enhancementNew feature or requestNew feature or requesthelp wantedExtra attention is neededExtra attention is needed
Activity
[-]Joins: one shared protocol for "the last probe partition emits the build-side rows"[/-][+]Use `ProbeCompletion` in `NestedLoopJoinExec`[/+]jeffoodchain commentedon Oct 9, 2026
@jayzhan211 I will be working on this.