chore: convert FinalHashAggregateStream to async generators and cleanup

[rluvaton]

Which issue does this PR close?

Related to:

• [Status Update] Simplifing Streams to be more textbook-like and have less state while keeping the same perf #23974

Rationale for this change

Cleanup the code and remove state

What changes are included in this PR?

changed FinalHashAggregateStream to async generator and remove unneeded code due to that

The first commit in this PR is FinalHashAggregateStream

Are these changes tested?

existing tests

Are there any user-facing changes?

nope

[codecov-commenter]

Codecov Report

❌ Patch coverage is 83.50515% with 16 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.62%. Comparing base (3a4c310) to head (a7fe361).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...fusion/physical-plan/src/aggregates/hash_stream.rs	83.33%	3 Missing and 13 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24874      +/-   ##
==========================================
- Coverage   81.62%   81.62%   -0.01%     
==========================================
  Files        1123     1123              
  Lines      409637   409488     -149     
  Branches   409637   409488     -149     
==========================================
- Hits       334383   334250     -133     
+ Misses      55624    55599      -25     
- Partials    19630    19639       +9     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[jayzhan211]

Thanks @rluvaton , here is a non-blocking suggestion:

The is_done() early-return duplicates the record_output + emit pair, and the loop's natural next_output_batch() == None exit no longer zeroes the reservation the way the old Ok(None) arm did. Harmless today — that exit is only reachable for an empty table — but it's a latent divergence worth closing while the code is being restructured anyway.

A single-exit loop removes both. This relies on two properties I checked in common.rs: memory_size() returns 0 in the Done state (common.rs:315), and next_output_batch_inner drops the MaterializedAggregateOutput when the last slice is taken (common.rs:520-522). So try_resize(hash_table.memory_size()) on the final batch is exactly try_resize(0), and the explicit drop(hash_table) is cosmetic. Poll counts are unchanged — both shapes need one extra poll after the last batch before the generator completes.

     /// Emit final aggregate value batches:
     /// Input was exhausted without spilling, or the soft group limit was reached.
     async fn produce_output_from_memory(
         &mut self,
         mut hash_table: AggregateHashTable<FinalMarker>,
         mut emitter: TryEmitter<RecordBatch, DataFusionError>,
     ) -> Result<()> {
         let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
-        let mut timer = elapsed_compute.timer();
-
-        hash_table.start_output()?;
-
-        while let Some(batch) = hash_table.next_output_batch()? {
-            if hash_table.is_done() {
-                drop(hash_table);
-                self.reservation.try_resize(0)?;
-                timer.done();
-
-                emitter
-                    .emit(batch.record_output(&self.baseline_metrics))
-                    .await;
-
-                return Ok(());
-            }
-
-            self.reservation.try_resize(hash_table.memory_size())?;
-
-            timer.done();
-            emitter
-                .emit(batch.record_output(&self.baseline_metrics))
-                .await;
-            timer = elapsed_compute.timer();
-        }
-
-        Ok(())
+
+        {
+            let _timer = elapsed_compute.timer();
+            hash_table.start_output()?;
+        }
+
+        loop {
+            let timer = elapsed_compute.timer();
+
+            let Some(batch) = hash_table.next_output_batch()? else {
+                // Only reachable when the table held no groups at all: a
+                // non-empty table always reports its last batch together with
+                // the `Done` state, which the `try_resize` below already zeroes.
+                self.reservation.try_resize(0)?;
+                return Ok(());
+            };
+
+            // The table hands over its groups as they are materialized and
+            // reports a size of 0 once it reaches `Done`, so this releases the
+            // reservation before the final batch goes downstream.
+            self.reservation.try_resize(hash_table.memory_size())?;
+            timer.done();
+
+            emitter
+                .emit(batch.record_output(&self.baseline_metrics))
+                .await;
+        }
     }