feat: Expose public method for optimizing physical plans (#11879)

* expose public method for optimizing physical plans using the default planner

* cargo fmt

datafusion-examples/examples/planner_api.rs

@@ -17,6 +17,7 @@
 
 use datafusion::error::Result;
 use datafusion::physical_plan::displayable;
+use datafusion::physical_planner::DefaultPhysicalPlanner;
 use datafusion::prelude::*;
 use datafusion_expr::{LogicalPlan, PlanType};
 
@@ -123,5 +124,20 @@ async fn to_physical_plan_step_by_step_demo(
             .plan
     );
 
+    // Call the physical optimizer with an existing physical plan (in this
+    // case the plan is already optimized, but an unoptimized plan would
+    // typically be used in this context)
+    // Note that this is not part of the trait but a public method
+    // on DefaultPhysicalPlanner. Not all planners will provide this feature.
+    let planner = DefaultPhysicalPlanner::default();
+    let physical_plan =
+        planner.optimize_physical_plan(physical_plan, &ctx.state(), |_, _| {})?;
+    println!(
+        "Optimized physical plan:\n\n{}\n\n",
+        displayable(physical_plan.as_ref())
+            .to_stringified(false, PlanType::InitialPhysicalPlan)
+            .plan
+    );
+
     Ok(())
 }

datafusion/core/src/physical_planner.rs

@@ -180,7 +180,7 @@ impl PhysicalPlanner for DefaultPhysicalPlanner {
                     .create_initial_plan(logical_plan, session_state)
                     .await?;
 
-                self.optimize_internal(plan, session_state, |_, _| {})
+                self.optimize_physical_plan(plan, session_state, |_, _| {})
             }
         }
     }
@@ -1732,7 +1732,7 @@ impl DefaultPhysicalPlanner {
                             }
                         }
 
-                        let optimized_plan = self.optimize_internal(
+                        let optimized_plan = self.optimize_physical_plan(
                             input,
                             session_state,
                             |plan, optimizer| {
@@ -1816,7 +1816,7 @@ impl DefaultPhysicalPlanner {
 
     /// Optimize a physical plan by applying each physical optimizer,
     /// calling observer(plan, optimizer after each one)
-    fn optimize_internal<F>(
+    pub fn optimize_physical_plan<F>(
         &self,
         plan: Arc<dyn ExecutionPlan>,
         session_state: &SessionState,