@@ -84,6 +84,24 @@ pub trait Accumulator: Send + Sync + Debug + std::any::Any {
     /// running sum.
     fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()>;
 
+    /// Returns an optional metric timed once per grouped adapter input batch.
+    ///
+    /// A grouped accumulator adapter uses this for aggregate-owned work it
+    /// dispatches to one accumulator per group. The default preserves the
+    /// usual per-accumulator update path.
+    fn grouped_update_batch_metric(&self) -> Option<Arc<dyn AggregateMetric>> {
+        None
+    }
+
+    /// Updates state when called by a grouped accumulator adapter.
+    ///
+    /// The default delegates to [`Self::update_batch`]. Implementations that
+    /// return a [`Self::grouped_update_batch_metric`] can avoid timing every
+    /// per-group call; the adapter records one interval for the full batch.
+    fn update_batch_grouped(&mut self, values: &[ArrayRef]) -> Result<()> {
+        self.update_batch(values)
+    }
+
     /// Returns the final aggregate value.
     ///
     /// For example, the `SUM` accumulator maintains a running sum,
 
@@ -24,6 +24,8 @@ pub mod nulls;
 pub mod prim_op;
 
 use std::mem::{size_of, size_of_val};
+use std::sync::Arc;
+use std::time::Instant;
 
 use arrow::array::new_empty_array;
 use arrow::{
@@ -33,7 +35,7 @@ use arrow::{
     datatypes::UInt32Type,
 };
 use datafusion_common::{Result, ScalarValue, arrow_datafusion_err};
-use datafusion_expr_common::accumulator::Accumulator;
+use datafusion_expr_common::accumulator::{Accumulator, AggregateMetric};
 use datafusion_expr_common::groups_accumulator::{
     EmitTo, GroupSelection, GroupsAccumulator,
 };
@@ -102,6 +104,9 @@ pub struct GroupsAccumulatorAdapter {
     /// bottleneck in earlier implementations when there were many
     /// distinct groups.
     allocation_bytes: usize,
+
+    /// Optional aggregate-owned metric timed once for a grouped update batch.
+    grouped_update_metric: Option<Arc<dyn AggregateMetric>>,
 }
 
 struct AccumulatorState {
@@ -139,6 +144,7 @@ impl GroupsAccumulatorAdapter {
             factory: Box::new(factory),
             states: vec![],
             allocation_bytes: 0,
+            grouped_update_metric: None,
         }
     }
 
@@ -152,6 +158,9 @@ impl GroupsAccumulatorAdapter {
         let new_accumulators = total_num_groups - self.states.len();
         for _ in 0..new_accumulators {
             let accumulator = (self.factory)()?;
+            if self.grouped_update_metric.is_none() {
+                self.grouped_update_metric = accumulator.grouped_update_batch_metric();
+            }
             let state = AccumulatorState::new(accumulator);
             self.add_allocation(state.size());
             self.states.push(state);
@@ -191,6 +200,7 @@ impl GroupsAccumulatorAdapter {
         group_indices: &[usize],
         opt_filter: Option<&BooleanArray>,
         total_num_groups: usize,
+        time_grouped_update: bool,
         f: F,
     ) -> Result<()>
     where
@@ -245,25 +255,37 @@ impl GroupsAccumulatorAdapter {
         // RecordBatch(es)
         let iter = groups_with_rows.iter().zip(offsets.windows(2));
 
+        let grouped_update_metric = time_grouped_update
+            .then(|| self.grouped_update_metric.as_ref().cloned())
+            .flatten();
+        let start = grouped_update_metric.as_ref().map(|_| Instant::now());
+
         let mut sizes_pre = 0;
         let mut sizes_post = 0;
-        for (&group_idx, offsets) in iter {
-            let state = &mut self.states[group_idx];
-            sizes_pre += state.size();
-
-            let values_to_accumulate = slice_and_maybe_filter(
-                &values,
-                opt_filter.as_ref().map(|f| f.as_boolean()),
-                offsets,
-            )?;
-            f(state.accumulator.as_mut(), &values_to_accumulate)?;
-
-            // clear out the state so they are empty for next
-            // iteration
-            state.indices.clear();
-            sizes_post += state.size();
-        }
+        let result: Result<()> = (|| {
+            for (&group_idx, offsets) in iter {
+                let state = &mut self.states[group_idx];
+                sizes_pre += state.size();
+
+                let values_to_accumulate = slice_and_maybe_filter(
+                    &values,
+                    opt_filter.as_ref().map(|f| f.as_boolean()),
+                    offsets,
+                )?;
+                f(state.accumulator.as_mut(), &values_to_accumulate)?;
+
+                // clear out the state so they are empty for next
+                // iteration
+                state.indices.clear();
+                sizes_post += state.size();
+            }
+            Ok(())
+        })();
 
+        if let (Some(metric), Some(start)) = (grouped_update_metric, start) {
+            metric.add_duration(start.elapsed());
+        }
+        result?;
         self.adjust_allocation(sizes_pre, sizes_post);
         Ok(())
     }
@@ -310,8 +332,9 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
             group_indices,
             opt_filter,
             total_num_groups,
+            true,
             |accumulator, values_to_accumulate| {
-                accumulator.update_batch(values_to_accumulate)
+                accumulator.update_batch_grouped(values_to_accumulate)
             },
         )?;
         Ok(())
@@ -412,6 +435,7 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
             group_indices,
             None,
             total_num_groups,
+            false,
             |accumulator, values_to_accumulate| {
                 accumulator.merge_batch(values_to_accumulate)?;
                 Ok(())
 
@@ -857,12 +857,6 @@ pub struct DistinctArrayAggAccumulator {
     distinct_metric: Option<Arc<dyn AggregateMetric>>,
 }
 
-/// Avoid per-group timer overhead when the adapter calls us with a tiny batch.
-///
-/// Batches below this threshold are not large enough for a useful timing
-/// sample, while timing each one adds two clock reads and an atomic update.
-const DISTINCT_METRIC_MIN_BATCH_SIZE: usize = 16;
-
 /// Returns `true` if `dt` is, or recursively contains, a `Dictionary` type.
 ///
 /// `RowConverter` always decodes to the physical (non-dictionary) type, so a
@@ -922,16 +916,12 @@ impl DistinctArrayAggAccumulator {
     }
 }
 
-impl Accumulator for DistinctArrayAggAccumulator {
-    fn set_metrics(&mut self, metrics: Arc<dyn AggregateMetrics>) {
-        self.distinct_metric = Some(metrics.metric("distinct"));
-    }
-
-    fn state(&mut self) -> Result<Vec<ScalarValue>> {
-        Ok(vec![self.evaluate()?])
-    }
-
-    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+impl DistinctArrayAggAccumulator {
+    fn update_batch_impl(
+        &mut self,
+        values: &[ArrayRef],
+        record_metric: bool,
+    ) -> Result<()> {
         if values.is_empty() {
             return Ok(());
         }
@@ -961,7 +951,7 @@ impl Accumulator for DistinctArrayAggAccumulator {
             return Ok(());
         }
 
-        let distinct_metric = (col.len() >= DISTINCT_METRIC_MIN_BATCH_SIZE)
+        let distinct_metric = record_metric
             .then(|| self.distinct_metric.as_ref().cloned())
             .flatten();
         let distinct_start = distinct_metric.as_ref().map(|_| Instant::now());
@@ -1016,6 +1006,28 @@ impl Accumulator for DistinctArrayAggAccumulator {
         }
         Ok(())
     }
+}
+
+impl Accumulator for DistinctArrayAggAccumulator {
+    fn set_metrics(&mut self, metrics: Arc<dyn AggregateMetrics>) {
+        self.distinct_metric = Some(metrics.metric("distinct"));
+    }
+
+    fn grouped_update_batch_metric(&self) -> Option<Arc<dyn AggregateMetric>> {
+        self.distinct_metric.as_ref().cloned()
+    }
+
+    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+        self.update_batch_impl(values, true)
+    }
+
+    fn update_batch_grouped(&mut self, values: &[ArrayRef]) -> Result<()> {
+        self.update_batch_impl(values, false)
+    }
+
+    fn state(&mut self) -> Result<Vec<ScalarValue>> {
+        Ok(vec![self.evaluate()?])
+    }
 
     fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
         if states.is_empty() {
@@ -1514,19 +1526,15 @@ mod tests {
     }
 
     #[test]
-    fn distinct_accumulator_skips_metric_for_small_batches() -> Result<()> {
+    fn distinct_accumulator_records_metric_for_small_batches() -> Result<()> {
         let metric_updates = Arc::new(AtomicUsize::new(0));
         let mut accumulator =
             DistinctArrayAggAccumulator::try_new(&DataType::Int32, None, false)?;
         accumulator.set_metrics(Arc::new(CountingMetrics(Arc::clone(&metric_updates))));
 
-        accumulator.update_batch(&[Arc::new(Int32Array::from(vec![
-            1;
-            DISTINCT_METRIC_MIN_BATCH_SIZE
-                - 1
-        ]))])?;
+        accumulator.update_batch(&[Arc::new(Int32Array::from(vec![1]))])?;
 
-        assert_eq!(metric_updates.load(Ordering::Relaxed), 0);
+        assert_eq!(metric_updates.load(Ordering::Relaxed), 1);
         Ok(())
     }
 
@@ -1537,10 +1545,7 @@ mod tests {
             DistinctArrayAggAccumulator::try_new(&DataType::Int32, None, false)?;
         accumulator.set_metrics(Arc::new(CountingMetrics(Arc::clone(&metric_updates))));
 
-        accumulator.update_batch(&[Arc::new(Int32Array::from(vec![
-            1;
-            DISTINCT_METRIC_MIN_BATCH_SIZE
-        ]))])?;
+        accumulator.update_batch(&[Arc::new(Int32Array::from(vec![1; 16]))])?;
 
         assert_eq!(metric_updates.load(Ordering::Relaxed), 1);
         Ok(())
 
@@ -1558,13 +1558,12 @@ mod tests {
             batch?;
         }
 
-        assert!(
-            aggregate_exec
-                .metrics()
-                .unwrap()
-                .sum_by_name("agg_expr_0_internal_distinct_time")
-                .is_some_and(|metric| metric.as_usize() > 0)
-        );
+        let metrics = aggregate_exec.metrics().unwrap();
+        let distinct_time = metrics
+            .iter()
+            .find(|metric| metric.value().name() == "agg_expr_0_internal_distinct_time")
+            .expect("internal distinct time metric");
+        assert!(distinct_time.value().as_usize() > 0);
 
         Ok(())
     }
 
@@ -165,9 +165,11 @@ accumulators in that partition share the same time, and normal metric display
 combines that time across partitions. An aggregate may request its submetric
 during accumulator construction, so it can appear even when its input is empty.
 For example, `array_agg(DISTINCT ...)` records the time spent deduplicating
-input values as `agg_expr_{index}_internal_distinct_time`. These submetrics
-complement the `update`, `merge`, `state`, and `evaluate` timers rather than
-subdividing or replacing them.
+input values as `agg_expr_{index}_internal_distinct_time`. Grouped accumulation
+records this once per input batch, rather than once per group, to avoid making
+metric collection proportional to group cardinality. These submetrics complement
+the `update`, `merge`, `state`, and `evaluate` timers rather than subdividing or
+replacing them.
 
 Except for the `Summary` metric `reduction_factor`, these operator-level and
 per-aggregate metrics are `Dev` metrics. They appear in `EXPLAIN ANALYZE` when