fix: count aggregation_time once per batch instead of once per accumulator

[ranflarion]

Which issue does this PR close?

• Closes aggregation_time metric is inflated up to (N+1)/2 times on GROUP BY nodes with N aggregate functions #24735.

Rationale for this change

aggregation_time on GroupedHashAggregateStream is added inside the per-accumulator loop while agg_start_time is taken once before it, so with N aggregate functions the k-th accumulator re-adds the elapsed time of all k accumulators so far, inflating the metric by up to (N+1)/2. On a 28-function GROUP BY a single task reported 4m22s of aggregation_time inside a 2m0s stage. Single-aggregate plans are unaffected. The newer aggregate_hash_table implementations already scope the timer around the whole loop.

What changes are included in this PR?

Move the add_elapsed call after the accumulator loop, once per interned batch. time_calculating_group_ids, which shares agg_start_time as its end point, is unchanged, and the per-accumulator aggregate_accumulator_metrics timings are unaffected.

Are these changes tested?

Covered by the existing aggregate tests (cargo test -p datafusion-physical-plan aggregates, 193 passed). The metric's value is wall-clock time, which the existing tests do not assert exactly; the inflation itself was measured on a 28-aggregate workload where the fixed metric now stays within the operator's elapsed time.

Are there any user-facing changes?

aggregation_time (shown in EXPLAIN ANALYZE as elapsed_compute subset time) now reports the actual time spent in accumulators on plans with more than one aggregate function; previously it over-reported on such plans.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.79592% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.52%. Comparing base (ee59f62) to head (5ae679d).
⚠️ Report is 27 commits behind head on main.

Files with missing lines	Patch %	Lines
...ysical-plan/src/aggregates/group_values/metrics.rs	89.13%	1 Missing and 4 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24736      +/-   ##
==========================================
+ Coverage   81.42%   81.52%   +0.09%     
==========================================
  Files        1121     1123       +2     
  Lines      402142   406194    +4052     
  Branches   402142   406194    +4052     
==========================================
+ Hits       327460   331153    +3693     
- Misses      55484    55668     +184     
- Partials    19198    19373     +175     

☔ 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.

[kumarUjjawal]

Thank you @ranflarion for the fix.

This looks correct. I was wondering if adding and assert or a test help?

[ranflarion]

Thank you @ranflarion for the fix.

This looks correct. I was wondering if adding and assert or a test help?

added a test. the assertion that works is a structural invariant rather than a value check: aggregation_time is accumulated strictly inside the elapsed_compute timer that wraps group_aggregate_batch, so aggregation_time <= elapsed_compute always holds and double counting breaks it. there's no wall-clock threshold so no reason for it to be flaky

[kumarUjjawal]

Thank you @ranflarion for the iterations.

Looks good!