Skip to main content

datafusion_expr/logical_plan/
tree_node.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//!  [`TreeNode`] based visiting and rewriting for [`LogicalPlan`]s
19//!
20//! Visiting (read only) APIs
21//! * [`LogicalPlan::visit`]: recursively visit the node and all of its inputs
22//! * [`LogicalPlan::visit_with_subqueries`]: recursively visit the node and all of its inputs, including subqueries
23//! * [`LogicalPlan::apply_children`]: recursively visit all inputs of this node
24//! * [`LogicalPlan::apply_expressions`]: (non recursively) visit all expressions of this node
25//! * [`LogicalPlan::apply_subqueries`]: (non recursively) visit all subqueries of this node
26//! * [`LogicalPlan::apply_with_subqueries`]: recursively visit all inputs and embedded subqueries.
27//!
28//! Rewriting (update) APIs:
29//! * [`LogicalPlan::exists`]: search for an expression in a plan
30//! * [`LogicalPlan::rewrite`]: recursively rewrite the node and all of its inputs
31//! * [`LogicalPlan::map_children`]: recursively rewrite all inputs of this node
32//! * [`LogicalPlan::map_expressions`]: (non recursively) visit all expressions of this node
33//! * [`LogicalPlan::map_subqueries`]: (non recursively) rewrite all subqueries of this node
34//! * [`LogicalPlan::rewrite_with_subqueries`]: recursively rewrite the node and all of its inputs, including subqueries
35//!
36//! (Re)creation APIs (these require substantial cloning and thus are slow):
37//! * [`LogicalPlan::with_new_exprs`]: Create a new plan with different expressions
38//! * [`LogicalPlan::expressions`]: Return a copy of the plan's expressions
39
40use std::sync::Arc;
41
42use crate::logical_plan::plan::RangePartitioning;
43use crate::{
44    Aggregate, Analyze, CreateMemoryTable, CreateView, DdlStatement, Distinct,
45    DistinctOn, DmlStatement, Execute, Explain, Expr, Extension, Filter, Join, Limit,
46    LogicalPlan, Partitioning, Prepare, Projection, RecursiveQuery, Repartition, Sort,
47    Statement, Subquery, SubqueryAlias, TableScan, Union, Unnest, UserDefinedLogicalNode,
48    Values, Window, WriteOp, builder::unnest_with_options, dml::CopyTo,
49};
50use datafusion_common::tree_node::TreeNodeRefContainer;
51
52use crate::expr::{Exists, InSubquery, SetComparison};
53use datafusion_common::tree_node::{
54    Transformed, TreeNode, TreeNodeContainer, TreeNodeIterator, TreeNodeRecursion,
55    TreeNodeRewriter, TreeNodeVisitor,
56};
57use datafusion_common::{Result, internal_err};
58
59impl TreeNode for LogicalPlan {
60    fn apply_children<'n, F: FnMut(&'n Self) -> Result<TreeNodeRecursion>>(
61        &'n self,
62        f: F,
63    ) -> Result<TreeNodeRecursion> {
64        self.inputs().apply_ref_elements(f)
65    }
66
67    /// Applies `f` to each child (input) of this plan node, rewriting them *in place.*
68    ///
69    /// # Notes
70    ///
71    /// Inputs include ONLY direct children, not embedded `LogicalPlan`s for
72    /// subqueries, for example such as are in [`Expr::Exists`].
73    ///
74    /// [`Expr::Exists`]: crate::Expr::Exists
75    fn map_children<F: FnMut(Self) -> Result<Transformed<Self>>>(
76        self,
77        f: F,
78    ) -> Result<Transformed<Self>> {
79        Ok(match self {
80            LogicalPlan::Projection(Projection {
81                expr,
82                input,
83                schema,
84            }) => input.map_elements(f)?.update_data(|input| {
85                LogicalPlan::Projection(Projection {
86                    expr,
87                    input,
88                    schema,
89                })
90            }),
91            LogicalPlan::Filter(Filter { predicate, input }) => input
92                .map_elements(f)?
93                .update_data(|input| LogicalPlan::Filter(Filter { predicate, input })),
94            LogicalPlan::Repartition(Repartition {
95                input,
96                partitioning_scheme,
97            }) => input.map_elements(f)?.update_data(|input| {
98                LogicalPlan::Repartition(Repartition {
99                    input,
100                    partitioning_scheme,
101                })
102            }),
103            LogicalPlan::Window(Window {
104                input,
105                window_expr,
106                schema,
107            }) => input.map_elements(f)?.update_data(|input| {
108                LogicalPlan::Window(Window {
109                    input,
110                    window_expr,
111                    schema,
112                })
113            }),
114            LogicalPlan::Aggregate(Aggregate {
115                input,
116                group_expr,
117                aggr_expr,
118                schema,
119            }) => input.map_elements(f)?.update_data(|input| {
120                LogicalPlan::Aggregate(Aggregate {
121                    input,
122                    group_expr,
123                    aggr_expr,
124                    schema,
125                })
126            }),
127            LogicalPlan::Sort(Sort { expr, input, fetch }) => input
128                .map_elements(f)?
129                .update_data(|input| LogicalPlan::Sort(Sort { expr, input, fetch })),
130            LogicalPlan::Join(Join {
131                left,
132                right,
133                on,
134                filter,
135                join_type,
136                join_constraint,
137                schema,
138                null_equality,
139                null_aware,
140            }) => (left, right).map_elements(f)?.update_data(|(left, right)| {
141                LogicalPlan::Join(Join {
142                    left,
143                    right,
144                    on,
145                    filter,
146                    join_type,
147                    join_constraint,
148                    schema,
149                    null_equality,
150                    null_aware,
151                })
152            }),
153            LogicalPlan::Limit(Limit { skip, fetch, input }) => input
154                .map_elements(f)?
155                .update_data(|input| LogicalPlan::Limit(Limit { skip, fetch, input })),
156            LogicalPlan::Subquery(Subquery {
157                subquery,
158                outer_ref_columns,
159                spans,
160            }) => subquery.map_elements(f)?.update_data(|subquery| {
161                LogicalPlan::Subquery(Subquery {
162                    subquery,
163                    outer_ref_columns,
164                    spans,
165                })
166            }),
167            LogicalPlan::SubqueryAlias(SubqueryAlias {
168                input,
169                alias,
170                schema,
171            }) => input.map_elements(f)?.update_data(|input| {
172                LogicalPlan::SubqueryAlias(SubqueryAlias {
173                    input,
174                    alias,
175                    schema,
176                })
177            }),
178            LogicalPlan::Extension(extension) => rewrite_extension_inputs(extension, f)?
179                .update_data(LogicalPlan::Extension),
180            LogicalPlan::Union(Union { inputs, schema }) => inputs
181                .map_elements(f)?
182                .update_data(|inputs| LogicalPlan::Union(Union { inputs, schema })),
183            LogicalPlan::Distinct(distinct) => match distinct {
184                Distinct::All(input) => input.map_elements(f)?.update_data(Distinct::All),
185                Distinct::On(DistinctOn {
186                    on_expr,
187                    select_expr,
188                    sort_expr,
189                    input,
190                    schema,
191                }) => input.map_elements(f)?.update_data(|input| {
192                    Distinct::On(DistinctOn {
193                        on_expr,
194                        select_expr,
195                        sort_expr,
196                        input,
197                        schema,
198                    })
199                }),
200            }
201            .update_data(LogicalPlan::Distinct),
202            LogicalPlan::Explain(Explain {
203                verbose,
204                explain_format: format,
205                plan,
206                stringified_plans,
207                schema,
208                logical_optimization_succeeded,
209                show_statistics,
210            }) => plan.map_elements(f)?.update_data(|plan| {
211                LogicalPlan::Explain(Explain {
212                    verbose,
213                    explain_format: format,
214                    plan,
215                    stringified_plans,
216                    schema,
217                    logical_optimization_succeeded,
218                    show_statistics,
219                })
220            }),
221            LogicalPlan::Analyze(Analyze {
222                verbose,
223                format,
224                input,
225                schema,
226                analyze_level,
227                analyze_categories,
228            }) => input.map_elements(f)?.update_data(|input| {
229                LogicalPlan::Analyze(Analyze {
230                    verbose,
231                    format,
232                    input,
233                    schema,
234                    analyze_level,
235                    analyze_categories,
236                })
237            }),
238            LogicalPlan::Dml(DmlStatement {
239                table_name,
240                target,
241                op,
242                input,
243                output_schema,
244            }) => input.map_elements(f)?.update_data(|input| {
245                LogicalPlan::Dml(DmlStatement {
246                    table_name,
247                    target,
248                    op,
249                    input,
250                    output_schema,
251                })
252            }),
253            LogicalPlan::Copy(CopyTo {
254                input,
255                output_url,
256                partition_by,
257                file_type,
258                options,
259                output_schema,
260            }) => input.map_elements(f)?.update_data(|input| {
261                LogicalPlan::Copy(CopyTo {
262                    input,
263                    output_url,
264                    partition_by,
265                    file_type,
266                    options,
267                    output_schema,
268                })
269            }),
270            LogicalPlan::Ddl(ddl) => {
271                match ddl {
272                    DdlStatement::CreateMemoryTable(CreateMemoryTable {
273                        name,
274                        constraints,
275                        input,
276                        if_not_exists,
277                        or_replace,
278                        column_defaults,
279                        temporary,
280                    }) => input.map_elements(f)?.update_data(|input| {
281                        DdlStatement::CreateMemoryTable(CreateMemoryTable {
282                            name,
283                            constraints,
284                            input,
285                            if_not_exists,
286                            or_replace,
287                            column_defaults,
288                            temporary,
289                        })
290                    }),
291                    DdlStatement::CreateView(CreateView {
292                        name,
293                        input,
294                        or_replace,
295                        definition,
296                        temporary,
297                    }) => input.map_elements(f)?.update_data(|input| {
298                        DdlStatement::CreateView(CreateView {
299                            name,
300                            input,
301                            or_replace,
302                            definition,
303                            temporary,
304                        })
305                    }),
306                    // no inputs in these statements
307                    DdlStatement::CreateExternalTable(_)
308                    | DdlStatement::CreateCatalogSchema(_)
309                    | DdlStatement::CreateCatalog(_)
310                    | DdlStatement::CreateIndex(_)
311                    | DdlStatement::DropTable(_)
312                    | DdlStatement::DropView(_)
313                    | DdlStatement::DropCatalogSchema(_)
314                    | DdlStatement::CreateFunction(_)
315                    | DdlStatement::DropFunction(_) => Transformed::no(ddl),
316                }
317                .update_data(LogicalPlan::Ddl)
318            }
319            LogicalPlan::Unnest(Unnest {
320                input,
321                exec_columns: input_columns,
322                list_type_columns,
323                struct_type_columns,
324                dependency_indices,
325                schema,
326                options,
327            }) => input.map_elements(f)?.update_data(|input| {
328                LogicalPlan::Unnest(Unnest {
329                    input,
330                    exec_columns: input_columns,
331                    list_type_columns,
332                    struct_type_columns,
333                    dependency_indices,
334                    schema,
335                    options,
336                })
337            }),
338            LogicalPlan::RecursiveQuery(RecursiveQuery {
339                name,
340                static_term,
341                recursive_term,
342                is_distinct,
343                schema,
344            }) => (static_term, recursive_term).map_elements(f)?.update_data(
345                |(static_term, recursive_term)| {
346                    // Ordinary child rewrites preserve derived schemas. Call
347                    // `LogicalPlan::recompute_schema` when child schemas should
348                    // be reconciled again.
349                    LogicalPlan::RecursiveQuery(RecursiveQuery {
350                        name,
351                        static_term,
352                        recursive_term,
353                        is_distinct,
354                        schema,
355                    })
356                },
357            ),
358            LogicalPlan::Statement(stmt) => match stmt {
359                Statement::Prepare(p) => p
360                    .input
361                    .map_elements(f)?
362                    .update_data(|input| Statement::Prepare(Prepare { input, ..p })),
363                _ => Transformed::no(stmt),
364            }
365            .update_data(LogicalPlan::Statement),
366            // plans without inputs
367            LogicalPlan::TableScan { .. }
368            | LogicalPlan::EmptyRelation { .. }
369            | LogicalPlan::Values { .. }
370            | LogicalPlan::DescribeTable(_) => Transformed::no(self),
371        })
372    }
373}
374
375/// Rewrites all inputs for an Extension node "in place"
376/// (it currently has to copy values because there are no APIs for in place modification)
377///
378/// Should be removed when we have an API for in place modifications of the
379/// extension to avoid these copies
380fn rewrite_extension_inputs<F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>>(
381    extension: Extension,
382    f: F,
383) -> Result<Transformed<Extension>> {
384    let Extension { node } = extension;
385
386    node.inputs()
387        .into_iter()
388        .cloned()
389        .map_until_stop_and_collect(f)?
390        .map_data(|new_inputs| {
391            let exprs = node.expressions();
392            Ok(Extension {
393                node: node.with_exprs_and_inputs(exprs, new_inputs)?,
394            })
395        })
396}
397
398/// This macro is used to determine continuation during combined transforming
399/// traversals.
400macro_rules! handle_transform_recursion {
401    ($F_DOWN:expr, $F_CHILD:expr, $F_UP:expr) => {{
402        $F_DOWN?
403            .transform_children(|n| {
404                n.map_subqueries($F_CHILD)?
405                    .transform_sibling(|n| n.map_children($F_CHILD))
406            })?
407            .transform_parent($F_UP)
408    }};
409}
410
411impl LogicalPlan {
412    /// Calls `f` on all expressions in the current `LogicalPlan` node.
413    ///
414    /// # Notes
415    /// * Similar to [`TreeNode::apply`] but for this node's expressions.
416    /// * Does not include expressions in input `LogicalPlan` nodes
417    /// * Visits only the top level expressions (Does not recurse into each expression)
418    pub fn apply_expressions<F: FnMut(&Expr) -> Result<TreeNodeRecursion>>(
419        &self,
420        mut f: F,
421    ) -> Result<TreeNodeRecursion> {
422        match self {
423            LogicalPlan::Projection(Projection { expr, .. }) => expr.apply_elements(f),
424            LogicalPlan::Values(Values { values, .. }) => values.apply_elements(f),
425            LogicalPlan::Filter(Filter { predicate, .. }) => f(predicate),
426            LogicalPlan::Repartition(Repartition {
427                partitioning_scheme,
428                ..
429            }) => match partitioning_scheme {
430                Partitioning::Hash(expr, _) | Partitioning::DistributeBy(expr) => {
431                    expr.apply_elements(f)
432                }
433                Partitioning::Range(range) => range.ordering().to_vec().apply_elements(f),
434                Partitioning::RoundRobinBatch(_) => Ok(TreeNodeRecursion::Continue),
435            },
436            LogicalPlan::Window(Window { window_expr, .. }) => {
437                window_expr.apply_elements(f)
438            }
439            LogicalPlan::Aggregate(Aggregate {
440                group_expr,
441                aggr_expr,
442                ..
443            }) => (group_expr, aggr_expr).apply_ref_elements(f),
444            // There are two part of expression for join, equijoin(on) and non-equijoin(filter).
445            // 1. the first part is `on.len()` equijoin expressions, and the struct of each expr is `left-on = right-on`.
446            // 2. the second part is non-equijoin(filter).
447            LogicalPlan::Join(Join { on, filter, .. }) => {
448                (on, filter).apply_ref_elements(f)
449            }
450            LogicalPlan::Sort(Sort { expr, .. }) => expr.apply_elements(f),
451            LogicalPlan::Extension(extension) => {
452                // would be nice to avoid this copy -- maybe can
453                // update extension to just observer Exprs
454                extension.node.expressions().apply_elements(f)
455            }
456            LogicalPlan::TableScan(TableScan { filters, .. }) => {
457                filters.apply_elements(f)
458            }
459            LogicalPlan::Unnest(unnest) => {
460                let exprs = unnest
461                    .exec_columns
462                    .iter()
463                    .cloned()
464                    .map(Expr::Column)
465                    .collect::<Vec<_>>();
466                exprs.apply_elements(f)
467            }
468            LogicalPlan::Distinct(Distinct::On(DistinctOn {
469                on_expr,
470                select_expr,
471                sort_expr,
472                ..
473            })) => (on_expr, select_expr, sort_expr).apply_ref_elements(f),
474            LogicalPlan::Limit(Limit { skip, fetch, .. }) => {
475                (skip, fetch).apply_ref_elements(f)
476            }
477            LogicalPlan::Statement(stmt) => match stmt {
478                Statement::Execute(Execute { parameters, .. }) => {
479                    parameters.apply_elements(f)
480                }
481                _ => Ok(TreeNodeRecursion::Continue),
482            },
483            LogicalPlan::Dml(DmlStatement {
484                op: WriteOp::MergeInto(merge_op),
485                ..
486            }) => merge_op.exprs().apply_ref_elements(f),
487            // plans without expressions
488            LogicalPlan::EmptyRelation(_)
489            | LogicalPlan::RecursiveQuery(_)
490            | LogicalPlan::Subquery(_)
491            | LogicalPlan::SubqueryAlias(_)
492            | LogicalPlan::Analyze(_)
493            | LogicalPlan::Explain(_)
494            | LogicalPlan::Union(_)
495            | LogicalPlan::Distinct(Distinct::All(_))
496            | LogicalPlan::Dml(_)
497            | LogicalPlan::Ddl(_)
498            | LogicalPlan::Copy(_)
499            | LogicalPlan::DescribeTable(_) => Ok(TreeNodeRecursion::Continue),
500        }
501    }
502
503    /// Rewrites all expressions in the current `LogicalPlan` node using `f`.
504    ///
505    /// Returns the current node.
506    ///
507    /// # Notes
508    /// * Similar to [`TreeNode::map_children`] but for this node's expressions.
509    /// * Visits only the top level expressions (Does not recurse into each expression)
510    pub fn map_expressions<F: FnMut(Expr) -> Result<Transformed<Expr>>>(
511        self,
512        mut f: F,
513    ) -> Result<Transformed<Self>> {
514        Ok(match self {
515            LogicalPlan::Projection(Projection {
516                expr,
517                input,
518                schema,
519            }) => expr.map_elements(f)?.update_data(|expr| {
520                LogicalPlan::Projection(Projection {
521                    expr,
522                    input,
523                    schema,
524                })
525            }),
526            LogicalPlan::Values(Values { schema, values }) => values
527                .map_elements(f)?
528                .update_data(|values| LogicalPlan::Values(Values { schema, values })),
529            LogicalPlan::Filter(Filter { predicate, input }) => f(predicate)?
530                .update_data(|predicate| {
531                    LogicalPlan::Filter(Filter { predicate, input })
532                }),
533            LogicalPlan::Repartition(Repartition {
534                input,
535                partitioning_scheme,
536            }) => match partitioning_scheme {
537                Partitioning::Hash(expr, usize) => expr
538                    .map_elements(f)?
539                    .update_data(|expr| Partitioning::Hash(expr, usize)),
540                Partitioning::DistributeBy(expr) => expr
541                    .map_elements(f)?
542                    .update_data(Partitioning::DistributeBy),
543                Partitioning::Range(range) => {
544                    let split_points = range.split_points().to_vec();
545                    range
546                        .ordering()
547                        .to_vec()
548                        .map_elements(f)?
549                        .map_data(|ordering| {
550                            Ok(Partitioning::Range(RangePartitioning::try_new(
551                                ordering,
552                                split_points,
553                            )?))
554                        })?
555                }
556                Partitioning::RoundRobinBatch(_) => Transformed::no(partitioning_scheme),
557            }
558            .update_data(|partitioning_scheme| {
559                LogicalPlan::Repartition(Repartition {
560                    input,
561                    partitioning_scheme,
562                })
563            }),
564            LogicalPlan::Window(Window {
565                input,
566                window_expr,
567                schema,
568            }) => window_expr.map_elements(f)?.update_data(|window_expr| {
569                LogicalPlan::Window(Window {
570                    input,
571                    window_expr,
572                    schema,
573                })
574            }),
575            LogicalPlan::Aggregate(Aggregate {
576                input,
577                group_expr,
578                aggr_expr,
579                schema,
580            }) => (group_expr, aggr_expr).map_elements(f)?.update_data(
581                |(group_expr, aggr_expr)| {
582                    LogicalPlan::Aggregate(Aggregate {
583                        input,
584                        group_expr,
585                        aggr_expr,
586                        schema,
587                    })
588                },
589            ),
590
591            // There are two part of expression for join, equijoin(on) and non-equijoin(filter).
592            // 1. the first part is `on.len()` equijoin expressions, and the struct of each expr is `left-on = right-on`.
593            // 2. the second part is non-equijoin(filter).
594            LogicalPlan::Join(Join {
595                left,
596                right,
597                on,
598                filter,
599                join_type,
600                join_constraint,
601                schema,
602                null_equality,
603                null_aware,
604            }) => (on, filter).map_elements(f)?.update_data(|(on, filter)| {
605                LogicalPlan::Join(Join {
606                    left,
607                    right,
608                    on,
609                    filter,
610                    join_type,
611                    join_constraint,
612                    schema,
613                    null_equality,
614                    null_aware,
615                })
616            }),
617            LogicalPlan::Sort(Sort { expr, input, fetch }) => expr
618                .map_elements(f)?
619                .update_data(|expr| LogicalPlan::Sort(Sort { expr, input, fetch })),
620            LogicalPlan::Extension(Extension { node }) => {
621                let raw_exprs = node.expressions();
622                if raw_exprs.is_empty() {
623                    // No expressions to transform — skip expensive clone of
624                    // all inputs and reconstruction via with_exprs_and_inputs.
625                    Transformed::no(LogicalPlan::Extension(Extension { node }))
626                } else {
627                    // TODO: a more general optimization would be to change
628                    // `UserDefinedLogicalNode::expressions()` to return
629                    // references (`&[Expr]`) instead of cloned `Vec<Expr>`,
630                    // and only clone + rebuild when the transform actually
631                    // modifies an expression. This would avoid the clone +
632                    // `with_exprs_and_inputs` rebuild even for non-empty
633                    // expression lists when the transform is a no-op.
634                    let exprs = raw_exprs.map_elements(f)?;
635                    let plan = LogicalPlan::Extension(Extension {
636                        node: UserDefinedLogicalNode::with_exprs_and_inputs(
637                            node.as_ref(),
638                            exprs.data,
639                            node.inputs().into_iter().cloned().collect::<Vec<_>>(),
640                        )?,
641                    });
642                    Transformed::new(plan, exprs.transformed, exprs.tnr)
643                }
644            }
645            LogicalPlan::TableScan(TableScan {
646                table_name,
647                source,
648                projection,
649                projected_schema,
650                filters,
651                fetch,
652                statistics_requests,
653            }) => filters.map_elements(f)?.update_data(|filters| {
654                LogicalPlan::TableScan(TableScan {
655                    table_name,
656                    source,
657                    projection,
658                    projected_schema,
659                    filters,
660                    fetch,
661                    statistics_requests,
662                })
663            }),
664            LogicalPlan::Distinct(Distinct::On(DistinctOn {
665                on_expr,
666                select_expr,
667                sort_expr,
668                input,
669                schema,
670            })) => (on_expr, select_expr, sort_expr)
671                .map_elements(f)?
672                .update_data(|(on_expr, select_expr, sort_expr)| {
673                    LogicalPlan::Distinct(Distinct::On(DistinctOn {
674                        on_expr,
675                        select_expr,
676                        sort_expr,
677                        input,
678                        schema,
679                    }))
680                }),
681            LogicalPlan::Limit(Limit { skip, fetch, input }) => {
682                (skip, fetch).map_elements(f)?.update_data(|(skip, fetch)| {
683                    LogicalPlan::Limit(Limit { skip, fetch, input })
684                })
685            }
686            LogicalPlan::Statement(stmt) => match stmt {
687                Statement::Execute(e) => {
688                    e.parameters.map_elements(f)?.update_data(|parameters| {
689                        Statement::Execute(Execute { parameters, ..e })
690                    })
691                }
692                _ => Transformed::no(stmt),
693            }
694            .update_data(LogicalPlan::Statement),
695            LogicalPlan::Unnest(Unnest {
696                input,
697                exec_columns,
698                options,
699                ..
700            }) => {
701                let exprs: Vec<Expr> =
702                    exec_columns.into_iter().map(Expr::Column).collect();
703                exprs.map_elements(f)?.map_data(|mapped_exprs| {
704                    let new_columns = mapped_exprs
705                        .into_iter()
706                        .map(|e| match e {
707                            Expr::Column(c) => Ok(c),
708                            other => internal_err!(
709                                "Expected Expr::Column for Unnest exec_columns, got {other:?}"
710                            ),
711                        })
712                        .collect::<Result<Vec<_>>>()?;
713                    // Rebuild through `unnest_with_options` so the derived
714                    // `list_type_columns`, `struct_type_columns`,
715                    // `dependency_indices`, and `schema` are recomputed from
716                    // the (possibly rewritten) columns rather than carried over
717                    // stale. This keeps `map_expressions` consistent with
718                    // `with_new_exprs`.
719                    unnest_with_options(
720                        Arc::unwrap_or_clone(input),
721                        new_columns,
722                        options,
723                    )
724                })?
725            }
726            LogicalPlan::Dml(DmlStatement {
727                table_name,
728                target,
729                op: WriteOp::MergeInto(merge_op),
730                input,
731                output_schema,
732            }) => {
733                let owned_exprs: Vec<Expr> =
734                    merge_op.exprs().into_iter().cloned().collect();
735                owned_exprs.map_elements(f)?.transform_data(|new_exprs| {
736                    Ok(Transformed::no(LogicalPlan::Dml(DmlStatement {
737                        table_name,
738                        target,
739                        op: WriteOp::MergeInto(Box::new(
740                            merge_op.with_new_exprs(new_exprs)?,
741                        )),
742                        input,
743                        output_schema,
744                    })))
745                })?
746            }
747            // plans without expressions
748            LogicalPlan::EmptyRelation(_)
749            | LogicalPlan::RecursiveQuery(_)
750            | LogicalPlan::Subquery(_)
751            | LogicalPlan::SubqueryAlias(_)
752            | LogicalPlan::Analyze(_)
753            | LogicalPlan::Explain(_)
754            | LogicalPlan::Union(_)
755            | LogicalPlan::Distinct(Distinct::All(_))
756            | LogicalPlan::Dml(_)
757            | LogicalPlan::Ddl(_)
758            | LogicalPlan::Copy(_)
759            | LogicalPlan::DescribeTable(_) => Transformed::no(self),
760        })
761    }
762
763    /// Visits a plan similarly to [`Self::visit`], including subqueries that
764    /// may appear in expressions such as `IN (SELECT ...)`.
765    #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
766    pub fn visit_with_subqueries<V: for<'n> TreeNodeVisitor<'n, Node = Self>>(
767        &self,
768        visitor: &mut V,
769    ) -> Result<TreeNodeRecursion> {
770        visitor
771            .f_down(self)?
772            .visit_children(|| {
773                self.apply_subqueries(|c| c.visit_with_subqueries(visitor))?
774                    .visit_sibling(|| {
775                        self.apply_children(|c| c.visit_with_subqueries(visitor))
776                    })
777            })?
778            .visit_parent(|| visitor.f_up(self))
779    }
780
781    /// Similarly to [`Self::rewrite`], rewrites this node and its inputs using `f`,
782    /// including subqueries that may appear in expressions such as `IN (SELECT
783    /// ...)`.
784    #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
785    pub fn rewrite_with_subqueries<R: TreeNodeRewriter<Node = Self>>(
786        self,
787        rewriter: &mut R,
788    ) -> Result<Transformed<Self>> {
789        handle_transform_recursion!(
790            rewriter.f_down(self),
791            |c| c.rewrite_with_subqueries(rewriter),
792            |n| rewriter.f_up(n)
793        )
794    }
795
796    /// Similarly to [`Self::apply`], calls `f` on this node and all its inputs,
797    /// including subqueries that may appear in expressions such as `IN (SELECT
798    /// ...)`.
799    pub fn apply_with_subqueries<F: FnMut(&Self) -> Result<TreeNodeRecursion>>(
800        &self,
801        mut f: F,
802    ) -> Result<TreeNodeRecursion> {
803        #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
804        fn apply_with_subqueries_impl<
805            F: FnMut(&LogicalPlan) -> Result<TreeNodeRecursion>,
806        >(
807            node: &LogicalPlan,
808            f: &mut F,
809        ) -> Result<TreeNodeRecursion> {
810            f(node)?.visit_children(|| {
811                node.apply_subqueries(|c| apply_with_subqueries_impl(c, f))?
812                    .visit_sibling(|| {
813                        node.apply_children(|c| apply_with_subqueries_impl(c, f))
814                    })
815            })
816        }
817
818        apply_with_subqueries_impl(self, &mut f)
819    }
820
821    /// Similarly to [`Self::transform`], rewrites this node and its inputs using `f`,
822    /// including subqueries that may appear in expressions such as `IN (SELECT
823    /// ...)`.
824    pub fn transform_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
825        self,
826        f: F,
827    ) -> Result<Transformed<Self>> {
828        self.transform_up_with_subqueries(f)
829    }
830
831    /// Similarly to [`Self::transform_down`], rewrites this node and its inputs using `f`,
832    /// including subqueries that may appear in expressions such as `IN (SELECT
833    /// ...)`.
834    pub fn transform_down_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
835        self,
836        mut f: F,
837    ) -> Result<Transformed<Self>> {
838        #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
839        fn transform_down_with_subqueries_impl<
840            F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
841        >(
842            node: LogicalPlan,
843            f: &mut F,
844        ) -> Result<Transformed<LogicalPlan>> {
845            f(node)?.transform_children(|n| {
846                n.map_subqueries(|c| transform_down_with_subqueries_impl(c, f))?
847                    .transform_sibling(|n| {
848                        n.map_children(|c| transform_down_with_subqueries_impl(c, f))
849                    })
850            })
851        }
852
853        transform_down_with_subqueries_impl(self, &mut f)
854    }
855
856    /// Similarly to [`Self::transform_up`], rewrites this node and its inputs using `f`,
857    /// including subqueries that may appear in expressions such as `IN (SELECT
858    /// ...)`.
859    pub fn transform_up_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
860        self,
861        mut f: F,
862    ) -> Result<Transformed<Self>> {
863        #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
864        fn transform_up_with_subqueries_impl<
865            F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
866        >(
867            node: LogicalPlan,
868            f: &mut F,
869        ) -> Result<Transformed<LogicalPlan>> {
870            node.map_subqueries(|c| transform_up_with_subqueries_impl(c, f))?
871                .transform_sibling(|n| {
872                    n.map_children(|c| transform_up_with_subqueries_impl(c, f))
873                })?
874                .transform_parent(f)
875        }
876
877        transform_up_with_subqueries_impl(self, &mut f)
878    }
879
880    /// Similarly to [`Self::transform_down`], rewrites this node and its inputs using `f`,
881    /// including subqueries that may appear in expressions such as `IN (SELECT
882    /// ...)`.
883    pub fn transform_down_up_with_subqueries<
884        FD: FnMut(Self) -> Result<Transformed<Self>>,
885        FU: FnMut(Self) -> Result<Transformed<Self>>,
886    >(
887        self,
888        mut f_down: FD,
889        mut f_up: FU,
890    ) -> Result<Transformed<Self>> {
891        #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
892        fn transform_down_up_with_subqueries_impl<
893            FD: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
894            FU: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
895        >(
896            node: LogicalPlan,
897            f_down: &mut FD,
898            f_up: &mut FU,
899        ) -> Result<Transformed<LogicalPlan>> {
900            handle_transform_recursion!(
901                f_down(node),
902                |c| transform_down_up_with_subqueries_impl(c, f_down, f_up),
903                f_up
904            )
905        }
906
907        transform_down_up_with_subqueries_impl(self, &mut f_down, &mut f_up)
908    }
909
910    /// Similarly to [`Self::apply`], calls `f` on this node and its inputs,
911    /// including subqueries that may appear in expressions such as `IN (SELECT
912    /// ...)`.
913    pub fn apply_subqueries<F: FnMut(&Self) -> Result<TreeNodeRecursion>>(
914        &self,
915        mut f: F,
916    ) -> Result<TreeNodeRecursion> {
917        self.apply_expressions(|expr| {
918            expr.apply(|expr| match expr {
919                Expr::Exists(Exists { subquery, .. })
920                | Expr::InSubquery(InSubquery { subquery, .. })
921                | Expr::SetComparison(SetComparison { subquery, .. })
922                | Expr::ScalarSubquery(subquery) => {
923                    // Wrap in LogicalPlan::Subquery to match f's signature
924                    f(&LogicalPlan::Subquery(subquery.clone()))
925                }
926                _ => Ok(TreeNodeRecursion::Continue),
927            })
928        })
929    }
930
931    /// Returns true if any expression in this node contains a subquery
932    /// (Exists, InSubquery, SetComparison, or ScalarSubquery).
933    fn has_subquery_expressions(&self) -> bool {
934        let mut found = false;
935        let _ = self.apply_expressions(|expr| {
936            if found {
937                return Ok(TreeNodeRecursion::Stop);
938            }
939            expr.apply(|e| {
940                if matches!(
941                    e,
942                    Expr::Exists(_)
943                        | Expr::InSubquery(_)
944                        | Expr::SetComparison(_)
945                        | Expr::ScalarSubquery(_)
946                ) {
947                    found = true;
948                    Ok(TreeNodeRecursion::Stop)
949                } else {
950                    Ok(TreeNodeRecursion::Continue)
951                }
952            })
953        });
954        found
955    }
956
957    /// Similarly to [`Self::map_children`], rewrites all subqueries that may
958    /// appear in expressions such as `IN (SELECT ...)` using `f`.
959    ///
960    /// Returns the current node.
961    pub fn map_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
962        self,
963        mut f: F,
964    ) -> Result<Transformed<Self>> {
965        // Fast path: skip the expensive ownership-based expression traversal
966        // when this node has no subquery expressions. This avoids
967        // map_expressions → transform_down walking every expression node
968        // via consume+recreate just to find no subqueries.
969        if !self.has_subquery_expressions() {
970            return Ok(Transformed::no(self));
971        }
972
973        self.map_expressions(|expr| {
974            expr.transform_down(|expr| match expr {
975                Expr::Exists(Exists { subquery, negated }) => {
976                    f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
977                        LogicalPlan::Subquery(subquery) => {
978                            Ok(Expr::Exists(Exists { subquery, negated }))
979                        }
980                        _ => internal_err!("Transformation should return Subquery"),
981                    })
982                }
983                Expr::InSubquery(InSubquery {
984                    expr,
985                    subquery,
986                    negated,
987                }) => f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
988                    LogicalPlan::Subquery(subquery) => Ok(Expr::InSubquery(InSubquery {
989                        expr,
990                        subquery,
991                        negated,
992                    })),
993                    _ => internal_err!("Transformation should return Subquery"),
994                }),
995                Expr::SetComparison(SetComparison {
996                    expr,
997                    subquery,
998                    op,
999                    quantifier,
1000                }) => f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
1001                    LogicalPlan::Subquery(subquery) => {
1002                        Ok(Expr::SetComparison(SetComparison {
1003                            expr,
1004                            subquery,
1005                            op,
1006                            quantifier,
1007                        }))
1008                    }
1009                    _ => internal_err!("Transformation should return Subquery"),
1010                }),
1011                Expr::ScalarSubquery(subquery) => f(LogicalPlan::Subquery(subquery))?
1012                    .map_data(|s| match s {
1013                        LogicalPlan::Subquery(subquery) => {
1014                            Ok(Expr::ScalarSubquery(subquery))
1015                        }
1016                        _ => internal_err!("Transformation should return Subquery"),
1017                    }),
1018                _ => Ok(Transformed::no(expr)),
1019            })
1020        })
1021    }
1022
1023    /// Similar to [`Self::map_subqueries`], but only applies `f` to
1024    /// uncorrelated subqueries (those with no outer column references).
1025    pub fn map_uncorrelated_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
1026        self,
1027        mut f: F,
1028    ) -> Result<Transformed<Self>> {
1029        self.map_subqueries(|subquery_plan| match &subquery_plan {
1030            LogicalPlan::Subquery(sq) if sq.outer_ref_columns.is_empty() => {
1031                f(subquery_plan)
1032            }
1033            _ => Ok(Transformed::no(subquery_plan)),
1034        })
1035    }
1036}