diff --git a/datafusion-examples/examples/function_factory.rs b/datafusion-examples/examples/function_factory.rs
index a7c8558c6da8..810361fa9c21 100644
--- a/datafusion-examples/examples/function_factory.rs
+++ b/datafusion-examples/examples/function_factory.rs
@@ -164,7 +164,7 @@ impl ScalarUDFImpl for ScalarFunctionWrapper {
 impl ScalarFunctionWrapper {
     // replaces placeholders such as $1 with actual arguments (args[0]
     fn replacement(expr: &Expr, args: &[Expr]) -> Result<Expr> {
-        let result = expr.clone().transform(&|e| {
+        let result = expr.clone().transform(&mut |e| {
             let r = match e {
                 Expr::Placeholder(placeholder) => {
                     let placeholder_position =
diff --git a/datafusion-examples/examples/rewrite_expr.rs b/datafusion-examples/examples/rewrite_expr.rs
index dcebbb55fb66..b8abc93916ab 100644
--- a/datafusion-examples/examples/rewrite_expr.rs
+++ b/datafusion-examples/examples/rewrite_expr.rs
@@ -91,7 +91,7 @@ impl AnalyzerRule for MyAnalyzerRule {
 
 impl MyAnalyzerRule {
     fn analyze_plan(plan: LogicalPlan) -> Result<LogicalPlan> {
-        plan.transform(&|plan| {
+        plan.transform(&mut |plan| {
             Ok(match plan {
                 LogicalPlan::Filter(filter) => {
                     let predicate = Self::analyze_expr(filter.predicate.clone())?;
@@ -107,7 +107,7 @@ impl MyAnalyzerRule {
     }
 
     fn analyze_expr(expr: Expr) -> Result<Expr> {
-        expr.transform(&|expr| {
+        expr.transform(&mut |expr| {
             // closure is invoked for all sub expressions
             Ok(match expr {
                 Expr::Literal(ScalarValue::Int64(i)) => {
@@ -163,7 +163,7 @@ impl OptimizerRule for MyOptimizerRule {
 
 /// use rewrite_expr to modify the expression tree.
 fn my_rewrite(expr: Expr) -> Result<Expr> {
-    expr.transform(&|expr| {
+    expr.transform(&mut |expr| {
         // closure is invoked for all sub expressions
         Ok(match expr {
             Expr::Between(Between {
diff --git a/datafusion/common/src/tree_node.rs b/datafusion/common/src/tree_node.rs
index dff22d495958..1e347b63a53d 100644
--- a/datafusion/common/src/tree_node.rs
+++ b/datafusion/common/src/tree_node.rs
@@ -145,9 +145,9 @@ pub trait TreeNode: Sized {
     /// Convenience utility for writing optimizer rules: Recursively apply the
     /// given function `f` to the tree in a bottom-up (post-order) fashion. When
     /// `f` does not apply to a given node, it is left unchanged.
-    fn transform<F: Fn(Self) -> Result<Transformed<Self>>>(
+    fn transform<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
-        f: &F,
+        f: &mut F,
     ) -> Result<Transformed<Self>> {
         self.transform_up(f)
     }
@@ -155,9 +155,9 @@ pub trait TreeNode: Sized {
     /// Convenience utility for writing optimizer rules: Recursively apply the
     /// given function `f` to a node and then to its children (pre-order traversal).
     /// When `f` does not apply to a given node, it is left unchanged.
-    fn transform_down<F: Fn(Self) -> Result<Transformed<Self>>>(
+    fn transform_down<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
-        f: &F,
+        f: &mut F,
     ) -> Result<Transformed<Self>> {
         handle_transform_recursion_down!(f(self), |c| c.transform_down(f))
     }
@@ -165,20 +165,21 @@ pub trait TreeNode: Sized {
     /// Convenience utility for writing optimizer rules: Recursively apply the
     /// given mutable function `f` to a node and then to its children (pre-order
     /// traversal). When `f` does not apply to a given node, it is left unchanged.
+    #[deprecated(since = "38.0.0", note = "Use `transform_down` instead")]
     fn transform_down_mut<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
         f: &mut F,
     ) -> Result<Transformed<Self>> {
-        handle_transform_recursion_down!(f(self), |c| c.transform_down_mut(f))
+        self.transform_down(f)
     }
 
     /// Convenience utility for writing optimizer rules: Recursively apply the
     /// given function `f` to all children of a node, and then to the node itself
     /// (post-order traversal). When `f` does not apply to a given node, it is
     /// left unchanged.
-    fn transform_up<F: Fn(Self) -> Result<Transformed<Self>>>(
+    fn transform_up<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
-        f: &F,
+        f: &mut F,
     ) -> Result<Transformed<Self>> {
         handle_transform_recursion_up!(self, |c| c.transform_up(f), f)
     }
@@ -187,11 +188,12 @@ pub trait TreeNode: Sized {
     /// given mutable function `f` to all children of a node, and then to the
     /// node itself (post-order traversal). When `f` does not apply to a given
     /// node, it is left unchanged.
+    #[deprecated(since = "38.0.0", note = "Use `transform_up` instead")]
     fn transform_up_mut<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
         f: &mut F,
     ) -> Result<Transformed<Self>> {
-        handle_transform_recursion_up!(self, |c| c.transform_up_mut(f), f)
+        self.transform_up(f)
     }
 
     /// Transforms the tree using `f_down` while traversing the tree top-down
@@ -200,8 +202,8 @@ pub trait TreeNode: Sized {
     ///
     /// Use this method if you want to start the `f_up` process right where `f_down` jumps.
     /// This can make the whole process faster by reducing the number of `f_up` steps.
-    /// If you don't need this, it's just like using `transform_down_mut` followed by
-    /// `transform_up_mut` on the same tree.
+    /// If you don't need this, it's just like using `transform_down` followed by
+    /// `transform_up` on the same tree.
     ///
     /// Consider the following tree structure:
     /// ```text
@@ -439,9 +441,7 @@ impl TreeNodeRecursion {
 /// This struct is used by tree transformation APIs such as
 /// - [`TreeNode::rewrite`],
 /// - [`TreeNode::transform_down`],
-/// - [`TreeNode::transform_down_mut`],
 /// - [`TreeNode::transform_up`],
-/// - [`TreeNode::transform_up_mut`],
 /// - [`TreeNode::transform_down_up`]
 ///
 /// to control the transformation and return the transformed result.
@@ -1466,7 +1466,7 @@ mod tests {
             #[test]
             fn $NAME() -> Result<()> {
                 let tree = test_tree();
-                assert_eq!(tree.transform_down_mut(&mut $F)?, $EXPECTED_TREE);
+                assert_eq!(tree.transform_down(&mut $F)?, $EXPECTED_TREE);
 
                 Ok(())
             }
@@ -1478,7 +1478,7 @@ mod tests {
             #[test]
             fn $NAME() -> Result<()> {
                 let tree = test_tree();
-                assert_eq!(tree.transform_up_mut(&mut $F)?, $EXPECTED_TREE);
+                assert_eq!(tree.transform_up(&mut $F)?, $EXPECTED_TREE);
 
                 Ok(())
             }
diff --git a/datafusion/core/src/physical_optimizer/coalesce_batches.rs b/datafusion/core/src/physical_optimizer/coalesce_batches.rs
index 7c0082037da0..b0b84aa6947b 100644
--- a/datafusion/core/src/physical_optimizer/coalesce_batches.rs
+++ b/datafusion/core/src/physical_optimizer/coalesce_batches.rs
@@ -54,7 +54,7 @@ impl PhysicalOptimizerRule for CoalesceBatches {
         }
 
         let target_batch_size = config.execution.batch_size;
-        plan.transform_up(&|plan| {
+        plan.transform_up(&mut |plan| {
             let plan_any = plan.as_any();
             // The goal here is to detect operators that could produce small batches and only
             // wrap those ones with a CoalesceBatchesExec operator. An alternate approach here
diff --git a/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs b/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs
index 1cba8f025895..4582d7293873 100644
--- a/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs
+++ b/datafusion/core/src/physical_optimizer/combine_partial_final_agg.rs
@@ -51,7 +51,7 @@ impl PhysicalOptimizerRule for CombinePartialFinalAggregate {
         plan: Arc<dyn ExecutionPlan>,
         _config: &ConfigOptions,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        plan.transform_down(&|plan| {
+        plan.transform_down(&mut |plan| {
             let transformed =
                 plan.as_any()
                     .downcast_ref::<AggregateExec>()
@@ -179,7 +179,7 @@ fn normalize_group_exprs(group_exprs: GroupExprsRef) -> GroupExprs {
 fn discard_column_index(group_expr: Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr> {
     group_expr
         .clone()
-        .transform(&|expr| {
+        .transform(&mut |expr| {
             let normalized_form: Option<Arc<dyn PhysicalExpr>> =
                 match expr.as_any().downcast_ref::<Column>() {
                     Some(column) => Some(Arc::new(Column::new(column.name(), 0))),
diff --git a/datafusion/core/src/physical_optimizer/convert_first_last.rs b/datafusion/core/src/physical_optimizer/convert_first_last.rs
index 4102313d3126..c9cec4fe38d6 100644
--- a/datafusion/core/src/physical_optimizer/convert_first_last.rs
+++ b/datafusion/core/src/physical_optimizer/convert_first_last.rs
@@ -60,7 +60,7 @@ impl PhysicalOptimizerRule for OptimizeAggregateOrder {
         plan: Arc<dyn ExecutionPlan>,
         _config: &ConfigOptions,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        plan.transform_up(&get_common_requirement_of_aggregate_input)
+        plan.transform_up(&mut get_common_requirement_of_aggregate_input)
             .data()
     }
 
diff --git a/datafusion/core/src/physical_optimizer/enforce_distribution.rs b/datafusion/core/src/physical_optimizer/enforce_distribution.rs
index c9c54a46bd1c..98a1ffae7fd2 100644
--- a/datafusion/core/src/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/src/physical_optimizer/enforce_distribution.rs
@@ -197,12 +197,12 @@ impl PhysicalOptimizerRule for EnforceDistribution {
             // Run a top-down process to adjust input key ordering recursively
             let plan_requirements = PlanWithKeyRequirements::new_default(plan);
             let adjusted = plan_requirements
-                .transform_down(&adjust_input_keys_ordering)
+                .transform_down(&mut adjust_input_keys_ordering)
                 .data()?;
             adjusted.plan
         } else {
             // Run a bottom-up process
-            plan.transform_up(&|plan| {
+            plan.transform_up(&mut |plan| {
                 Ok(Transformed::yes(reorder_join_keys_to_inputs(plan)?))
             })
             .data()?
@@ -211,7 +211,7 @@ impl PhysicalOptimizerRule for EnforceDistribution {
         let distribution_context = DistributionContext::new_default(adjusted);
         // Distribution enforcement needs to be applied bottom-up.
         let distribution_context = distribution_context
-            .transform_up(&|distribution_context| {
+            .transform_up(&mut |distribution_context| {
                 ensure_distribution(distribution_context, config)
             })
             .data()?;
@@ -1768,14 +1768,14 @@ pub(crate) mod tests {
                     let plan_requirements =
                         PlanWithKeyRequirements::new_default($PLAN.clone());
                     let adjusted = plan_requirements
-                        .transform_down(&adjust_input_keys_ordering)
+                        .transform_down(&mut adjust_input_keys_ordering)
                         .data()
                         .and_then(check_integrity)?;
                     // TODO: End state payloads will be checked here.
                     adjusted.plan
                 } else {
                     // Run reorder_join_keys_to_inputs rule
-                    $PLAN.clone().transform_up(&|plan| {
+                    $PLAN.clone().transform_up(&mut |plan| {
                         Ok(Transformed::yes(reorder_join_keys_to_inputs(plan)?))
                     })
                     .data()?
@@ -1783,7 +1783,7 @@ pub(crate) mod tests {
 
                 // Then run ensure_distribution rule
                 DistributionContext::new_default(adjusted)
-                    .transform_up(&|distribution_context| {
+                    .transform_up(&mut |distribution_context| {
                         ensure_distribution(distribution_context, &config)
                     })
                     .data()
diff --git a/datafusion/core/src/physical_optimizer/enforce_sorting.rs b/datafusion/core/src/physical_optimizer/enforce_sorting.rs
index 5bf21c3dfab5..c6a8890e27c3 100644
--- a/datafusion/core/src/physical_optimizer/enforce_sorting.rs
+++ b/datafusion/core/src/physical_optimizer/enforce_sorting.rs
@@ -160,12 +160,12 @@ impl PhysicalOptimizerRule for EnforceSorting {
         let plan_requirements = PlanWithCorrespondingSort::new_default(plan);
         // Execute a bottom-up traversal to enforce sorting requirements,
         // remove unnecessary sorts, and optimize sort-sensitive operators:
-        let adjusted = plan_requirements.transform_up(&ensure_sorting)?.data;
+        let adjusted = plan_requirements.transform_up(&mut ensure_sorting)?.data;
         let new_plan = if config.optimizer.repartition_sorts {
             let plan_with_coalesce_partitions =
                 PlanWithCorrespondingCoalescePartitions::new_default(adjusted.plan);
             let parallel = plan_with_coalesce_partitions
-                .transform_up(&parallelize_sorts)
+                .transform_up(&mut parallelize_sorts)
                 .data()?;
             parallel.plan
         } else {
@@ -174,7 +174,7 @@ impl PhysicalOptimizerRule for EnforceSorting {
 
         let plan_with_pipeline_fixer = OrderPreservationContext::new_default(new_plan);
         let updated_plan = plan_with_pipeline_fixer
-            .transform_up(&|plan_with_pipeline_fixer| {
+            .transform_up(&mut |plan_with_pipeline_fixer| {
                 replace_with_order_preserving_variants(
                     plan_with_pipeline_fixer,
                     false,
@@ -188,11 +188,13 @@ impl PhysicalOptimizerRule for EnforceSorting {
         // missed by the bottom-up traversal:
         let mut sort_pushdown = SortPushDown::new_default(updated_plan.plan);
         assign_initial_requirements(&mut sort_pushdown);
-        let adjusted = sort_pushdown.transform_down(&pushdown_sorts)?.data;
+        let adjusted = sort_pushdown.transform_down(&mut pushdown_sorts)?.data;
 
         adjusted
             .plan
-            .transform_up(&|plan| Ok(Transformed::yes(replace_with_partial_sort(plan)?)))
+            .transform_up(&mut |plan| {
+                Ok(Transformed::yes(replace_with_partial_sort(plan)?))
+            })
             .data()
     }
 
@@ -681,7 +683,7 @@ mod tests {
             {
                 let plan_requirements = PlanWithCorrespondingSort::new_default($PLAN.clone());
                 let adjusted = plan_requirements
-                    .transform_up(&ensure_sorting)
+                    .transform_up(&mut ensure_sorting)
                     .data()
                     .and_then(check_integrity)?;
                 // TODO: End state payloads will be checked here.
@@ -690,7 +692,7 @@ mod tests {
                     let plan_with_coalesce_partitions =
                         PlanWithCorrespondingCoalescePartitions::new_default(adjusted.plan);
                     let parallel = plan_with_coalesce_partitions
-                        .transform_up(&parallelize_sorts)
+                        .transform_up(&mut parallelize_sorts)
                         .data()
                         .and_then(check_integrity)?;
                     // TODO: End state payloads will be checked here.
@@ -701,7 +703,7 @@ mod tests {
 
                 let plan_with_pipeline_fixer = OrderPreservationContext::new_default(new_plan);
                 let updated_plan = plan_with_pipeline_fixer
-                    .transform_up(&|plan_with_pipeline_fixer| {
+                    .transform_up(&mut |plan_with_pipeline_fixer| {
                         replace_with_order_preserving_variants(
                             plan_with_pipeline_fixer,
                             false,
@@ -716,7 +718,7 @@ mod tests {
                 let mut sort_pushdown = SortPushDown::new_default(updated_plan.plan);
                 assign_initial_requirements(&mut sort_pushdown);
                 sort_pushdown
-                    .transform_down(&pushdown_sorts)
+                    .transform_down(&mut pushdown_sorts)
                     .data()
                     .and_then(check_integrity)?;
                 // TODO: End state payloads will be checked here.
diff --git a/datafusion/core/src/physical_optimizer/join_selection.rs b/datafusion/core/src/physical_optimizer/join_selection.rs
index 72174b0e6e2f..fa08a210cc6a 100644
--- a/datafusion/core/src/physical_optimizer/join_selection.rs
+++ b/datafusion/core/src/physical_optimizer/join_selection.rs
@@ -263,7 +263,7 @@ impl PhysicalOptimizerRule for JoinSelection {
             Box::new(hash_join_swap_subrule),
         ];
         let new_plan = plan
-            .transform_up(&|p| apply_subrules(p, &subrules, config))
+            .transform_up(&mut |p| apply_subrules(p, &subrules, config))
             .data()?;
         // Next, we apply another subrule that tries to optimize joins using any
         // statistics their inputs might have.
@@ -280,7 +280,7 @@ impl PhysicalOptimizerRule for JoinSelection {
         let collect_threshold_byte_size = config.hash_join_single_partition_threshold;
         let collect_threshold_num_rows = config.hash_join_single_partition_threshold_rows;
         new_plan
-            .transform_up(&|plan| {
+            .transform_up(&mut |plan| {
                 statistical_join_selection_subrule(
                     plan,
                     collect_threshold_byte_size,
@@ -842,13 +842,13 @@ mod tests_statistical {
             Box::new(hash_join_swap_subrule),
         ];
         let new_plan = plan
-            .transform_up(&|p| apply_subrules(p, &subrules, &ConfigOptions::new()))
+            .transform_up(&mut |p| apply_subrules(p, &subrules, &ConfigOptions::new()))
             .data()?;
         // TODO: End state payloads will be checked here.
         let config = ConfigOptions::new().optimizer;
         let collect_left_threshold = config.hash_join_single_partition_threshold;
         let collect_threshold_num_rows = config.hash_join_single_partition_threshold_rows;
-        let _ = new_plan.transform_up(&|plan| {
+        let _ = new_plan.transform_up(&mut |plan| {
             statistical_join_selection_subrule(
                 plan,
                 collect_left_threshold,
diff --git a/datafusion/core/src/physical_optimizer/limited_distinct_aggregation.rs b/datafusion/core/src/physical_optimizer/limited_distinct_aggregation.rs
index 9509d4e4c828..db4ea3396eed 100644
--- a/datafusion/core/src/physical_optimizer/limited_distinct_aggregation.rs
+++ b/datafusion/core/src/physical_optimizer/limited_distinct_aggregation.rs
@@ -138,7 +138,7 @@ impl LimitedDistinctAggregation {
             rewrite_applicable = false;
             Ok(Transformed::no(plan))
         };
-        let child = child.clone().transform_down_mut(&mut closure).data().ok()?;
+        let child = child.clone().transform_down(&mut closure).data().ok()?;
         if is_global_limit {
             return Some(Arc::new(GlobalLimitExec::new(
                 child,
@@ -163,7 +163,7 @@ impl PhysicalOptimizerRule for LimitedDistinctAggregation {
         config: &ConfigOptions,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         if config.optimizer.enable_distinct_aggregation_soft_limit {
-            plan.transform_down(&|plan| {
+            plan.transform_down(&mut |plan| {
                 Ok(
                     if let Some(plan) =
                         LimitedDistinctAggregation::transform_limit(plan.clone())
diff --git a/datafusion/core/src/physical_optimizer/output_requirements.rs b/datafusion/core/src/physical_optimizer/output_requirements.rs
index 829d523c990c..7be536559d8c 100644
--- a/datafusion/core/src/physical_optimizer/output_requirements.rs
+++ b/datafusion/core/src/physical_optimizer/output_requirements.rs
@@ -198,7 +198,7 @@ impl PhysicalOptimizerRule for OutputRequirements {
         match self.mode {
             RuleMode::Add => require_top_ordering(plan),
             RuleMode::Remove => plan
-                .transform_up(&|plan| {
+                .transform_up(&mut |plan| {
                     if let Some(sort_req) =
                         plan.as_any().downcast_ref::<OutputRequirementExec>()
                     {
diff --git a/datafusion/core/src/physical_optimizer/pipeline_checker.rs b/datafusion/core/src/physical_optimizer/pipeline_checker.rs
index 1dc8bc5042bf..f853327bb395 100644
--- a/datafusion/core/src/physical_optimizer/pipeline_checker.rs
+++ b/datafusion/core/src/physical_optimizer/pipeline_checker.rs
@@ -50,7 +50,7 @@ impl PhysicalOptimizerRule for PipelineChecker {
         plan: Arc<dyn ExecutionPlan>,
         config: &ConfigOptions,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        plan.transform_up(&|p| check_finiteness_requirements(p, &config.optimizer))
+        plan.transform_up(&mut |p| check_finiteness_requirements(p, &config.optimizer))
             .data()
     }
 
diff --git a/datafusion/core/src/physical_optimizer/projection_pushdown.rs b/datafusion/core/src/physical_optimizer/projection_pushdown.rs
index ed445e6d48b8..e0da9b1dbacc 100644
--- a/datafusion/core/src/physical_optimizer/projection_pushdown.rs
+++ b/datafusion/core/src/physical_optimizer/projection_pushdown.rs
@@ -75,7 +75,8 @@ impl PhysicalOptimizerRule for ProjectionPushdown {
         plan: Arc<dyn ExecutionPlan>,
         _config: &ConfigOptions,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        plan.transform_down(&remove_unnecessary_projections).data()
+        plan.transform_down(&mut remove_unnecessary_projections)
+            .data()
     }
 
     fn name(&self) -> &str {
@@ -977,7 +978,7 @@ fn update_expr(
 
     let new_expr = expr
         .clone()
-        .transform_up_mut(&mut |expr: Arc<dyn PhysicalExpr>| {
+        .transform_up(&mut |expr: Arc<dyn PhysicalExpr>| {
             if state == RewriteState::RewrittenInvalid {
                 return Ok(Transformed::no(expr));
             }
@@ -1120,7 +1121,7 @@ fn new_columns_for_join_on(
             // Rewrite all columns in `on`
             (*on)
                 .clone()
-                .transform(&|expr| {
+                .transform(&mut |expr| {
                     if let Some(column) = expr.as_any().downcast_ref::<Column>() {
                         // Find the column in the projection expressions
                         let new_column = projection_exprs
diff --git a/datafusion/core/src/physical_optimizer/pruning.rs b/datafusion/core/src/physical_optimizer/pruning.rs
index d8a3814d77e1..6b0aba541613 100644
--- a/datafusion/core/src/physical_optimizer/pruning.rs
+++ b/datafusion/core/src/physical_optimizer/pruning.rs
@@ -1175,7 +1175,7 @@ fn rewrite_column_expr(
     column_old: &phys_expr::Column,
     column_new: &phys_expr::Column,
 ) -> Result<Arc<dyn PhysicalExpr>> {
-    e.transform(&|expr| {
+    e.transform(&mut |expr| {
         if let Some(column) = expr.as_any().downcast_ref::<phys_expr::Column>() {
             if column == column_old {
                 return Ok(Transformed::yes(Arc::new(column_new.clone())));
diff --git a/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs b/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs
index ad19215fbf67..6db431c0aa58 100644
--- a/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs
+++ b/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs
@@ -395,7 +395,7 @@ mod tests {
             // Run the rule top-down
             let config = SessionConfig::new().with_prefer_existing_sort($PREFER_EXISTING_SORT);
             let plan_with_pipeline_fixer = OrderPreservationContext::new_default(physical_plan);
-            let parallel = plan_with_pipeline_fixer.transform_up(&|plan_with_pipeline_fixer| replace_with_order_preserving_variants(plan_with_pipeline_fixer, false, false, config.options())).data().and_then(check_integrity)?;
+            let parallel = plan_with_pipeline_fixer.transform_up(&mut |plan_with_pipeline_fixer| replace_with_order_preserving_variants(plan_with_pipeline_fixer, false, false, config.options())).data().and_then(check_integrity)?;
             let optimized_physical_plan = parallel.plan;
 
             // Get string representation of the plan
diff --git a/datafusion/core/src/physical_optimizer/test_utils.rs b/datafusion/core/src/physical_optimizer/test_utils.rs
index 2e6e3af5dfe2..6c75c3490eb4 100644
--- a/datafusion/core/src/physical_optimizer/test_utils.rs
+++ b/datafusion/core/src/physical_optimizer/test_utils.rs
@@ -380,7 +380,7 @@ pub fn sort_exec(
 /// replaced with direct plan equality checks.
 pub fn check_integrity<T: Clone>(context: PlanContext<T>) -> Result<PlanContext<T>> {
     context
-        .transform_up(&|node| {
+        .transform_up(&mut |node| {
             let children_plans = node.plan.children();
             assert_eq!(node.children.len(), children_plans.len());
             for (child_plan, child_node) in
diff --git a/datafusion/core/src/physical_optimizer/topk_aggregation.rs b/datafusion/core/src/physical_optimizer/topk_aggregation.rs
index c47e5e25d143..f7f0285fcb93 100644
--- a/datafusion/core/src/physical_optimizer/topk_aggregation.rs
+++ b/datafusion/core/src/physical_optimizer/topk_aggregation.rs
@@ -120,7 +120,7 @@ impl TopKAggregation {
             }
             Ok(Transformed::no(plan))
         };
-        let child = child.clone().transform_down_mut(&mut closure).data().ok()?;
+        let child = child.clone().transform_down(&mut closure).data().ok()?;
         let sort = SortExec::new(sort.expr().to_vec(), child)
             .with_fetch(sort.fetch())
             .with_preserve_partitioning(sort.preserve_partitioning());
@@ -141,7 +141,7 @@ impl PhysicalOptimizerRule for TopKAggregation {
         config: &ConfigOptions,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         if config.optimizer.enable_topk_aggregation {
-            plan.transform_down(&|plan| {
+            plan.transform_down(&mut |plan| {
                 Ok(
                     if let Some(plan) = TopKAggregation::transform_sort(plan.clone()) {
                         Transformed::yes(plan)
diff --git a/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs b/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs
index 86be887198ae..efbe390c0933 100644
--- a/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs
+++ b/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs
@@ -817,7 +817,7 @@ impl ScalarUDFImpl for ScalarFunctionWrapper {
 impl ScalarFunctionWrapper {
     // replaces placeholders with actual arguments
     fn replacement(expr: &Expr, args: &[Expr]) -> Result<Expr> {
-        let result = expr.clone().transform(&|e| {
+        let result = expr.clone().transform(&mut |e| {
             let r = match e {
                 Expr::Placeholder(placeholder) => {
                     let placeholder_position =
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
index cffb58dadd8e..f8c3d2b622fb 100644
--- a/datafusion/expr/src/expr.rs
+++ b/datafusion/expr/src/expr.rs
@@ -1250,7 +1250,7 @@ impl Expr {
     /// For example, gicen an expression like `<int32> = $0` will infer `$0` to
     /// have type `int32`.
     pub fn infer_placeholder_types(self, schema: &DFSchema) -> Result<Expr> {
-        self.transform(&|mut expr| {
+        self.transform(&mut |mut expr| {
             // Default to assuming the arguments are the same type
             if let Expr::BinaryExpr(BinaryExpr { left, op: _, right }) = &mut expr {
                 rewrite_placeholder(left.as_mut(), right.as_ref(), schema)?;
diff --git a/datafusion/expr/src/expr_rewriter/mod.rs b/datafusion/expr/src/expr_rewriter/mod.rs
index c11619fc0ea2..96e048fa583b 100644
--- a/datafusion/expr/src/expr_rewriter/mod.rs
+++ b/datafusion/expr/src/expr_rewriter/mod.rs
@@ -62,7 +62,7 @@ pub trait FunctionRewrite {
 /// Recursively call [`Column::normalize_with_schemas`] on all [`Column`] expressions
 /// in the `expr` expression tree.
 pub fn normalize_col(expr: Expr, plan: &LogicalPlan) -> Result<Expr> {
-    expr.transform(&|expr| {
+    expr.transform(&mut |expr| {
         Ok({
             if let Expr::Column(c) = expr {
                 let col = LogicalPlanBuilder::normalize(plan, c)?;
@@ -91,7 +91,7 @@ pub fn normalize_col_with_schemas_and_ambiguity_check(
         return Ok(Expr::Unnest(Unnest { expr: Box::new(e) }));
     }
 
-    expr.transform(&|expr| {
+    expr.transform(&mut |expr| {
         Ok({
             if let Expr::Column(c) = expr {
                 let col =
@@ -119,7 +119,7 @@ pub fn normalize_cols(
 /// Recursively replace all [`Column`] expressions in a given expression tree with
 /// `Column` expressions provided by the hash map argument.
 pub fn replace_col(expr: Expr, replace_map: &HashMap<&Column, &Column>) -> Result<Expr> {
-    expr.transform(&|expr| {
+    expr.transform(&mut |expr| {
         Ok({
             if let Expr::Column(c) = &expr {
                 match replace_map.get(c) {
@@ -140,7 +140,7 @@ pub fn replace_col(expr: Expr, replace_map: &HashMap<&Column, &Column>) -> Resul
 /// For example, if there were expressions like `foo.bar` this would
 /// rewrite it to just `bar`.
 pub fn unnormalize_col(expr: Expr) -> Expr {
-    expr.transform(&|expr| {
+    expr.transform(&mut |expr| {
         Ok({
             if let Expr::Column(c) = expr {
                 let col = Column {
@@ -190,7 +190,7 @@ pub fn unnormalize_cols(exprs: impl IntoIterator<Item = Expr>) -> Vec<Expr> {
 /// Recursively remove all the ['OuterReferenceColumn'] and return the inside Column
 /// in the expression tree.
 pub fn strip_outer_reference(expr: Expr) -> Expr {
-    expr.transform(&|expr| {
+    expr.transform(&mut |expr| {
         Ok({
             if let Expr::OuterReferenceColumn(_, col) = expr {
                 Transformed::yes(Expr::Column(col))
@@ -318,7 +318,7 @@ mod test {
     #[test]
     fn rewriter_rewrite() {
         // rewrites all "foo" string literals to "bar"
-        let transformer = |expr: Expr| -> Result<Transformed<Expr>> {
+        let mut transformer = |expr: Expr| -> Result<Transformed<Expr>> {
             match expr {
                 Expr::Literal(ScalarValue::Utf8(Some(utf8_val))) => {
                     let utf8_val = if utf8_val == "foo" {
@@ -336,7 +336,7 @@ mod test {
         // rewrites "foo" --> "bar"
         let rewritten = col("state")
             .eq(lit("foo"))
-            .transform(&transformer)
+            .transform(&mut transformer)
             .data()
             .unwrap();
         assert_eq!(rewritten, col("state").eq(lit("bar")));
@@ -344,7 +344,7 @@ mod test {
         // doesn't rewrite
         let rewritten = col("state")
             .eq(lit("baz"))
-            .transform(&transformer)
+            .transform(&mut transformer)
             .data()
             .unwrap();
         assert_eq!(rewritten, col("state").eq(lit("baz")));
diff --git a/datafusion/expr/src/expr_rewriter/order_by.rs b/datafusion/expr/src/expr_rewriter/order_by.rs
index 2fb522b979b0..dff179f3db00 100644
--- a/datafusion/expr/src/expr_rewriter/order_by.rs
+++ b/datafusion/expr/src/expr_rewriter/order_by.rs
@@ -84,7 +84,7 @@ fn rewrite_in_terms_of_projection(
 ) -> Result<Expr> {
     // assumption is that each item in exprs, such as "b + c" is
     // available as an output column named "b + c"
-    expr.transform(&|expr| {
+    expr.transform(&mut |expr| {
         // search for unnormalized names first such as "c1" (such as aliases)
         if let Some(found) = proj_exprs.iter().find(|a| (**a) == expr) {
             let col = Expr::Column(
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index dbff5046013b..c5dd1f04641d 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -554,7 +554,7 @@ impl LogicalPlan {
                 // AND lineitem.l_quantity < Decimal128(Some(2400),15,2)
 
                 let predicate = predicate
-                    .transform_down(&|expr| {
+                    .transform_down(&mut |expr| {
                         match expr {
                             Expr::Exists { .. }
                             | Expr::ScalarSubquery(_)
@@ -1017,10 +1017,10 @@ impl LogicalPlan {
         self,
         param_values: &ParamValues,
     ) -> Result<LogicalPlan> {
-        self.transform_up_with_subqueries(&|plan| {
+        self.transform_up_with_subqueries(&mut |plan| {
             let schema = plan.schema().clone();
             plan.map_expressions(|e| {
-                e.infer_placeholder_types(&schema)?.transform_up(&|e| {
+                e.infer_placeholder_types(&schema)?.transform_up(&mut |e| {
                     if let Expr::Placeholder(Placeholder { id, .. }) = e {
                         let value = param_values.get_placeholders_with_values(&id)?;
                         Ok(Transformed::yes(Expr::Literal(value)))
@@ -3170,7 +3170,7 @@ digraph {
         // after transformation, because plan is not the same anymore,
         // the parent plan is built again with call to LogicalPlan::with_new_inputs -> with_new_exprs
         let plan = plan
-            .transform(&|plan| match plan {
+            .transform(&mut |plan| match plan {
                 LogicalPlan::TableScan(table) => {
                     let filter = Filter::try_new(
                         external_filter.clone(),
diff --git a/datafusion/expr/src/logical_plan/tree_node.rs b/datafusion/expr/src/logical_plan/tree_node.rs
index 48f047c070dd..22a369579950 100644
--- a/datafusion/expr/src/logical_plan/tree_node.rs
+++ b/datafusion/expr/src/logical_plan/tree_node.rs
@@ -797,9 +797,9 @@ impl LogicalPlan {
     /// Similarly to [`Self::transform`], rewrites this node and its inputs using `f`,
     /// including subqueries that may appear in expressions such as `IN (SELECT
     /// ...)`.
-    pub fn transform_with_subqueries<F: Fn(Self) -> Result<Transformed<Self>>>(
+    pub fn transform_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
-        f: &F,
+        f: &mut F,
     ) -> Result<Transformed<Self>> {
         self.transform_up_with_subqueries(f)
     }
@@ -807,43 +807,21 @@ impl LogicalPlan {
     /// Similarly to [`Self::transform_down`], rewrites this node and its inputs using `f`,
     /// including subqueries that may appear in expressions such as `IN (SELECT
     /// ...)`.
-    pub fn transform_down_with_subqueries<F: Fn(Self) -> Result<Transformed<Self>>>(
-        self,
-        f: &F,
-    ) -> Result<Transformed<Self>> {
-        handle_transform_recursion_down!(f(self), |c| c.transform_down_with_subqueries(f))
-    }
-
-    /// Similarly to [`Self::transform_down_mut`], rewrites this node and its inputs using `f`,
-    /// including subqueries that may appear in expressions such as `IN (SELECT
-    /// ...)`.
-    pub fn transform_down_mut_with_subqueries<
-        F: FnMut(Self) -> Result<Transformed<Self>>,
-    >(
+    pub fn transform_down_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
         f: &mut F,
     ) -> Result<Transformed<Self>> {
-        handle_transform_recursion_down!(f(self), |c| c
-            .transform_down_mut_with_subqueries(f))
+        handle_transform_recursion_down!(f(self), |c| c.transform_down_with_subqueries(f))
     }
 
     /// Similarly to [`Self::transform_up`], rewrites this node and its inputs using `f`,
     /// including subqueries that may appear in expressions such as `IN (SELECT
     /// ...)`.
-    pub fn transform_up_with_subqueries<F: Fn(Self) -> Result<Transformed<Self>>>(
-        self,
-        f: &F,
-    ) -> Result<Transformed<Self>> {
-        handle_transform_recursion_up!(self, |c| c.transform_up_with_subqueries(f), f)
-    }
-
-    pub fn transform_up_mut_with_subqueries<
-        F: FnMut(Self) -> Result<Transformed<Self>>,
-    >(
+    pub fn transform_up_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
         self,
         f: &mut F,
     ) -> Result<Transformed<Self>> {
-        handle_transform_recursion_up!(self, |c| c.transform_up_mut_with_subqueries(f), f)
+        handle_transform_recursion_up!(self, |c| c.transform_up_with_subqueries(f), f)
     }
 
     /// Similarly to [`Self::transform_down`], rewrites this node and its inputs using `f`,
@@ -895,7 +873,7 @@ impl LogicalPlan {
         mut f: F,
     ) -> Result<Transformed<Self>> {
         self.map_expressions(|expr| {
-            expr.transform_down_mut(&mut |expr| match expr {
+            expr.transform_down(&mut |expr| match expr {
                 Expr::Exists(Exists { subquery, negated }) => {
                     f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
                         LogicalPlan::Subquery(subquery) => {
diff --git a/datafusion/optimizer/src/analyzer/count_wildcard_rule.rs b/datafusion/optimizer/src/analyzer/count_wildcard_rule.rs
index 080ec074d3c3..e3e9f959e042 100644
--- a/datafusion/optimizer/src/analyzer/count_wildcard_rule.rs
+++ b/datafusion/optimizer/src/analyzer/count_wildcard_rule.rs
@@ -41,7 +41,7 @@ impl CountWildcardRule {
 
 impl AnalyzerRule for CountWildcardRule {
     fn analyze(&self, plan: LogicalPlan, _: &ConfigOptions) -> Result<LogicalPlan> {
-        plan.transform_down_with_subqueries(&analyze_internal)
+        plan.transform_down_with_subqueries(&mut analyze_internal)
             .data()
     }
 
@@ -78,7 +78,7 @@ fn analyze_internal(plan: LogicalPlan) -> Result<Transformed<LogicalPlan>> {
     let name_preserver = NamePreserver::new(&plan);
     plan.map_expressions(|expr| {
         let original_name = name_preserver.save(&expr)?;
-        let transformed_expr = expr.transform_up(&|expr| match expr {
+        let transformed_expr = expr.transform_up(&mut |expr| match expr {
             Expr::WindowFunction(mut window_function)
                 if is_count_star_window_aggregate(&window_function) =>
             {
diff --git a/datafusion/optimizer/src/analyzer/function_rewrite.rs b/datafusion/optimizer/src/analyzer/function_rewrite.rs
index 4dd3222a32cf..38d018d1627d 100644
--- a/datafusion/optimizer/src/analyzer/function_rewrite.rs
+++ b/datafusion/optimizer/src/analyzer/function_rewrite.rs
@@ -64,7 +64,7 @@ impl ApplyFunctionRewrites {
             let original_name = name_preserver.save(&expr)?;
 
             // recursively transform the expression, applying the rewrites at each step
-            let transformed_expr = expr.transform_up(&|expr| {
+            let transformed_expr = expr.transform_up(&mut |expr| {
                 let mut result = Transformed::no(expr);
                 for rewriter in self.function_rewrites.iter() {
                     result = result.transform_data(|expr| {
@@ -85,7 +85,7 @@ impl AnalyzerRule for ApplyFunctionRewrites {
     }
 
     fn analyze(&self, plan: LogicalPlan, options: &ConfigOptions) -> Result<LogicalPlan> {
-        plan.transform_up_with_subqueries(&|plan| self.rewrite_plan(plan, options))
+        plan.transform_up_with_subqueries(&mut |plan| self.rewrite_plan(plan, options))
             .map(|res| res.data)
     }
 }
diff --git a/datafusion/optimizer/src/analyzer/inline_table_scan.rs b/datafusion/optimizer/src/analyzer/inline_table_scan.rs
index cc5f870a9c73..a3efafe6e314 100644
--- a/datafusion/optimizer/src/analyzer/inline_table_scan.rs
+++ b/datafusion/optimizer/src/analyzer/inline_table_scan.rs
@@ -38,7 +38,7 @@ impl InlineTableScan {
 
 impl AnalyzerRule for InlineTableScan {
     fn analyze(&self, plan: LogicalPlan, _: &ConfigOptions) -> Result<LogicalPlan> {
-        plan.transform_up(&analyze_internal).data()
+        plan.transform_up(&mut analyze_internal).data()
     }
 
     fn name(&self) -> &str {
@@ -49,7 +49,7 @@ impl AnalyzerRule for InlineTableScan {
 fn analyze_internal(plan: LogicalPlan) -> Result<Transformed<LogicalPlan>> {
     // rewrite any subqueries in the plan first
     let transformed_plan =
-        plan.map_subqueries(|plan| plan.transform_up(&analyze_internal))?;
+        plan.map_subqueries(|plan| plan.transform_up(&mut analyze_internal))?;
 
     let transformed_plan = transformed_plan.transform_data(|plan| {
         match plan {
diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs
index 7eda45fb563c..884f001fce32 100644
--- a/datafusion/optimizer/src/decorrelate.rs
+++ b/datafusion/optimizer/src/decorrelate.rs
@@ -381,7 +381,7 @@ fn agg_exprs_evaluation_result_on_empty_batch(
     for e in agg_expr.iter() {
         let result_expr = e
             .clone()
-            .transform_up(&|expr| {
+            .transform_up(&mut |expr| {
                 let new_expr = match expr {
                     Expr::AggregateFunction(expr::AggregateFunction {
                         func_def, ..
@@ -429,7 +429,7 @@ fn proj_exprs_evaluation_result_on_empty_batch(
     for expr in proj_expr.iter() {
         let result_expr = expr
             .clone()
-            .transform_up(&|expr| {
+            .transform_up(&mut |expr| {
                 if let Expr::Column(Column { name, .. }) = &expr {
                     if let Some(result_expr) =
                         input_expr_result_map_for_count_bug.get(name)
@@ -468,7 +468,7 @@ fn filter_exprs_evaluation_result_on_empty_batch(
 ) -> Result<Option<Expr>> {
     let result_expr = filter_expr
         .clone()
-        .transform_up(&|expr| {
+        .transform_up(&mut |expr| {
             if let Expr::Column(Column { name, .. }) = &expr {
                 if let Some(result_expr) = input_expr_result_map_for_count_bug.get(name) {
                     Ok(Transformed::yes(result_expr.clone()))
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index 2b123e3559f5..0caaad1d02d6 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -993,7 +993,7 @@ pub fn replace_cols_by_name(
     e: Expr,
     replace_map: &HashMap<String, Expr>,
 ) -> Result<Expr> {
-    e.transform_up(&|expr| {
+    e.transform_up(&mut |expr| {
         Ok(if let Expr::Column(c) = &expr {
             match replace_map.get(&c.flat_name()) {
                 Some(new_c) => Transformed::yes(new_c.clone()),
diff --git a/datafusion/optimizer/src/scalar_subquery_to_join.rs b/datafusion/optimizer/src/scalar_subquery_to_join.rs
index a8999f9c1d3c..b6ca7f23b448 100644
--- a/datafusion/optimizer/src/scalar_subquery_to_join.rs
+++ b/datafusion/optimizer/src/scalar_subquery_to_join.rs
@@ -95,7 +95,7 @@ impl OptimizerRule for ScalarSubqueryToJoin {
                         if !expr_check_map.is_empty() {
                             rewrite_expr = rewrite_expr
                                 .clone()
-                                .transform_up(&|expr| {
+                                .transform_up(&mut |expr| {
                                     if let Expr::Column(col) = &expr {
                                         if let Some(map_expr) =
                                             expr_check_map.get(&col.name)
@@ -152,7 +152,7 @@ impl OptimizerRule for ScalarSubqueryToJoin {
                                 {
                                     let new_expr = rewrite_expr
                                         .clone()
-                                        .transform_up(&|expr| {
+                                        .transform_up(&mut |expr| {
                                             if let Expr::Column(col) = &expr {
                                                 if let Some(map_expr) =
                                                     expr_check_map.get(&col.name)
diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs
index 58519c61cf1f..4cc0a3e3d304 100644
--- a/datafusion/physical-expr/src/equivalence/class.rs
+++ b/datafusion/physical-expr/src/equivalence/class.rs
@@ -262,7 +262,7 @@ impl EquivalenceGroup {
     /// class it matches with (if any).
     pub fn normalize_expr(&self, expr: Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr> {
         expr.clone()
-            .transform(&|expr| {
+            .transform(&mut |expr| {
                 for cls in self.iter() {
                     if cls.contains(&expr) {
                         return Ok(Transformed::yes(cls.canonical_expr().unwrap()));
@@ -452,7 +452,7 @@ impl EquivalenceGroup {
                         // Rewrite rhs to point to the right side of the join:
                         let new_rhs = rhs
                             .clone()
-                            .transform(&|expr| {
+                            .transform(&mut |expr| {
                                 if let Some(column) =
                                     expr.as_any().downcast_ref::<Column>()
                                 {
diff --git a/datafusion/physical-expr/src/equivalence/mod.rs b/datafusion/physical-expr/src/equivalence/mod.rs
index fd8123c45b06..60aa242351bd 100644
--- a/datafusion/physical-expr/src/equivalence/mod.rs
+++ b/datafusion/physical-expr/src/equivalence/mod.rs
@@ -97,7 +97,7 @@ pub fn add_offset_to_expr(
     expr: Arc<dyn PhysicalExpr>,
     offset: usize,
 ) -> Arc<dyn PhysicalExpr> {
-    expr.transform_down(&|e| match e.as_any().downcast_ref::<Column>() {
+    expr.transform_down(&mut |e| match e.as_any().downcast_ref::<Column>() {
         Some(col) => Ok(Transformed::yes(Arc::new(Column::new(
             col.name(),
             offset + col.index(),
diff --git a/datafusion/physical-expr/src/equivalence/projection.rs b/datafusion/physical-expr/src/equivalence/projection.rs
index 92772e4623be..e5086b4f644a 100644
--- a/datafusion/physical-expr/src/equivalence/projection.rs
+++ b/datafusion/physical-expr/src/equivalence/projection.rs
@@ -59,7 +59,7 @@ impl ProjectionMapping {
                 let target_expr = Arc::new(Column::new(name, expr_idx)) as _;
                 expression
                     .clone()
-                    .transform_down(&|e| match e.as_any().downcast_ref::<Column>() {
+                    .transform_down(&mut |e| match e.as_any().downcast_ref::<Column>() {
                         Some(col) => {
                             // Sometimes, an expression and its name in the input_schema
                             // doesn't match. This can cause problems, so we make sure
diff --git a/datafusion/physical-expr/src/equivalence/properties.rs b/datafusion/physical-expr/src/equivalence/properties.rs
index 58ef5ec797b8..2d5247100225 100644
--- a/datafusion/physical-expr/src/equivalence/properties.rs
+++ b/datafusion/physical-expr/src/equivalence/properties.rs
@@ -857,7 +857,7 @@ impl EquivalenceProperties {
     /// the given expression.
     pub fn get_expr_ordering(&self, expr: Arc<dyn PhysicalExpr>) -> ExprOrdering {
         ExprOrdering::new_default(expr.clone())
-            .transform_up(&|expr| Ok(update_ordering(expr, self)))
+            .transform_up(&mut |expr| Ok(update_ordering(expr, self)))
             .data()
             // Guaranteed to always return `Ok`.
             .unwrap()
diff --git a/datafusion/physical-expr/src/expressions/case.rs b/datafusion/physical-expr/src/expressions/case.rs
index 609349509b86..ed3b3f84ab8b 100644
--- a/datafusion/physical-expr/src/expressions/case.rs
+++ b/datafusion/physical-expr/src/expressions/case.rs
@@ -958,7 +958,7 @@ mod tests {
 
         let expr2 = expr
             .clone()
-            .transform(&|e| {
+            .transform(&mut |e| {
                 let transformed =
                     match e.as_any().downcast_ref::<crate::expressions::Literal>() {
                         Some(lit_value) => match lit_value.value() {
@@ -980,7 +980,7 @@ mod tests {
 
         let expr3 = expr
             .clone()
-            .transform_down(&|e| {
+            .transform_down(&mut |e| {
                 let transformed =
                     match e.as_any().downcast_ref::<crate::expressions::Literal>() {
                         Some(lit_value) => match lit_value.value() {
diff --git a/datafusion/physical-expr/src/utils/mod.rs b/datafusion/physical-expr/src/utils/mod.rs
index a0d6436586a2..44994ec922bb 100644
--- a/datafusion/physical-expr/src/utils/mod.rs
+++ b/datafusion/physical-expr/src/utils/mod.rs
@@ -194,9 +194,7 @@ where
         constructor,
     };
     // Use the builder to transform the expression tree node into a DAG.
-    let root = init
-        .transform_up_mut(&mut |node| builder.mutate(node))
-        .data()?;
+    let root = init.transform_up(&mut |node| builder.mutate(node)).data()?;
     // Return a tuple containing the root node index and the DAG.
     Ok((root.data.unwrap(), builder.graph))
 }
@@ -224,7 +222,7 @@ pub fn reassign_predicate_columns(
     schema: &SchemaRef,
     ignore_not_found: bool,
 ) -> Result<Arc<dyn PhysicalExpr>> {
-    pred.transform_down(&|expr| {
+    pred.transform_down(&mut |expr| {
         let expr_any = expr.as_any();
 
         if let Some(column) = expr_any.downcast_ref::<Column>() {
diff --git a/datafusion/physical-plan/src/joins/stream_join_utils.rs b/datafusion/physical-plan/src/joins/stream_join_utils.rs
index 9824c723d9d1..7c7dd01e1b90 100644
--- a/datafusion/physical-plan/src/joins/stream_join_utils.rs
+++ b/datafusion/physical-plan/src/joins/stream_join_utils.rs
@@ -285,7 +285,7 @@ pub fn convert_sort_expr_with_filter_schema(
         // Since we are sure that one to one column mapping includes all columns, we convert
         // the sort expression into a filter expression.
         let converted_filter_expr = expr
-            .transform_up(&|p| {
+            .transform_up(&mut |p| {
                 convert_filter_columns(p.as_ref(), &column_map).map(|transformed| {
                     match transformed {
                         Some(transformed) => Transformed::yes(transformed),
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index a3d20b97d1ab..f6ab027915f0 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -478,7 +478,7 @@ fn replace_on_columns_of_right_ordering(
             let new_expr = item
                 .expr
                 .clone()
-                .transform(&|e| {
+                .transform(&mut |e| {
                     if e.eq(right_col) {
                         Ok(Transformed::yes(left_col.clone()))
                     } else {
diff --git a/datafusion/physical-plan/src/recursive_query.rs b/datafusion/physical-plan/src/recursive_query.rs
index ba7d1a54548a..67440242fe8e 100644
--- a/datafusion/physical-plan/src/recursive_query.rs
+++ b/datafusion/physical-plan/src/recursive_query.rs
@@ -325,7 +325,7 @@ fn assign_work_table(
     work_table: Arc<WorkTable>,
 ) -> Result<Arc<dyn ExecutionPlan>> {
     let mut work_table_refs = 0;
-    plan.transform_down_mut(&mut |plan| {
+    plan.transform_down(&mut |plan| {
         if let Some(exec) = plan.as_any().downcast_ref::<WorkTableExec>() {
             if work_table_refs > 0 {
                 not_impl_err!(
@@ -353,7 +353,7 @@ fn assign_work_table(
 /// However, if the data of the left table is derived from the work table, it will become outdated
 /// as the work table changes. When the next iteration executes this plan again, we must clear the left table.
 fn reset_plan_states(plan: Arc<dyn ExecutionPlan>) -> Result<Arc<dyn ExecutionPlan>> {
-    plan.transform_up(&|plan| {
+    plan.transform_up(&mut |plan| {
         // WorkTableExec's states have already been updated correctly.
         if plan.as_any().is::<WorkTableExec>() {
             Ok(Transformed::no(plan))
diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs
index 30eacdb44c4a..d3353774dcab 100644
--- a/datafusion/sql/src/select.rs
+++ b/datafusion/sql/src/select.rs
@@ -293,7 +293,7 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                     data: transformed_expr,
                     transformed,
                     tnr: _,
-                } = expr.transform_up_mut(&mut |expr: Expr| {
+                } = expr.transform_up(&mut |expr: Expr| {
                     if let Expr::Unnest(Unnest { expr: ref arg }) = expr {
                         let column_name = expr.display_name()?;
                         unnest_columns.push(column_name.clone());
diff --git a/datafusion/sql/src/unparser/utils.rs b/datafusion/sql/src/unparser/utils.rs
index 9d098c494599..4d62e6b2b40d 100644
--- a/datafusion/sql/src/unparser/utils.rs
+++ b/datafusion/sql/src/unparser/utils.rs
@@ -60,7 +60,7 @@ pub(crate) fn find_agg_node_within_select(
 /// into an actual aggregate expression COUNT(*) as identified in the aggregate node.
 pub(crate) fn unproject_agg_exprs(expr: &Expr, agg: &Aggregate) -> Result<Expr> {
     expr.clone()
-        .transform(&|sub_expr| {
+        .transform(&mut |sub_expr| {
             if let Expr::Column(c) = sub_expr {
                 // find the column in the agg schmea
                 if let Ok(n) = agg.schema.index_of_column(&c) {
diff --git a/datafusion/sql/src/utils.rs b/datafusion/sql/src/utils.rs
index d2f1982d5418..d59549232f4b 100644
--- a/datafusion/sql/src/utils.rs
+++ b/datafusion/sql/src/utils.rs
@@ -34,7 +34,7 @@ use sqlparser::ast::Ident;
 /// Make a best-effort attempt at resolving all columns in the expression tree
 pub(crate) fn resolve_columns(expr: &Expr, plan: &LogicalPlan) -> Result<Expr> {
     expr.clone()
-        .transform_up(&|nested_expr| {
+        .transform_up(&mut |nested_expr| {
             match nested_expr {
                 Expr::Column(col) => {
                     let (qualifier, field) =
@@ -72,7 +72,7 @@ pub(crate) fn rebase_expr(
     plan: &LogicalPlan,
 ) -> Result<Expr> {
     expr.clone()
-        .transform_down(&|nested_expr| {
+        .transform_down(&mut |nested_expr| {
             if base_exprs.contains(&nested_expr) {
                 Ok(Transformed::yes(expr_as_column_expr(&nested_expr, plan)?))
             } else {
@@ -178,7 +178,7 @@ pub(crate) fn resolve_aliases_to_exprs(
     aliases: &HashMap<String, Expr>,
 ) -> Result<Expr> {
     expr.clone()
-        .transform_up(&|nested_expr| match nested_expr {
+        .transform_up(&mut |nested_expr| match nested_expr {
             Expr::Column(c) if c.relation.is_none() => {
                 if let Some(aliased_expr) = aliases.get(&c.name) {
                     Ok(Transformed::yes(aliased_expr.clone()))