## Coverage Report

#### Created: 2025-11-22 06:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /Users/andrewlamb/Software/datafusion/datafusion/expr/src/logical_plan/tree_node.rs

    Line
    Count
    Source
    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
    40
    use crate::{
    41
        dml::CopyTo, Aggregate, Analyze, CreateMemoryTable, CreateView, DdlStatement,
    42
        Distinct, DistinctOn, DmlStatement, Execute, Explain, Expr, Extension, Filter, Join,
    43
        Limit, LogicalPlan, Partitioning, Prepare, Projection, RecursiveQuery, Repartition,
    44
        Sort, Statement, Subquery, SubqueryAlias, TableScan, Union, Unnest,
    45
        UserDefinedLogicalNode, Values, Window,
    46
    };
    47
    use datafusion_common::tree_node::TreeNodeRefContainer;
    48
    49
    use crate::expr::{Exists, InSubquery};
    50
    use datafusion_common::tree_node::{
    51
        Transformed, TreeNode, TreeNodeContainer, TreeNodeIterator, TreeNodeRecursion,
    52
        TreeNodeRewriter, TreeNodeVisitor,
    53
    };
    54
    use datafusion_common::{internal_err, Result};
    55
    56
    impl TreeNode for LogicalPlan {
    57
    0
        fn apply_children<'n, F: FnMut(&'n Self) -> Result<TreeNodeRecursion>>(
    58
    0
            &'n self,
    59
    0
            f: F,
    60
    0
        ) -> Result<TreeNodeRecursion> {
    61
    0
            self.inputs().apply_ref_elements(f)
    62
    0
        }
    63
    64
        /// Applies `f` to each child (input) of this plan node, rewriting them *in place.*
    65
        ///
    66
        /// # Notes
    67
        ///
    68
        /// Inputs include ONLY direct children, not embedded `LogicalPlan`s for
    69
        /// subqueries, for example such as are in [`Expr::Exists`].
    70
        ///
    71
        /// [`Expr::Exists`]: crate::Expr::Exists
    72
    0
        fn map_children<F: FnMut(Self) -> Result<Transformed<Self>>>(
    73
    0
            self,
    74
    0
            f: F,
    75
    0
        ) -> Result<Transformed<Self>> {
    76
    0
            Ok(match self {
    77
                LogicalPlan::Projection(Projection {
    78
    0
                    expr,
    79
    0
                    input,
    80
    0
                    schema,
    81
    0
                }) => input.map_elements(f)?.update_data(|input| {
    82
    0
                    LogicalPlan::Projection(Projection {
    83
    0
                        expr,
    84
    0
                        input,
    85
    0
                        schema,
    86
    0
                    })
    87
    0
                }),
    88
    0
                LogicalPlan::Filter(Filter { predicate, input }) => input
    89
    0
                    .map_elements(f)?
    90
    0
                    .update_data(|input| LogicalPlan::Filter(Filter { predicate, input })),
    91
                LogicalPlan::Repartition(Repartition {
    92
    0
                    input,
    93
    0
                    partitioning_scheme,
    94
    0
                }) => input.map_elements(f)?.update_data(|input| {
    95
    0
                    LogicalPlan::Repartition(Repartition {
    96
    0
                        input,
    97
    0
                        partitioning_scheme,
    98
    0
                    })
    99
    0
                }),
    100
                LogicalPlan::Window(Window {
    101
    0
                    input,
    102
    0
                    window_expr,
    103
    0
                    schema,
    104
    0
                }) => input.map_elements(f)?.update_data(|input| {
    105
    0
                    LogicalPlan::Window(Window {
    106
    0
                        input,
    107
    0
                        window_expr,
    108
    0
                        schema,
    109
    0
                    })
    110
    0
                }),
    111
                LogicalPlan::Aggregate(Aggregate {
    112
    0
                    input,
    113
    0
                    group_expr,
    114
    0
                    aggr_expr,
    115
    0
                    schema,
    116
    0
                }) => input.map_elements(f)?.update_data(|input| {
    117
    0
                    LogicalPlan::Aggregate(Aggregate {
    118
    0
                        input,
    119
    0
                        group_expr,
    120
    0
                        aggr_expr,
    121
    0
                        schema,
    122
    0
                    })
    123
    0
                }),
    124
    0
                LogicalPlan::Sort(Sort { expr, input, fetch }) => input
    125
    0
                    .map_elements(f)?
    126
    0
                    .update_data(|input| LogicalPlan::Sort(Sort { expr, input, fetch })),
    127
                LogicalPlan::Join(Join {
    128
    0
                    left,
    129
    0
                    right,
    130
    0
                    on,
    131
    0
                    filter,
    132
    0
                    join_type,
    133
    0
                    join_constraint,
    134
    0
                    schema,
    135
    0
                    null_equality,
    136
    0
                }) => (left, right).map_elements(f)?.update_data(|(left, right)| {
    137
    0
                    LogicalPlan::Join(Join {
    138
    0
                        left,
    139
    0
                        right,
    140
    0
                        on,
    141
    0
                        filter,
    142
    0
                        join_type,
    143
    0
                        join_constraint,
    144
    0
                        schema,
    145
    0
                        null_equality,
    146
    0
                    })
    147
    0
                }),
    148
    0
                LogicalPlan::Limit(Limit { skip, fetch, input }) => input
    149
    0
                    .map_elements(f)?
    150
    0
                    .update_data(|input| LogicalPlan::Limit(Limit { skip, fetch, input })),
    151
                LogicalPlan::Subquery(Subquery {
    152
    0
                    subquery,
    153
    0
                    outer_ref_columns,
    154
    0
                    spans,
    155
    0
                }) => subquery.map_elements(f)?.update_data(|subquery| {
    156
    0
                    LogicalPlan::Subquery(Subquery {
    157
    0
                        subquery,
    158
    0
                        outer_ref_columns,
    159
    0
                        spans,
    160
    0
                    })
    161
    0
                }),
    162
                LogicalPlan::SubqueryAlias(SubqueryAlias {
    163
    0
                    input,
    164
    0
                    alias,
    165
    0
                    schema,
    166
    0
                }) => input.map_elements(f)?.update_data(|input| {
    167
    0
                    LogicalPlan::SubqueryAlias(SubqueryAlias {
    168
    0
                        input,
    169
    0
                        alias,
    170
    0
                        schema,
    171
    0
                    })
    172
    0
                }),
    173
    0
                LogicalPlan::Extension(extension) => rewrite_extension_inputs(extension, f)?
    174
    0
                    .update_data(LogicalPlan::Extension),
    175
    0
                LogicalPlan::Union(Union { inputs, schema }) => inputs
    176
    0
                    .map_elements(f)?
    177
    0
                    .update_data(|inputs| LogicalPlan::Union(Union { inputs, schema })),
    178
    0
                LogicalPlan::Distinct(distinct) => match distinct {
    179
    0
                    Distinct::All(input) => input.map_elements(f)?.update_data(Distinct::All),
    180
                    Distinct::On(DistinctOn {
    181
    0
                        on_expr,
    182
    0
                        select_expr,
    183
    0
                        sort_expr,
    184
    0
                        input,
    185
    0
                        schema,
    186
    0
                    }) => input.map_elements(f)?.update_data(|input| {
    187
    0
                        Distinct::On(DistinctOn {
    188
    0
                            on_expr,
    189
    0
                            select_expr,
    190
    0
                            sort_expr,
    191
    0
                            input,
    192
    0
                            schema,
    193
    0
                        })
    194
    0
                    }),
    195
                }
    196
    0
                .update_data(LogicalPlan::Distinct),
    197
                LogicalPlan::Explain(Explain {
    198
    0
                    verbose,
    199
    0
                    explain_format: format,
    200
    0
                    plan,
    201
    0
                    stringified_plans,
    202
    0
                    schema,
    203
    0
                    logical_optimization_succeeded,
    204
    0
                }) => plan.map_elements(f)?.update_data(|plan| {
    205
    0
                    LogicalPlan::Explain(Explain {
    206
    0
                        verbose,
    207
    0
                        explain_format: format,
    208
    0
                        plan,
    209
    0
                        stringified_plans,
    210
    0
                        schema,
    211
    0
                        logical_optimization_succeeded,
    212
    0
                    })
    213
    0
                }),
    214
                LogicalPlan::Analyze(Analyze {
    215
    0
                    verbose,
    216
    0
                    input,
    217
    0
                    schema,
    218
    0
                }) => input.map_elements(f)?.update_data(|input| {
    219
    0
                    LogicalPlan::Analyze(Analyze {
    220
    0
                        verbose,
    221
    0
                        input,
    222
    0
                        schema,
    223
    0
                    })
    224
    0
                }),
    225
                LogicalPlan::Dml(DmlStatement {
    226
    0
                    table_name,
    227
    0
                    target,
    228
    0
                    op,
    229
    0
                    input,
    230
    0
                    output_schema,
    231
    0
                }) => input.map_elements(f)?.update_data(|input| {
    232
    0
                    LogicalPlan::Dml(DmlStatement {
    233
    0
                        table_name,
    234
    0
                        target,
    235
    0
                        op,
    236
    0
                        input,
    237
    0
                        output_schema,
    238
    0
                    })
    239
    0
                }),
    240
                LogicalPlan::Copy(CopyTo {
    241
    0
                    input,
    242
    0
                    output_url,
    243
    0
                    partition_by,
    244
    0
                    file_type,
    245
    0
                    options,
    246
    0
                    output_schema,
    247
    0
                }) => input.map_elements(f)?.update_data(|input| {
    248
    0
                    LogicalPlan::Copy(CopyTo {
    249
    0
                        input,
    250
    0
                        output_url,
    251
    0
                        partition_by,
    252
    0
                        file_type,
    253
    0
                        options,
    254
    0
                        output_schema,
    255
    0
                    })
    256
    0
                }),
    257
    0
                LogicalPlan::Ddl(ddl) => {
    258
    0
                    match ddl {
    259
                        DdlStatement::CreateMemoryTable(CreateMemoryTable {
    260
    0
                            name,
    261
    0
                            constraints,
    262
    0
                            input,
    263
    0
                            if_not_exists,
    264
    0
                            or_replace,
    265
    0
                            column_defaults,
    266
    0
                            temporary,
    267
    0
                        }) => input.map_elements(f)?.update_data(|input| {
    268
    0
                            DdlStatement::CreateMemoryTable(CreateMemoryTable {
    269
    0
                                name,
    270
    0
                                constraints,
    271
    0
                                input,
    272
    0
                                if_not_exists,
    273
    0
                                or_replace,
    274
    0
                                column_defaults,
    275
    0
                                temporary,
    276
    0
                            })
    277
    0
                        }),
    278
                        DdlStatement::CreateView(CreateView {
    279
    0
                            name,
    280
    0
                            input,
    281
    0
                            or_replace,
    282
    0
                            definition,
    283
    0
                            temporary,
    284
    0
                        }) => input.map_elements(f)?.update_data(|input| {
    285
    0
                            DdlStatement::CreateView(CreateView {
    286
    0
                                name,
    287
    0
                                input,
    288
    0
                                or_replace,
    289
    0
                                definition,
    290
    0
                                temporary,
    291
    0
                            })
    292
    0
                        }),
    293
                        // no inputs in these statements
    294
                        DdlStatement::CreateExternalTable(_)
    295
                        | DdlStatement::CreateCatalogSchema(_)
    296
                        | DdlStatement::CreateCatalog(_)
    297
                        | DdlStatement::CreateIndex(_)
    298
                        | DdlStatement::DropTable(_)
    299
                        | DdlStatement::DropView(_)
    300
                        | DdlStatement::DropCatalogSchema(_)
    301
                        | DdlStatement::CreateFunction(_)
    302
    0
                        | DdlStatement::DropFunction(_) => Transformed::no(ddl),
    303
                    }
    304
    0
                    .update_data(LogicalPlan::Ddl)
    305
                }
    306
                LogicalPlan::Unnest(Unnest {
    307
    0
                    input,
    308
    0
                    exec_columns: input_columns,
    309
    0
                    list_type_columns,
    310
    0
                    struct_type_columns,
    311
    0
                    dependency_indices,
    312
    0
                    schema,
    313
    0
                    options,
    314
    0
                }) => input.map_elements(f)?.update_data(|input| {
    315
    0
                    LogicalPlan::Unnest(Unnest {
    316
    0
                        input,
    317
    0
                        exec_columns: input_columns,
    318
    0
                        list_type_columns,
    319
    0
                        struct_type_columns,
    320
    0
                        dependency_indices,
    321
    0
                        schema,
    322
    0
                        options,
    323
    0
                    })
    324
    0
                }),
    325
                LogicalPlan::RecursiveQuery(RecursiveQuery {
    326
    0
                    name,
    327
    0
                    static_term,
    328
    0
                    recursive_term,
    329
    0
                    is_distinct,
    330
    0
                }) => (static_term, recursive_term).map_elements(f)?.update_data(
    331
    0
                    |(static_term, recursive_term)| {
    332
    0
                        LogicalPlan::RecursiveQuery(RecursiveQuery {
    333
    0
                            name,
    334
    0
                            static_term,
    335
    0
                            recursive_term,
    336
    0
                            is_distinct,
    337
    0
                        })
    338
    0
                    },
    339
                ),
    340
    0
                LogicalPlan::Statement(stmt) => match stmt {
    341
    0
                    Statement::Prepare(p) => p
    342
    0
                        .input
    343
    0
                        .map_elements(f)?
    344
    0
                        .update_data(|input| Statement::Prepare(Prepare { input, ..p })),
    345
    0
                    _ => Transformed::no(stmt),
    346
                }
    347
    0
                .update_data(LogicalPlan::Statement),
    348
                // plans without inputs
    349
                LogicalPlan::TableScan { .. }
    350
                | LogicalPlan::EmptyRelation { .. }
    351
                | LogicalPlan::Values { .. }
    352
    0
                | LogicalPlan::DescribeTable(_) => Transformed::no(self),
    353
            })
    354
    0
        }
    355
    }
    356
    357
    /// Rewrites all inputs for an Extension node "in place"
    358
    /// (it currently has to copy values because there are no APIs for in place modification)
    359
    ///
    360
    /// Should be removed when we have an API for in place modifications of the
    361
    /// extension to avoid these copies
    362
    0
    fn rewrite_extension_inputs<F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>>(
    363
    0
        extension: Extension,
    364
    0
        f: F,
    365
    0
    ) -> Result<Transformed<Extension>> {
    366
    0
        let Extension { node } = extension;
    367
    368
    0
        node.inputs()
    369
    0
            .into_iter()
    370
    0
            .cloned()
    371
    0
            .map_until_stop_and_collect(f)?
    372
    0
            .map_data(|new_inputs| {
    373
    0
                let exprs = node.expressions();
    374
                Ok(Extension {
    375
    0
                    node: node.with_exprs_and_inputs(exprs, new_inputs)?,
    376
                })
    377
    0
            })
    378
    0
    }
    379
    380
    /// This macro is used to determine continuation during combined transforming
    381
    /// traversals.
    382
    macro_rules! handle_transform_recursion {
    383
        ($F_DOWN:expr, $F_CHILD:expr, $F_UP:expr) => {{
    384
            $F_DOWN?
    385
    0
                .transform_children(|n| {
    386
    0
                    n.map_subqueries($F_CHILD)?
    387
    0
                        .transform_sibling(|n| n.map_children($F_CHILD))
    388
    0
                })?
    389
                .transform_parent($F_UP)
    390
        }};
    391
    }
    392
    393
    impl LogicalPlan {
    394
        /// Calls `f` on all expressions in the current `LogicalPlan` node.
    395
        ///
    396
        /// # Notes
    397
        /// * Similar to [`TreeNode::apply`] but for this node's expressions.
    398
        /// * Does not include expressions in input `LogicalPlan` nodes
    399
        /// * Visits only the top level expressions (Does not recurse into each expression)
    400
    0
        pub fn apply_expressions<F: FnMut(&Expr) -> Result<TreeNodeRecursion>>(
    401
    0
            &self,
    402
    0
            mut f: F,
    403
    0
        ) -> Result<TreeNodeRecursion> {
    404
    0
            match self {
    405
    0
                LogicalPlan::Projection(Projection { expr, .. }) => expr.apply_elements(f),
    406
    0
                LogicalPlan::Values(Values { values, .. }) => values.apply_elements(f),
    407
    0
                LogicalPlan::Filter(Filter { predicate, .. }) => f(predicate),
    408
                LogicalPlan::Repartition(Repartition {
    409
    0
                    partitioning_scheme,
    410
                    ..
    411
    0
                }) => match partitioning_scheme {
    412
    0
                    Partitioning::Hash(expr, _) | Partitioning::DistributeBy(expr) => {
    413
    0
                        expr.apply_elements(f)
    414
                    }
    415
    0
                    Partitioning::RoundRobinBatch(_) => Ok(TreeNodeRecursion::Continue),
    416
                },
    417
    0
                LogicalPlan::Window(Window { window_expr, .. }) => {
    418
    0
                    window_expr.apply_elements(f)
    419
                }
    420
                LogicalPlan::Aggregate(Aggregate {
    421
    0
                    group_expr,
    422
    0
                    aggr_expr,
    423
                    ..
    424
    0
                }) => (group_expr, aggr_expr).apply_ref_elements(f),
    425
                // There are two part of expression for join, equijoin(on) and non-equijoin(filter).
    426
                // 1. the first part is `on.len()` equijoin expressions, and the struct of each expr is `left-on = right-on`.
    427
                // 2. the second part is non-equijoin(filter).
    428
    0
                LogicalPlan::Join(Join { on, filter, .. }) => {
    429
    0
                    (on, filter).apply_ref_elements(f)
    430
                }
    431
    0
                LogicalPlan::Sort(Sort { expr, .. }) => expr.apply_elements(f),
    432
    0
                LogicalPlan::Extension(extension) => {
    433
                    // would be nice to avoid this copy -- maybe can
    434
                    // update extension to just observer Exprs
    435
    0
                    extension.node.expressions().apply_elements(f)
    436
                }
    437
    0
                LogicalPlan::TableScan(TableScan { filters, .. }) => {
    438
    0
                    filters.apply_elements(f)
    439
                }
    440
    0
                LogicalPlan::Unnest(unnest) => {
    441
    0
                    let exprs = unnest
    442
    0
                        .exec_columns
    443
    0
                        .iter()
    444
    0
                        .cloned()
    445
    0
                        .map(Expr::Column)
    446
    0
                        .collect::<Vec<_>>();
    447
    0
                    exprs.apply_elements(f)
    448
                }
    449
                LogicalPlan::Distinct(Distinct::On(DistinctOn {
    450
    0
                    on_expr,
    451
    0
                    select_expr,
    452
    0
                    sort_expr,
    453
                    ..
    454
    0
                })) => (on_expr, select_expr, sort_expr).apply_ref_elements(f),
    455
    0
                LogicalPlan::Limit(Limit { skip, fetch, .. }) => {
    456
    0
                    (skip, fetch).apply_ref_elements(f)
    457
                }
    458
    0
                LogicalPlan::Statement(stmt) => match stmt {
    459
    0
                    Statement::Execute(Execute { parameters, .. }) => {
    460
    0
                        parameters.apply_elements(f)
    461
                    }
    462
    0
                    _ => Ok(TreeNodeRecursion::Continue),
    463
                },
    464
                // plans without expressions
    465
                LogicalPlan::EmptyRelation(_)
    466
                | LogicalPlan::RecursiveQuery(_)
    467
                | LogicalPlan::Subquery(_)
    468
                | LogicalPlan::SubqueryAlias(_)
    469
                | LogicalPlan::Analyze(_)
    470
                | LogicalPlan::Explain(_)
    471
                | LogicalPlan::Union(_)
    472
                | LogicalPlan::Distinct(Distinct::All(_))
    473
                | LogicalPlan::Dml(_)
    474
                | LogicalPlan::Ddl(_)
    475
                | LogicalPlan::Copy(_)
    476
    0
                | LogicalPlan::DescribeTable(_) => Ok(TreeNodeRecursion::Continue),
    477
            }
    478
    0
        }
    479
    480
        /// Rewrites all expressions in the current `LogicalPlan` node using `f`.
    481
        ///
    482
        /// Returns the current node.
    483
        ///
    484
        /// # Notes
    485
        /// * Similar to [`TreeNode::map_children`] but for this node's expressions.
    486
        /// * Visits only the top level expressions (Does not recurse into each expression)
    487
    0
        pub fn map_expressions<F: FnMut(Expr) -> Result<Transformed<Expr>>>(
    488
    0
            self,
    489
    0
            mut f: F,
    490
    0
        ) -> Result<Transformed<Self>> {
    491
    0
            Ok(match self {
    492
                LogicalPlan::Projection(Projection {
    493
    0
                    expr,
    494
    0
                    input,
    495
    0
                    schema,
    496
    0
                }) => expr.map_elements(f)?.update_data(|expr| {
    497
    0
                    LogicalPlan::Projection(Projection {
    498
    0
                        expr,
    499
    0
                        input,
    500
    0
                        schema,
    501
    0
                    })
    502
    0
                }),
    503
    0
                LogicalPlan::Values(Values { schema, values }) => values
    504
    0
                    .map_elements(f)?
    505
    0
                    .update_data(|values| LogicalPlan::Values(Values { schema, values })),
    506
    0
                LogicalPlan::Filter(Filter { predicate, input }) => f(predicate)?
    507
    0
                    .update_data(|predicate| {
    508
    0
                        LogicalPlan::Filter(Filter { predicate, input })
    509
    0
                    }),
    510
                LogicalPlan::Repartition(Repartition {
    511
    0
                    input,
    512
    0
                    partitioning_scheme,
    513
    0
                }) => match partitioning_scheme {
    514
    0
                    Partitioning::Hash(expr, usize) => expr
    515
    0
                        .map_elements(f)?
    516
    0
                        .update_data(|expr| Partitioning::Hash(expr, usize)),
    517
    0
                    Partitioning::DistributeBy(expr) => expr
    518
    0
                        .map_elements(f)?
    519
    0
                        .update_data(Partitioning::DistributeBy),
    520
    0
                    Partitioning::RoundRobinBatch(_) => Transformed::no(partitioning_scheme),
    521
                }
    522
    0
                .update_data(|partitioning_scheme| {
    523
    0
                    LogicalPlan::Repartition(Repartition {
    524
    0
                        input,
    525
    0
                        partitioning_scheme,
    526
    0
                    })
    527
    0
                }),
    528
                LogicalPlan::Window(Window {
    529
    0
                    input,
    530
    0
                    window_expr,
    531
    0
                    schema,
    532
    0
                }) => window_expr.map_elements(f)?.update_data(|window_expr| {
    533
    0
                    LogicalPlan::Window(Window {
    534
    0
                        input,
    535
    0
                        window_expr,
    536
    0
                        schema,
    537
    0
                    })
    538
    0
                }),
    539
                LogicalPlan::Aggregate(Aggregate {
    540
    0
                    input,
    541
    0
                    group_expr,
    542
    0
                    aggr_expr,
    543
    0
                    schema,
    544
    0
                }) => (group_expr, aggr_expr).map_elements(f)?.update_data(
    545
    0
                    |(group_expr, aggr_expr)| {
    546
    0
                        LogicalPlan::Aggregate(Aggregate {
    547
    0
                            input,
    548
    0
                            group_expr,
    549
    0
                            aggr_expr,
    550
    0
                            schema,
    551
    0
                        })
    552
    0
                    },
    553
                ),
    554
    555
                // There are two part of expression for join, equijoin(on) and non-equijoin(filter).
    556
                // 1. the first part is `on.len()` equijoin expressions, and the struct of each expr is `left-on = right-on`.
    557
                // 2. the second part is non-equijoin(filter).
    558
                LogicalPlan::Join(Join {
    559
    0
                    left,
    560
    0
                    right,
    561
    0
                    on,
    562
    0
                    filter,
    563
    0
                    join_type,
    564
    0
                    join_constraint,
    565
    0
                    schema,
    566
    0
                    null_equality,
    567
    0
                }) => (on, filter).map_elements(f)?.update_data(|(on, filter)| {
    568
    0
                    LogicalPlan::Join(Join {
    569
    0
                        left,
    570
    0
                        right,
    571
    0
                        on,
    572
    0
                        filter,
    573
    0
                        join_type,
    574
    0
                        join_constraint,
    575
    0
                        schema,
    576
    0
                        null_equality,
    577
    0
                    })
    578
    0
                }),
    579
    0
                LogicalPlan::Sort(Sort { expr, input, fetch }) => expr
    580
    0
                    .map_elements(f)?
    581
    0
                    .update_data(|expr| LogicalPlan::Sort(Sort { expr, input, fetch })),
    582
    0
                LogicalPlan::Extension(Extension { node }) => {
    583
                    // would be nice to avoid this copy -- maybe can
    584
                    // update extension to just observer Exprs
    585
    0
                    let exprs = node.expressions().map_elements(f)?;
    586
    0
                    let plan = LogicalPlan::Extension(Extension {
    587
    0
                        node: UserDefinedLogicalNode::with_exprs_and_inputs(
    588
    0
                            node.as_ref(),
    589
    0
                            exprs.data,
    590
    0
                            node.inputs().into_iter().cloned().collect::<Vec<_>>(),
    591
    0
                        )?,
    592
                    });
    593
    0
                    Transformed::new(plan, exprs.transformed, exprs.tnr)
    594
                }
    595
                LogicalPlan::TableScan(TableScan {
    596
    0
                    table_name,
    597
    0
                    source,
    598
    0
                    projection,
    599
    0
                    projected_schema,
    600
    0
                    filters,
    601
    0
                    fetch,
    602
    0
                }) => filters.map_elements(f)?.update_data(|filters| {
    603
    0
                    LogicalPlan::TableScan(TableScan {
    604
    0
                        table_name,
    605
    0
                        source,
    606
    0
                        projection,
    607
    0
                        projected_schema,
    608
    0
                        filters,
    609
    0
                        fetch,
    610
    0
                    })
    611
    0
                }),
    612
                LogicalPlan::Distinct(Distinct::On(DistinctOn {
    613
    0
                    on_expr,
    614
    0
                    select_expr,
    615
    0
                    sort_expr,
    616
    0
                    input,
    617
    0
                    schema,
    618
    0
                })) => (on_expr, select_expr, sort_expr)
    619
    0
                    .map_elements(f)?
    620
    0
                    .update_data(|(on_expr, select_expr, sort_expr)| {
    621
    0
                        LogicalPlan::Distinct(Distinct::On(DistinctOn {
    622
    0
                            on_expr,
    623
    0
                            select_expr,
    624
    0
                            sort_expr,
    625
    0
                            input,
    626
    0
                            schema,
    627
    0
                        }))
    628
    0
                    }),
    629
    0
                LogicalPlan::Limit(Limit { skip, fetch, input }) => {
    630
    0
                    (skip, fetch).map_elements(f)?.update_data(|(skip, fetch)| {
    631
    0
                        LogicalPlan::Limit(Limit { skip, fetch, input })
    632
    0
                    })
    633
                }
    634
    0
                LogicalPlan::Statement(stmt) => match stmt {
    635
    0
                    Statement::Execute(e) => {
    636
    0
                        e.parameters.map_elements(f)?.update_data(|parameters| {
    637
    0
                            Statement::Execute(Execute { parameters, ..e })
    638
    0
                        })
    639
                    }
    640
    0
                    _ => Transformed::no(stmt),
    641
                }
    642
    0
                .update_data(LogicalPlan::Statement),
    643
                // plans without expressions
    644
                LogicalPlan::EmptyRelation(_)
    645
                | LogicalPlan::Unnest(_)
    646
                | LogicalPlan::RecursiveQuery(_)
    647
                | LogicalPlan::Subquery(_)
    648
                | LogicalPlan::SubqueryAlias(_)
    649
                | LogicalPlan::Analyze(_)
    650
                | LogicalPlan::Explain(_)
    651
                | LogicalPlan::Union(_)
    652
                | LogicalPlan::Distinct(Distinct::All(_))
    653
                | LogicalPlan::Dml(_)
    654
                | LogicalPlan::Ddl(_)
    655
                | LogicalPlan::Copy(_)
    656
    0
                | LogicalPlan::DescribeTable(_) => Transformed::no(self),
    657
            })
    658
    0
        }
    659
    660
        /// Visits a plan similarly to [`Self::visit`], including subqueries that
    661
        /// may appear in expressions such as `IN (SELECT ...)`.
    662
        #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    663
    0
        pub fn visit_with_subqueries<V: for<'n> TreeNodeVisitor<'n, Node = Self>>(
    664
    0
            &self,
    665
    0
            visitor: &mut V,
    666
    0
        ) -> Result<TreeNodeRecursion> {
    667
    0
            visitor
    668
    0
                .f_down(self)?
    669
    0
                .visit_children(|| {
    670
    0
                    self.apply_subqueries(|c| c.visit_with_subqueries(visitor))?
    671
    0
                        .visit_sibling(|| {
    672
    0
                            self.apply_children(|c| c.visit_with_subqueries(visitor))
    673
    0
                        })
    674
    0
                })?
    675
    0
                .visit_parent(|| visitor.f_up(self))
    676
    0
        }
    677
    678
        /// Similarly to [`Self::rewrite`], rewrites this node and its inputs using `f`,
    679
        /// including subqueries that may appear in expressions such as `IN (SELECT
    680
        /// ...)`.
    681
        #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    682
    0
        pub fn rewrite_with_subqueries<R: TreeNodeRewriter<Node = Self>>(
    683
    0
            self,
    684
    0
            rewriter: &mut R,
    685
    0
        ) -> Result<Transformed<Self>> {
    686
    0
            handle_transform_recursion!(
    687
    0
                rewriter.f_down(self),
    688
    0
                |c| c.rewrite_with_subqueries(rewriter),
    689
    0
                |n| rewriter.f_up(n)
    690
            )
    691
    0
        }
    692
    693
        /// Similarly to [`Self::apply`], calls `f` on this node and all its inputs,
    694
        /// including subqueries that may appear in expressions such as `IN (SELECT
    695
        /// ...)`.
    696
    0
        pub fn apply_with_subqueries<F: FnMut(&Self) -> Result<TreeNodeRecursion>>(
    697
    0
            &self,
    698
    0
            mut f: F,
    699
    0
        ) -> Result<TreeNodeRecursion> {
    700
            #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    701
    0
            fn apply_with_subqueries_impl<
    702
    0
                F: FnMut(&LogicalPlan) -> Result<TreeNodeRecursion>,
    703
    0
            >(
    704
    0
                node: &LogicalPlan,
    705
    0
                f: &mut F,
    706
    0
            ) -> Result<TreeNodeRecursion> {
    707
    0
                f(node)?.visit_children(|| {
    708
    0
                    node.apply_subqueries(|c| apply_with_subqueries_impl(c, f))?
    709
    0
                        .visit_sibling(|| {
    710
    0
                            node.apply_children(|c| apply_with_subqueries_impl(c, f))
    711
    0
                        })
    712
    0
                })
    713
    0
            }
    714
    715
    0
            apply_with_subqueries_impl(self, &mut f)
    716
    0
        }
    717
    718
        /// Similarly to [`Self::transform`], rewrites this node and its inputs using `f`,
    719
        /// including subqueries that may appear in expressions such as `IN (SELECT
    720
        /// ...)`.
    721
    0
        pub fn transform_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
    722
    0
            self,
    723
    0
            f: F,
    724
    0
        ) -> Result<Transformed<Self>> {
    725
    0
            self.transform_up_with_subqueries(f)
    726
    0
        }
    727
    728
        /// Similarly to [`Self::transform_down`], rewrites this node and its inputs using `f`,
    729
        /// including subqueries that may appear in expressions such as `IN (SELECT
    730
        /// ...)`.
    731
    0
        pub fn transform_down_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
    732
    0
            self,
    733
    0
            mut f: F,
    734
    0
        ) -> Result<Transformed<Self>> {
    735
            #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    736
    0
            fn transform_down_with_subqueries_impl<
    737
    0
                F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
    738
    0
            >(
    739
    0
                node: LogicalPlan,
    740
    0
                f: &mut F,
    741
    0
            ) -> Result<Transformed<LogicalPlan>> {
    742
    0
                f(node)?.transform_children(|n| {
    743
    0
                    n.map_subqueries(|c| transform_down_with_subqueries_impl(c, f))?
    744
    0
                        .transform_sibling(|n| {
    745
    0
                            n.map_children(|c| transform_down_with_subqueries_impl(c, f))
    746
    0
                        })
    747
    0
                })
    748
    0
            }
    749
    750
    0
            transform_down_with_subqueries_impl(self, &mut f)
    751
    0
        }
    752
    753
        /// Similarly to [`Self::transform_up`], rewrites this node and its inputs using `f`,
    754
        /// including subqueries that may appear in expressions such as `IN (SELECT
    755
        /// ...)`.
    756
    0
        pub fn transform_up_with_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
    757
    0
            self,
    758
    0
            mut f: F,
    759
    0
        ) -> Result<Transformed<Self>> {
    760
            #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    761
    0
            fn transform_up_with_subqueries_impl<
    762
    0
                F: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
    763
    0
            >(
    764
    0
                node: LogicalPlan,
    765
    0
                f: &mut F,
    766
    0
            ) -> Result<Transformed<LogicalPlan>> {
    767
    0
                node.map_subqueries(|c| transform_up_with_subqueries_impl(c, f))?
    768
    0
                    .transform_sibling(|n| {
    769
    0
                        n.map_children(|c| transform_up_with_subqueries_impl(c, f))
    770
    0
                    })?
    771
    0
                    .transform_parent(f)
    772
    0
            }
    773
    774
    0
            transform_up_with_subqueries_impl(self, &mut f)
    775
    0
        }
    776
    777
        /// Similarly to [`Self::transform_down`], rewrites this node and its inputs using `f`,
    778
        /// including subqueries that may appear in expressions such as `IN (SELECT
    779
        /// ...)`.
    780
    0
        pub fn transform_down_up_with_subqueries<
    781
    0
            FD: FnMut(Self) -> Result<Transformed<Self>>,
    782
    0
            FU: FnMut(Self) -> Result<Transformed<Self>>,
    783
    0
        >(
    784
    0
            self,
    785
    0
            mut f_down: FD,
    786
    0
            mut f_up: FU,
    787
    0
        ) -> Result<Transformed<Self>> {
    788
            #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    789
    0
            fn transform_down_up_with_subqueries_impl<
    790
    0
                FD: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
    791
    0
                FU: FnMut(LogicalPlan) -> Result<Transformed<LogicalPlan>>,
    792
    0
            >(
    793
    0
                node: LogicalPlan,
    794
    0
                f_down: &mut FD,
    795
    0
                f_up: &mut FU,
    796
    0
            ) -> Result<Transformed<LogicalPlan>> {
    797
    0
                handle_transform_recursion!(
    798
    0
                    f_down(node),
    799
    0
                    |c| transform_down_up_with_subqueries_impl(c, f_down, f_up),
    800
    0
                    f_up
    801
                )
    802
    0
            }
    803
    804
    0
            transform_down_up_with_subqueries_impl(self, &mut f_down, &mut f_up)
    805
    0
        }
    806
    807
        /// Similarly to [`Self::apply`], calls `f` on  this node and its inputs
    808
        /// including subqueries that may appear in expressions such as `IN (SELECT
    809
        /// ...)`.
    810
    0
        pub fn apply_subqueries<F: FnMut(&Self) -> Result<TreeNodeRecursion>>(
    811
    0
            &self,
    812
    0
            mut f: F,
    813
    0
        ) -> Result<TreeNodeRecursion> {
    814
    0
            self.apply_expressions(|expr| {
    815
    0
                expr.apply(|expr| match expr {
    816
    0
                    Expr::Exists(Exists { subquery, .. })
    817
    0
                    | Expr::InSubquery(InSubquery { subquery, .. })
    818
    0
                    | Expr::ScalarSubquery(subquery) => {
    819
                        // use a synthetic plan so the collector sees a
    820
                        // LogicalPlan::Subquery (even though it is
    821
                        // actually a Subquery alias)
    822
    0
                        f(&LogicalPlan::Subquery(subquery.clone()))
    823
                    }
    824
    0
                    _ => Ok(TreeNodeRecursion::Continue),
    825
    0
                })
    826
    0
            })
    827
    0
        }
    828
    829
        /// Similarly to [`Self::map_children`], rewrites all subqueries that may
    830
        /// appear in expressions such as `IN (SELECT ...)` using `f`.
    831
        ///
    832
        /// Returns the current node.
    833
    0
        pub fn map_subqueries<F: FnMut(Self) -> Result<Transformed<Self>>>(
    834
    0
            self,
    835
    0
            mut f: F,
    836
    0
        ) -> Result<Transformed<Self>> {
    837
    0
            self.map_expressions(|expr| {
    838
    0
                expr.transform_down(|expr| match expr {
    839
    0
                    Expr::Exists(Exists { subquery, negated }) => {
    840
    0
                        f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
    841
    0
                            LogicalPlan::Subquery(subquery) => {
    842
    0
                                Ok(Expr::Exists(Exists { subquery, negated }))
    843
                            }
    844
    0
                            _ => internal_err!("Transformation should return Subquery"),
    845
    0
                        })
    846
                    }
    847
                    Expr::InSubquery(InSubquery {
    848
    0
                        expr,
    849
    0
                        subquery,
    850
    0
                        negated,
    851
    0
                    }) => f(LogicalPlan::Subquery(subquery))?.map_data(|s| match s {
    852
    0
                        LogicalPlan::Subquery(subquery) => Ok(Expr::InSubquery(InSubquery {
    853
    0
                            expr,
    854
    0
                            subquery,
    855
    0
                            negated,
    856
    0
                        })),
    857
    0
                        _ => internal_err!("Transformation should return Subquery"),
    858
    0
                    }),
    859
    0
                    Expr::ScalarSubquery(subquery) => f(LogicalPlan::Subquery(subquery))?
    860
    0
                        .map_data(|s| match s {
    861
    0
                            LogicalPlan::Subquery(subquery) => {
    862
    0
                                Ok(Expr::ScalarSubquery(subquery))
    863
                            }
    864
    0
                            _ => internal_err!("Transformation should return Subquery"),
    865
    0
                        }),
    866
    0
                    _ => Ok(Transformed::no(expr)),
    867
    0
                })
    868
    0
            })
    869
    0
        }
    870
    }