NestedLoopJoinExec spill path: untracked allocation overshoots memory pool

[avantgardnerio]

Describe the bug

NestedLoopJoinExec's memory-limited (spill) path allocates more memory than it reserves through the MemoryPool. The accounting-pool framework added in #22626 surfaces this if HEADROOM_FACTOR in the SLT pool is tightened — see #22721, which lowers it from 8.0 → 5.0 and trips this test.

To Reproduce

In datafusion/sqllogictest/src/accounting_pool.rs, set:

```rust
const HEADROOM_FACTOR: f64 = 5.0;
```

Then run nested_loop_join_spill.slt. First query at line 33:

```sql
SET datafusion.execution.target_partitions = 1;
SET datafusion.runtime.memory_limit = '150K';

SELECT count(*) as cnt, min(v1) as mn, max(v1) as mx
FROM generate_series(1, 100000) AS t1(v1)
INNER JOIN generate_series(1, 1) AS t2(v2)
ON (t1.v1 + t2.v2) > 0;
```

Expected behavior

Query succeeds with allocator usage bounded by the configured memory_limit (×~10% slop for legitimate untracked overhead like Tokio/Rayon thread state).

Actual behavior

```
External error: 1 errors in file datafusion/sqllogictest/test_files/nested_loop_join_spill.slt

1. query failed: Other Error: allocator overdraft: account balance at panic = -20245 bytes
   at nested_loop_join_spill.slt:33
   ```

Peak allocator usage reaches ~770KB against a declared 150KB pool — 5.13× the budget. The first ~750KB (5×) is absorbed by the headroom factor; the additional 20245 bytes is allocator usage that was never reserved through MemoryPool::try_grow.

Likely root cause

Spill path in NestedLoopJoinExec has at least these untracked allocation sites (need verification with targeted instrumentation):

• generate_next_batch buffering — RecordBatches accumulated for the build side before the spill decision
• concat_batches at the spill boundary — copies into a coalesced batch before the IPC writer sees it
• take_native during the probe phase — gather kernels allocate output buffers
• IPC reader path on spill re-read — the StreamReader/FileReader decoder owns buffers that aren't accounted

The first query in the test stays small (100K × i32 = ~400KB build side), so the overshoot is whatever lives off-pool in the spill setup, not the bulk data itself.

Additional context

• Surfaced by the allocator-vs-pool tracker added in Track allocator-level memory vs MemoryPool during SLTs to prevent OOMs #22626.
• Demonstration PR: Lower SLT HEADROOM_FACTOR 8.0 -> 5.0 to surface nested_loop_join_spill leak #22721 (test-only: lowers HEADROOM_FACTOR; expected to fail until this is fixed).
• Fix should reserve memory through MemoryReservation before each of the sites above, not bump the headroom constant — the constant is a leak detector, not a budget.

Component

• Physical Plan

[avantgardnerio]

@comphead @andygrove the first of many issues that will probably be found tightening the ratchet. @alamb do we want an epic?

[2010YOUY01]

I think, given the current implementation conventions, this kind of validation is not feasible in this failing SLT.

Issue 1 -- operators don't track in-progress batches

Operators don't account for in-progress batch memory for simplicity. This is typically one batch inside each operator per partition. If the batches are wide or the partition count is high, this can become significant.

SET datafusion.execution.target_partitions = 1;
SET datafusion.runtime.memory_limit = '150K';

SELECT count(*) as cnt, min(v1) as mn, max(v1) as mx
FROM generate_series(1, 100000) AS t1(v1)
INNER JOIN generate_series(1, 1) AS t2(v2)
ON (t1.v1 + t2.v2) > 0;

This query already contains a substantial amount of untracked in-progress memory:

• generate_series holds one temporary batch: 8K rows * 8 bytes = 64KB
• NLJ holds a batch for evaluating the filter (v1, v2, 0), plus a result batch: roughly 8K rows * 8 bytes * 3

Given the current implementation, this behavior is expected and not specific to NLJ. Addressing it would require updating all operators to account for in-progress batches.

Issue 2 -- memory allocator noise

I'm assuming the 770KB measurement is allocator RSS. This may also include internal fragmentation, or memory that the allocator has not yet returned to the OS. In such cases RSS can appear high, even though the allocator could potentially release the memory immediately if the OS requests it.

Ideas on follow up works

Overall, I don't think issue 1 is easy to solve, and perhaps these very small memory-limited queries should not be used for allocator-vs-pool validation. They are still useful as correctness tests.

This kind of validation is more feasible when memory pool limit >> single batch size. That likely means using memory limits in the hundreds of MB range and adding a separate extended test suite with dedicated test cases.

Issue 2 remains a challenge, and I'm not sure how to make the validation more accurate. It likely requires a deeper investigation into allocator and OS memory accounting behavior.

[avantgardnerio]

Thanks @2010YOUY01 , you raise some valid points. I'm working on a full response, but I wanted to just respond now to say I'm not ignoring you :)