chore: Move OptimizeAggregateOrder from core to optimizer crate (#13284)

* move OptimizeAggregateOrder

* clippy fix

Author: jonathanc-n

datafusion/core/src/physical_optimizer/mod.rs

@@ -32,7 +32,6 @@ pub mod replace_with_order_preserving_variants;
 pub mod sanity_checker;
 #[cfg(test)]
 pub mod test_utils;
-pub mod update_aggr_exprs;
 
 mod sort_pushdown;
 mod utils;

datafusion/physical-optimizer/src/lib.rs

@@ -24,5 +24,6 @@ pub mod limited_distinct_aggregation;
 mod optimizer;
 pub mod output_requirements;
 pub mod topk_aggregation;
+pub mod update_aggr_exprs;
 
 pub use optimizer::PhysicalOptimizerRule;

datafusion/core/src/physical_optimizer/update_aggr_exprs.rs renamed to datafusion/physical-optimizer/src/update_aggr_exprs.rs

@@ -27,14 +27,15 @@ use datafusion_physical_expr::aggregate::AggregateFunctionExpr;
 use datafusion_physical_expr::{
     reverse_order_bys, EquivalenceProperties, PhysicalSortRequirement,
 };
-use datafusion_physical_expr_common::sort_expr::{LexOrdering, LexRequirement};
-use datafusion_physical_optimizer::PhysicalOptimizerRule;
+use datafusion_physical_expr::{LexOrdering, LexRequirement};
 use datafusion_physical_plan::aggregates::concat_slices;
 use datafusion_physical_plan::windows::get_ordered_partition_by_indices;
 use datafusion_physical_plan::{
     aggregates::AggregateExec, ExecutionPlan, ExecutionPlanProperties,
 };
 
+use crate::PhysicalOptimizerRule;
+
 /// This optimizer rule checks ordering requirements of aggregate expressions.
 ///
 /// There are 3 kinds of aggregators in terms of ordering requirements:
@@ -60,6 +61,20 @@ impl OptimizeAggregateOrder {
 }
 
 impl PhysicalOptimizerRule for OptimizeAggregateOrder {
+    /// Applies the `OptimizeAggregateOrder` rule to the provided execution plan.
+    ///
+    /// This function traverses the execution plan tree, identifies `AggregateExec` nodes,
+    /// and optimizes their aggregate expressions based on existing input orderings.
+    /// If optimizations are applied, it returns a modified execution plan.
+    ///
+    /// # Arguments
+    ///
+    /// * `plan` - The root of the execution plan to optimize.
+    /// * `_config` - Configuration options (currently unused).
+    ///
+    /// # Returns
+    ///
+    /// A `Result` containing the potentially optimized execution plan or an error.
     fn optimize(
         &self,
         plan: Arc<dyn ExecutionPlan>,
@@ -85,7 +100,12 @@ impl PhysicalOptimizerRule for OptimizeAggregateOrder {
                 let requirement = indices
                     .iter()
                     .map(|&idx| {
-                        PhysicalSortRequirement::new(groupby_exprs[idx].clone(), None)
+                        PhysicalSortRequirement::new(
+                            Arc::<dyn datafusion_physical_plan::PhysicalExpr>::clone(
+                                &groupby_exprs[idx],
+                            ),
+                            None,
+                        )
                     })
                     .collect::<Vec<_>>();