fix: refuse memory-limited NestedLoopJoin fallback for left-emission joins with a multi-partition probe side

[ranflarion]

Which issue does this PR close?

• Closes Correctness guard: disable memory-limited NestedLoopJoin fallback for LEFT-family joins with multi-partition right side #22641.

Rationale for this change

The memory-limited fallback returns wrong rows for join types whose final emission reads the visited-left bitmap (LEFT, LEFT SEMI, LEFT ANTI, LEFT MARK) when the probe side has more than one partition. The fallback rebuilds a per-partition JoinLeftData per left chunk with probe_threads_counter == 1 and a per-partition bitmap, so every partition believes it is the last to finish probing and emits its own unmatched set, from a bitmap that only saw its own right rows. Unmatched rows are emitted once per partition, and rows matched only in another partition are emitted as unmatched.

Proof, using this file's own build_table / prepare_join_filter / multi_partitioned_join_collect fixtures (4 right partitions) with a 100-byte memory limit versus unbounded:

for join_type in [JoinType::Left, JoinType::LeftAnti] {
    // unbounded memory -> correct; 100-byte pool -> memory-limited fallback
    let (_, correct, _) = multi_partitioned_join_collect(left, right, &join_type, Some(filter), ample_ctx).await?;
    let (_, wrong, _)   = multi_partitioned_join_collect(left, right, &join_type, Some(filter), tight_ctx).await?;
}
// Left:     correct=19  memory_limited=49
// LeftAnti: correct=1   memory_limited=31

The existing gate covers only FULL. This extends it to every need_produce_result_in_final type, so an over-budget build side fails with ResourcesExhausted instead of returning wrong rows. Right-side emission types (RIGHT, RIGHT SEMI, RIGHT ANTI, RIGHT MARK) keep the fallback: each partition owns its right rows exclusively, so their bitmaps are complete per partition. Proper cross-partition coordination of the left bitmap, which would re-enable the fallback for these types, is #22038.

What changes are included in this PR?

The full_join_multi_partition condition in NestedLoopJoinExec::execute becomes need_produce_result_in_final(self.join_type) && right_partition_count > 1, with the comment updated to describe the failure mode. test_overallocation moves LEFT/LEFT SEMI/LEFT ANTI/LEFT MARK from the succeed-via-fallback group (which collected but never checked the row values) into the must-OOM group alongside FULL.

Are these changes tested?

test_overallocation now asserts the ResourcesExhausted refusal for all five gated join types with a multi-partition probe side, and still asserts fallback success for Inner and the right-emission types. All 42 nested_loop_join tests pass; ./dev/rust_lint.sh is clean.

Are there any user-facing changes?

Left-family nested loop joins with a multi-partition probe side whose build side exceeds the memory budget now fail with ResourcesExhausted instead of returning incorrect results. No API changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.70588% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.47%. Comparing base (3cabfa9) to head (ce9440f).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...fusion/physical-plan/src/joins/nested_loop_join.rs	89.70%	0 Missing and 7 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24675      +/-   ##
==========================================
+ Coverage   81.45%   81.47%   +0.01%     
==========================================
  Files        1122     1122              
  Lines      403051   403629     +578     
  Branches   403051   403629     +578     
==========================================
+ Hits       328317   328855     +538     
- Misses      55498    55517      +19     
- Partials    19236    19257      +21     

☔ 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.

[kosiew]

@ranflarion,

Thanks for working on this. The change looks good to me. Expanding the fallback gate to the join types that depend on the visited-left bitmap avoids returning incorrect results when there are multiple right partitions, while keeping the fallback available for the join types where it is safe.

I have one non-blocking suggestion for some additional test coverage.

[kosiew]

Could we add a tight-memory, single-right-partition spill test for LeftSemi, LeftAnti, and LeftMark as well? The new gate specifically applies when there is more than one right partition, but the current single-partition success coverage only exercises Left and Full. Adding these cases would help make sure the supported single-partition fallback continues to work while #22038 remains open.

[ranflarion]

sure, added in ab71c9c: test_nlj_memory_limited_left_semi_join, test_nlj_memory_limited_left_anti_join, test_nlj_memory_limited_left_mark_join, same shape as the existing single-partition tests, 50-byte pool with spill_count asserted > 0 so the fallback path is actually exercised, results snapshot-checked against the unlimited-memory expectations.

[jayzhan211]

Thanks @ranflarion and @kosiew, LGTM

[kosiew]

@ranflarion
Can you resolve the merge conflicts?

[ranflarion]

@ranflarion Can you resolve the merge conflicts?

done