diff --git a/datafusion/core/tests/execution/coop.rs b/datafusion/core/tests/execution/coop.rs index e02364a0530cc..7c5a4f646dea1 100644 --- a/datafusion/core/tests/execution/coop.rs +++ b/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, diff --git a/datafusion/core/tests/physical_optimizer/combine_partial_final_agg.rs b/datafusion/core/tests/physical_optimizer/combine_partial_final_agg.rs index 9e63c341c92d9..12dee363b10bc 100644 --- a/datafusion/core/tests/physical_optimizer/combine_partial_final_agg.rs +++ b/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![]; diff --git a/datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs b/datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs index 323dfd6183306..7de4c97bee67e 100644 --- a/datafusion/core/tests/physical_optimizer/limited_distinct_aggregation.rs +++ b/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, metrics: &mut Vec, diff --git a/datafusion/physical-optimizer/src/combine_partial_final_agg.rs b/datafusion/physical-optimizer/src/combine_partial_final_agg.rs index 297a92c45a16d..81d3df7125df6 100644 --- a/datafusion/physical-optimizer/src/combine_partial_final_agg.rs +++ b/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, diff --git a/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs b/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs index 192a139f36021..89719ef591be6 100644 --- a/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs +++ b/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs @@ -44,6 +44,7 @@ impl LimitedDistinctAggregation { Self {} } + #[expect(deprecated)] fn transform_agg( aggr: &AggregateExec, limit: usize, diff --git a/datafusion/physical-optimizer/src/topk_aggregation.rs b/datafusion/physical-optimizer/src/topk_aggregation.rs index 0eddb5d5507e4..3559fdd567f5a 100644 --- a/datafusion/physical-optimizer/src/topk_aggregation.rs +++ b/datafusion/physical-optimizer/src/topk_aggregation.rs @@ -42,6 +42,7 @@ impl TopKAggregation { Self {} } + #[expect(deprecated)] fn transform_agg( aggr: &AggregateExec, order_by: &str, diff --git a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs index d8f2543996f0e..f03028916f766 100644 --- a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs +++ b/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, }) diff --git a/datafusion/physical-plan/src/aggregates/grouped_topk_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_topk_stream.rs index 13ead739309d3..a4a4bcc41fcff 100644 --- a/datafusion/physical-plan/src/aggregates/grouped_topk_stream.rs +++ b/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), diff --git a/datafusion/physical-plan/src/aggregates/hash_stream.rs b/datafusion/physical-plan/src/aggregates/hash_stream.rs index 152082d88dd90..c4eda5e8664b5 100644 --- a/datafusion/physical-plan/src/aggregates/hash_stream.rs +++ b/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, }) diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 3ed93e09ce4f4..4ff57ff7b9b4b 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/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) -> 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) -> 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 { + 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) -> Self { - self.limit_options = limit_options; - self - } - - /// Get the limit options (if set) - pub fn limit_options(&self) -> Option { - 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, diff --git a/datafusion/proto/tests/cases/plans/aggregates.rs b/datafusion/proto/tests/cases/plans/aggregates.rs index e57ac9fb5045b..9c91a958c1404 100644 --- a/datafusion/proto/tests/cases/plans/aggregates.rs +++ b/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);