refactor(hash-aggr): Migrate partial-reduce hash aggregation

[2010YOUY01]

Which issue does this PR close?

• Closes #.

Rationale for this change

Part of #22710

This PRs implements the partial-reduce aggregation.

Comments at datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs explains the high-level idea.

What changes are included in this PR?

• Adds an AggregateHashTable<PartialReduceMarker> variant to handle partial-reduce aggregation.
• Adds PartialReduceHashAggregateStream to implement the partial-reduce state machine.

Are these changes tested?

Covered by existing tests

Are there any user-facing changes?

No

[alamb]

This is on my review radar

[alamb]

I merged up from main to resolve a conflict

[alamb]

Looks good to me -- thank you @2010YOUY01

The only thing I found (with codex) is that PartialReduceHashAggregateStream is also used regardless of the memory pool sizes even though it doesn't implement memory reservations / OOM behavior yet

So I suggest we disable this special stream when there is a memory limit -- similarly to Partial hash stream guard:

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

Lines 1056 to 1060 in 8f77c61

	fn should_use_partial_hash_stream(&self, context: &TaskContext) -> bool {
	// TODO: implement memory-limited path and remove this limitation
	if matches!(context.memory_pool().memory_limit(), MemoryLimit::Finite(_)) {
	return false;
	}

This is similar to the other refactor (and the fix is 2010YOUY01#4)

[alamb]

👍

[alamb]

I verified that this code is covered by existing tests:

cargo llvm-cov -p datafusion-physical-plan --lib --html -- partial_reduce

[2010YOUY01]

The only thing I found (with codex) is that PartialReduceHashAggregateStream is also used regardless of the memory pool sizes even though it doesn't implement memory reservations / OOM behavior yet

So I suggest we disable this special stream when there is a memory limit -- similarly to Partial hash stream guard:

Thanks for the review. I agree we should not use the new path if there is the memory limit.

One thing I noticed is that the original PR added spilling and streaming merge support for PartialReduce aggregation, but I don’t see any tests for the PartialReduce + memory limit case, or much discussion about the expected OOM behavior.

• feat: add AggregateMode::PartialReduce for tree-reduce aggregation #20019

For this use case, I think early emit is probably both more efficient and simpler than supporting spilling / streaming merge.

Should we make this path explicitly error on OOM instead? That would keep the behavior more constrained for now. Since the existing feature does not seem to have test coverage, relying on the current implementation feels a bit risky, and we can reopen the discussion later if users depend on this behavior. 🤔

[2010YOUY01]

So I suggest we disable this special stream when there is a memory limit -- similarly to Partial hash stream guard:

disabled in 08558c6

[alamb]

Shall we merge it?