refactor(query): Remove PhysicalOptimizer trait

src/query/src/datafusion.rs

@@ -57,7 +57,6 @@ use crate::error::{
 use crate::executor::QueryExecutor;
 use crate::logical_optimizer::LogicalOptimizer;
 use crate::metrics::{OnDone, QUERY_STAGE_ELAPSED};
-use crate::physical_optimizer::PhysicalOptimizer;
 use crate::physical_planner::PhysicalPlanner;
 use crate::physical_wrapper::PhysicalPlanWrapperRef;
 use crate::plan::LogicalPlan;
@@ -256,6 +255,39 @@ impl DatafusionQueryEngine {
             .context(CatalogSnafu)?
             .with_context(|| TableNotFoundSnafu { table: table_name })
     }
+
+    #[tracing::instrument(skip_all)]
+    fn optimize_physical_plan(
+        &self,
+        ctx: &mut QueryEngineContext,
+        plan: Arc<dyn ExecutionPlan>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let _timer = metrics::OPTIMIZE_PHYSICAL_ELAPSED.start_timer();
+
+        let state = ctx.state();
+        let config = state.config_options();
+        // skip optimize AnalyzeExec plan
+        let optimized_plan = if let Some(analyze_plan) = plan.as_any().downcast_ref::<AnalyzeExec>()
+        {
+            let mut new_plan = analyze_plan.input().clone();
+            for optimizer in state.physical_optimizers() {
+                new_plan = optimizer
+                    .optimize(new_plan, config)
+                    .context(DataFusionSnafu)?;
+            }
+            Arc::new(DistAnalyzeExec::new(new_plan))
+        } else {
+            let mut new_plan = plan;
+            for optimizer in state.physical_optimizers() {
+                new_plan = optimizer
+                    .optimize(new_plan, config)
+                    .context(DataFusionSnafu)?;
+            }
+            new_plan
+        };
+
+        Ok(optimized_plan)
+    }
 }
 
 #[async_trait]
@@ -420,41 +452,6 @@ impl PhysicalPlanner for DatafusionQueryEngine {
     }
 }
 
-impl PhysicalOptimizer for DatafusionQueryEngine {
-    #[tracing::instrument(skip_all)]
-    fn optimize_physical_plan(
-        &self,
-        ctx: &mut QueryEngineContext,
-        plan: Arc<dyn ExecutionPlan>,
-    ) -> Result<Arc<dyn ExecutionPlan>> {
-        let _timer = metrics::OPTIMIZE_PHYSICAL_ELAPSED.start_timer();
-
-        let state = ctx.state();
-        let config = state.config_options();
-        // skip optimize AnalyzeExec plan
-        let optimized_plan = if let Some(analyze_plan) = plan.as_any().downcast_ref::<AnalyzeExec>()
-        {
-            let mut new_plan = analyze_plan.input().clone();
-            for optimizer in state.physical_optimizers() {
-                new_plan = optimizer
-                    .optimize(new_plan, config)
-                    .context(DataFusionSnafu)?;
-            }
-            Arc::new(DistAnalyzeExec::new(new_plan))
-        } else {
-            let mut new_plan = plan;
-            for optimizer in state.physical_optimizers() {
-                new_plan = optimizer
-                    .optimize(new_plan, config)
-                    .context(DataFusionSnafu)?;
-            }
-            new_plan
-        };
-
-        Ok(optimized_plan)
-    }
-}
-
 impl QueryExecutor for DatafusionQueryEngine {
     #[tracing::instrument(skip_all)]
     fn execute_stream(

src/query/src/lib.rs

@@ -28,7 +28,6 @@ pub mod logical_optimizer;
 pub mod metrics;
 mod optimizer;
 pub mod parser;
-pub mod physical_optimizer;
 pub mod physical_planner;
 pub mod physical_wrapper;
 pub mod plan;