@@ -660,6 +660,15 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
 
     fn shrink_to_fit(&mut self) {
         self.free_scratch();
+        // Each group keeps the capacity of its row indices after a batch, so
+        // a group that had many rows in the last batch still holds them
+        let mut freed = 0;
+        for state in &mut self.states {
+            if state.indices.is_empty() {
+                freed += std::mem::take(&mut state.indices).allocated_size();
+            }
+        }
+        self.free_allocation(freed);
     }
 
     fn convert_to_state(
 
@@ -882,6 +882,18 @@ impl<S: ValueState + 'static> GroupsAccumulator for FirstLastGroupsAccumulator<S
             + self.extreme_of_each_group_buf.0.capacity() * size_of::<usize>()
             + self.extreme_of_each_group_buf.1.capacity() / 8
     }
+
+    fn shrink_to_fit(&mut self) {
+        // `take_state` truncates the per-batch scoreboard, which keeps its
+        // capacity for every group seen before the emit. Every user of the
+        // scoreboard accepts it all-false, so rebuild it at the current size.
+        let num_groups = self.extreme_of_each_group_buf.1.len();
+        let mut scoreboard = BooleanBufferBuilder::new(num_groups);
+        scoreboard.append_n(num_groups, false);
+        self.extreme_of_each_group_buf.1 = scoreboard;
+        self.extreme_buf_dirty = false;
+    }
+
     fn convert_to_state(
         &self,
         values: &[ArrayRef],
 
@@ -424,9 +424,9 @@ mod tests {
     use crate::aggregates::GroupClusteringMode;
     use crate::aggregates::PhysicalGroupBy;
     use crate::aggregates::clustered_test_util::{
-        aggregate_over_sorted_input, assert_completed_groups_emitted_under_limit,
-        assert_output_unchanged_under_limit, collect_under_limit, distinct_key_batches,
-        full_batches, sum_over_sorted_input,
+        aggregate_expr_over_sorted_input, aggregate_over_sorted_input,
+        assert_completed_groups_emitted_under_limit, assert_output_unchanged_under_limit,
+        collect_under_limit, distinct_key_batches, full_batches, sum_over_sorted_input,
     };
     use crate::test::TestMemoryExec;
     use arrow::array::{ArrayRef, Float64Array, Int64Array, ListArray};
@@ -438,6 +438,7 @@ mod tests {
     use datafusion_execution::runtime_env::RuntimeEnvBuilder;
     use datafusion_functions_aggregate::any_value::any_value_udaf;
     use datafusion_functions_aggregate::approx_distinct::approx_distinct_udaf;
+    use datafusion_functions_aggregate::first_last::first_value_udaf;
     use datafusion_functions_aggregate::sum::sum_udaf;
     use datafusion_physical_expr::PhysicalSortExpr;
     use datafusion_physical_expr::aggregate::AggregateExprBuilder;
@@ -586,6 +587,45 @@ mod tests {
         Ok(())
     }
 
+    /// Like [`completed_groups_release_accumulator_scratch`], but the active
+    /// group holds most rows of the batch. The `GroupsAccumulatorAdapter` of
+    /// `ANY_VALUE` keeps the row indices it collected for each group (32 KB for
+    /// the 8191 rows of the active group), so it must release them too.
+    #[tokio::test]
+    async fn active_group_releases_adapter_indices() -> Result<()> {
+        let keys = std::iter::once(0)
+            .chain(std::iter::repeat_n(1, 8191))
+            .collect();
+        let plan = aggregate_over_sorted_input(
+            AggregateMode::Partial,
+            &["k"],
+            vec![vec![keys]],
+            any_value_udaf(),
+        )?;
+        assert_eq!(plan.group_clustering_mode(), &GroupClusteringMode::Full);
+        assert_output_unchanged_under_limit(&plan, 16 * 1024).await
+    }
+
+    /// Like [`completed_groups_release_accumulator_scratch`], with the native
+    /// `FIRST_VALUE(v ORDER BY v)` accumulator. It keeps a bitmap of the
+    /// groups each batch updated, which grows to one bit per group (16 KB for
+    /// 131072 groups), so it must release that bitmap too.
+    #[tokio::test]
+    async fn completed_groups_release_first_value_bitmap() -> Result<()> {
+        let plan = aggregate_expr_over_sorted_input(
+            AggregateMode::Partial,
+            &["k"],
+            vec![vec![(0..131072).collect()]],
+            first_value_udaf(),
+            |builder, schema| {
+                Ok(builder
+                    .order_by(vec![PhysicalSortExpr::new_default(col("v", schema)?)]))
+            },
+        )?;
+        assert_eq!(plan.group_clustering_mode(), &GroupClusteringMode::Full);
+        assert_output_unchanged_under_limit(&plan, 16 * 1024).await
+    }
+
     /// With partial group clustering, the stream emits every state under
     /// memory pressure, so the output is not compared. After the emit, the
     /// accumulators must release their per-batch scratch, or reserving the
 
@@ -7,7 +7,7 @@
 use std::sync::Arc;
 
 use arrow::array::{ArrayRef, Int64Array};
-use arrow::datatypes::{DataType, Field, Schema};
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
 use datafusion_common::test_util::batches_to_string;
 use datafusion_common::{DataFusionError, Result};
@@ -75,6 +75,19 @@ pub(super) fn aggregate_over_sorted_input(
     keys: &[&str],
     batches: Vec<Vec<Vec<i64>>>,
     udaf: Arc<AggregateUDF>,
+) -> Result<AggregateExec> {
+    aggregate_expr_over_sorted_input(mode, keys, batches, udaf, |builder, _| Ok(builder))
+}
+
+/// Like [`aggregate_over_sorted_input`], but `configure` can change the
+/// aggregate expression, given the input schema. For example, it can add
+/// `ORDER BY v`.
+pub(super) fn aggregate_expr_over_sorted_input(
+    mode: AggregateMode,
+    keys: &[&str],
+    batches: Vec<Vec<Vec<i64>>>,
+    udaf: Arc<AggregateUDF>,
+    configure: impl FnOnce(AggregateExprBuilder, &SchemaRef) -> Result<AggregateExprBuilder>,
 ) -> Result<AggregateExec> {
     let fields = keys
         .iter()
@@ -88,10 +101,13 @@ pub(super) fn aggregate_over_sorted_input(
             .collect::<Result<_>>()?,
     );
     let aggregate_exprs = vec![Arc::new(
-        AggregateExprBuilder::new(Arc::clone(&udaf), vec![col("v", &schema)?])
-            .schema(Arc::clone(&schema))
-            .alias(udaf.name())
-            .build()?,
+        configure(
+            AggregateExprBuilder::new(Arc::clone(&udaf), vec![col("v", &schema)?]),
+            &schema,
+        )?
+        .schema(Arc::clone(&schema))
+        .alias(udaf.name())
+        .build()?,
     )];
 
     let (input_schema, group_by) = if mode == AggregateMode::Final {
 
@@ -843,6 +843,7 @@ mod tests {
 
     use super::*;
     use crate::aggregates::GroupClusteringMode;
+    use crate::aggregates::clustered_test_util::collect_under_limit;
     use crate::aggregates::{AggregateMode, PhysicalGroupBy};
     use crate::common::collect;
     use crate::execution_plan::ExecutionPlan;
@@ -1691,21 +1692,17 @@ mod tests {
             aggregate.group_clustering_mode(),
             &GroupClusteringMode::None
         );
-        let context =
-            Arc::new(
-                TaskContext::default()
-                    .with_session_config(SessionConfig::new().set_bool(
-                        "datafusion.execution.enable_migration_aggregate",
-                        true,
-                    ))
-                    .with_runtime(
-                        RuntimeEnvBuilder::new()
-                            .with_memory_pool(Arc::new(GreedyMemoryPool::new(64 * 1024)))
-                            .build_arc()?,
-                    ),
-            );
-        let output = collect(aggregate.execute(0, context)?).await?;
+        let output = collect_under_limit(&aggregate, 64 * 1024).await?;
         assert_eq!(output.iter().map(|b| b.num_rows()).sum::<usize>(), 24576);
+        assert!(
+            aggregate
+                .metrics()
+                .unwrap()
+                .sum_by_name("early_emit_count")
+                .unwrap()
+                .as_usize()
+                > 0
+        );
         Ok(())
     }
 }