NestedLoopJoinExec memory-limited fallback returns an empty result (wrong count = 0) instead of spilling or erroring #25701

Describe the bug

A three-way self-join on inequality predicates over a single 500,000-row Parquet file returns
count = 0 in well under a second when the session has a bounded memory pool (FairSpillPool or
GreedyMemoryPool, 256 MiB or 1 GiB). The correct answer is 500,000 × 499,999 × 499,999 ≈ 1.25 × 10¹⁷.
There is no error and no warning. With DataFusion's default, unbounded pool the same query runs and keeps
running, which is what it should do.

EXPLAIN ANALYZE shows the outer NestedLoopJoinExec buffering ~33 M build rows (≈252 MB, the pool's
size), switching to its spill fallback (spill_count=1) — and then spilling zero rows (spilled_rows=0,
spilled_bytes=1096 B), reading zero rows from its probe side (input_rows=0; the probe-side
DataSourceExec reports files_opened=0) and emitting zero rows. The query completes successfully with
an empty join.

To reproduce

DataFusion 55.1.0 (arrow/parquet 59.2), Rust 1.95. Reproduced on macOS arm64 in a debug build; first
seen on Linux arm64 (Docker Desktop) in a release build, through datafusion-postgres.

use std::sync::Arc;
use arrow::array::Int64Array;
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use datafusion::execution::disk_manager::{DiskManagerBuilder, DiskManagerMode};
use datafusion::execution::memory_pool::FairSpillPool;
use datafusion::execution::runtime_env::RuntimeEnvBuilder;
use datafusion::prelude::*;
use parquet::arrow::ArrowWriter;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    // one Parquet file, one Int64 column, ids 1..=500_000
    let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
    let path = std::env::temp_dir().join("t.parquet");
    let mut w = ArrowWriter::try_new(std::fs::File::create(&path)?, schema.clone(), None)?;
    w.write(&RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from_iter_values(1..=500_000))])?)?;
    w.close()?;

    // a bounded pool with a spill directory
    let spill = std::env::temp_dir().join("spill");
    std::fs::create_dir_all(&spill)?;
    let runtime = RuntimeEnvBuilder::new()
        .with_memory_pool(Arc::new(FairSpillPool::new(256 * 1024 * 1024)))
        .with_disk_manager_builder(
            DiskManagerBuilder::default().with_mode(DiskManagerMode::Directories(vec![spill])),
        )
        .build_arc()?;
    let ctx = SessionContext::new_with_config_rt(SessionConfig::new(), runtime);
    ctx.register_parquet("t", path.to_str().unwrap(), ParquetReadOptions::default()).await?;

    ctx.sql("SELECT count(*) FROM t a, t b, t c WHERE a.id <> b.id AND b.id <> c.id")
        .await?
        .show()
        .await?;
    Ok(())
}

Expected behavior

Either the correct count (after a very long time), or a ResourcesExhausted error. Never a successful,
empty join.

Actual behavior

+----------+
| count(*) |
+----------+
| 0        |
+----------+

in ~0.6 s (256 MiB pool) or ~2.5 s (1 GiB pool). Same result with GreedyMemoryPool, and with
target_partitions = 1.

Plan (identical under both pools; EXPLAIN ANALYZE metrics of the 256 MiB run on the outer join):

ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
  AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
    CoalescePartitionsExec
      AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
        NestedLoopJoinExec: join_type=Inner, filter=id@1 != id@0, projection=[]
            metrics=[output_rows=0, output_batches=8, spill_count=1, spilled_bytes=1096.0 B,
                     spilled_rows=0, build_mem_used=252.1 MB, build_input_batches=4.03 K,
                     build_input_rows=32.99 M, input_batches=0, input_rows=0, ...]
          CoalescePartitionsExec
            NestedLoopJoinExec: join_type=Inner, filter=id@1 != id@0, projection=[id@1]
              CoalescePartitionsExec
                DataSourceExec: file_groups={8 groups: [...]}, projection=[id], file_type=parquet
              DataSourceExec: file_groups={8 groups: [...]}, projection=[id], file_type=parquet
          DataSourceExec: file_groups={8 groups: [...]}, projection=[id], file_type=parquet
              metrics=[output_rows=0, files_opened=0, ...]

Additional context

• Related, same fallback path, different symptoms: NestedLoopJoinExec spill fallback evaluates the left input twice #24661 (the left input is evaluated twice in the fallback), NestedLoopJoin buffers the build side into a single concat_batches allocation: 2x transient peak, invisible to the memory pool #24819 (the build side's transient peak is invisible to the pool), NestedLoopJoin coordinated fallback hangs surviving partitions when one is dropped unfinished #25003 (a hang when a partition is dropped). This report is about the fallback completing successfully with an empty result.

• The two-way form (t a, t b WHERE a.id <> b.id) over the same file does not come back early under the
  same pool; it runs until cancelled. The inner join's output — tens of millions of rows — is what fills
  the pool and sends the outer join into the fallback.

• A guess, not a diagnosis: the path is NestedLoopJoinStream::initiate_fallback /
  handle_buffering_left_memory_limited (datafusion-physical-plan 55.1.0,
  src/joins/nested_loop_join.rs), which re-executes the left plan into a spill file
  (spill_record_batch_stream_and_return_max_batch_memory). The metrics say what reaches that file is
  empty and the probe side is never read.

• In the soak the defect appeared through deltalake's scan and datafusion-postgres; tables rebuilt
  into many small files ran into the statement timeout instead, most likely because the inner join did
  not fill the pool before the clock ran out (not measured). The reproduction above has no Delta in it.

• Downstream, Pumice now refuses such plans before execution (a NestedLoopJoinExec with both inputs
  estimated above a bound), but that is a guard, not a fix: any bounded-pool deployment can get a silent
  wrong answer from this shape.