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

[rluvaton]

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