fix: emit deferred unmatched rows when memory-limited NestedLoopJoin exhausts its left side

[viirya]

Which issue does this PR close?

• Closes Memory-limited NestedLoopJoin silently drops unmatched rows when it spills #24745.

Rationale for this change

A memory-limited NestedLoopJoinExec silently returned fewer rows than the same query with an ample memory pool. No error was raised, so the loss is invisible to the caller.

In memory-limited mode the left side is processed in chunks and the right side is replayed for each chunk. A per-right-batch match bitmap therefore cannot be emitted as soon as one chunk finishes probing it — a later chunk may still match those rows. The operator already handles this by merging each bitmap into SpillStateActive::global_right_bitmaps and deferring emission to NLJState::EmitGlobalRightUnmatched, which is reached from handle_emit_left_unmatched once the left side is exhausted.

handle_buffering_left had a separate early exit:

if active.pending_batches.is_empty() {
    // No data at all — go directly to Done
    self.left_exhausted = true;
    self.state = NLJState::Done;
    return ControlFlow::Continue(());
}

Despite the comment, this also fires on the load that follows the final left chunk, when the left side is exhausted and there is nothing more to buffer. Ending the stream there discards the accumulated bitmaps, so every probe-side row that no chunk matched is dropped.

Instrumenting a failing LEFT JOIN showed 39 bitmap merges and zero emissions, with the completion path only ever running while left_exhausted was still false.

The affected queries reach the operator as swapped join types (Right, RightAnti, RightSemi) or as Full, so the rows needing unmatched emission are on the operator's probe side. Measured against an unlimited pool (l = 200 rows, r = 90 rows, target_partitions = 1, batch_size = 16, 64-byte pool):

query	ample	memory-limited (before)
NOT EXISTS (... l.v > r.w)	2	0
LEFT JOIN ... ON l.v > r.w	9336	9334
FULL JOIN ... ON l.v > r.w	9339	9337

INNER and explicit RIGHT JOIN were already correct.

This is pre-existing rather than a regression from #24675: that PR only changes the spill gate in the same file, and its condition (need_produce_result_in_final(join_type) && right_partition_count > 1) does not cover this path.

What changes are included in this PR?

Route that early exit to EmitGlobalRightUnmatched instead of Done when the join tracks unmatched probe-side rows, clearing right_data so a fresh replay pass is opened. A left side that was empty from the start keeps its previous behaviour: no bitmaps have been accumulated, so that state reports nothing unmatched and finishes immediately.

Spilling is preserved rather than refused — the fix corrects the emission instead of turning these queries into ResourcesExhausted errors.

Are these changes tested?

Yes, datafusion/core/tests/memory_limit/nlj_spill_unmatched.rs adds three tests comparing memory-limited results against an ample pool for LEFT JOIN, LEFT ANTI, and FULL JOIN. Without the production change they fail with 0 instead of 2, 9334 instead of 9336, and 9337 instead of 9339; with it they pass.

The existing nested_loop_join unit tests (46) and the wider joins suite (1116) pass unchanged, including test_nlj_memory_limited_right_join, which asserts that a spilling RIGHT join still returns its unmatched rows.

Are there any user-facing changes?

No API changes. Queries that previously lost rows under a memory limit now return the correct result.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 75.00000% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 81.47%. Comparing base (c4910e0) to head (9fef857).

Files with missing lines	Patch %	Lines
...fusion/physical-plan/src/joins/nested_loop_join.rs	75.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24746      +/-   ##
==========================================
- Coverage   81.47%   81.47%   -0.01%     
==========================================
  Files        1122     1122              
  Lines      403629   403632       +3     
  Branches   403629   403632       +3     
==========================================
- Hits       328866   328863       -3     
- Misses      55510    55511       +1     
- Partials    19253    19258       +5     

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

[sunchao]

Thanks, Liang-Chi. LGTM.

I reviewed 9fef857 against c4910e0 and found no blocking issues. The terminal empty-load path now emits the accumulated right-side results, and clearing right_data correctly opens a fresh replay after the previous pass has finished.

Validation:

• All three new regression tests fail on the base and pass with this fix.
• All 1,116 existing join tests pass.
• All 540 additional direct-NLJ cases match an independent row oracle, including 420 spilling cases. These cover the five right-emission join types, NULLs, duplicates, empty inputs/projections, and chunk boundaries. The 164 failing base cases all pass on this head.

All 41 current CI checks are green. I did not rerun the full workspace suite locally.

[viirya]

Thank you @sunchao !