Add a fine grain memory usage tracking / break down of memory usage *within* each operator

Is your feature request related to a problem or challenge?

Yes, debugging memory problems are hard, when running DF in production and the memory pool is unable to grow the memory it will return ResourcesExhausted error or panic depending on the memory pool implementation, however we don't know what takes all the memory.

In my case I had to manually patch DataFusion row_hash file to print on error what every accumulator takes and the internals

Describe the solution you'd like

I want to add new API - that has size function (not required just because I think it should be combined with explain_memory) and explain_memory function or something (similar to Debug trait) to get string with breakdown of the size with every thing that takes memory.

So for GroupedHashAggregateStream it would say how much group values takes, and for each accumulators it would call the explain memory on each

Describe alternatives you've considered

No response

Additional context

No response

Within the DataFusion CLI there is a flag called:

--top-memory-consumers <TOP_MEMORY_CONSUMERS>
    The number of top memory consumers to display when query fails due to memory exhaustion. To disable memory consumer tracking, set this value to 0 [default: 3]

This helped me identify large consumers of memory in: #16776. Perhaps the same solution can be implemented for your request.

The problem with that is there is no breakdown on what the memory is actually spent on in each consumer

so if I have a aggregation that takes a lot of memory I won't know about it

I agree more details are good -- I think the trick will be figuring out how to do so in such as way that is reasonable to maintain as well as doesn't slow down the normal operation

Also possibly related:

[DISCUSSION] Memory accounting model discussion #16841

Related Add Memory Profiling Functionality #14510

With this feature, issues like this are much easier to debug:

Disproportionate memory use for DISTINCT ON query #17169

@kosiew

@kosiew has a PR that adds this capability -- please check it out if you have time

Add memory profiling support to DataFusion CLI and memory pool metrics #17021

Basically it uses the existing TrackConsumersPool that was contributed by @wiedld to get memory usage per operator

Maybe we need to add some more documentation / examples of how to use that API

@alamb

@alamb -- as for more documentation, do you means adding code examples with the TrackConsumersPool docs? And/or in the datafusion-examples?

@rluvaton

Specifically I think @rluvaton is requesting something that is already possible with TrackConsumersPool -- and if that is the case, then we could potentially resolve the ticket by adding some more docs / examples to the TrackConsumersPool.

We might add an example (ideally in the docstrings) of how to get the largest consumers of a plan / sample the current memory users while the plan is running

@alamb

@alamb -- here is an initial draft of some docs for the current functionality of TrackConsumersPool:
#17213

The new/incoming functionality (PR: #17021) will require an expansion of the docs to include it's usage in realtime monitoring (not just error reporting). That other PR adds other TrackConsumersPool APIs and cumulative metrics for realtime reporting and monitoring.

Please lmk @rluvaton if these docs, combined with the incoming CLI (same PR above) fulfills what you are seeking?

Specifically I think @rluvaton is requesting something that is already possible with TrackConsumersPool -- and if that is the case, then we could potentially resolve the ticket by adding some more docs / examples to the TrackConsumersPool.

nope, this is a new functionality.

Example:
in the case of:

Disproportionate memory use for DISTINCT ON query #17169

I want when memory failed and by some API that I can call it will print not only all the consumers but the consumers size breakdown.

for:

Resources exhausted: Failed to allocate additional 55.0 MB for GroupedHashAggregateStream[3]

I don't really know what takes a lot of memory, is it group values? accumulators? which accumulator?. this makes it really hard to debug memory issues.

I want to get something like for all existing consumers (not only the one that failed):

GroupedHashAggregateStream[3]:
  accumulators: total size <>
    accumulator[1] <name>:  <size>
    accumulator[2] <name>:  <size>

  group_values: <size>
  group_ordering: <size>
  current_group_indices: <size>

as this is the code that calculate the size in row_hash:
https://github.com/apache/datafusion/blob/8f15991f33bf6aca9d4da8958141b59d196b2ed6/datafusion/physical-plan/src/aggregates/row_hash.rs#L916C9-L922C1

this will require almost everything that implement size to also implement some sort of explain memory

@kosiew

This PR by @kosiew was more what I meant:

Add ExplainMemory Trait for Memory Usage Reporting and Enhance Memory Debugging Tools #16926

the changes where deleted but this is gist:

pub trait MemoryUsage {
    /// Returns the total bytes used by `self`.
    fn size(&self) -> usize;

    /// Returns a breakdown of memory usage for `self`.
    fn explain_memory(&self, formatter: &mut std::fmt::Formatter,) -> std::fmt::Result;
}

everything that have a size function will be replaced to force implement this trait for ensuring greater memory explain coverage.

so Accumulator trait will no longer have the size function:

datafusion/datafusion/expr-common/src/accumulator.rs

Lines 74 to 82 in 8f15991

    
               /// Returns the allocated size required for this accumulator, in 
        
               /// bytes, including `Self`. 
        
               /// 
        
               /// This value is used to calculate the memory used during 
        
               /// execution so DataFusion can stay within its allotted limit. 
        
               /// 
        
               /// "Allocated" means that for internal containers such as `Vec`, 
        
               /// the `capacity` should be used not the `len`. 
        
               fn size(&self) -> usize;

and instead it will be forced to implement the MemoryUsage trait.

same for GroupValues:

datafusion/datafusion/physical-plan/src/aggregates/group_values/mod.rs

Lines 99 to 100 in 8f15991

    
               /// Returns the number of bytes of memory used by this [`GroupValues`] 
        
               fn size(&self) -> usize;

and for MemoryReservation

datafusion/datafusion/execution/src/memory_pool/mod.rs

Lines 357 to 360 in 8f15991

    
               /// Returns the size of this reservation in bytes 
        
               pub fn size(&self) -> usize { 
        
                   self.size 
        
               }

@rluvaton

I see -- thank you @rluvaton -- I have updated the title of this ticket to reflect what I think is the ask

I've marked the docs PR as ready for review: #17213

And I'll separately start work on the proposed new API.

@rluvaton

I've marked the docs PR as ready for review: #17213

And I'll separately start work on the proposed new API.

FYI about the API, I suggest you do it incrementally:

Figure out how to add a new API that DOES NOT require changing all existing call sites (this will also be the upgrade path for existing code)
Update one operator (e.g. GroupByHashStream) to use the new API
Add an example (ideally in doc comments) that shows how to use the new API to produce the report @rluvaton desires

Using the example, we can then make sure the API looks good / does what we need, and then we can figure out how to roll it out to the rest of the code

@alamb

@alamb -- I feel like we need a new epic to delineate the different needs, and then consider whether we have solutions to address these needs. (Please lmk, and I can create this Epic.)

I'm going to take a shot at consolidating the identified needs:

more fine grained memory tracking, per Add a fine grain memory usage tracking / break down of memory usage *within* each operator #16904 (comment) (THIS TICKET)
- This may be about tags or "sub-consumers" under a MemoryConsumer. Or maybe a graph/connections between existing consumers? (unclear here)
- as mentioned here, try not to have all callers need to change
provide APIs for memory tracking in non-error situations (a.k.a. Runtime Monitoring), per Add Memory Profiling Functionality #14510
- imo, this expands beyond the existing report_top, and is more aligned with the ConsumerMemoryMetrics in this open PR: Add memory profiling support to DataFusion CLI and memory pool metrics #17021
provide CLI tooling for doing the memory profiling
- e.g. the \memory_profiling in the same PR referenced above: Add memory profiling support to DataFusion CLI and memory pool metrics #17021
consider how memory is accounted for in shared data buffers
- current datafusion ticket: [DISCUSSION] Memory accounting model discussion #16841
- perhaps building on work done for this epic in arrow-rs: [EPIC] Memory Pool for Memory Tracking arrow-rs#8137

@kosiew

@kosiew -- your linked PR #17021 (which is great btw, thank you) states that it closes this issue ticket. I haven't reviewed it yet, altho on quick skim I don't think it address the itemization (e.g. breakdown into sub consumers etc) as denoted here?

Please feel free to point out if I missed something. 🙏🏼

@wiedld

@wiedld
You're right.
The PR now summarizes, rather than displays the fine grain.
During development, I found that the fine grain reporting might be too detailed for everybody and so opted to report a summary.
I will amend the PR description accordingly.

@kosiew

The open PR from @kosiew includes some new CLI tooling, which may take longer in the code review process.

To unblock progress on this ticket, here is a proposed solution which brings us closer to the example output mentioned by @rluvaton here. Specifically, it provides a bit more context and a lineage of how these memory consumers were created (to help debug). Please let me know if this solution is useful:
https://github.com/apache/datafusion/pull/17943/files#r2408421334

Navigation Menu

Repository navigation

Add a fine grain memory usage tracking / break down of memory usage within each operator #16904

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	/// Returns the allocated size required for this accumulator, in
	/// bytes, including `Self`.
	///
	/// This value is used to calculate the memory used during
	/// execution so DataFusion can stay within its allotted limit.
	///
	/// "Allocated" means that for internal containers such as `Vec`,
	/// the `capacity` should be used not the `len`.
	fn size(&self) -> usize;

	/// Returns the number of bytes of memory used by this [`GroupValues`]
	fn size(&self) -> usize;

	/// Returns the size of this reservation in bytes
	pub fn size(&self) -> usize {
	self.size
	}

Add a fine grain memory usage tracking / break down of memory usage *within* each operator #16904

Description

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Activity

zheniasigayev commented on Jul 24, 2025

rluvaton commented on Jul 24, 2025

alamb commented on Jul 25, 2025

alamb commented on Jul 25, 2025

alamb commented on Aug 12, 2025

rluvaton commented on Aug 14, 2025

alamb commented on Aug 15, 2025

wiedld commented on Aug 15, 2025

alamb commented on Aug 15, 2025

wiedld commented on Aug 16, 2025

rluvaton commented on Aug 16, 2025

rluvaton commented on Aug 16, 2025

alamb commented on Aug 18, 2025

wiedld commented on Aug 18, 2025

alamb commented on Aug 19, 2025

wiedld commented on Aug 21, 2025

wiedld commented on Aug 21, 2025

kosiew commented on Aug 22, 2025

wiedld commented on Oct 6, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Add a fine grain memory usage tracking / break down of memory usage within each operator #16904