chore: Move CoalesceBatches from core to optimizer crate (#13283)

* move coalesce batches

* small fix

datafusion/core/src/physical_optimizer/mod.rs

@@ -21,7 +21,7 @@
 //! "Repartition" or "Sortedness"
 //!
 //! [`ExecutionPlan`]: crate::physical_plan::ExecutionPlan
-pub mod coalesce_batches;
+
 pub mod enforce_distribution;
 pub mod enforce_sorting;
 pub mod join_selection;

datafusion/core/src/physical_optimizer/coalesce_batches.rs → datafusion/physical-optimizer/src/coalesce_batches.rs

@@ -18,19 +18,19 @@
 //! CoalesceBatches optimizer that groups batches together rows
 //! in bigger batches to avoid overhead with small batches
 
+use crate::PhysicalOptimizerRule;
+
 use std::sync::Arc;
 
-use crate::{
-    config::ConfigOptions,
-    error::Result,
-    physical_plan::{
-        coalesce_batches::CoalesceBatchesExec, filter::FilterExec, joins::HashJoinExec,
-        repartition::RepartitionExec, Partitioning,
-    },
+use datafusion_common::config::ConfigOptions;
+use datafusion_common::error::Result;
+use datafusion_physical_expr::Partitioning;
+use datafusion_physical_plan::{
+    coalesce_batches::CoalesceBatchesExec, filter::FilterExec, joins::HashJoinExec,
+    repartition::RepartitionExec, ExecutionPlan,
 };
 
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
-use datafusion_physical_optimizer::PhysicalOptimizerRule;
 
 /// Optimizer rule that introduces CoalesceBatchesExec to avoid overhead with small batches that
 /// are produced by highly selective filters
@@ -46,9 +46,9 @@ impl CoalesceBatches {
 impl PhysicalOptimizerRule for CoalesceBatches {
     fn optimize(
         &self,
-        plan: Arc<dyn crate::physical_plan::ExecutionPlan>,
+        plan: Arc<dyn ExecutionPlan>,
         config: &ConfigOptions,
-    ) -> Result<Arc<dyn crate::physical_plan::ExecutionPlan>> {
+    ) -> Result<Arc<dyn ExecutionPlan>> {
         if !config.execution.coalesce_batches {
             return Ok(plan);
         }

datafusion/physical-optimizer/src/lib.rs

@@ -19,6 +19,7 @@
 #![deny(clippy::clone_on_ref_ptr)]
 
 pub mod aggregate_statistics;
+pub mod coalesce_batches;
 pub mod combine_partial_final_agg;
 pub mod limit_pushdown;
 pub mod limited_distinct_aggregation;