Untracked memory in FilterExec, RepartitionExec, HashJoinExec and AsyncFuncExec and more due to LimitedBatchCoalescer and arrow BatchCoalescer usage

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:

datafusion/datafusion/physical-plan/src/filter.rs

Lines 580 to 584 in f34a676

    
           batch_coalescer: LimitedBatchCoalescer::new( 
        
               self.schema(), 
        
               self.batch_size, 
        
               self.fetch, 
        
           ),

RepartitionExec:

datafusion/datafusion/physical-plan/src/repartition/mod.rs

Lines 479 to 483 in f34a676

    
           SharedCoalescer::new( 
        
               input.schema(), 
        
               context.session_config().batch_size(), 
        
               num_input_partitions, 
        
           )

datafusion/datafusion/physical-plan/src/repartition/mod.rs

Lines 226 to 229 in f34a676

    
           struct SharedCoalescer { 
        
               inner: Arc<Mutex<LimitedBatchCoalescer>>, 
        
               active_senders: Arc<AtomicUsize>, 
        
           }

even though RepartitionExec have some memory reservation, it does not count the LimitedBatchCoalescer memory

HashJoinExec:

datafusion/datafusion/physical-plan/src/joins/hash_join/stream.rs

Lines 491 to 493 in f34a676

    
           // Create output buffer with coalescing and optional fetch limit. 
        
           let output_buffer = 
        
               LimitedBatchCoalescer::new(Arc::clone(&schema), batch_size, fetch);

AsyncFuncExec

datafusion/datafusion/physical-plan/src/async_func.rs

Lines 207 to 214 in f34a676

    
           let coalesced_input_stream = CoalesceInputStream { 
        
               input_stream, 
        
               batch_coalescer: LimitedBatchCoalescer::new( 
        
                   Arc::clone(&self.input.schema()), 
        
                   config_options_ref.execution.batch_size.get(), 
        
                   None, 
        
               ), 
        
           };

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:

datafusion/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs

Lines 384 to 386 in f34a676

    
               /// Output buffer. Currently used by filtering as it requires double buffering 
        
               /// to avoid small/empty batches. Non-filtered join outputs directly from `staging_output_record_batches.batches` 
        
               pub output: BatchCoalescer,

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

take

I opened a PR in arrow-rs to add size function to the underlying BatchCoalescer so once it is merged and release we can use that here

feat(coalesce): add size function arrow-rs#10331

@rluvaton

I opened a PR in arrow-rs to add size function to the underlying BatchCoalescer so once it is merged and release we can use that here

feat(coalesce): add size function arrow-rs#10331

Thanks @rluvaton , I've explored implementation but was not sure because of this. I can implement these when its landed in arrow upstream as you've said or leave it to you if you prefer that way

Yeah, I was going back and forth with this but decided to just expose memory size of BatchCoalescer in arrow, in any case we need to wait for a merge and release before we can continue here

tagging this issue with #22758 since its related

feat(coalesce): add size function (#10331)

Navigation Menu

Repository navigation

Untracked memory in `FilterExec`, `RepartitionExec`, `HashJoinExec` and `AsyncFuncExec` and more due to `LimitedBatchCoalescer` and arrow `BatchCoalescer` usage #23385

Describe the bug

To Reproduce

Expected behavior

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	batch_coalescer: LimitedBatchCoalescer::new(
	self.schema(),
	self.batch_size,
	self.fetch,
	),

	SharedCoalescer::new(
	input.schema(),
	context.session_config().batch_size(),
	num_input_partitions,
	)

	struct SharedCoalescer {
	inner: Arc<Mutex<LimitedBatchCoalescer>>,
	active_senders: Arc<AtomicUsize>,
	}

	// Create output buffer with coalescing and optional fetch limit.
	let output_buffer =
	LimitedBatchCoalescer::new(Arc::clone(&schema), batch_size, fetch);

	let coalesced_input_stream = CoalesceInputStream {
	input_stream,
	batch_coalescer: LimitedBatchCoalescer::new(
	Arc::clone(&self.input.schema()),
	config_options_ref.execution.batch_size.get(),
	None,
	),
	};

	/// Output buffer. Currently used by filtering as it requires double buffering
	/// to avoid small/empty batches. Non-filtered join outputs directly from `staging_output_record_batches.batches`
	pub output: BatchCoalescer,

Untracked memory in FilterExec, RepartitionExec, HashJoinExec and AsyncFuncExec and more due to LimitedBatchCoalescer and arrow BatchCoalescer usage #23385

Description

Describe the bug

To Reproduce

Expected behavior

Additional context

Activity

buraksenn commented on Jul 10, 2026

rluvaton commented on Jul 14, 2026

buraksenn commented on Jul 14, 2026

rluvaton commented on Jul 14, 2026

Rich-T-kid commented on Jul 15, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

Untracked memory in `FilterExec`, `RepartitionExec`, `HashJoinExec` and `AsyncFuncExec` and more due to `LimitedBatchCoalescer` and arrow `BatchCoalescer` usage #23385