chore: Deprecate internal-only APIs in AggregateExec

datafusion/core/tests/execution/coop.rs

@@ -242,6 +242,7 @@ async fn agg_grouped_topk_yields(
     let value_col = col("value", &inf.schema())?;
     let group = binary(value_col.clone(), Divide, lit(1000000i64), &inf.schema())?;
 
+    #[expect(deprecated)]
     let aggr = Arc::new(
         AggregateExec::try_new(
             AggregateMode::Single,

datafusion/core/tests/physical_optimizer/combine_partial_final_agg.rs

@@ -234,6 +234,7 @@ fn aggregations_with_group_combined() -> datafusion_common::Result<()> {
 }
 
 #[test]
+#[expect(deprecated)]
 fn aggregations_with_limit_combined() -> datafusion_common::Result<()> {
     let schema = schema();
     let aggr_expr = vec![];

datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs

@@ -129,6 +129,7 @@ async fn limited_distinct_aggregate_stream_respects_soft_limit() -> Result<()> {
         output_rows: usize,
     }
 
+    #[expect(deprecated)]
     fn collect_aggregate_runtime_metrics(
         plan: &Arc<dyn ExecutionPlan>,
         metrics: &mut Vec<AggregateRuntimeMetric>,

datafusion/physical-optimizer/src/combine_partial_final_agg.rs

@@ -48,6 +48,7 @@ impl CombinePartialFinalAggregate {
 }
 
 impl PhysicalOptimizerRule for CombinePartialFinalAggregate {
+    #[expect(deprecated)]
     fn optimize(
         &self,
         plan: Arc<dyn ExecutionPlan>,

datafusion/physical-optimizer/src/limited_distinct_aggregation.rs

@@ -44,6 +44,7 @@ impl LimitedDistinctAggregation {
         Self {}
     }
 
+    #[expect(deprecated)]
     fn transform_agg(
         aggr: &AggregateExec,
         limit: usize,

datafusion/physical-optimizer/src/topk_aggregation.rs

@@ -42,6 +42,7 @@ impl TopKAggregation {
         Self {}
     }
 
+    #[expect(deprecated)]
     fn transform_agg(
         aggr: &AggregateExec,
         order_by: &str,

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

@@ -639,7 +639,7 @@ impl GroupedHashAggregateStream {
             group_ordering,
             input_done: false,
             spill_state,
-            group_values_soft_limit: agg.limit_options().map(|config| config.limit()),
+            group_values_soft_limit: agg.limit_options.map(|config| config.limit()),
             skip_aggregation_probe,
             reduction_factor,
         })

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

@@ -96,7 +96,7 @@ impl GroupedTopKAggregateStream {
             // DISTINCT case: use the group key type and get ordering from limit_order_descending
             // The ordering direction is set by the optimizer when it pushes down the limit
             let desc = aggr
-                .limit_options()
+                .limit_options
                 .and_then(|config| config.descending)
                 .ok_or_else(|| {
                     internal_datafusion_err!(
@@ -342,6 +342,7 @@ mod tests {
     use datafusion_physical_expr::expressions::col;
 
     #[tokio::test]
+    #[expect(deprecated)]
     async fn test_topk_aggregate_argument_metrics() -> Result<()> {
         let schema = Arc::new(Schema::new(vec![
             Field::new("k", DataType::UInt32, false),

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

@@ -440,7 +440,7 @@ impl PartialHashAggregateStream {
             reduction_factor,
             early_emit_count,
             skip_aggregation_probe,
-            group_values_soft_limit: agg.limit_options().map(|config| config.limit()),
+            group_values_soft_limit: agg.limit_options.map(|config| config.limit()),
             hash_table: Some(hash_table),
         })
     }
@@ -773,7 +773,7 @@ impl FinalHashAggregateStream {
             input,
             baseline_metrics,
             reservation,
-            group_values_soft_limit: agg.limit_options().map(|config| config.limit()),
+            group_values_soft_limit: agg.limit_options.map(|config| config.limit()),
             hash_table: Some(hash_table),
             spill_context,
         })

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

@@ -927,6 +927,22 @@ impl AggregateExec {
     }
 
     /// Clone this exec, overriding only the limit hint.
+    ///
+    /// This is public for internal use only and should not be treated as a
+    /// public API.
+    ///
+    /// It is marked deprecated so that external callers get a warning, but it
+    /// will not be removed because the physical optimizer depends on it.
+    ///
+    /// It is used by the physical optimizer for certain optimizations. See the
+    /// following rules in `datafusion-physical-optimizer` for details:
+    /// - `topk_aggregation.rs`
+    /// - `limited_distinct_aggregation.rs`
+    /// - `combine_partial_final_agg.rs`
+    #[doc(hidden)]
+    #[deprecated(
+        note = "public for internal use only and not a public API; deprecated only to warn external callers, it will not be removed since the physical optimizer depends on it"
+    )]
     pub fn with_new_limit_options(&self, limit_options: Option<LimitOptions>) -> Self {
         Self {
             limit_options,
@@ -946,6 +962,31 @@ impl AggregateExec {
         }
     }
 
+    /// Set the limit options
+    ///
+    /// This is public for internal use only and should not be treated as a
+    /// public API. See [`Self::with_new_limit_options`] for details.
+    #[doc(hidden)]
+    #[deprecated(
+        note = "public for internal use only and not a public API; deprecated only to warn external callers, it will not be removed since the physical optimizer depends on it"
+    )]
+    pub fn with_limit_options(mut self, limit_options: Option<LimitOptions>) -> Self {
+        self.limit_options = limit_options;
+        self
+    }
+
+    /// Get the limit options (if set)
+    ///
+    /// This is public for internal use only and should not be treated as a
+    /// public API. See [`Self::with_new_limit_options`] for details.
+    #[doc(hidden)]
+    #[deprecated(
+        note = "public for internal use only and not a public API; deprecated only to warn external callers, it will not be removed since the physical optimizer depends on it"
+    )]
+    pub fn limit_options(&self) -> Option<LimitOptions> {
+        self.limit_options
+    }
+
     pub fn cache(&self) -> &PlanProperties {
         &self.cache
     }
@@ -1100,17 +1141,6 @@ impl AggregateExec {
         &self.mode
     }
 
-    /// Set the limit options for this AggExec
-    pub fn with_limit_options(mut self, limit_options: Option<LimitOptions>) -> Self {
-        self.limit_options = limit_options;
-        self
-    }
-
-    /// Get the limit options (if set)
-    pub fn limit_options(&self) -> Option<LimitOptions> {
-        self.limit_options
-    }
-
     /// Grouping expressions
     pub fn group_expr(&self) -> &PhysicalGroupBy {
         &self.group_by
@@ -1375,7 +1405,7 @@ impl AggregateExec {
     /// on an AggregateExec.
     pub fn is_unordered_unfiltered_group_by_distinct(&self) -> bool {
         if self
-            .limit_options()
+            .limit_options
             .and_then(|config| config.descending)
             .is_some()
         {
@@ -2541,6 +2571,7 @@ impl AggregateExec {
     /// Grouping expressions are decoded against the child schema. Aggregate
     /// arguments, ordering, filters, and the dynamic filter are decoded against
     /// the aggregate input schema carried in the protobuf node.
+    #[expect(deprecated)]
     pub fn try_from_proto(
         node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
         ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
@@ -4425,6 +4456,7 @@ mod tests {
     }
 
     #[tokio::test]
+    #[expect(deprecated)]
     async fn limited_distinct_aggregate_uses_migrated_hash_streams() -> Result<()> {
         let schema =
             Arc::new(Schema::new(vec![Field::new("a", DataType::UInt32, false)]));
@@ -6867,6 +6899,7 @@ mod tests {
         ))
     }
 
+    #[expect(deprecated)]
     fn build_test_aggregate_with_mode(
         schema: &SchemaRef,
         stats: Statistics,

datafusion/proto/tests/cases/plans/aggregates.rs

@@ -197,6 +197,7 @@ fn decode_aggregate_without_output_schema() -> Result<()> {
 }
 
 #[test]
+#[expect(deprecated)]
 fn roundtrip_aggregate_with_limit() -> Result<()> {
     let field_a = Field::new("a", DataType::Int64, false);
     let field_b = Field::new("b", DataType::Int64, false);