parthchandra opened a new issue, #7279:
URL: https://github.com/apache/datafusion-comet/issues/7279
Part of #7267.
## Problem
`FilterExec` merges its output with DataFusion's `LimitedBatchCoalescer`,
which
wraps arrow's `BatchCoalescer`. None of the memory that coalescer holds is
registered with the DataFusion memory pool: neither `FilterExec` nor
`LimitedBatchCoalescer` creates a `MemoryReservation` (DataFusion 55.2.0,
`filter.rs`, `coalesce/mod.rs`).
What goes untracked:
- **Buffered copies.** For string/binary and nested columns,
`GenericInProgressArray` keeps the filtered copy of every buffered input
batch
until the output batch is full (`arrow-select coalesce/generic.rs`).
- **The merge output.** `finish()` calls `concat`, which reserves the full
output
buffer before copying. Peak memory is about twice the buffered data. For an
oversized batch, the over-2 GiB reservation happens before the offset
overflow
error.
- **Up-front reservations.** For primitive columns,
`InProgressPrimitiveArray`
reserves `batch_size` slots per column on first write
(`coalesce/primitive.rs:58-62`), held until the output is emitted.
In Comet this memory sits outside Spark's off-heap accounting, so a task can
use
more native memory than its pool allows without spilling or failing cleanly.
It
grows with `spark.comet.batchSize` (sub-issue 1 makes the filter follow that
setting, so users who raised it above 8192 now hold larger buffers) and with
value
size.
The same coalescer is used by `CoalesceBatchesExec`, `RepartitionExec`, the
hash
join output stream, sort-merge join, nested loop join and TopK. Comet's
native
shuffle writer uses `BatchCoalescer` directly (`buf_batch_writer.rs:133`);
whether
its buffers are tracked is not checked yet.
## Proposed fix
- **Upstream (DataFusion):** give `LimitedBatchCoalescer` a
`MemoryReservation` and
resize it after each push from `BatchCoalescer::size()` (`arrow-select
coalesce.rs:588`), which already reports buffered bytes. Account for the
`concat`
output before calling `finish()`, so the pool can refuse it instead of
allocating it. `FilterExec` would register the reservation in `execute`.
- **Comet:** check the shuffle writer's coalescer and track it the same way
if it
isn't.
- Pairs well with sub-issue 4 (byte limit): once buffered bytes are tracked,
the
same number can trigger an early flush.
## Tests
- DataFusion unit test: push batches into a `FilterExec` under a small
`GreedyMemoryPool` and check the reservation grows and is released after
emit,
and that a pool smaller than one merged batch returns a
resources-exhausted error
instead of allocating.
- Comet: a native test with a small task memory limit and a filter over a
binary
column, checking the query fails with Comet's memory error, not an
over-allocation.
## Notes
- Existing DataFusion behavior, not introduced by sub-issue 1. Sub-issue 1
only
changes how many rows the filter buffers.
--
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]
