cetra3 commented on issue #22758:
URL: https://github.com/apache/datafusion/issues/22758#issuecomment-4909507722
I will speak about a couple of paths we have taken in production to curb
memory usage, because I think it's probably important context for this
discussion.
The first attempt at limiting memory (besides setting the DF memory limit)
was to introduce a record batch memory limiter. This is an `ExecutionPlan`
wrapper struct that wraps an inner plan node, uses `claim` on all record
batches in the tree, and applies backpressure/limits when memory reaches a
certain threshold. There is an optimizer that does a `transform_down`
wrapping *every* node in the tree with this limiter.
I.e, this is essentially:
* Wait for a Record Batch
* Claim the Memory
* Check if we're above the watermark:
* If we're above, wait up to 10 seconds for memory to be freed
* If we're below, just allow the Record Batch through
And so with this at *every* node in the tree, because reclaiming old memory
will free the reservation, this gives a pretty holistic approach that doesn't
depend on certain execution plans to limit memory (There is one outstanding
[race condition in arrow](https://github.com/apache/arrow-rs/issues/10139)
which makes this inaccurate).
You would *think* that such a blunt tool would be enough to capture memory
usage, or at least a large swathe of it. This is not the case:
* There is a fair amount of allocations that happen in arrow before a record
batch is created. For instance the predicate cache that is *not globally
capped* and instead defaults to a 100mb *per file* which, if you have big files
and are querying lots of them can blow up memory *very quick*
* There are a fair amount of *internal* allocations within the datafusion
physical execution nodes that are not part of record batches. Things like
intermediate sorts and other data structures that it uses. Like
`GroupsAccumulatorAdapter` and
* The `Bytes` struct from the bytes crate, which is used for `object_store`
among other things, famously does not include a `capacity()`, only `length()`,
and so you have no real way of knowing how much memory it is actually taking up.
Most of the time, the allocations for queries is much higher than the
reported watermark.
So, the next thing we've tried, and had *better* success, albeit not
perfect, is to use a [custom global
allocator](https://github.com/cetra3/thresher), and essentially do the same
thing we were doing at the record batch level:
* Override the global allocator
* Set a memory watermark and register a callback
* When that callback hits:
* Stop any active queries
* Record a heap dump so we can see what's going on
* Cancel any active queries
With this approach, this is once again not perfect, because it happens
*after* allocations already, but it does catch a lot more memory pressure
events, that the pure record batch one does not catch. In fact, I hardly
*ever* see the recordbatch-level limiter hit its threshold, because most
allocations are happening elsewhere.
I feel that, pretty much, the only way to really knock this problem on its
head is that we need to account for all allocations within datafusion and
arrow. It is a bit of a painful, but probably needed, refactor. If we don't
do this, we are going to end up in our current whack-a-mole state.
--
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]
