DamiaoCastro commented on issue #26140:
URL: https://github.com/apache/datafusion/issues/26140#issuecomment-6106884055
A concrete data point for TopK, one of the operators this issue lists as not
able to use the counter yet.
TopK charges each retained input batch its `get_record_batch_memory_size`
(`RecordBatchStore::insert`), so N zero-copy slices of one allocation are
charged N × that allocation. You can hit this without any slicing inside
DataFusion: `arrow_ipc::reader::StreamDecoder` decodes zero-copy, so every
record batch of a multi-batch IPC stream points into the one stream buffer.
Repro on 53.1.0: a 128 MiB pool, and 2 000 rows passed as 2 000 one-row
slices of one batch:
```rust
use std::sync::Arc;
use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::datasource::MemTable;
use datafusion::execution::runtime_env::RuntimeEnvBuilder;
use datafusion::prelude::{SessionConfig, SessionContext};
#[tokio::main(flavor = "current_thread")]
async fn main() -> datafusion::error::Result<()> {
// 2 000 rows, passed as 2 000 one-row zero-copy slices of one batch
let n = 2_000;
let schema = Arc::new(Schema::new(vec![
Field::new("k", DataType::Int64, false),
Field::new("v", DataType::Utf8, false),
]));
let batch = RecordBatch::try_new(
Arc::clone(&schema),
vec![
Arc::new(Int64Array::from_iter_values(0..n as i64)),
Arc::new(StringArray::from_iter_values((0..n).map(|i|
format!("{i:0>100}")))),
],
)?;
let slices: Vec<RecordBatch> = (0..n).map(|i| batch.slice(i,
1)).collect();
let runtime = RuntimeEnvBuilder::new()
.with_memory_limit(128 * 1024 * 1024, 1.0)
.build_arc()?;
let ctx = SessionContext::new_with_config_rt(SessionConfig::new(),
runtime);
ctx.register_table("t", Arc::new(MemTable::try_new(schema,
vec![slices])?))?;
// k = 100 000 > n: every slice keeps a row, so TopK retains all of them
ctx.sql("SELECT k, v FROM t ORDER BY k DESC LIMIT 100000")
.await?
.collect()
.await?;
Ok(())
}
```
```
ResourcesExhausted("Additional allocation failed for TopK[0] with top memory
consumers (across reservations) as:
TopK[0]#0(can spill: false) consumed 127.8 MB, peak 127.8 MB.
Error: Failed to allocate additional 279.5 KB for TopK[0] with 127.8 MB
already allocated for this reservation - 243.8 KB remain available for the
total pool")
```
Each one-row slice is charged the parent's full ~280 KB of buffer capacity.
I haven't run this on `main`, but the path looks unchanged there:
- `register_batch` keeps the input batch as-is.
- `RecordBatchStore::insert` adds `get_record_batch_memory_size` per entry.
- `maybe_compact` only runs once unused rows reach `20 * batch_size + k`, so
with a large `k` it never compacts.
We ran into this on stored IPC streams of hundreds of one-row batches. An
`ORDER BY` over 577 rows with a large `LIMIT` was charged ~0.75 MB per batch
and failed at 128 MiB. On 53.1 the hash-join build side failed the same way: a
join of two ~1 000-row tables was charged ~0.6 MB per one-row batch. #22862
already fixes that on `main`. Our workaround is to `concat_batches` each
decoded IPC stream before handing it to DataFusion.
--
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]
