Untracked memory in `FilterExec`, `RepartitionExec`, `HashJoinExec` and `AsyncFuncExec` and more due to `LimitedBatchCoalescer` and arrow `BatchCoalescer` usage
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Describe the bug
According to the datafusion memory model, you need to reserve memory for things you hold on, `LimitedBatchCoalescer` which hold on the batches to coalesce and hold on batches that not yet polled does not reserve any memory
`LimitedBatchCoalescer` is used by:
`FilterExec`:
https://github.com/apache/datafusion/blob/f34a676302e2320526172705503a5ac8222804ea/datafusion/physical-plan/src/filter.rs#L580-L584
`RepartitionExec`:
https://github.com/apache/datafusion/blob/f34a676302e2320526172705503a5ac8222804ea/datafusion/physical-plan/src/repartition/mod.rs#L479-L483
https://github.com/apache/datafusion/blob/f34a676302e2320526172705503a5ac8222804ea/datafusion/physical-plan/src/repartition/mod.rs#L226-L229
even though `RepartitionExec` have some memory reservation, it does not count the `LimitedBatchCoalescer` memory
`HashJoinExec`:
https://github.com/apache/datafusion/blob/f34a676302e2320526172705503a5ac8222804ea/datafusion/physical-plan/src/joins/hash_join/stream.rs#L491-L493
`AsyncFuncExec`
https://github.com/apache/datafusion/blob/f34a676302e2320526172705503a5ac8222804ea/datafusion/physical-plan/src/async_func.rs#L207-L214
and more that use arrow `BatchCoalescer` directly, since it does not expose a way to know which batches hold which result, you have no way to reserve and free memory for them
Like `SortMergeJoin`:
https://github.com/apache/datafusion/blob/f34a676302e2320526172705503a5ac8222804ea/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs#L384-L386
### To Reproduce
Looking at the code you can see no reservation, so just create `FilterExec` for example and see no reservation being made
### Expected behavior
everything that hold on memory should reserve for it
### Additional context
Ideally `CoalesceBatchStream` should output small batches when unable to reach the batch size
Contributor guide
Assessment
This issue has not been assessed yet.