hosseinsha opened a new issue, #25701:
URL: https://github.com/apache/datafusion/issues/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`.
```rust
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: #24661 (the left input is
evaluated twice in the fallback), #24819 (the build side's transient peak is
invisible to the pool), #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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
