simplify: trim verbose comments, restore #8296 link, tidy loop

- Cut WHAT-comments and PR-message-style prose from the diff; keep only
  non-obvious WHY (the #8296 link is back on the merging-beneficial guard,
  one-line note on why we hand the unified projection back to
  `remove_unnecessary_projections` for leaf absorption).
- Drop the orphaned `/// Unifies projection with its input ...` doc line
  left over from the removed `try_unifying_projections`.
- Hoist `column_ref_map` out of the per-iteration allocation and `.clear()`
  it each step.
- Replace the `can_unify` flag with a labeled `break 'outer` and destructure
  `Transformed { data, transformed, .. }` in the optimizer-side loop.

No behaviour change. All projection unit + integration tests still pass
(24 + 54 + 23).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Dandandan

datafusion/core/benches/sql_planner_extended.rs

@@ -324,13 +324,9 @@ fn build_non_case_left_join_df_with_push_down_filter(
     rt.block_on(async { ctx.sql(&query).await.unwrap() })
 }
 
-/// Build a query of the shape:
-///   join + wide-OR filter + N chained CTEs, each adding one column
-///   defined by a depth-K nested CASE ladder over the same input column.
-///
-/// The chained projections force the physical planner to walk a stack of
-/// `ProjectionExec`s whose expression trees all reference the same upstream
-/// column, which exercises the `ProjectionPushdown` physical rule heavily.
+/// Join + wide-OR filter + N chained CTEs, each adding one column defined
+/// by a depth-K nested CASE ladder over the same input column. Exercises
+/// the physical `ProjectionPushdown` rule on long projection chains.
 fn build_chained_case_projection_query(
     chained_steps: usize,
     case_depth: usize,
@@ -516,9 +512,6 @@ fn criterion_benchmark(c: &mut Criterion) {
     }
     control_group.finish();
 
-    // Hot-spot bench: N chained CTE projections, each adding one column
-    // defined by a depth-K nested CASE ladder, on top of a join + wide-OR
-    // filter. Exercises the physical `ProjectionPushdown` rule.
     let chained_df = build_chained_case_projection_df(&rt, 80, 23, 30);
     c.bench_function(
         "physical_plan_chained_case_projection_hotspot",

datafusion/optimizer/src/optimize_projections/mod.rs

@@ -536,26 +536,25 @@ fn optimize_subqueries(
 /// - `Ok(None)`: Signals that merge is not beneficial (and has not taken place).
 /// - `Err(error)`: An error occurred during the function call.
 fn merge_consecutive_projections(proj: Projection) -> Result<Transformed<Projection>> {
-    // Iteratively merge as long as the current Projection's input is another
-    // Projection that can be folded. Without this, a chain of N consecutive
-    // projections (e.g. SQL of the form `SELECT *, <expr> AS dN FROM ...`
-    // stacked N times) requires N applications of this rule (re-run by the
-    // outer optimizer loop, each walking the full plan tree) before the
-    // chain is fully collapsed.
+    // Collapse the whole chain in one pass; otherwise an N-deep chain needs
+    // N outer optimizer passes to fully fold.
     let mut current = proj;
     let mut transformed_any = false;
     loop {
-        let result = merge_consecutive_projections_one_level(current)?;
-        current = result.data;
-        if !result.transformed {
-            return Ok(if transformed_any {
-                Transformed::yes(current)
-            } else {
-                Transformed::no(current)
-            });
+        let Transformed {
+            data, transformed, ..
+        } = merge_consecutive_projections_one_level(current)?;
+        current = data;
+        if !transformed {
+            break;
         }
         transformed_any = true;
     }
+    Ok(if transformed_any {
+        Transformed::yes(current)
+    } else {
+        Transformed::no(current)
+    })
 }
 
 fn merge_consecutive_projections_one_level(

datafusion/physical-plan/src/projection.rs

@@ -382,12 +382,6 @@ impl ExecutionPlan for ProjectionExec {
         &self,
         projection: &ProjectionExec,
     ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
-        // Collapse the entire run of consecutive `ProjectionExec`s in a single
-        // pass instead of unifying one level at a time. The pairwise approach
-        // had to allocate (and recompute equivalence properties for) one
-        // intermediate `ProjectionExec` per level, which is the dominant cost
-        // for plans with many chained projections (e.g. a long pipeline of
-        // `SELECT *, <expr> AS dN FROM ...`).
         match try_collapse_projection_chain(projection)? {
             Some(plan) => Ok(Some(plan)),
             None => Ok(Some(Arc::new(projection.clone()))),
@@ -1017,37 +1011,25 @@ pub fn update_join_filter(
     })
 }
 
-/// Unifies `projection` with its input (which is also a [`ProjectionExec`]).
-/// Iteratively collapse a chain of consecutive [`ProjectionExec`]s starting
-/// at `outer` into a single `ProjectionExec`. Returns `None` if the very
-/// first level cannot be unified (matching the previous behavior of
-/// [`try_unifying_projections`] + recursive `remove_unnecessary_projections`).
-///
-/// Functionally equivalent to repeatedly applying [`try_unifying_projections`]
-/// and re-entering [`remove_unnecessary_projections`] on the result, but
-/// avoids constructing an intermediate `ProjectionExec` (and recomputing its
-/// [`PlanProperties`] / equivalence properties) for every level it walks
-/// through. For long projection chains the saved work scales linearly with
-/// the chain length.
+/// Collapse a chain of consecutive [`ProjectionExec`]s into one. Returns
+/// `None` if nothing could be merged.
 fn try_collapse_projection_chain(
     outer: &ProjectionExec,
 ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
     let mut current_exprs: Vec<ProjectionExpr> = outer.expr().to_vec();
     let mut current_input: Arc<dyn ExecutionPlan> = Arc::clone(outer.input());
+    let mut column_ref_map: HashMap<Column, usize> = HashMap::new();
     let mut collapsed_any = false;
 
-    loop {
-        // Only continue if the current input is itself a projection.
+    'outer: loop {
         let Some(inner_proj) = current_input.downcast_ref::<ProjectionExec>()
         else {
             break;
         };
 
-        // Replicate the "merging is beneficial" guard from
-        // `try_unifying_projections`: if any column referenced more than once
-        // in `current_exprs` resolves to a non-trivial expression in the
-        // inner projection, fusing them would duplicate that computation.
-        let mut column_ref_map: HashMap<Column, usize> = HashMap::new();
+        // Merging a referenced-more-than-once column whose inner expression is
+        // non-trivial would duplicate that computation. See #8296.
+        column_ref_map.clear();
         for proj_expr in &current_exprs {
             proj_expr.expr.apply(|expr| {
                 if let Some(column) = expr.downcast_ref::<Column>() {
@@ -1068,24 +1050,15 @@ fn try_collapse_projection_chain(
             break;
         }
 
-        // Substitute each outer expression through the inner projection.
-        let mut new_exprs: Vec<ProjectionExpr> =
-            Vec::with_capacity(current_exprs.len());
-        let mut can_unify = true;
+        let mut new_exprs = Vec::with_capacity(current_exprs.len());
         for proj_expr in &current_exprs {
-            match update_expr(&proj_expr.expr, inner_exprs, true)? {
-                Some(expr) => new_exprs.push(ProjectionExpr {
-                    expr,
-                    alias: proj_expr.alias.clone(),
-                }),
-                None => {
-                    can_unify = false;
-                    break;
-                }
-            }
-        }
-        if !can_unify {
-            break;
+            let Some(expr) = update_expr(&proj_expr.expr, inner_exprs, true)? else {
+                break 'outer;
+            };
+            new_exprs.push(ProjectionExpr {
+                expr,
+                alias: proj_expr.alias.clone(),
+            });
         }
 
         current_exprs = new_exprs;
@@ -1097,11 +1070,8 @@ fn try_collapse_projection_chain(
         return Ok(None);
     }
 
-    // After collapsing the projection chain, hand the unified projection back
-    // to `remove_unnecessary_projections` once. This gives the non-Projection
-    // input (e.g. `DataSourceExec`) the chance to absorb the projection via
-    // its own `try_swapping_with_projection` impl — the same behaviour the
-    // pairwise recursion used to provide for the final step.
+    // Give the (now non-Projection) input a chance to absorb the unified
+    // projection via its own `try_swapping_with_projection` impl.
     let unified: Arc<dyn ExecutionPlan> =
         Arc::new(ProjectionExec::try_new(current_exprs, current_input)?);
     remove_unnecessary_projections(unified).data().map(Some)