perf(optimizer): generic in-place mutable rewrite to obsolete per-rule fast-paths

datafusion/expr/src/logical_plan/tree_node.rs

@@ -858,7 +858,7 @@ impl LogicalPlan {
 
     /// Returns true if any expression in this node contains a subquery
     /// (Exists, InSubquery, SetComparison, or ScalarSubquery).
-    fn has_subquery_expressions(&self) -> bool {
+    pub fn has_subquery_expressions(&self) -> bool {
         let mut found = false;
         let _ = self.apply_expressions(|expr| {
             if found {

datafusion/optimizer/src/common_subexpr_eliminate.rs

@@ -28,7 +28,7 @@ use crate::utils::NamePreserver;
 use datafusion_common::alias::AliasGenerator;
 
 use datafusion_common::cse::{CSE, CSEController, FoundCommonNodes};
-use datafusion_common::tree_node::{Transformed, TreeNode};
+use datafusion_common::tree_node::Transformed;
 use datafusion_common::{Column, DFSchema, DFSchemaRef, Result, qualified_name};
 use datafusion_expr::expr::{Alias, HigherOrderFunction, ScalarFunction};
 use datafusion_expr::logical_plan::{
@@ -586,12 +586,19 @@ impl OptimizerRule for CommonSubexprEliminate {
             | LogicalPlan::Unnest(_)
             | LogicalPlan::RecursiveQuery(_) => {
                 // This rule handles recursion itself in a `ApplyOrder::TopDown` like
-                // manner. Process uncorrelated subqueries in expressions
-                // (e.g., Expr::ScalarSubquery), then direct children.
-                plan.map_uncorrelated_subqueries(|c| self.rewrite(c, config))?
-                    .transform_sibling(|plan| {
-                        plan.map_children(|c| self.rewrite(c, config))
-                    })?
+                // manner. In-place recursion via `Arc::make_mut` + `mem::take`
+                // — when a child returns `Transformed::no` its `Arc` is reused
+                // without rebuilding the parent variant. Same shape as the
+                // previous `map_uncorrelated_subqueries` + `map_children` pair
+                // (subqueries first, then direct children).
+                let mut plan = plan;
+                let changed =
+                    crate::optimizer::rewrite_children_in_place(&mut plan, self, config)?;
+                if changed {
+                    Transformed::yes(plan)
+                } else {
+                    Transformed::no(plan)
+                }
             }
         };
 

datafusion/optimizer/src/eliminate_cross_join.rs

@@ -20,7 +20,7 @@ use crate::{OptimizerConfig, OptimizerRule};
 use std::sync::Arc;
 
 use crate::join_key_set::JoinKeySet;
-use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
+use datafusion_common::tree_node::{Transformed, TreeNodeRecursion};
 use datafusion_common::{NullEquality, Result};
 use datafusion_expr::expr::{BinaryExpr, Expr};
 use datafusion_expr::logical_plan::{
@@ -251,18 +251,19 @@ fn rewrite_children(
     plan: LogicalPlan,
     config: &dyn OptimizerConfig,
 ) -> Result<Transformed<LogicalPlan>> {
-    // Process uncorrelated subqueries in expressions, then direct children.
-    let transformed_plan = plan
-        .map_uncorrelated_subqueries(|input| optimizer.rewrite(input, config))?
-        .transform_sibling(|plan| {
-            plan.map_children(|input| optimizer.rewrite(input, config))
-        })?;
-
-    // recompute schema if the plan was transformed
-    if transformed_plan.transformed {
-        transformed_plan.map_data(|plan| plan.recompute_schema())
+    // In-place recursion via `Arc::make_mut` + `std::mem::take`. When the
+    // recursive `optimizer.rewrite(child, config)` call returns
+    // `Transformed::no`, the child's `Arc` is reused without rebuilding
+    // the parent's `LogicalPlan` variant — same shape as the
+    // ownership-based `map_uncorrelated_subqueries` + `map_children` it
+    // replaces, just without the per-node destructure/restructure cost.
+    let mut plan = plan;
+    let changed =
+        crate::optimizer::rewrite_children_in_place(&mut plan, optimizer, config)?;
+    if changed {
+        plan.recompute_schema().map(Transformed::yes)
     } else {
-        Ok(transformed_plan)
+        Ok(Transformed::no(plan))
     }
 }