## 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/plan.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
    //! Logical plan types
    19
    20
    use std::cmp::Ordering;
    21
    use std::collections::{HashMap, HashSet};
    22
    use std::fmt::{self, Debug, Display, Formatter};
    23
    use std::hash::{Hash, Hasher};
    24
    use std::sync::{Arc, LazyLock};
    25
    26
    use super::dml::CopyTo;
    27
    use super::invariants::{
    28
        assert_always_invariants_at_current_node, assert_executable_invariants,
    29
        InvariantLevel,
    30
    };
    31
    use super::DdlStatement;
    32
    use crate::builder::{unique_field_aliases, unnest_with_options};
    33
    use crate::expr::{
    34
        intersect_metadata_for_union, Alias, Placeholder, Sort as SortExpr, WindowFunction,
    35
        WindowFunctionParams,
    36
    };
    37
    use crate::expr_rewriter::{
    38
        create_col_from_scalar_expr, normalize_cols, normalize_sorts, NamePreserver,
    39
    };
    40
    use crate::logical_plan::display::{GraphvizVisitor, IndentVisitor};
    41
    use crate::logical_plan::extension::UserDefinedLogicalNode;
    42
    use crate::logical_plan::{DmlStatement, Statement};
    43
    use crate::utils::{
    44
        enumerate_grouping_sets, exprlist_to_fields, find_out_reference_exprs,
    45
        grouping_set_expr_count, grouping_set_to_exprlist, split_conjunction,
    46
    };
    47
    use crate::{
    48
        build_join_schema, expr_vec_fmt, requalify_sides_if_needed, BinaryExpr,
    49
        CreateMemoryTable, CreateView, Execute, Expr, ExprSchemable, LogicalPlanBuilder,
    50
        Operator, Prepare, TableProviderFilterPushDown, TableSource,
    51
        WindowFunctionDefinition,
    52
    };
    53
    54
    use arrow::datatypes::{DataType, Field, FieldRef, Schema, SchemaRef};
    55
    use datafusion_common::cse::{NormalizeEq, Normalizeable};
    56
    use datafusion_common::format::ExplainFormat;
    57
    use datafusion_common::metadata::check_metadata_with_storage_equal;
    58
    use datafusion_common::tree_node::{
    59
        Transformed, TreeNode, TreeNodeContainer, TreeNodeRecursion,
    60
    };
    61
    use datafusion_common::{
    62
        aggregate_functional_dependencies, assert_eq_or_internal_err, assert_or_internal_err,
    63
        internal_err, plan_err, Column, Constraints, DFSchema, DFSchemaRef, DataFusionError,
    64
        Dependency, FunctionalDependence, FunctionalDependencies, NullEquality, ParamValues,
    65
        Result, ScalarValue, Spans, TableReference, UnnestOptions,
    66
    };
    67
    use indexmap::IndexSet;
    68
    69
    // backwards compatibility
    70
    use crate::display::PgJsonVisitor;
    71
    pub use datafusion_common::display::{PlanType, StringifiedPlan, ToStringifiedPlan};
    72
    pub use datafusion_common::{JoinConstraint, JoinType};
    73
    74
    /// A `LogicalPlan` is a node in a tree of relational operators (such as
    75
    /// Projection or Filter).
    76
    ///
    77
    /// Represents transforming an input relation (table) to an output relation
    78
    /// (table) with a potentially different schema. Plans form a dataflow tree
    79
    /// where data flows from leaves up to the root to produce the query result.
    80
    ///
    81
    /// `LogicalPlan`s can be created by the SQL query planner, the DataFrame API,
    82
    /// or programmatically (for example custom query languages).
    83
    ///
    84
    /// # See also:
    85
    /// * [`Expr`]: For the expressions that are evaluated by the plan
    86
    /// * [`LogicalPlanBuilder`]: For building `LogicalPlan`s
    87
    /// * [`tree_node`]: To inspect and rewrite `LogicalPlan`s
    88
    ///
    89
    /// [`tree_node`]: crate::logical_plan::tree_node
    90
    ///
    91
    /// # Examples
    92
    ///
    93
    /// ## Creating a LogicalPlan from SQL:
    94
    ///
    95
    /// See [`SessionContext::sql`](https://docs.rs/datafusion/latest/datafusion/execution/context/struct.SessionContext.html#method.sql)
    96
    ///
    97
    /// ## Creating a LogicalPlan from the DataFrame API:
    98
    ///
    99
    /// See [`DataFrame::logical_plan`](https://docs.rs/datafusion/latest/datafusion/dataframe/struct.DataFrame.html#method.logical_plan)
    100
    ///
    101
    /// ## Creating a LogicalPlan programmatically:
    102
    ///
    103
    /// See [`LogicalPlanBuilder`]
    104
    ///
    105
    /// # Visiting and Rewriting `LogicalPlan`s
    106
    ///
    107
    /// Using the [`tree_node`] API, you can recursively walk all nodes in a
    108
    /// `LogicalPlan`. For example, to find all column references in a plan:
    109
    ///
    110
    /// ```
    111
    /// # use std::collections::HashSet;
    112
    /// # use arrow::datatypes::{DataType, Field, Schema};
    113
    /// # use datafusion_expr::{Expr, col, lit, LogicalPlan, LogicalPlanBuilder, table_scan};
    114
    /// # use datafusion_common::tree_node::{TreeNodeRecursion, TreeNode};
    115
    /// # use datafusion_common::{Column, Result};
    116
    /// # fn employee_schema() -> Schema {
    117
    /// #    Schema::new(vec![
    118
    /// #           Field::new("name", DataType::Utf8, false),
    119
    /// #           Field::new("salary", DataType::Int32, false),
    120
    /// #       ])
    121
    /// #   }
    122
    /// // Projection(name, salary)
    123
    /// //   Filter(salary > 1000)
    124
    /// //     TableScan(employee)
    125
    /// # fn main() -> Result<()> {
    126
    /// let plan = table_scan(Some("employee"), &employee_schema(), None)?
    127
    ///  .filter(col("salary").gt(lit(1000)))?
    128
    ///  .project(vec![col("name")])?
    129
    ///  .build()?;
    130
    ///
    131
    /// // use apply to walk the plan and collect all expressions
    132
    /// let mut expressions = HashSet::new();
    133
    /// plan.apply(|node| {
    134
    ///   // collect all expressions in the plan
    135
    ///   node.apply_expressions(|expr| {
    136
    ///    expressions.insert(expr.clone());
    137
    ///    Ok(TreeNodeRecursion::Continue) // control walk of expressions
    138
    ///   })?;
    139
    ///   Ok(TreeNodeRecursion::Continue) // control walk of plan nodes
    140
    /// }).unwrap();
    141
    ///
    142
    /// // we found the expression in projection and filter
    143
    /// assert_eq!(expressions.len(), 2);
    144
    /// println!("Found expressions: {:?}", expressions);
    145
    /// // found predicate in the Filter: employee.salary > 1000
    146
    /// let salary = Expr::Column(Column::new(Some("employee"), "salary"));
    147
    /// assert!(expressions.contains(&salary.gt(lit(1000))));
    148
    /// // found projection in the Projection: employee.name
    149
    /// let name = Expr::Column(Column::new(Some("employee"), "name"));
    150
    /// assert!(expressions.contains(&name));
    151
    /// # Ok(())
    152
    /// # }
    153
    /// ```
    154
    ///
    155
    /// You can also rewrite plans using the [`tree_node`] API. For example, to
    156
    /// replace the filter predicate in a plan:
    157
    ///
    158
    /// ```
    159
    /// # use std::collections::HashSet;
    160
    /// # use arrow::datatypes::{DataType, Field, Schema};
    161
    /// # use datafusion_expr::{Expr, col, lit, LogicalPlan, LogicalPlanBuilder, table_scan};
    162
    /// # use datafusion_common::tree_node::{TreeNodeRecursion, TreeNode};
    163
    /// # use datafusion_common::{Column, Result};
    164
    /// # fn employee_schema() -> Schema {
    165
    /// #    Schema::new(vec![
    166
    /// #           Field::new("name", DataType::Utf8, false),
    167
    /// #           Field::new("salary", DataType::Int32, false),
    168
    /// #       ])
    169
    /// #   }
    170
    /// // Projection(name, salary)
    171
    /// //   Filter(salary > 1000)
    172
    /// //     TableScan(employee)
    173
    /// # fn main() -> Result<()> {
    174
    /// use datafusion_common::tree_node::Transformed;
    175
    /// let plan = table_scan(Some("employee"), &employee_schema(), None)?
    176
    ///  .filter(col("salary").gt(lit(1000)))?
    177
    ///  .project(vec![col("name")])?
    178
    ///  .build()?;
    179
    ///
    180
    /// // use transform to rewrite the plan
    181
    /// let transformed_result = plan.transform(|node| {
    182
    ///   // when we see the filter node
    183
    ///   if let LogicalPlan::Filter(mut filter) = node {
    184
    ///     // replace predicate with salary < 2000
    185
    ///     filter.predicate = Expr::Column(Column::new(Some("employee"), "salary")).lt(lit(2000));
    186
    ///     let new_plan = LogicalPlan::Filter(filter);
    187
    ///     return Ok(Transformed::yes(new_plan)); // communicate the node was changed
    188
    ///   }
    189
    ///   // return the node unchanged
    190
    ///   Ok(Transformed::no(node))
    191
    /// }).unwrap();
    192
    ///
    193
    /// // Transformed result contains rewritten plan and information about
    194
    /// // whether the plan was changed
    195
    /// assert!(transformed_result.transformed);
    196
    /// let rewritten_plan = transformed_result.data;
    197
    ///
    198
    /// // we found the filter
    199
    /// assert_eq!(rewritten_plan.display_indent().to_string(),
    200
    /// "Projection: employee.name\
    201
    /// \n  Filter: employee.salary < Int32(2000)\
    202
    /// \n    TableScan: employee");
    203
    /// # Ok(())
    204
    /// # }
    205
    /// ```
    206
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    207
    pub enum LogicalPlan {
    208
        /// Evaluates an arbitrary list of expressions (essentially a
    209
        /// SELECT with an expression list) on its input.
    210
        Projection(Projection),
    211
        /// Filters rows from its input that do not match an
    212
        /// expression (essentially a WHERE clause with a predicate
    213
        /// expression).
    214
        ///
    215
        /// Semantically, `<predicate>` is evaluated for each row of the
    216
        /// input; If the value of `<predicate>` is true, the input row is
    217
        /// passed to the output. If the value of `<predicate>` is false
    218
        /// (or null), the row is discarded.
    219
        Filter(Filter),
    220
        /// Windows input based on a set of window spec and window
    221
        /// function (e.g. SUM or RANK).  This is used to implement SQL
    222
        /// window functions, and the `OVER` clause.
    223
        ///
    224
        /// See [`Window`] for more details
    225
        Window(Window),
    226
        /// Aggregates its input based on a set of grouping and aggregate
    227
        /// expressions (e.g. SUM). This is used to implement SQL aggregates
    228
        /// and `GROUP BY`.
    229
        ///
    230
        /// See [`Aggregate`] for more details
    231
        Aggregate(Aggregate),
    232
        /// Sorts its input according to a list of sort expressions. This
    233
        /// is used to implement SQL `ORDER BY`
    234
        Sort(Sort),
    235
        /// Join two logical plans on one or more join columns.
    236
        /// This is used to implement SQL `JOIN`
    237
        Join(Join),
    238
        /// Repartitions the input based on a partitioning scheme. This is
    239
        /// used to add parallelism and is sometimes referred to as an
    240
        /// "exchange" operator in other systems
    241
        Repartition(Repartition),
    242
        /// Union multiple inputs with the same schema into a single
    243
        /// output stream. This is used to implement SQL `UNION [ALL]` and
    244
        /// `INTERSECT [ALL]`.
    245
        Union(Union),
    246
        /// Produces rows from a [`TableSource`], used to implement SQL
    247
        /// `FROM` tables or views.
    248
        TableScan(TableScan),
    249
        /// Produces no rows: An empty relation with an empty schema that
    250
        /// produces 0 or 1 row. This is used to implement SQL `SELECT`
    251
        /// that has no values in the `FROM` clause.
    252
        EmptyRelation(EmptyRelation),
    253
        /// Produces the output of running another query.  This is used to
    254
        /// implement SQL subqueries
    255
        Subquery(Subquery),
    256
        /// Aliased relation provides, or changes, the name of a relation.
    257
        SubqueryAlias(SubqueryAlias),
    258
        /// Skip some number of rows, and then fetch some number of rows.
    259
        Limit(Limit),
    260
        /// A DataFusion [`Statement`] such as `SET VARIABLE` or `START TRANSACTION`
    261
        Statement(Statement),
    262
        /// Values expression. See
    263
        /// [Postgres VALUES](https://www.postgresql.org/docs/current/queries-values.html)
    264
        /// documentation for more details. This is used to implement SQL such as
    265
        /// `VALUES (1, 2), (3, 4)`
    266
        Values(Values),
    267
        /// Produces a relation with string representations of
    268
        /// various parts of the plan. This is used to implement SQL `EXPLAIN`.
    269
        Explain(Explain),
    270
        /// Runs the input, and prints annotated physical plan as a string
    271
        /// with execution metric. This is used to implement SQL
    272
        /// `EXPLAIN ANALYZE`.
    273
        Analyze(Analyze),
    274
        /// Extension operator defined outside of DataFusion. This is used
    275
        /// to extend DataFusion with custom relational operations that
    276
        Extension(Extension),
    277
        /// Remove duplicate rows from the input. This is used to
    278
        /// implement SQL `SELECT DISTINCT ...`.
    279
        Distinct(Distinct),
    280
        /// Data Manipulation Language (DML): Insert / Update / Delete
    281
        Dml(DmlStatement),
    282
        /// Data Definition Language (DDL): CREATE / DROP TABLES / VIEWS / SCHEMAS
    283
        Ddl(DdlStatement),
    284
        /// `COPY TO` for writing plan results to files
    285
        Copy(CopyTo),
    286
        /// Describe the schema of the table. This is used to implement the
    287
        /// SQL `DESCRIBE` command from MySQL.
    288
        DescribeTable(DescribeTable),
    289
        /// Unnest a column that contains a nested list type such as an
    290
        /// ARRAY. This is used to implement SQL `UNNEST`
    291
        Unnest(Unnest),
    292
        /// A variadic query (e.g. "Recursive CTEs")
    293
        RecursiveQuery(RecursiveQuery),
    294
    }
    295
    296
    impl Default for LogicalPlan {
    297
    0
        fn default() -> Self {
    298
    0
            LogicalPlan::EmptyRelation(EmptyRelation {
    299
    0
                produce_one_row: false,
    300
    0
                schema: Arc::new(DFSchema::empty()),
    301
    0
            })
    302
    0
        }
    303
    }
    304
    305
    impl<'a> TreeNodeContainer<'a, Self> for LogicalPlan {
    306
    0
        fn apply_elements<F: FnMut(&'a Self) -> Result<TreeNodeRecursion>>(
    307
    0
            &'a self,
    308
    0
            mut f: F,
    309
    0
        ) -> Result<TreeNodeRecursion> {
    310
    0
            f(self)
    311
    0
        }
    312
    313
    0
        fn map_elements<F: FnMut(Self) -> Result<Transformed<Self>>>(
    314
    0
            self,
    315
    0
            mut f: F,
    316
    0
        ) -> Result<Transformed<Self>> {
    317
    0
            f(self)
    318
    0
        }
    319
    }
    320
    321
    impl LogicalPlan {
    322
        /// Get a reference to the logical plan's schema
    323
    0
        pub fn schema(&self) -> &DFSchemaRef {
    324
    0
            match self {
    325
    0
                LogicalPlan::EmptyRelation(EmptyRelation { schema, .. }) => schema,
    326
    0
                LogicalPlan::Values(Values { schema, .. }) => schema,
    327
                LogicalPlan::TableScan(TableScan {
    328
    0
                    projected_schema, ..
    329
    0
                }) => projected_schema,
    330
    0
                LogicalPlan::Projection(Projection { schema, .. }) => schema,
    331
    0
                LogicalPlan::Filter(Filter { input, .. }) => input.schema(),
    332
    0
                LogicalPlan::Distinct(Distinct::All(input)) => input.schema(),
    333
    0
                LogicalPlan::Distinct(Distinct::On(DistinctOn { schema, .. })) => schema,
    334
    0
                LogicalPlan::Window(Window { schema, .. }) => schema,
    335
    0
                LogicalPlan::Aggregate(Aggregate { schema, .. }) => schema,
    336
    0
                LogicalPlan::Sort(Sort { input, .. }) => input.schema(),
    337
    0
                LogicalPlan::Join(Join { schema, .. }) => schema,
    338
    0
                LogicalPlan::Repartition(Repartition { input, .. }) => input.schema(),
    339
    0
                LogicalPlan::Limit(Limit { input, .. }) => input.schema(),
    340
    0
                LogicalPlan::Statement(statement) => statement.schema(),
    341
    0
                LogicalPlan::Subquery(Subquery { subquery, .. }) => subquery.schema(),
    342
    0
                LogicalPlan::SubqueryAlias(SubqueryAlias { schema, .. }) => schema,
    343
    0
                LogicalPlan::Explain(explain) => &explain.schema,
    344
    0
                LogicalPlan::Analyze(analyze) => &analyze.schema,
    345
    0
                LogicalPlan::Extension(extension) => extension.node.schema(),
    346
    0
                LogicalPlan::Union(Union { schema, .. }) => schema,
    347
    0
                LogicalPlan::DescribeTable(DescribeTable { output_schema, .. }) => {
    348
    0
                    output_schema
    349
                }
    350
    0
                LogicalPlan::Dml(DmlStatement { output_schema, .. }) => output_schema,
    351
    0
                LogicalPlan::Copy(CopyTo { output_schema, .. }) => output_schema,
    352
    0
                LogicalPlan::Ddl(ddl) => ddl.schema(),
    353
    0
                LogicalPlan::Unnest(Unnest { schema, .. }) => schema,
    354
    0
                LogicalPlan::RecursiveQuery(RecursiveQuery { static_term, .. }) => {
    355
                    // we take the schema of the static term as the schema of the entire recursive query
    356
    0
                    static_term.schema()
    357
                }
    358
            }
    359
    0
        }
    360
    361
        /// Used for normalizing columns, as the fallback schemas to the main schema
    362
        /// of the plan.
    363
    0
        pub fn fallback_normalize_schemas(&self) -> Vec<&DFSchema> {
    364
    0
            match self {
    365
                LogicalPlan::Window(_)
    366
                | LogicalPlan::Projection(_)
    367
                | LogicalPlan::Aggregate(_)
    368
                | LogicalPlan::Unnest(_)
    369
    0
                | LogicalPlan::Join(_) => self
    370
    0
                    .inputs()
    371
    0
                    .iter()
    372
    0
                    .map(|input| input.schema().as_ref())
    373
    0
                    .collect(),
    374
    0
                _ => vec![],
    375
            }
    376
    0
        }
    377
    378
        /// Returns the (fixed) output schema for explain plans
    379
    0
        pub fn explain_schema() -> SchemaRef {
    380
    0
            SchemaRef::new(Schema::new(vec![
    381
    0
                Field::new("plan_type", DataType::Utf8, false),
    382
    0
                Field::new("plan", DataType::Utf8, false),
    383
            ]))
    384
    0
        }
    385
    386
        /// Returns the (fixed) output schema for `DESCRIBE` plans
    387
    0
        pub fn describe_schema() -> Schema {
    388
    0
            Schema::new(vec![
    389
    0
                Field::new("column_name", DataType::Utf8, false),
    390
    0
                Field::new("data_type", DataType::Utf8, false),
    391
    0
                Field::new("is_nullable", DataType::Utf8, false),
    392
            ])
    393
    0
        }
    394
    395
        /// Returns all expressions (non-recursively) evaluated by the current
    396
        /// logical plan node. This does not include expressions in any children.
    397
        ///
    398
        /// Note this method `clone`s all the expressions. When possible, the
    399
        /// [`tree_node`] API should be used instead of this API.
    400
        ///
    401
        /// The returned expressions do not necessarily represent or even
    402
        /// contributed to the output schema of this node. For example,
    403
        /// `LogicalPlan::Filter` returns the filter expression even though the
    404
        /// output of a Filter has the same columns as the input.
    405
        ///
    406
        /// The expressions do contain all the columns that are used by this plan,
    407
        /// so if there are columns not referenced by these expressions then
    408
        /// DataFusion's optimizer attempts to optimize them away.
    409
        ///
    410
        /// [`tree_node`]: crate::logical_plan::tree_node
    411
    0
        pub fn expressions(self: &LogicalPlan) -> Vec<Expr> {
    412
    0
            let mut exprs = vec![];
    413
    0
            self.apply_expressions(|e| {
    414
    0
                exprs.push(e.clone());
    415
    0
                Ok(TreeNodeRecursion::Continue)
    416
    0
            })
    417
            // closure always returns OK
    418
    0
            .unwrap();
    419
    0
            exprs
    420
    0
        }
    421
    422
        /// Returns all the out reference(correlated) expressions (recursively) in the current
    423
        /// logical plan nodes and all its descendant nodes.
    424
    0
        pub fn all_out_ref_exprs(self: &LogicalPlan) -> Vec<Expr> {
    425
    0
            let mut exprs = vec![];
    426
    0
            self.apply_expressions(|e| {
    427
    0
                find_out_reference_exprs(e).into_iter().for_each(|e| {
    428
    0
                    if !exprs.contains(&e) {
    429
    0
                        exprs.push(e)
    430
    0
                    }
    431
    0
                });
    432
    0
                Ok(TreeNodeRecursion::Continue)
    433
    0
            })
    434
            // closure always returns OK
    435
    0
            .unwrap();
    436
    0
            self.inputs()
    437
    0
                .into_iter()
    438
    0
                .flat_map(|child| child.all_out_ref_exprs())
    439
    0
                .for_each(|e| {
    440
    0
                    if !exprs.contains(&e) {
    441
    0
                        exprs.push(e)
    442
    0
                    }
    443
    0
                });
    444
    0
            exprs
    445
    0
        }
    446
    447
        /// Returns all inputs / children of this `LogicalPlan` node.
    448
        ///
    449
        /// Note does not include inputs to inputs, or subqueries.
    450
    0
        pub fn inputs(&self) -> Vec<&LogicalPlan> {
    451
    0
            match self {
    452
    0
                LogicalPlan::Projection(Projection { input, .. }) => vec![input],
    453
    0
                LogicalPlan::Filter(Filter { input, .. }) => vec![input],
    454
    0
                LogicalPlan::Repartition(Repartition { input, .. }) => vec![input],
    455
    0
                LogicalPlan::Window(Window { input, .. }) => vec![input],
    456
    0
                LogicalPlan::Aggregate(Aggregate { input, .. }) => vec![input],
    457
    0
                LogicalPlan::Sort(Sort { input, .. }) => vec![input],
    458
    0
                LogicalPlan::Join(Join { left, right, .. }) => vec![left, right],
    459
    0
                LogicalPlan::Limit(Limit { input, .. }) => vec![input],
    460
    0
                LogicalPlan::Subquery(Subquery { subquery, .. }) => vec![subquery],
    461
    0
                LogicalPlan::SubqueryAlias(SubqueryAlias { input, .. }) => vec![input],
    462
    0
                LogicalPlan::Extension(extension) => extension.node.inputs(),
    463
    0
                LogicalPlan::Union(Union { inputs, .. }) => {
    464
    0
                    inputs.iter().map(|arc| arc.as_ref()).collect()
    465
                }
    466
                LogicalPlan::Distinct(
    467
    0
                    Distinct::All(input) | Distinct::On(DistinctOn { input, .. }),
    468
    0
                ) => vec![input],
    469
    0
                LogicalPlan::Explain(explain) => vec![&explain.plan],
    470
    0
                LogicalPlan::Analyze(analyze) => vec![&analyze.input],
    471
    0
                LogicalPlan::Dml(write) => vec![&write.input],
    472
    0
                LogicalPlan::Copy(copy) => vec![&copy.input],
    473
    0
                LogicalPlan::Ddl(ddl) => ddl.inputs(),
    474
    0
                LogicalPlan::Unnest(Unnest { input, .. }) => vec![input],
    475
                LogicalPlan::RecursiveQuery(RecursiveQuery {
    476
    0
                    static_term,
    477
    0
                    recursive_term,
    478
                    ..
    479
    0
                }) => vec![static_term, recursive_term],
    480
    0
                LogicalPlan::Statement(stmt) => stmt.inputs(),
    481
                // plans without inputs
    482
                LogicalPlan::TableScan { .. }
    483
                | LogicalPlan::EmptyRelation { .. }
    484
                | LogicalPlan::Values { .. }
    485
    0
                | LogicalPlan::DescribeTable(_) => vec![],
    486
            }
    487
    0
        }
    488
    489
        /// returns all `Using` join columns in a logical plan
    490
    0
        pub fn using_columns(&self) -> Result<Vec<HashSet<Column>>, DataFusionError> {
    491
    0
            let mut using_columns: Vec<HashSet<Column>> = vec![];
    492
    493
    0
            self.apply_with_subqueries(|plan| {
    494
                if let LogicalPlan::Join(Join {
    495
                    join_constraint: JoinConstraint::Using,
    496
    0
                    on,
    497
                    ..
    498
    0
                }) = plan
    499
                {
    500
                    // The join keys in using-join must be columns.
    501
    0
                    let columns =
    502
    0
                        on.iter().try_fold(HashSet::new(), |mut accumu, (l, r)| {
    503
    0
                            let Some(l) = l.get_as_join_column() else {
    504
    0
                                return internal_err!(
    505
    0
                                    "Invalid join key. Expected column, found {l:?}"
    506
                                );
    507
                            };
    508
    0
                            let Some(r) = r.get_as_join_column() else {
    509
    0
                                return internal_err!(
    510
    0
                                    "Invalid join key. Expected column, found {r:?}"
    511
                                );
    512
                            };
    513
    0
                            accumu.insert(l.to_owned());
    514
    0
                            accumu.insert(r.to_owned());
    515
    0
                            Result::<_, DataFusionError>::Ok(accumu)
    516
    0
                        })?;
    517
    0
                    using_columns.push(columns);
    518
    0
                }
    519
    0
                Ok(TreeNodeRecursion::Continue)
    520
    0
            })?;
    521
    522
    0
            Ok(using_columns)
    523
    0
        }
    524
    525
        /// returns the first output expression of this `LogicalPlan` node.
    526
    0
        pub fn head_output_expr(&self) -> Result<Option<Expr>> {
    527
    0
            match self {
    528
    0
                LogicalPlan::Projection(projection) => {
    529
    0
                    Ok(Some(projection.expr.as_slice()[0].clone()))
    530
                }
    531
    0
                LogicalPlan::Aggregate(agg) => {
    532
    0
                    if agg.group_expr.is_empty() {
    533
    0
                        Ok(Some(agg.aggr_expr.as_slice()[0].clone()))
    534
                    } else {
    535
    0
                        Ok(Some(agg.group_expr.as_slice()[0].clone()))
    536
                    }
    537
                }
    538
    0
                LogicalPlan::Distinct(Distinct::On(DistinctOn { select_expr, .. })) => {
    539
    0
                    Ok(Some(select_expr[0].clone()))
    540
                }
    541
    0
                LogicalPlan::Filter(Filter { input, .. })
    542
    0
                | LogicalPlan::Distinct(Distinct::All(input))
    543
    0
                | LogicalPlan::Sort(Sort { input, .. })
    544
    0
                | LogicalPlan::Limit(Limit { input, .. })
    545
    0
                | LogicalPlan::Repartition(Repartition { input, .. })
    546
    0
                | LogicalPlan::Window(Window { input, .. }) => input.head_output_expr(),
    547
                LogicalPlan::Join(Join {
    548
    0
                    left,
    549
    0
                    right,
    550
    0
                    join_type,
    551
                    ..
    552
    0
                }) => match join_type {
    553
                    JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full => {
    554
    0
                        if left.schema().fields().is_empty() {
    555
    0
                            right.head_output_expr()
    556
                        } else {
    557
    0
                            left.head_output_expr()
    558
                        }
    559
                    }
    560
                    JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
    561
    0
                        left.head_output_expr()
    562
                    }
    563
                    JoinType::RightSemi | JoinType::RightAnti | JoinType::RightMark => {
    564
    0
                        right.head_output_expr()
    565
                    }
    566
                },
    567
    0
                LogicalPlan::RecursiveQuery(RecursiveQuery { static_term, .. }) => {
    568
    0
                    static_term.head_output_expr()
    569
                }
    570
    0
                LogicalPlan::Union(union) => Ok(Some(Expr::Column(Column::from(
    571
    0
                    union.schema.qualified_field(0),
    572
    0
                )))),
    573
    0
                LogicalPlan::TableScan(table) => Ok(Some(Expr::Column(Column::from(
    574
    0
                    table.projected_schema.qualified_field(0),
    575
    0
                )))),
    576
    0
                LogicalPlan::SubqueryAlias(subquery_alias) => {
    577
    0
                    let expr_opt = subquery_alias.input.head_output_expr()?;
    578
    0
                    expr_opt
    579
    0
                        .map(|expr| {
    580
    0
                            Ok(Expr::Column(create_col_from_scalar_expr(
    581
    0
                                &expr,
    582
    0
                                subquery_alias.alias.to_string(),
    583
    0
                            )?))
    584
    0
                        })
    585
    0
                        .map_or(Ok(None), |v| v.map(Some))
    586
                }
    587
    0
                LogicalPlan::Subquery(_) => Ok(None),
    588
                LogicalPlan::EmptyRelation(_)
    589
                | LogicalPlan::Statement(_)
    590
                | LogicalPlan::Values(_)
    591
                | LogicalPlan::Explain(_)
    592
                | LogicalPlan::Analyze(_)
    593
                | LogicalPlan::Extension(_)
    594
                | LogicalPlan::Dml(_)
    595
                | LogicalPlan::Copy(_)
    596
                | LogicalPlan::Ddl(_)
    597
                | LogicalPlan::DescribeTable(_)
    598
    0
                | LogicalPlan::Unnest(_) => Ok(None),
    599
            }
    600
    0
        }
    601
    602
        /// Recomputes schema and type information for this LogicalPlan if needed.
    603
        ///
    604
        /// Some `LogicalPlan`s may need to recompute their schema if the number or
    605
        /// type of expressions have been changed (for example due to type
    606
        /// coercion). For example [`LogicalPlan::Projection`]s schema depends on
    607
        /// its expressions.
    608
        ///
    609
        /// Some `LogicalPlan`s schema is unaffected by any changes to their
    610
        /// expressions. For example [`LogicalPlan::Filter`] schema is always the
    611
        /// same as its input schema.
    612
        ///
    613
        /// This is useful after modifying a plans `Expr`s (or input plans) via
    614
        /// methods such as [Self::map_children] and [Self::map_expressions]. Unlike
    615
        /// [Self::with_new_exprs], this method does not require a new set of
    616
        /// expressions or inputs plans.
    617
        ///
    618
        /// # Return value
    619
        /// Returns an error if there is some issue recomputing the schema.
    620
        ///
    621
        /// # Notes
    622
        ///
    623
        /// * Does not recursively recompute schema for input (child) plans.
    624
    0
        pub fn recompute_schema(self) -> Result<Self> {
    625
    0
            match self {
    626
                // Since expr may be different than the previous expr, schema of the projection
    627
                // may change. We need to use try_new method instead of try_new_with_schema method.
    628
                LogicalPlan::Projection(Projection {
    629
    0
                    expr,
    630
    0
                    input,
    631
                    schema: _,
    632
    0
                }) => Projection::try_new(expr, input).map(LogicalPlan::Projection),
    633
    0
                LogicalPlan::Dml(_) => Ok(self),
    634
    0
                LogicalPlan::Copy(_) => Ok(self),
    635
    0
                LogicalPlan::Values(Values { schema, values }) => {
    636
                    // todo it isn't clear why the schema is not recomputed here
    637
    0
                    Ok(LogicalPlan::Values(Values { schema, values }))
    638
                }
    639
    0
                LogicalPlan::Filter(Filter { predicate, input }) => {
    640
    0
                    Filter::try_new(predicate, input).map(LogicalPlan::Filter)
    641
                }
    642
    0
                LogicalPlan::Repartition(_) => Ok(self),
    643
                LogicalPlan::Window(Window {
    644
    0
                    input,
    645
    0
                    window_expr,
    646
                    schema: _,
    647
    0
                }) => Window::try_new(window_expr, input).map(LogicalPlan::Window),
    648
                LogicalPlan::Aggregate(Aggregate {
    649
    0
                    input,
    650
    0
                    group_expr,
    651
    0
                    aggr_expr,
    652
                    schema: _,
    653
    0
                }) => Aggregate::try_new(input, group_expr, aggr_expr)
    654
    0
                    .map(LogicalPlan::Aggregate),
    655
    0
                LogicalPlan::Sort(_) => Ok(self),
    656
                LogicalPlan::Join(Join {
    657
    0
                    left,
    658
    0
                    right,
    659
    0
                    filter,
    660
    0
                    join_type,
    661
    0
                    join_constraint,
    662
    0
                    on,
    663
                    schema: _,
    664
    0
                    null_equality,
    665
                }) => {
    666
    0
                    let schema =
    667
    0
                        build_join_schema(left.schema(), right.schema(), &join_type)?;
    668
    669
    0
                    let new_on: Vec<_> = on
    670
    0
                        .into_iter()
    671
    0
                        .map(|equi_expr| {
    672
                            // SimplifyExpression rule may add alias to the equi_expr.
    673
    0
                            (equi_expr.0.unalias(), equi_expr.1.unalias())
    674
    0
                        })
    675
    0
                        .collect();
    676
    677
    0
                    Ok(LogicalPlan::Join(Join {
    678
    0
                        left,
    679
    0
                        right,
    680
    0
                        join_type,
    681
    0
                        join_constraint,
    682
    0
                        on: new_on,
    683
    0
                        filter,
    684
    0
                        schema: DFSchemaRef::new(schema),
    685
    0
                        null_equality,
    686
    0
                    }))
    687
                }
    688
    0
                LogicalPlan::Subquery(_) => Ok(self),
    689
                LogicalPlan::SubqueryAlias(SubqueryAlias {
    690
    0
                    input,
    691
    0
                    alias,
    692
                    schema: _,
    693
    0
                }) => SubqueryAlias::try_new(input, alias).map(LogicalPlan::SubqueryAlias),
    694
    0
                LogicalPlan::Limit(_) => Ok(self),
    695
    0
                LogicalPlan::Ddl(_) => Ok(self),
    696
    0
                LogicalPlan::Extension(Extension { node }) => {
    697
                    // todo make an API that does not require cloning
    698
                    // This requires a copy of the extension nodes expressions and inputs
    699
    0
                    let expr = node.expressions();
    700
    0
                    let inputs: Vec<_> = node.inputs().into_iter().cloned().collect();
    701
                    Ok(LogicalPlan::Extension(Extension {
    702
    0
                        node: node.with_exprs_and_inputs(expr, inputs)?,
    703
                    }))
    704
                }
    705
    0
                LogicalPlan::Union(Union { inputs, schema }) => {
    706
    0
                    let first_input_schema = inputs[0].schema();
    707
    0
                    if schema.fields().len() == first_input_schema.fields().len() {
    708
                        // If inputs are not pruned do not change schema
    709
    0
                        Ok(LogicalPlan::Union(Union { inputs, schema }))
    710
                    } else {
    711
                        // A note on `Union`s constructed via `try_new_by_name`:
    712
                        //
    713
                        // At this point, the schema for each input should have
    714
                        // the same width. Thus, we do not need to save whether a
    715
                        // `Union` was created `BY NAME`, and can safely rely on the
    716
                        // `try_new` initializer to derive the new schema based on
    717
                        // column positions.
    718
    0
                        Ok(LogicalPlan::Union(Union::try_new(inputs)?))
    719
                    }
    720
                }
    721
    0
                LogicalPlan::Distinct(distinct) => {
    722
    0
                    let distinct = match distinct {
    723
    0
                        Distinct::All(input) => Distinct::All(input),
    724
                        Distinct::On(DistinctOn {
    725
    0
                            on_expr,
    726
    0
                            select_expr,
    727
    0
                            sort_expr,
    728
    0
                            input,
    729
                            schema: _,
    730
    0
                        }) => Distinct::On(DistinctOn::try_new(
    731
    0
                            on_expr,
    732
    0
                            select_expr,
    733
    0
                            sort_expr,
    734
    0
                            input,
    735
    0
                        )?),
    736
                    };
    737
    0
                    Ok(LogicalPlan::Distinct(distinct))
    738
                }
    739
    0
                LogicalPlan::RecursiveQuery(_) => Ok(self),
    740
    0
                LogicalPlan::Analyze(_) => Ok(self),
    741
    0
                LogicalPlan::Explain(_) => Ok(self),
    742
    0
                LogicalPlan::TableScan(_) => Ok(self),
    743
    0
                LogicalPlan::EmptyRelation(_) => Ok(self),
    744
    0
                LogicalPlan::Statement(_) => Ok(self),
    745
    0
                LogicalPlan::DescribeTable(_) => Ok(self),
    746
                LogicalPlan::Unnest(Unnest {
    747
    0
                    input,
    748
    0
                    exec_columns,
    749
    0
                    options,
    750
                    ..
    751
                }) => {
    752
                    // Update schema with unnested column type.
    753
    0
                    unnest_with_options(Arc::unwrap_or_clone(input), exec_columns, options)
    754
                }
    755
            }
    756
    0
        }
    757
    758
        /// Returns a new `LogicalPlan` based on `self` with inputs and
    759
        /// expressions replaced.
    760
        ///
    761
        /// Note this method creates an entirely new node, which requires a large
    762
        /// amount of clone'ing. When possible, the [`tree_node`] API should be used
    763
        /// instead of this API.
    764
        ///
    765
        /// The exprs correspond to the same order of expressions returned
    766
        /// by [`Self::expressions`]. This function is used by optimizers
    767
        /// to rewrite plans using the following pattern:
    768
        ///
    769
        /// [`tree_node`]: crate::logical_plan::tree_node
    770
        ///
    771
        /// ```text
    772
        /// let new_inputs = optimize_children(..., plan, props);
    773
        ///
    774
        /// // get the plans expressions to optimize
    775
        /// let exprs = plan.expressions();
    776
        ///
    777
        /// // potentially rewrite plan expressions
    778
        /// let rewritten_exprs = rewrite_exprs(exprs);
    779
        ///
    780
        /// // create new plan using rewritten_exprs in same position
    781
        /// let new_plan = plan.new_with_exprs(rewritten_exprs, new_inputs);
    782
        /// ```
    783
    0
        pub fn with_new_exprs(
    784
    0
            &self,
    785
    0
            mut expr: Vec<Expr>,
    786
    0
            inputs: Vec<LogicalPlan>,
    787
    0
        ) -> Result<LogicalPlan> {
    788
    0
            match self {
    789
                // Since expr may be different than the previous expr, schema of the projection
    790
                // may change. We need to use try_new method instead of try_new_with_schema method.
    791
                LogicalPlan::Projection(Projection { .. }) => {
    792
    0
                    let input = self.only_input(inputs)?;
    793
    0
                    Projection::try_new(expr, Arc::new(input)).map(LogicalPlan::Projection)
    794
                }
    795
                LogicalPlan::Dml(DmlStatement {
    796
    0
                    table_name,
    797
    0
                    target,
    798
    0
                    op,
    799
                    ..
    800
                }) => {
    801
    0
                    self.assert_no_expressions(expr)?;
    802
    0
                    let input = self.only_input(inputs)?;
    803
    0
                    Ok(LogicalPlan::Dml(DmlStatement::new(
    804
    0
                        table_name.clone(),
    805
    0
                        Arc::clone(target),
    806
    0
                        op.clone(),
    807
    0
                        Arc::new(input),
    808
    0
                    )))
    809
                }
    810
                LogicalPlan::Copy(CopyTo {
    811
                    input: _,
    812
    0
                    output_url,
    813
    0
                    file_type,
    814
    0
                    options,
    815
    0
                    partition_by,
    816
                    output_schema: _,
    817
                }) => {
    818
    0
                    self.assert_no_expressions(expr)?;
    819
    0
                    let input = self.only_input(inputs)?;
    820
    0
                    Ok(LogicalPlan::Copy(CopyTo::new(
    821
    0
                        Arc::new(input),
    822
    0
                        output_url.clone(),
    823
    0
                        partition_by.clone(),
    824
    0
                        Arc::clone(file_type),
    825
    0
                        options.clone(),
    826
    0
                    )))
    827
                }
    828
    0
                LogicalPlan::Values(Values { schema, .. }) => {
    829
    0
                    self.assert_no_inputs(inputs)?;
    830
                    Ok(LogicalPlan::Values(Values {
    831
    0
                        schema: Arc::clone(schema),
    832
    0
                        values: expr
    833
    0
                            .chunks_exact(schema.fields().len())
    834
    0
                            .map(|s| s.to_vec())
    835
    0
                            .collect(),
    836
                    }))
    837
                }
    838
                LogicalPlan::Filter { .. } => {
    839
    0
                    let predicate = self.only_expr(expr)?;
    840
    0
                    let input = self.only_input(inputs)?;
    841
    842
    0
                    Filter::try_new(predicate, Arc::new(input)).map(LogicalPlan::Filter)
    843
                }
    844
                LogicalPlan::Repartition(Repartition {
    845
    0
                    partitioning_scheme,
    846
                    ..
    847
    0
                }) => match partitioning_scheme {
    848
    0
                    Partitioning::RoundRobinBatch(n) => {
    849
    0
                        self.assert_no_expressions(expr)?;
    850
    0
                        let input = self.only_input(inputs)?;
    851
    0
                        Ok(LogicalPlan::Repartition(Repartition {
    852
    0
                            partitioning_scheme: Partitioning::RoundRobinBatch(*n),
    853
    0
                            input: Arc::new(input),
    854
    0
                        }))
    855
                    }
    856
    0
                    Partitioning::Hash(_, n) => {
    857
    0
                        let input = self.only_input(inputs)?;
    858
    0
                        Ok(LogicalPlan::Repartition(Repartition {
    859
    0
                            partitioning_scheme: Partitioning::Hash(expr, *n),
    860
    0
                            input: Arc::new(input),
    861
    0
                        }))
    862
                    }
    863
                    Partitioning::DistributeBy(_) => {
    864
    0
                        let input = self.only_input(inputs)?;
    865
    0
                        Ok(LogicalPlan::Repartition(Repartition {
    866
    0
                            partitioning_scheme: Partitioning::DistributeBy(expr),
    867
    0
                            input: Arc::new(input),
    868
    0
                        }))
    869
                    }
    870
                },
    871
    0
                LogicalPlan::Window(Window { window_expr, .. }) => {
    872
    0
                    assert_eq!(window_expr.len(), expr.len());
    873
    0
                    let input = self.only_input(inputs)?;
    874
    0
                    Window::try_new(expr, Arc::new(input)).map(LogicalPlan::Window)
    875
                }
    876
    0
                LogicalPlan::Aggregate(Aggregate { group_expr, .. }) => {
    877
    0
                    let input = self.only_input(inputs)?;
    878
                    // group exprs are the first expressions
    879
    0
                    let agg_expr = expr.split_off(group_expr.len());
    880
    881
    0
                    Aggregate::try_new(Arc::new(input), expr, agg_expr)
    882
    0
                        .map(LogicalPlan::Aggregate)
    883
                }
    884
                LogicalPlan::Sort(Sort {
    885
    0
                    expr: sort_expr,
    886
    0
                    fetch,
    887
                    ..
    888
                }) => {
    889
    0
                    let input = self.only_input(inputs)?;
    890
                    Ok(LogicalPlan::Sort(Sort {
    891
    0
                        expr: expr
    892
    0
                            .into_iter()
    893
    0
                            .zip(sort_expr.iter())
    894
    0
                            .map(|(expr, sort)| sort.with_expr(expr))
    895
    0
                            .collect(),
    896
    0
                        input: Arc::new(input),
    897
    0
                        fetch: *fetch,
    898
                    }))
    899
                }
    900
                LogicalPlan::Join(Join {
    901
    0
                    join_type,
    902
    0
                    join_constraint,
    903
    0
                    on,
    904
    0
                    null_equality,
    905
                    ..
    906
                }) => {
    907
    0
                    let (left, right) = self.only_two_inputs(inputs)?;
    908
    0
                    let schema = build_join_schema(left.schema(), right.schema(), join_type)?;
    909
    910
    0
                    let equi_expr_count = on.len() * 2;
    911
    0
                    assert!(expr.len() >= equi_expr_count);
    912
    913
                    // Assume that the last expr, if any,
    914
                    // is the filter_expr (non equality predicate from ON clause)
    915
    0
                    let filter_expr = if expr.len() > equi_expr_count {
    916
    0
                        expr.pop()
    917
                    } else {
    918
    0
                        None
    919
                    };
    920
    921
                    // The first part of expr is equi-exprs,
    922
                    // and the struct of each equi-expr is like `left-expr = right-expr`.
    923
    0
                    assert_eq!(expr.len(), equi_expr_count);
    924
    0
                    let mut new_on = Vec::with_capacity(on.len());
    925
    0
                    let mut iter = expr.into_iter();
    926
    0
                    while let Some(left) = iter.next() {
    927
    0
                        let Some(right) = iter.next() else {
    928
    0
                            internal_err!("Expected a pair of expressions to construct the join on expression")?
    929
                        };
    930
    931
                        // SimplifyExpression rule may add alias to the equi_expr.
    932
    0
                        new_on.push((left.unalias(), right.unalias()));
    933
                    }
    934
    935
    0
                    Ok(LogicalPlan::Join(Join {
    936
    0
                        left: Arc::new(left),
    937
    0
                        right: Arc::new(right),
    938
    0
                        join_type: *join_type,
    939
    0
                        join_constraint: *join_constraint,
    940
    0
                        on: new_on,
    941
    0
                        filter: filter_expr,
    942
    0
                        schema: DFSchemaRef::new(schema),
    943
    0
                        null_equality: *null_equality,
    944
    0
                    }))
    945
                }
    946
                LogicalPlan::Subquery(Subquery {
    947
    0
                    outer_ref_columns,
    948
    0
                    spans,
    949
                    ..
    950
                }) => {
    951
    0
                    self.assert_no_expressions(expr)?;
    952
    0
                    let input = self.only_input(inputs)?;
    953
    0
                    let subquery = LogicalPlanBuilder::from(input).build()?;
    954
    0
                    Ok(LogicalPlan::Subquery(Subquery {
    955
    0
                        subquery: Arc::new(subquery),
    956
    0
                        outer_ref_columns: outer_ref_columns.clone(),
    957
    0
                        spans: spans.clone(),
    958
    0
                    }))
    959
                }
    960
    0
                LogicalPlan::SubqueryAlias(SubqueryAlias { alias, .. }) => {
    961
    0
                    self.assert_no_expressions(expr)?;
    962
    0
                    let input = self.only_input(inputs)?;
    963
    0
                    SubqueryAlias::try_new(Arc::new(input), alias.clone())
    964
    0
                        .map(LogicalPlan::SubqueryAlias)
    965
                }
    966
    0
                LogicalPlan::Limit(Limit { skip, fetch, .. }) => {
    967
    0
                    let old_expr_len = skip.iter().chain(fetch.iter()).count();
    968
    0
                    assert_eq_or_internal_err!(
    969
                        old_expr_len,
    970
    0
                        expr.len(),
    971
    0
                        "Invalid number of new Limit expressions: expected {}, got {}",
    972
                        old_expr_len,
    973
    0
                        expr.len()
    974
                    );
    975
                    // `LogicalPlan::expressions()` returns in [skip, fetch] order, so we can pop from the end.
    976
    0
                    let new_fetch = fetch.as_ref().and_then(|_| expr.pop());
    977
    0
                    let new_skip = skip.as_ref().and_then(|_| expr.pop());
    978
    0
                    let input = self.only_input(inputs)?;
    979
    0
                    Ok(LogicalPlan::Limit(Limit {
    980
    0
                        skip: new_skip.map(Box::new),
    981
    0
                        fetch: new_fetch.map(Box::new),
    982
    0
                        input: Arc::new(input),
    983
    0
                    }))
    984
                }
    985
                LogicalPlan::Ddl(DdlStatement::CreateMemoryTable(CreateMemoryTable {
    986
    0
                    name,
    987
    0
                    if_not_exists,
    988
    0
                    or_replace,
    989
    0
                    column_defaults,
    990
    0
                    temporary,
    991
                    ..
    992
                })) => {
    993
    0
                    self.assert_no_expressions(expr)?;
    994
    0
                    let input = self.only_input(inputs)?;
    995
    0
                    Ok(LogicalPlan::Ddl(DdlStatement::CreateMemoryTable(
    996
    0
                        CreateMemoryTable {
    997
    0
                            input: Arc::new(input),
    998
    0
                            constraints: Constraints::default(),
    999
    0
                            name: name.clone(),
    1000
    0
                            if_not_exists: *if_not_exists,
    1001
    0
                            or_replace: *or_replace,
    1002
    0
                            column_defaults: column_defaults.clone(),
    1003
    0
                            temporary: *temporary,
    1004
    0
                        },
    1005
    0
                    )))
    1006
                }
    1007
                LogicalPlan::Ddl(DdlStatement::CreateView(CreateView {
    1008
    0
                    name,
    1009
    0
                    or_replace,
    1010
    0
                    definition,
    1011
    0
                    temporary,
    1012
                    ..
    1013
                })) => {
    1014
    0
                    self.assert_no_expressions(expr)?;
    1015
    0
                    let input = self.only_input(inputs)?;
    1016
    0
                    Ok(LogicalPlan::Ddl(DdlStatement::CreateView(CreateView {
    1017
    0
                        input: Arc::new(input),
    1018
    0
                        name: name.clone(),
    1019
    0
                        or_replace: *or_replace,
    1020
    0
                        temporary: *temporary,
    1021
    0
                        definition: definition.clone(),
    1022
    0
                    })))
    1023
                }
    1024
    0
                LogicalPlan::Extension(e) => Ok(LogicalPlan::Extension(Extension {
    1025
    0
                    node: e.node.with_exprs_and_inputs(expr, inputs)?,
    1026
                })),
    1027
    0
                LogicalPlan::Union(Union { schema, .. }) => {
    1028
    0
                    self.assert_no_expressions(expr)?;
    1029
    0
                    let input_schema = inputs[0].schema();
    1030
                    // If inputs are not pruned do not change schema.
    1031
    0
                    let schema = if schema.fields().len() == input_schema.fields().len() {
    1032
    0
                        Arc::clone(schema)
    1033
                    } else {
    1034
    0
                        Arc::clone(input_schema)
    1035
                    };
    1036
    0
                    Ok(LogicalPlan::Union(Union {
    1037
    0
                        inputs: inputs.into_iter().map(Arc::new).collect(),
    1038
    0
                        schema,
    1039
    0
                    }))
    1040
                }
    1041
    0
                LogicalPlan::Distinct(distinct) => {
    1042
    0
                    let distinct = match distinct {
    1043
                        Distinct::All(_) => {
    1044
    0
                            self.assert_no_expressions(expr)?;
    1045
    0
                            let input = self.only_input(inputs)?;
    1046
    0
                            Distinct::All(Arc::new(input))
    1047
                        }
    1048
                        Distinct::On(DistinctOn {
    1049
    0
                            on_expr,
    1050
    0
                            select_expr,
    1051
                            ..
    1052
                        }) => {
    1053
    0
                            let input = self.only_input(inputs)?;
    1054
    0
                            let sort_expr = expr.split_off(on_expr.len() + select_expr.len());
    1055
    0
                            let select_expr = expr.split_off(on_expr.len());
    1056
    0
                            assert!(sort_expr.is_empty(), "with_new_exprs for Distinct does not support sort expressions");
    1057
    0
                            Distinct::On(DistinctOn::try_new(
    1058
    0
                                expr,
    1059
    0
                                select_expr,
    1060
    0
                                None, // no sort expressions accepted
    1061
    0
                                Arc::new(input),
    1062
    0
                            )?)
    1063
                        }
    1064
                    };
    1065
    0
                    Ok(LogicalPlan::Distinct(distinct))
    1066
                }
    1067
                LogicalPlan::RecursiveQuery(RecursiveQuery {
    1068
    0
                    name, is_distinct, ..
    1069
                }) => {
    1070
    0
                    self.assert_no_expressions(expr)?;
    1071
    0
                    let (static_term, recursive_term) = self.only_two_inputs(inputs)?;
    1072
    0
                    Ok(LogicalPlan::RecursiveQuery(RecursiveQuery {
    1073
    0
                        name: name.clone(),
    1074
    0
                        static_term: Arc::new(static_term),
    1075
    0
                        recursive_term: Arc::new(recursive_term),
    1076
    0
                        is_distinct: *is_distinct,
    1077
    0
                    }))
    1078
                }
    1079
    0
                LogicalPlan::Analyze(a) => {
    1080
    0
                    self.assert_no_expressions(expr)?;
    1081
    0
                    let input = self.only_input(inputs)?;
    1082
    0
                    Ok(LogicalPlan::Analyze(Analyze {
    1083
    0
                        verbose: a.verbose,
    1084
    0
                        schema: Arc::clone(&a.schema),
    1085
    0
                        input: Arc::new(input),
    1086
    0
                    }))
    1087
                }
    1088
    0
                LogicalPlan::Explain(e) => {
    1089
    0
                    self.assert_no_expressions(expr)?;
    1090
    0
                    let input = self.only_input(inputs)?;
    1091
    0
                    Ok(LogicalPlan::Explain(Explain {
    1092
    0
                        verbose: e.verbose,
    1093
    0
                        plan: Arc::new(input),
    1094
    0
                        explain_format: e.explain_format.clone(),
    1095
    0
                        stringified_plans: e.stringified_plans.clone(),
    1096
    0
                        schema: Arc::clone(&e.schema),
    1097
    0
                        logical_optimization_succeeded: e.logical_optimization_succeeded,
    1098
    0
                    }))
    1099
                }
    1100
                LogicalPlan::Statement(Statement::Prepare(Prepare {
    1101
    0
                    name, fields, ..
    1102
                })) => {
    1103
    0
                    self.assert_no_expressions(expr)?;
    1104
    0
                    let input = self.only_input(inputs)?;
    1105
    0
                    Ok(LogicalPlan::Statement(Statement::Prepare(Prepare {
    1106
    0
                        name: name.clone(),
    1107
    0
                        fields: fields.clone(),
    1108
    0
                        input: Arc::new(input),
    1109
    0
                    })))
    1110
                }
    1111
    0
                LogicalPlan::Statement(Statement::Execute(Execute { name, .. })) => {
    1112
    0
                    self.assert_no_inputs(inputs)?;
    1113
    0
                    Ok(LogicalPlan::Statement(Statement::Execute(Execute {
    1114
    0
                        name: name.clone(),
    1115
    0
                        parameters: expr,
    1116
    0
                    })))
    1117
                }
    1118
    0
                LogicalPlan::TableScan(ts) => {
    1119
    0
                    self.assert_no_inputs(inputs)?;
    1120
    0
                    Ok(LogicalPlan::TableScan(TableScan {
    1121
    0
                        filters: expr,
    1122
    0
                        ..ts.clone()
    1123
    0
                    }))
    1124
                }
    1125
                LogicalPlan::EmptyRelation(_)
    1126
                | LogicalPlan::Ddl(_)
    1127
                | LogicalPlan::Statement(_)
    1128
                | LogicalPlan::DescribeTable(_) => {
    1129
                    // All of these plan types have no inputs / exprs so should not be called
    1130
    0
                    self.assert_no_expressions(expr)?;
    1131
    0
                    self.assert_no_inputs(inputs)?;
    1132
    0
                    Ok(self.clone())
    1133
                }
    1134
                LogicalPlan::Unnest(Unnest {
    1135
    0
                    exec_columns: columns,
    1136
    0
                    options,
    1137
                    ..
    1138
                }) => {
    1139
    0
                    self.assert_no_expressions(expr)?;
    1140
    0
                    let input = self.only_input(inputs)?;
    1141
                    // Update schema with unnested column type.
    1142
    0
                    let new_plan =
    1143
    0
                        unnest_with_options(input, columns.clone(), options.clone())?;
    1144
    0
                    Ok(new_plan)
    1145
                }
    1146
            }
    1147
    0
        }
    1148
    1149
        /// checks that the plan conforms to the listed invariant level, returning an Error if not
    1150
    0
        pub fn check_invariants(&self, check: InvariantLevel) -> Result<()> {
    1151
    0
            match check {
    1152
    0
                InvariantLevel::Always => assert_always_invariants_at_current_node(self),
    1153
    0
                InvariantLevel::Executable => assert_executable_invariants(self),
    1154
            }
    1155
    0
        }
    1156
    1157
        /// Helper for [Self::with_new_exprs] to use when no expressions are expected.
    1158
        #[inline]
    1159
        #[expect(clippy::needless_pass_by_value)] // expr is moved intentionally to ensure it's not used again
    1160
    0
        fn assert_no_expressions(&self, expr: Vec<Expr>) -> Result<()> {
    1161
    0
            assert_or_internal_err!(
    1162
    0
                expr.is_empty(),
    1163
    0
                "{self:?} should have no exprs, got {:?}",
    1164
                expr
    1165
            );
    1166
    0
            Ok(())
    1167
    0
        }
    1168
    1169
        /// Helper for [Self::with_new_exprs] to use when no inputs are expected.
    1170
        #[inline]
    1171
        #[expect(clippy::needless_pass_by_value)] // inputs is moved intentionally to ensure it's not used again
    1172
    0
        fn assert_no_inputs(&self, inputs: Vec<LogicalPlan>) -> Result<()> {
    1173
    0
            assert_or_internal_err!(
    1174
    0
                inputs.is_empty(),
    1175
    0
                "{self:?} should have no inputs, got: {:?}",
    1176
                inputs
    1177
            );
    1178
    0
            Ok(())
    1179
    0
        }
    1180
    1181
        /// Helper for [Self::with_new_exprs] to use when exactly one expression is expected.
    1182
        #[inline]
    1183
    0
        fn only_expr(&self, mut expr: Vec<Expr>) -> Result<Expr> {
    1184
    0
            assert_eq_or_internal_err!(
    1185
    0
                expr.len(),
    1186
                1,
    1187
    0
                "{self:?} should have exactly one expr, got {:?}",
    1188
    0
                &expr
    1189
            );
    1190
    0
            Ok(expr.remove(0))
    1191
    0
        }
    1192
    1193
        /// Helper for [Self::with_new_exprs] to use when exactly one input is expected.
    1194
        #[inline]
    1195
    0
        fn only_input(&self, mut inputs: Vec<LogicalPlan>) -> Result<LogicalPlan> {
    1196
    0
            assert_eq_or_internal_err!(
    1197
    0
                inputs.len(),
    1198
                1,
    1199
    0
                "{self:?} should have exactly one input, got {:?}",
    1200
    0
                &inputs
    1201
            );
    1202
    0
            Ok(inputs.remove(0))
    1203
    0
        }
    1204
    1205
        /// Helper for [Self::with_new_exprs] to use when exactly two inputs are expected.
    1206
        #[inline]
    1207
    0
        fn only_two_inputs(
    1208
    0
            &self,
    1209
    0
            mut inputs: Vec<LogicalPlan>,
    1210
    0
        ) -> Result<(LogicalPlan, LogicalPlan)> {
    1211
    0
            assert_eq_or_internal_err!(
    1212
    0
                inputs.len(),
    1213
                2,
    1214
    0
                "{self:?} should have exactly two inputs, got {:?}",
    1215
    0
                &inputs
    1216
            );
    1217
    0
            let right = inputs.remove(1);
    1218
    0
            let left = inputs.remove(0);
    1219
    0
            Ok((left, right))
    1220
    0
        }
    1221
    1222
        /// Replaces placeholder param values (like `$1`, `$2`) in [`LogicalPlan`]
    1223
        /// with the specified `param_values`.
    1224
        ///
    1225
        /// [`Prepare`] statements are converted to
    1226
        /// their inner logical plan for execution.
    1227
        ///
    1228
        /// # Example
    1229
        /// ```
    1230
        /// # use arrow::datatypes::{Field, Schema, DataType};
    1231
        /// use datafusion_common::ScalarValue;
    1232
        /// # use datafusion_expr::{lit, col, LogicalPlanBuilder, logical_plan::table_scan, placeholder};
    1233
        /// # let schema = Schema::new(vec![
    1234
        /// #     Field::new("id", DataType::Int32, false),
    1235
        /// # ]);
    1236
        /// // Build SELECT * FROM t1 WHERE id = $1
    1237
        /// let plan = table_scan(Some("t1"), &schema, None).unwrap()
    1238
        ///     .filter(col("id").eq(placeholder("$1"))).unwrap()
    1239
        ///     .build().unwrap();
    1240
        ///
    1241
        /// assert_eq!(
    1242
        ///   "Filter: t1.id = $1\
    1243
        ///   \n  TableScan: t1",
    1244
        ///   plan.display_indent().to_string()
    1245
        /// );
    1246
        ///
    1247
        /// // Fill in the parameter $1 with a literal 3
    1248
        /// let plan = plan.with_param_values(vec![
    1249
        ///   ScalarValue::from(3i32) // value at index 0 --> $1
    1250
        /// ]).unwrap();
    1251
        ///
    1252
        /// assert_eq!(
    1253
        ///    "Filter: t1.id = Int32(3)\
    1254
        ///    \n  TableScan: t1",
    1255
        ///    plan.display_indent().to_string()
    1256
        ///  );
    1257
        ///
    1258
        /// // Note you can also used named parameters
    1259
        /// // Build SELECT * FROM t1 WHERE id = $my_param
    1260
        /// let plan = table_scan(Some("t1"), &schema, None).unwrap()
    1261
        ///     .filter(col("id").eq(placeholder("$my_param"))).unwrap()
    1262
        ///     .build().unwrap()
    1263
        ///     // Fill in the parameter $my_param with a literal 3
    1264
        ///     .with_param_values(vec![
    1265
        ///       ("my_param", ScalarValue::from(3i32)),
    1266
        ///     ]).unwrap();
    1267
        ///
    1268
        /// assert_eq!(
    1269
        ///    "Filter: t1.id = Int32(3)\
    1270
        ///    \n  TableScan: t1",
    1271
        ///    plan.display_indent().to_string()
    1272
        ///  );
    1273
        /// ```
    1274
    0
        pub fn with_param_values(
    1275
    0
            self,
    1276
    0
            param_values: impl Into<ParamValues>,
    1277
    0
        ) -> Result<LogicalPlan> {
    1278
    0
            let param_values = param_values.into();
    1279
    0
            let plan_with_values = self.replace_params_with_values(&param_values)?;
    1280
    1281
            // unwrap Prepare
    1282
            Ok(
    1283
    0
                if let LogicalPlan::Statement(Statement::Prepare(prepare_lp)) =
    1284
    0
                    plan_with_values
    1285
                {
    1286
    0
                    param_values.verify_fields(&prepare_lp.fields)?;
    1287
                    // try and take ownership of the input if is not shared, clone otherwise
    1288
    0
                    Arc::unwrap_or_clone(prepare_lp.input)
    1289
                } else {
    1290
    0
                    plan_with_values
    1291
                },
    1292
            )
    1293
    0
        }
    1294
    1295
        /// Returns the maximum number of rows that this plan can output, if known.
    1296
        ///
    1297
        /// If `None`, the plan can return any number of rows.
    1298
        /// If `Some(n)` then the plan can return at most `n` rows but may return fewer.
    1299
    0
        pub fn max_rows(self: &LogicalPlan) -> Option<usize> {
    1300
    0
            match self {
    1301
    0
                LogicalPlan::Projection(Projection { input, .. }) => input.max_rows(),
    1302
    0
                LogicalPlan::Filter(filter) => {
    1303
    0
                    if filter.is_scalar() {
    1304
    0
                        Some(1)
    1305
                    } else {
    1306
    0
                        filter.input.max_rows()
    1307
                    }
    1308
                }
    1309
    0
                LogicalPlan::Window(Window { input, .. }) => input.max_rows(),
    1310
                LogicalPlan::Aggregate(Aggregate {
    1311
    0
                    input, group_expr, ..
    1312
                }) => {
    1313
                    // Empty group_expr will return Some(1)
    1314
    0
                    if group_expr
    1315
    0
                        .iter()
    1316
    0
                        .all(|expr| matches!(expr, Expr::Literal(_, _)))
    1317
                    {
    1318
    0
                        Some(1)
    1319
                    } else {
    1320
    0
                        input.max_rows()
    1321
                    }
    1322
                }
    1323
    0
                LogicalPlan::Sort(Sort { input, fetch, .. }) => {
    1324
    0
                    match (fetch, input.max_rows()) {
    1325
    0
                        (Some(fetch_limit), Some(input_max)) => {
    1326
    0
                            Some(input_max.min(*fetch_limit))
    1327
                        }
    1328
    0
                        (Some(fetch_limit), None) => Some(*fetch_limit),
    1329
    0
                        (None, Some(input_max)) => Some(input_max),
    1330
    0
                        (None, None) => None,
    1331
                    }
    1332
                }
    1333
                LogicalPlan::Join(Join {
    1334
    0
                    left,
    1335
    0
                    right,
    1336
    0
                    join_type,
    1337
                    ..
    1338
    0
                }) => match join_type {
    1339
    0
                    JoinType::Inner => Some(left.max_rows()? * right.max_rows()?),
    1340
                    JoinType::Left | JoinType::Right | JoinType::Full => {
    1341
    0
                        match (left.max_rows()?, right.max_rows()?, join_type) {
    1342
    0
                            (0, 0, _) => Some(0),
    1343
    0
                            (max_rows, 0, JoinType::Left | JoinType::Full) => Some(max_rows),
    1344
    0
                            (0, max_rows, JoinType::Right | JoinType::Full) => Some(max_rows),
    1345
    0
                            (left_max, right_max, _) => Some(left_max * right_max),
    1346
                        }
    1347
                    }
    1348
                    JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
    1349
    0
                        left.max_rows()
    1350
                    }
    1351
                    JoinType::RightSemi | JoinType::RightAnti | JoinType::RightMark => {
    1352
    0
                        right.max_rows()
    1353
                    }
    1354
                },
    1355
    0
                LogicalPlan::Repartition(Repartition { input, .. }) => input.max_rows(),
    1356
    0
                LogicalPlan::Union(Union { inputs, .. }) => {
    1357
    0
                    inputs.iter().try_fold(0usize, |mut acc, plan| {
    1358
    0
                        acc += plan.max_rows()?;
    1359
    0
                        Some(acc)
    1360
    0
                    })
    1361
                }
    1362
    0
                LogicalPlan::TableScan(TableScan { fetch, .. }) => *fetch,
    1363
    0
                LogicalPlan::EmptyRelation(_) => Some(0),
    1364
    0
                LogicalPlan::RecursiveQuery(_) => None,
    1365
    0
                LogicalPlan::Subquery(_) => None,
    1366
    0
                LogicalPlan::SubqueryAlias(SubqueryAlias { input, .. }) => input.max_rows(),
    1367
    0
                LogicalPlan::Limit(limit) => match limit.get_fetch_type() {
    1368
    0
                    Ok(FetchType::Literal(s)) => s,
    1369
    0
                    _ => None,
    1370
                },
    1371
                LogicalPlan::Distinct(
    1372
    0
                    Distinct::All(input) | Distinct::On(DistinctOn { input, .. }),
    1373
    0
                ) => input.max_rows(),
    1374
    0
                LogicalPlan::Values(v) => Some(v.values.len()),
    1375
    0
                LogicalPlan::Unnest(_) => None,
    1376
                LogicalPlan::Ddl(_)
    1377
                | LogicalPlan::Explain(_)
    1378
                | LogicalPlan::Analyze(_)
    1379
                | LogicalPlan::Dml(_)
    1380
                | LogicalPlan::Copy(_)
    1381
                | LogicalPlan::DescribeTable(_)
    1382
                | LogicalPlan::Statement(_)
    1383
    0
                | LogicalPlan::Extension(_) => None,
    1384
            }
    1385
    0
        }
    1386
    1387
        /// If this node's expressions contains any references to an outer subquery
    1388
    0
        pub fn contains_outer_reference(&self) -> bool {
    1389
    0
            let mut contains = false;
    1390
    0
            self.apply_expressions(|expr| {
    1391
    0
                Ok(if expr.contains_outer() {
    1392
    0
                    contains = true;
    1393
    0
                    TreeNodeRecursion::Stop
    1394
                } else {
    1395
    0
                    TreeNodeRecursion::Continue
    1396
                })
    1397
    0
            })
    1398
    0
            .unwrap();
    1399
    0
            contains
    1400
    0
        }
    1401
    1402
        /// Get the output expressions and their corresponding columns.
    1403
        ///
    1404
        /// The parent node may reference the output columns of the plan by expressions, such as
    1405
        /// projection over aggregate or window functions. This method helps to convert the
    1406
        /// referenced expressions into columns.
    1407
        ///
    1408
        /// See also: [`crate::utils::columnize_expr`]
    1409
    0
        pub fn columnized_output_exprs(&self) -> Result<Vec<(&Expr, Column)>> {
    1410
    0
            match self {
    1411
    0
                LogicalPlan::Aggregate(aggregate) => Ok(aggregate
    1412
    0
                    .output_expressions()?
    1413
    0
                    .into_iter()
    1414
    0
                    .zip(self.schema().columns())
    1415
    0
                    .collect()),
    1416
                LogicalPlan::Window(Window {
    1417
    0
                    window_expr,
    1418
    0
                    input,
    1419
    0
                    schema,
    1420
                }) => {
    1421
                    // The input could be another Window, so the result should also include the input's. For Example:
    1422
                    // `EXPLAIN SELECT RANK() OVER (PARTITION BY a ORDER BY b), SUM(b) OVER (PARTITION BY a) FROM t`
    1423
                    // Its plan is:
    1424
                    // Projection: RANK() PARTITION BY [t.a] ORDER BY [t.b ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW, SUM(t.b) PARTITION BY [t.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
    1425
                    //   WindowAggr: windowExpr=[[SUM(CAST(t.b AS Int64)) PARTITION BY [t.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]]
    1426
                    //     WindowAggr: windowExpr=[[RANK() PARTITION BY [t.a] ORDER BY [t.b ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]/
    1427
                    //       TableScan: t projection=[a, b]
    1428
    0
                    let mut output_exprs = input.columnized_output_exprs()?;
    1429
    0
                    let input_len = input.schema().fields().len();
    1430
    0
                    output_exprs.extend(
    1431
    0
                        window_expr
    1432
    0
                            .iter()
    1433
    0
                            .zip(schema.columns().into_iter().skip(input_len)),
    1434
                    );
    1435
    0
                    Ok(output_exprs)
    1436
                }
    1437
    0
                _ => Ok(vec![]),
    1438
            }
    1439
    0
        }
    1440
    }
    1441
    1442
    impl LogicalPlan {
    1443
        /// Return a `LogicalPlan` with all placeholders (e.g $1 $2,
    1444
        /// ...) replaced with corresponding values provided in
    1445
        /// `params_values`
    1446
        ///
    1447
        /// See [`Self::with_param_values`] for examples and usage with an owned
    1448
        /// `ParamValues`
    1449
    0
        pub fn replace_params_with_values(
    1450
    0
            self,
    1451
    0
            param_values: &ParamValues,
    1452
    0
        ) -> Result<LogicalPlan> {
    1453
    0
            self.transform_up_with_subqueries(|plan| {
    1454
    0
                let schema = Arc::clone(plan.schema());
    1455
    0
                let name_preserver = NamePreserver::new(&plan);
    1456
    0
                plan.map_expressions(|e| {
    1457
    0
                    let (e, has_placeholder) = e.infer_placeholder_types(&schema)?;
    1458
    0
                    if !has_placeholder {
    1459
                        // Performance optimization:
    1460
                        // avoid NamePreserver copy and second pass over expression
    1461
                        // if no placeholders.
    1462
    0
                        Ok(Transformed::no(e))
    1463
                    } else {
    1464
    0
                        let original_name = name_preserver.save(&e);
    1465
    0
                        let transformed_expr = e.transform_up(|e| {
    1466
    0
                            if let Expr::Placeholder(Placeholder { id, .. }) = e {
    1467
    0
                                let (value, metadata) = param_values
    1468
    0
                                    .get_placeholders_with_values(&id)?
    1469
    0
                                    .into_inner();
    1470
    0
                                Ok(Transformed::yes(Expr::Literal(value, metadata)))
    1471
                            } else {
    1472
    0
                                Ok(Transformed::no(e))
    1473
                            }
    1474
    0
                        })?;
    1475
                        // Preserve name to avoid breaking column references to this expression
    1476
    0
                        Ok(transformed_expr.update_data(|expr| original_name.restore(expr)))
    1477
                    }
    1478
    0
                })?
    1479
    0
                .map_data(|plan| plan.update_schema_data_type())
    1480
    0
            })
    1481
    0
            .map(|res| res.data)
    1482
    0
        }
    1483
    1484
        /// Recompute schema fields' data type after replacing params, ensuring fields data type can be
    1485
        /// updated according to the new parameters.
    1486
        ///
    1487
        /// Unlike `recompute_schema()`, this method rebuilds VALUES plans entirely to properly infer
    1488
        /// types types from literal values after placeholder substitution.
    1489
    0
        fn update_schema_data_type(self) -> Result<LogicalPlan> {
    1490
    0
            match self {
    1491
                // Build `LogicalPlan::Values` from the values for type inference.
    1492
                // We can't use `recompute_schema` because it skips recomputing for
    1493
                // `LogicalPlan::Values`.
    1494
    0
                LogicalPlan::Values(Values { values, schema: _ }) => {
    1495
    0
                    LogicalPlanBuilder::values(values)?.build()
    1496
                }
    1497
                // other plans can just use `recompute_schema` directly.
    1498
    0
                plan => plan.recompute_schema(),
    1499
            }
    1500
    0
        }
    1501
    1502
        /// Walk the logical plan, find any `Placeholder` tokens, and return a set of their names.
    1503
    0
        pub fn get_parameter_names(&self) -> Result<HashSet<String>> {
    1504
    0
            let mut param_names = HashSet::new();
    1505
    0
            self.apply_with_subqueries(|plan| {
    1506
    0
                plan.apply_expressions(|expr| {
    1507
    0
                    expr.apply(|expr| {
    1508
    0
                        if let Expr::Placeholder(Placeholder { id, .. }) = expr {
    1509
    0
                            param_names.insert(id.clone());
    1510
    0
                        }
    1511
    0
                        Ok(TreeNodeRecursion::Continue)
    1512
    0
                    })
    1513
    0
                })
    1514
    0
            })
    1515
    0
            .map(|_| param_names)
    1516
    0
        }
    1517
    1518
        /// Walk the logical plan, find any `Placeholder` tokens, and return a map of their IDs and DataTypes
    1519
        ///
    1520
        /// Note that this will drop any extension or field metadata attached to parameters. Use
    1521
        /// [`LogicalPlan::get_parameter_fields`] to keep extension metadata.
    1522
    0
        pub fn get_parameter_types(
    1523
    0
            &self,
    1524
    0
        ) -> Result<HashMap<String, Option<DataType>>, DataFusionError> {
    1525
    0
            let mut parameter_fields = self.get_parameter_fields()?;
    1526
    0
            Ok(parameter_fields
    1527
    0
                .drain()
    1528
    0
                .map(|(name, maybe_field)| {
    1529
    0
                    (name, maybe_field.map(|field| field.data_type().clone()))
    1530
    0
                })
    1531
    0
                .collect())
    1532
    0
        }
    1533
    1534
        /// Walk the logical plan, find any `Placeholder` tokens, and return a map of their IDs and FieldRefs
    1535
    0
        pub fn get_parameter_fields(
    1536
    0
            &self,
    1537
    0
        ) -> Result<HashMap<String, Option<FieldRef>>, DataFusionError> {
    1538
    0
            let mut param_types: HashMap<String, Option<FieldRef>> = HashMap::new();
    1539
    1540
    0
            self.apply_with_subqueries(|plan| {
    1541
    0
                plan.apply_expressions(|expr| {
    1542
    0
                    expr.apply(|expr| {
    1543
    0
                        if let Expr::Placeholder(Placeholder { id, field }) = expr {
    1544
    0
                            let prev = param_types.get(id);
    1545
    0
                            match (prev, field) {
    1546
    0
                                (Some(Some(prev)), Some(field)) => {
    1547
    0
                                    check_metadata_with_storage_equal(
    1548
    0
                                        (field.data_type(), Some(field.metadata())),
    1549
    0
                                        (prev.data_type(), Some(prev.metadata())),
    1550
    0
                                        "parameter",
    1551
    0
                                        &format!(": Conflicting types for id {id}"),
    1552
    0
                                    )?;
    1553
                                }
    1554
    0
                                (_, Some(field)) => {
    1555
    0
                                    param_types.insert(id.clone(), Some(Arc::clone(field)));
    1556
    0
                                }
    1557
    0
                                _ => {
    1558
    0
                                    param_types.insert(id.clone(), None);
    1559
    0
                                }
    1560
                            }
    1561
    0
                        }
    1562
    0
                        Ok(TreeNodeRecursion::Continue)
    1563
    0
                    })
    1564
    0
                })
    1565
    0
            })
    1566
    0
            .map(|_| param_types)
    1567
    0
        }
    1568
    1569
        // ------------
    1570
        // Various implementations for printing out LogicalPlans
    1571
        // ------------
    1572
    1573
        /// Return a `format`able structure that produces a single line
    1574
        /// per node.
    1575
        ///
    1576
        /// # Example
    1577
        ///
    1578
        /// ```text
    1579
        /// Projection: employee.id
    1580
        ///    Filter: employee.state Eq Utf8(\"CO\")\
    1581
        ///       CsvScan: employee projection=Some([0, 3])
    1582
        /// ```
    1583
        ///
    1584
        /// ```
    1585
        /// use arrow::datatypes::{DataType, Field, Schema};
    1586
        /// use datafusion_expr::{col, lit, logical_plan::table_scan, LogicalPlanBuilder};
    1587
        /// let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    1588
        /// let plan = table_scan(Some("t1"), &schema, None)
    1589
        ///     .unwrap()
    1590
        ///     .filter(col("id").eq(lit(5)))
    1591
        ///     .unwrap()
    1592
        ///     .build()
    1593
        ///     .unwrap();
    1594
        ///
    1595
        /// // Format using display_indent
    1596
        /// let display_string = format!("{}", plan.display_indent());
    1597
        ///
    1598
        /// assert_eq!("Filter: t1.id = Int32(5)\n  TableScan: t1", display_string);
    1599
        /// ```
    1600
    0
        pub fn display_indent(&self) -> impl Display + '_ {
    1601
            // Boilerplate structure to wrap LogicalPlan with something
    1602
            // that that can be formatted
    1603
            struct Wrapper<'a>(&'a LogicalPlan);
    1604
            impl Display for Wrapper<'_> {
    1605
    0
                fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    1606
    0
                    let with_schema = false;
    1607
    0
                    let mut visitor = IndentVisitor::new(f, with_schema);
    1608
    0
                    match self.0.visit_with_subqueries(&mut visitor) {
    1609
    0
                        Ok(_) => Ok(()),
    1610
    0
                        Err(_) => Err(fmt::Error),
    1611
                    }
    1612
    0
                }
    1613
            }
    1614
    0
            Wrapper(self)
    1615
    0
        }
    1616
    1617
        /// Return a `format`able structure that produces a single line
    1618
        /// per node that includes the output schema. For example:
    1619
        ///
    1620
        /// ```text
    1621
        /// Projection: employee.id [id:Int32]\
    1622
        ///    Filter: employee.state = Utf8(\"CO\") [id:Int32, state:Utf8]\
    1623
        ///      TableScan: employee projection=[0, 3] [id:Int32, state:Utf8]";
    1624
        /// ```
    1625
        ///
    1626
        /// ```
    1627
        /// use arrow::datatypes::{DataType, Field, Schema};
    1628
        /// use datafusion_expr::{col, lit, logical_plan::table_scan, LogicalPlanBuilder};
    1629
        /// let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    1630
        /// let plan = table_scan(Some("t1"), &schema, None)
    1631
        ///     .unwrap()
    1632
        ///     .filter(col("id").eq(lit(5)))
    1633
        ///     .unwrap()
    1634
        ///     .build()
    1635
        ///     .unwrap();
    1636
        ///
    1637
        /// // Format using display_indent_schema
    1638
        /// let display_string = format!("{}", plan.display_indent_schema());
    1639
        ///
    1640
        /// assert_eq!(
    1641
        ///     "Filter: t1.id = Int32(5) [id:Int32]\
    1642
        ///             \n  TableScan: t1 [id:Int32]",
    1643
        ///     display_string
    1644
        /// );
    1645
        /// ```
    1646
    0
        pub fn display_indent_schema(&self) -> impl Display + '_ {
    1647
            // Boilerplate structure to wrap LogicalPlan with something
    1648
            // that that can be formatted
    1649
            struct Wrapper<'a>(&'a LogicalPlan);
    1650
            impl Display for Wrapper<'_> {
    1651
    0
                fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    1652
    0
                    let with_schema = true;
    1653
    0
                    let mut visitor = IndentVisitor::new(f, with_schema);
    1654
    0
                    match self.0.visit_with_subqueries(&mut visitor) {
    1655
    0
                        Ok(_) => Ok(()),
    1656
    0
                        Err(_) => Err(fmt::Error),
    1657
                    }
    1658
    0
                }
    1659
            }
    1660
    0
            Wrapper(self)
    1661
    0
        }
    1662
    1663
        /// Return a displayable structure that produces plan in postgresql JSON format.
    1664
        ///
    1665
        /// Users can use this format to visualize the plan in existing plan visualization tools, for example [dalibo](https://explain.dalibo.com/)
    1666
    0
        pub fn display_pg_json(&self) -> impl Display + '_ {
    1667
            // Boilerplate structure to wrap LogicalPlan with something
    1668
            // that that can be formatted
    1669
            struct Wrapper<'a>(&'a LogicalPlan);
    1670
            impl Display for Wrapper<'_> {
    1671
    0
                fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    1672
    0
                    let mut visitor = PgJsonVisitor::new(f);
    1673
    0
                    visitor.with_schema(true);
    1674
    0
                    match self.0.visit_with_subqueries(&mut visitor) {
    1675
    0
                        Ok(_) => Ok(()),
    1676
    0
                        Err(_) => Err(fmt::Error),
    1677
                    }
    1678
    0
                }
    1679
            }
    1680
    0
            Wrapper(self)
    1681
    0
        }
    1682
    1683
        /// Return a `format`able structure that produces lines meant for
    1684
        /// graphical display using the `DOT` language. This format can be
    1685
        /// visualized using software from
    1686
        /// [`graphviz`](https://graphviz.org/)
    1687
        ///
    1688
        /// This currently produces two graphs -- one with the basic
    1689
        /// structure, and one with additional details such as schema.
    1690
        ///
    1691
        /// ```
    1692
        /// use arrow::datatypes::{DataType, Field, Schema};
    1693
        /// use datafusion_expr::{col, lit, logical_plan::table_scan, LogicalPlanBuilder};
    1694
        /// let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    1695
        /// let plan = table_scan(Some("t1"), &schema, None)
    1696
        ///     .unwrap()
    1697
        ///     .filter(col("id").eq(lit(5)))
    1698
        ///     .unwrap()
    1699
        ///     .build()
    1700
        ///     .unwrap();
    1701
        ///
    1702
        /// // Format using display_graphviz
    1703
        /// let graphviz_string = format!("{}", plan.display_graphviz());
    1704
        /// ```
    1705
        ///
    1706
        /// If graphviz string is saved to a file such as `/tmp/example.dot`, the following
    1707
        /// commands can be used to render it as a pdf:
    1708
        ///
    1709
        /// ```bash
    1710
        ///   dot -Tpdf < /tmp/example.dot  > /tmp/example.pdf
    1711
        /// ```
    1712
    0
        pub fn display_graphviz(&self) -> impl Display + '_ {
    1713
            // Boilerplate structure to wrap LogicalPlan with something
    1714
            // that that can be formatted
    1715
            struct Wrapper<'a>(&'a LogicalPlan);
    1716
            impl Display for Wrapper<'_> {
    1717
    0
                fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    1718
    0
                    let mut visitor = GraphvizVisitor::new(f);
    1719
    1720
    0
                    visitor.start_graph()?;
    1721
    1722
    0
                    visitor.pre_visit_plan("LogicalPlan")?;
    1723
    0
                    self.0
    1724
    0
                        .visit_with_subqueries(&mut visitor)
    1725
    0
                        .map_err(|_| fmt::Error)?;
    1726
    0
                    visitor.post_visit_plan()?;
    1727
    1728
    0
                    visitor.set_with_schema(true);
    1729
    0
                    visitor.pre_visit_plan("Detailed LogicalPlan")?;
    1730
    0
                    self.0
    1731
    0
                        .visit_with_subqueries(&mut visitor)
    1732
    0
                        .map_err(|_| fmt::Error)?;
    1733
    0
                    visitor.post_visit_plan()?;
    1734
    1735
    0
                    visitor.end_graph()?;
    1736
    0
                    Ok(())
    1737
    0
                }
    1738
            }
    1739
    0
            Wrapper(self)
    1740
    0
        }
    1741
    1742
        /// Return a `format`able structure with the a human readable
    1743
        /// description of this LogicalPlan node per node, not including
    1744
        /// children. For example:
    1745
        ///
    1746
        /// ```text
    1747
        /// Projection: id
    1748
        /// ```
    1749
        /// ```
    1750
        /// use arrow::datatypes::{DataType, Field, Schema};
    1751
        /// use datafusion_expr::{col, lit, logical_plan::table_scan, LogicalPlanBuilder};
    1752
        /// let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    1753
        /// let plan = table_scan(Some("t1"), &schema, None)
    1754
        ///     .unwrap()
    1755
        ///     .build()
    1756
        ///     .unwrap();
    1757
        ///
    1758
        /// // Format using display
    1759
        /// let display_string = format!("{}", plan.display());
    1760
        ///
    1761
        /// assert_eq!("TableScan: t1", display_string);
    1762
        /// ```
    1763
    0
        pub fn display(&self) -> impl Display + '_ {
    1764
            // Boilerplate structure to wrap LogicalPlan with something
    1765
            // that that can be formatted
    1766
            struct Wrapper<'a>(&'a LogicalPlan);
    1767
            impl Display for Wrapper<'_> {
    1768
    0
                fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    1769
    0
                    match self.0 {
    1770
    0
                        LogicalPlan::EmptyRelation(EmptyRelation { produce_one_row, schema: _ }) => {
    1771
    0
                            let rows = if *produce_one_row { 1 } else { 0 };
    1772
    0
                            write!(f, "EmptyRelation: rows={rows}")
    1773
                        },
    1774
                        LogicalPlan::RecursiveQuery(RecursiveQuery {
    1775
    0
                            is_distinct, ..
    1776
                        }) => {
    1777
    0
                            write!(f, "RecursiveQuery: is_distinct={is_distinct}")
    1778
                        }
    1779
    0
                        LogicalPlan::Values(Values { ref values, .. }) => {
    1780
    0
                            let str_values: Vec<_> = values
    1781
    0
                                .iter()
    1782
                                // limit to only 5 values to avoid horrible display
    1783
    0
                                .take(5)
    1784
    0
                                .map(|row| {
    1785
    0
                                    let item = row
    1786
    0
                                        .iter()
    1787
    0
                                        .map(|expr| expr.to_string())
    1788
    0
                                        .collect::<Vec<_>>()
    1789
    0
                                        .join(", ");
    1790
    0
                                    format!("({item})")
    1791
    0
                                })
    1792
    0
                                .collect();
    1793
    1794
    0
                            let eclipse = if values.len() > 5 { "..." } else { "" };
    1795
    0
                            write!(f, "Values: {}{}", str_values.join(", "), eclipse)
    1796
                        }
    1797
    1798
                        LogicalPlan::TableScan(TableScan {
    1799
    0
                            ref source,
    1800
    0
                            ref table_name,
    1801
    0
                            ref projection,
    1802
    0
                            ref filters,
    1803
    0
                            ref fetch,
    1804
                            ..
    1805
                        }) => {
    1806
    0
                            let projected_fields = match projection {
    1807
    0
                                Some(indices) => {
    1808
    0
                                    let schema = source.schema();
    1809
    0
                                    let names: Vec<&str> = indices
    1810
    0
                                        .iter()
    1811
    0
                                        .map(|i| schema.field(*i).name().as_str())
    1812
    0
                                        .collect();
    1813
    0
                                    format!(" projection=[{}]", names.join(", "))
    1814
                                }
    1815
    0
                                _ => "".to_string(),
    1816
                            };
    1817
    1818
    0
                            write!(f, "TableScan: {table_name}{projected_fields}")?;
    1819
    1820
    0
                            if !filters.is_empty() {
    1821
    0
                                let mut full_filter = vec![];
    1822
    0
                                let mut partial_filter = vec![];
    1823
    0
                                let mut unsupported_filters = vec![];
    1824
    0
                                let filters: Vec<&Expr> = filters.iter().collect();
    1825
    1826
    0
                                if let Ok(results) =
    1827
    0
                                    source.supports_filters_pushdown(&filters)
    1828
                                {
    1829
    0
                                    filters.iter().zip(results.iter()).for_each(
    1830
    0
                                        |(x, res)| match res {
    1831
                                            TableProviderFilterPushDown::Exact => {
    1832
    0
                                                full_filter.push(x)
    1833
                                            }
    1834
                                            TableProviderFilterPushDown::Inexact => {
    1835
    0
                                                partial_filter.push(x)
    1836
                                            }
    1837
                                            TableProviderFilterPushDown::Unsupported => {
    1838
    0
                                                unsupported_filters.push(x)
    1839
                                            }
    1840
    0
                                        },
    1841
                                    );
    1842
    0
                                }
    1843
    1844
    0
                                if !full_filter.is_empty() {
    1845
    0
                                    write!(
    1846
    0
                                        f,
    1847
    0
                                        ", full_filters=[{}]",
    1848
    0
                                        expr_vec_fmt!(full_filter)
    1849
    0
                                    )?;
    1850
    0
                                };
    1851
    0
                                if !partial_filter.is_empty() {
    1852
    0
                                    write!(
    1853
    0
                                        f,
    1854
    0
                                        ", partial_filters=[{}]",
    1855
    0
                                        expr_vec_fmt!(partial_filter)
    1856
    0
                                    )?;
    1857
    0
                                }
    1858
    0
                                if !unsupported_filters.is_empty() {
    1859
    0
                                    write!(
    1860
    0
                                        f,
    1861
    0
                                        ", unsupported_filters=[{}]",
    1862
    0
                                        expr_vec_fmt!(unsupported_filters)
    1863
    0
                                    )?;
    1864
    0
                                }
    1865
    0
                            }
    1866
    1867
    0
                            if let Some(n) = fetch {
    1868
    0
                                write!(f, ", fetch={n}")?;
    1869
    0
                            }
    1870
    1871
    0
                            Ok(())
    1872
                        }
    1873
    0
                        LogicalPlan::Projection(Projection { ref expr, .. }) => {
    1874
    0
                            write!(f, "Projection:")?;
    1875
    0
                            for (i, expr_item) in expr.iter().enumerate() {
    1876
    0
                                if i > 0 {
    1877
    0
                                    write!(f, ",")?;
    1878
    0
                                }
    1879
    0
                                write!(f, " {expr_item}")?;
    1880
                            }
    1881
    0
                            Ok(())
    1882
                        }
    1883
    0
                        LogicalPlan::Dml(DmlStatement { table_name, op, .. }) => {
    1884
    0
                            write!(f, "Dml: op=[{op}] table=[{table_name}]")
    1885
                        }
    1886
                        LogicalPlan::Copy(CopyTo {
    1887
                            input: _,
    1888
    0
                            output_url,
    1889
    0
                            file_type,
    1890
    0
                            options,
    1891
                            ..
    1892
                        }) => {
    1893
    0
                            let op_str = options
    1894
    0
                                .iter()
    1895
    0
                                .map(|(k, v)| format!("{k} {v}"))
    1896
    0
                                .collect::<Vec<String>>()
    1897
    0
                                .join(", ");
    1898
    1899
    0
                            write!(f, "CopyTo: format={} output_url={output_url} options: ({op_str})", file_type.get_ext())
    1900
                        }
    1901
    0
                        LogicalPlan::Ddl(ddl) => {
    1902
    0
                            write!(f, "{}", ddl.display())
    1903
                        }
    1904
                        LogicalPlan::Filter(Filter {
    1905
    0
                            predicate: ref expr,
    1906
                            ..
    1907
    0
                        }) => write!(f, "Filter: {expr}"),
    1908
                        LogicalPlan::Window(Window {
    1909
    0
                            ref window_expr, ..
    1910
                        }) => {
    1911
    0
                            write!(
    1912
    0
                                f,
    1913
    0
                                "WindowAggr: windowExpr=[[{}]]",
    1914
    0
                                expr_vec_fmt!(window_expr)
    1915
                            )
    1916
                        }
    1917
                        LogicalPlan::Aggregate(Aggregate {
    1918
    0
                            ref group_expr,
    1919
    0
                            ref aggr_expr,
    1920
                            ..
    1921
    0
                        }) => write!(
    1922
    0
                            f,
    1923
    0
                            "Aggregate: groupBy=[[{}]], aggr=[[{}]]",
    1924
    0
                            expr_vec_fmt!(group_expr),
    1925
    0
                            expr_vec_fmt!(aggr_expr)
    1926
                        ),
    1927
    0
                        LogicalPlan::Sort(Sort { expr, fetch, .. }) => {
    1928
    0
                            write!(f, "Sort: ")?;
    1929
    0
                            for (i, expr_item) in expr.iter().enumerate() {
    1930
    0
                                if i > 0 {
    1931
    0
                                    write!(f, ", ")?;
    1932
    0
                                }
    1933
    0
                                write!(f, "{expr_item}")?;
    1934
                            }
    1935
    0
                            if let Some(a) = fetch {
    1936
    0
                                write!(f, ", fetch={a}")?;
    1937
    0
                            }
    1938
    1939
    0
                            Ok(())
    1940
                        }
    1941
                        LogicalPlan::Join(Join {
    1942
    0
                            on: ref keys,
    1943
    0
                            filter,
    1944
    0
                            join_constraint,
    1945
    0
                            join_type,
    1946
                            ..
    1947
                        }) => {
    1948
    0
                            let join_expr: Vec<String> =
    1949
    0
                                keys.iter().map(|(l, r)| format!("{l} = {r}")).collect();
    1950
    0
                            let filter_expr = filter
    1951
    0
                                .as_ref()
    1952
    0
                                .map(|expr| format!(" Filter: {expr}"))
    1953
    0
                                .unwrap_or_else(|| "".to_string());
    1954
    0
                            let join_type = if filter.is_none() && keys.is_empty() && matches!(join_type, JoinType::Inner) {
    1955
    0
                                "Cross".to_string()
    1956
                            } else {
    1957
    0
                                join_type.to_string()
    1958
                            };
    1959
    0
                            match join_constraint {
    1960
                                JoinConstraint::On => {
    1961
    0
                                    write!(
    1962
    0
                                        f,
    1963
    0
                                        "{} Join: {}{}",
    1964
                                        join_type,
    1965
    0
                                        join_expr.join(", "),
    1966
                                        filter_expr
    1967
                                    )
    1968
                                }
    1969
                                JoinConstraint::Using => {
    1970
    0
                                    write!(
    1971
    0
                                        f,
    1972
    0
                                        "{} Join: Using {}{}",
    1973
                                        join_type,
    1974
    0
                                        join_expr.join(", "),
    1975
                                        filter_expr,
    1976
                                    )
    1977
                                }
    1978
                            }
    1979
                        }
    1980
                        LogicalPlan::Repartition(Repartition {
    1981
    0
                            partitioning_scheme,
    1982
                            ..
    1983
    0
                        }) => match partitioning_scheme {
    1984
    0
                            Partitioning::RoundRobinBatch(n) => {
    1985
    0
                                write!(f, "Repartition: RoundRobinBatch partition_count={n}")
    1986
                            }
    1987
    0
                            Partitioning::Hash(expr, n) => {
    1988
    0
                                let hash_expr: Vec<String> =
    1989
    0
                                    expr.iter().map(|e| format!("{e}")).collect();
    1990
    0
                                write!(
    1991
    0
                                    f,
    1992
    0
                                    "Repartition: Hash({}) partition_count={}",
    1993
    0
                                    hash_expr.join(", "),
    1994
                                    n
    1995
                                )
    1996
                            }
    1997
    0
                            Partitioning::DistributeBy(expr) => {
    1998
    0
                                let dist_by_expr: Vec<String> =
    1999
    0
                                    expr.iter().map(|e| format!("{e}")).collect();
    2000
    0
                                write!(
    2001
    0
                                    f,
    2002
    0
                                    "Repartition: DistributeBy({})",
    2003
    0
                                    dist_by_expr.join(", "),
    2004
                                )
    2005
                            }
    2006
                        },
    2007
    0
                        LogicalPlan::Limit(limit) => {
    2008
                            // Attempt to display `skip` and `fetch` as literals if possible, otherwise as expressions.
    2009
    0
                            let skip_str = match limit.get_skip_type() {
    2010
    0
                                Ok(SkipType::Literal(n)) => n.to_string(),
    2011
    0
                                _ => limit.skip.as_ref().map_or_else(|| "None".to_string(), |x| x.to_string()),
    2012
                            };
    2013
    0
                            let fetch_str = match limit.get_fetch_type() {
    2014
    0
                                Ok(FetchType::Literal(Some(n))) => n.to_string(),
    2015
    0
                                Ok(FetchType::Literal(None)) => "None".to_string(),
    2016
    0
                                _ => limit.fetch.as_ref().map_or_else(|| "None".to_string(), |x| x.to_string())
    2017
                            };
    2018
    0
                            write!(
    2019
    0
                                f,
    2020
    0
                                "Limit: skip={skip_str}, fetch={fetch_str}",
    2021
                            )
    2022
                        }
    2023
                        LogicalPlan::Subquery(Subquery { .. }) => {
    2024
    0
                            write!(f, "Subquery:")
    2025
                        }
    2026
    0
                        LogicalPlan::SubqueryAlias(SubqueryAlias { ref alias, .. }) => {
    2027
    0
                            write!(f, "SubqueryAlias: {alias}")
    2028
                        }
    2029
    0
                        LogicalPlan::Statement(statement) => {
    2030
    0
                            write!(f, "{}", statement.display())
    2031
                        }
    2032
    0
                        LogicalPlan::Distinct(distinct) => match distinct {
    2033
    0
                            Distinct::All(_) => write!(f, "Distinct:"),
    2034
                            Distinct::On(DistinctOn {
    2035
    0
                                on_expr,
    2036
    0
                                select_expr,
    2037
    0
                                sort_expr,
    2038
                                ..
    2039
    0
                            }) => write!(
    2040
    0
                                f,
    2041
    0
                                "DistinctOn: on_expr=[[{}]], select_expr=[[{}]], sort_expr=[[{}]]",
    2042
    0
                                expr_vec_fmt!(on_expr),
    2043
    0
                                expr_vec_fmt!(select_expr),
    2044
    0
                                if let Some(sort_expr) = sort_expr { expr_vec_fmt!(sort_expr) } else { "".to_string() },
    2045
                            ),
    2046
                        },
    2047
    0
                        LogicalPlan::Explain { .. } => write!(f, "Explain"),
    2048
    0
                        LogicalPlan::Analyze { .. } => write!(f, "Analyze"),
    2049
    0
                        LogicalPlan::Union(_) => write!(f, "Union"),
    2050
    0
                        LogicalPlan::Extension(e) => e.node.fmt_for_explain(f),
    2051
                        LogicalPlan::DescribeTable(DescribeTable { .. }) => {
    2052
    0
                            write!(f, "DescribeTable")
    2053
                        }
    2054
                        LogicalPlan::Unnest(Unnest {
    2055
    0
                            input: plan,
    2056
    0
                            list_type_columns: list_col_indices,
    2057
    0
                            struct_type_columns: struct_col_indices, .. }) => {
    2058
    0
                            let input_columns = plan.schema().columns();
    2059
    0
                            let list_type_columns = list_col_indices
    2060
    0
                                .iter()
    2061
    0
                                .map(|(i,unnest_info)|
    2062
    0
                                    format!("{}|depth={}", &input_columns[*i].to_string(),
    2063
                                    unnest_info.depth))
    2064
    0
                                .collect::<Vec<String>>();
    2065
    0
                            let struct_type_columns = struct_col_indices
    2066
    0
                                .iter()
    2067
    0
                                .map(|i| &input_columns[*i])
    2068
    0
                                .collect::<Vec<&Column>>();
    2069
                            // get items from input_columns indexed by list_col_indices
    2070
    0
                            write!(f, "Unnest: lists[{}] structs[{}]",
    2071
    0
                            expr_vec_fmt!(list_type_columns),
    2072
    0
                            expr_vec_fmt!(struct_type_columns))
    2073
                        }
    2074
                    }
    2075
    0
                }
    2076
            }
    2077
    0
            Wrapper(self)
    2078
    0
        }
    2079
    }
    2080
    2081
    impl Display for LogicalPlan {
    2082
    0
        fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    2083
    0
            self.display_indent().fmt(f)
    2084
    0
        }
    2085
    }
    2086
    2087
    impl ToStringifiedPlan for LogicalPlan {
    2088
    0
        fn to_stringified(&self, plan_type: PlanType) -> StringifiedPlan {
    2089
    0
            StringifiedPlan::new(plan_type, self.display_indent().to_string())
    2090
    0
        }
    2091
    }
    2092
    2093
    /// Relationship produces 0 or 1 placeholder rows with specified output schema
    2094
    /// In most cases the output schema for `EmptyRelation` would be empty,
    2095
    /// however, it can be non-empty typically for optimizer rules
    2096
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    2097
    pub struct EmptyRelation {
    2098
        /// Whether to produce a placeholder row
    2099
        pub produce_one_row: bool,
    2100
        /// The schema description of the output
    2101
        pub schema: DFSchemaRef,
    2102
    }
    2103
    2104
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    2105
    impl PartialOrd for EmptyRelation {
    2106
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    2107
    0
            self.produce_one_row
    2108
    0
                .partial_cmp(&other.produce_one_row)
    2109
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    2110
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    2111
    0
        }
    2112
    }
    2113
    2114
    /// A variadic query operation, Recursive CTE.
    2115
    ///
    2116
    /// # Recursive Query Evaluation
    2117
    ///
    2118
    /// From the [Postgres Docs]:
    2119
    ///
    2120
    /// 1. Evaluate the non-recursive term. For `UNION` (but not `UNION ALL`),
    2121
    ///    discard duplicate rows. Include all remaining rows in the result of the
    2122
    ///    recursive query, and also place them in a temporary working table.
    2123
    ///
    2124
    /// 2. So long as the working table is not empty, repeat these steps:
    2125
    ///
    2126
    /// * Evaluate the recursive term, substituting the current contents of the
    2127
    ///   working table for the recursive self-reference. For `UNION` (but not `UNION
    2128
    ///   ALL`), discard duplicate rows and rows that duplicate any previous result
    2129
    ///   row. Include all remaining rows in the result of the recursive query, and
    2130
    ///   also place them in a temporary intermediate table.
    2131
    ///
    2132
    /// * Replace the contents of the working table with the contents of the
    2133
    ///   intermediate table, then empty the intermediate table.
    2134
    ///
    2135
    /// [Postgres Docs]: https://www.postgresql.org/docs/current/queries-with.html#QUERIES-WITH-RECURSIVE
    2136
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    2137
    pub struct RecursiveQuery {
    2138
        /// Name of the query
    2139
        pub name: String,
    2140
        /// The static term (initial contents of the working table)
    2141
        pub static_term: Arc<LogicalPlan>,
    2142
        /// The recursive term (evaluated on the contents of the working table until
    2143
        /// it returns an empty set)
    2144
        pub recursive_term: Arc<LogicalPlan>,
    2145
        /// Should the output of the recursive term be deduplicated (`UNION`) or
    2146
        /// not (`UNION ALL`).
    2147
        pub is_distinct: bool,
    2148
    }
    2149
    2150
    /// Values expression. See
    2151
    /// [Postgres VALUES](https://www.postgresql.org/docs/current/queries-values.html)
    2152
    /// documentation for more details.
    2153
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    2154
    pub struct Values {
    2155
        /// The table schema
    2156
        pub schema: DFSchemaRef,
    2157
        /// Values
    2158
        pub values: Vec<Vec<Expr>>,
    2159
    }
    2160
    2161
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    2162
    impl PartialOrd for Values {
    2163
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    2164
    0
            self.values
    2165
    0
                .partial_cmp(&other.values)
    2166
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    2167
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    2168
    0
        }
    2169
    }
    2170
    2171
    /// Evaluates an arbitrary list of expressions (essentially a
    2172
    /// SELECT with an expression list) on its input.
    2173
    #[derive(Clone, PartialEq, Eq, Hash, Debug)]
    2174
    // mark non_exhaustive to encourage use of try_new/new()
    2175
    #[non_exhaustive]
    2176
    pub struct Projection {
    2177
        /// The list of expressions
    2178
        pub expr: Vec<Expr>,
    2179
        /// The incoming logical plan
    2180
        pub input: Arc<LogicalPlan>,
    2181
        /// The schema description of the output
    2182
        pub schema: DFSchemaRef,
    2183
    }
    2184
    2185
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    2186
    impl PartialOrd for Projection {
    2187
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    2188
    0
            match self.expr.partial_cmp(&other.expr) {
    2189
    0
                Some(Ordering::Equal) => self.input.partial_cmp(&other.input),
    2190
    0
                cmp => cmp,
    2191
            }
    2192
            // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    2193
    0
            .filter(|cmp| *cmp != Ordering::Equal || self == other)
    2194
    0
        }
    2195
    }
    2196
    2197
    impl Projection {
    2198
        /// Create a new Projection
    2199
    0
        pub fn try_new(expr: Vec<Expr>, input: Arc<LogicalPlan>) -> Result<Self> {
    2200
    0
            let projection_schema = projection_schema(&input, &expr)?;
    2201
    0
            Self::try_new_with_schema(expr, input, projection_schema)
    2202
    0
        }
    2203
    2204
        /// Create a new Projection using the specified output schema
    2205
    0
        pub fn try_new_with_schema(
    2206
    0
            expr: Vec<Expr>,
    2207
    0
            input: Arc<LogicalPlan>,
    2208
    0
            schema: DFSchemaRef,
    2209
    0
        ) -> Result<Self> {
    2210
            #[expect(deprecated)]
    2211
    0
            if !expr.iter().any(|e| matches!(e, Expr::Wildcard { .. }))
    2212
    0
                && expr.len() != schema.fields().len()
    2213
            {
    2214
    0
                return plan_err!("Projection has mismatch between number of expressions ({}) and number of fields in schema ({})", expr.len(), schema.fields().len());
    2215
    0
            }
    2216
    0
            Ok(Self {
    2217
    0
                expr,
    2218
    0
                input,
    2219
    0
                schema,
    2220
    0
            })
    2221
    0
        }
    2222
    2223
        /// Create a new Projection using the specified output schema
    2224
    0
        pub fn new_from_schema(input: Arc<LogicalPlan>, schema: DFSchemaRef) -> Self {
    2225
    0
            let expr: Vec<Expr> = schema.columns().into_iter().map(Expr::Column).collect();
    2226
    0
            Self {
    2227
    0
                expr,
    2228
    0
                input,
    2229
    0
                schema,
    2230
    0
            }
    2231
    0
        }
    2232
    }
    2233
    2234
    /// Computes the schema of the result produced by applying a projection to the input logical plan.
    2235
    ///
    2236
    /// # Arguments
    2237
    ///
    2238
    /// * `input`: A reference to the input `LogicalPlan` for which the projection schema
    2239
    ///   will be computed.
    2240
    /// * `exprs`: A slice of `Expr` expressions representing the projection operation to apply.
    2241
    ///
    2242
    /// # Metadata Handling
    2243
    ///
    2244
    /// - **Schema-level metadata**: Passed through unchanged from the input schema
    2245
    /// - **Field-level metadata**: Determined by each expression via [`exprlist_to_fields`], which
    2246
    ///   calls [`Expr::to_field`] to handle expression-specific metadata (literals, aliases, etc.)
    2247
    ///
    2248
    /// # Returns
    2249
    ///
    2250
    /// A `Result` containing an `Arc<DFSchema>` representing the schema of the result
    2251
    /// produced by the projection operation. If the schema computation is successful,
    2252
    /// the `Result` will contain the schema; otherwise, it will contain an error.
    2253
    0
    pub fn projection_schema(input: &LogicalPlan, exprs: &[Expr]) -> Result<Arc<DFSchema>> {
    2254
        // Preserve input schema metadata at the schema level
    2255
    0
        let metadata = input.schema().metadata().clone();
    2256
    2257
        // Convert expressions to fields with Field properties determined by `Expr::to_field`
    2258
    0
        let schema =
    2259
    0
            DFSchema::new_with_metadata(exprlist_to_fields(exprs, input)?, metadata)?
    2260
    0
                .with_functional_dependencies(calc_func_dependencies_for_project(
    2261
    0
                    exprs, input,
    2262
    0
                )?)?;
    2263
    2264
    0
        Ok(Arc::new(schema))
    2265
    0
    }
    2266
    2267
    /// Aliased subquery
    2268
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    2269
    // mark non_exhaustive to encourage use of try_new/new()
    2270
    #[non_exhaustive]
    2271
    pub struct SubqueryAlias {
    2272
        /// The incoming logical plan
    2273
        pub input: Arc<LogicalPlan>,
    2274
        /// The alias for the input relation
    2275
        pub alias: TableReference,
    2276
        /// The schema with qualified field names
    2277
        pub schema: DFSchemaRef,
    2278
    }
    2279
    2280
    impl SubqueryAlias {
    2281
    0
        pub fn try_new(
    2282
    0
            plan: Arc<LogicalPlan>,
    2283
    0
            alias: impl Into<TableReference>,
    2284
    0
        ) -> Result<Self> {
    2285
    0
            let alias = alias.into();
    2286
    2287
            // Since SubqueryAlias will replace all field qualification for the output schema of `plan`,
    2288
            // no field must share the same column name as this would lead to ambiguity when referencing
    2289
            // columns in parent logical nodes.
    2290
    2291
            // Compute unique aliases, if any, for each column of the input's schema.
    2292
    0
            let aliases = unique_field_aliases(plan.schema().fields());
    2293
    0
            let is_projection_needed = aliases.iter().any(Option::is_some);
    2294
    2295
            // Insert a projection node, if needed, to make sure aliases are applied.
    2296
    0
            let plan = if is_projection_needed {
    2297
    0
                let projection_expressions = aliases
    2298
    0
                    .iter()
    2299
    0
                    .zip(plan.schema().iter())
    2300
    0
                    .map(|(alias, (qualifier, field))| {
    2301
    0
                        let column =
    2302
    0
                            Expr::Column(Column::new(qualifier.cloned(), field.name()));
    2303
    0
                        match alias {
    2304
    0
                            None => column,
    2305
    0
                            Some(alias) => {
    2306
    0
                                Expr::Alias(Alias::new(column, qualifier.cloned(), alias))
    2307
                            }
    2308
                        }
    2309
    0
                    })
    2310
    0
                    .collect();
    2311
    0
                let projection = Projection::try_new(projection_expressions, plan)?;
    2312
    0
                Arc::new(LogicalPlan::Projection(projection))
    2313
            } else {
    2314
    0
                plan
    2315
            };
    2316
    2317
            // Requalify fields with the new `alias`.
    2318
    0
            let fields = plan.schema().fields().clone();
    2319
    0
            let meta_data = plan.schema().metadata().clone();
    2320
    0
            let func_dependencies = plan.schema().functional_dependencies().clone();
    2321
    2322
    0
            let schema = DFSchema::from_unqualified_fields(fields, meta_data)?;
    2323
    0
            let schema = schema.as_arrow();
    2324
    2325
    0
            let schema = DFSchemaRef::new(
    2326
    0
                DFSchema::try_from_qualified_schema(alias.clone(), schema)?
    2327
    0
                    .with_functional_dependencies(func_dependencies)?,
    2328
            );
    2329
    0
            Ok(SubqueryAlias {
    2330
    0
                input: plan,
    2331
    0
                alias,
    2332
    0
                schema,
    2333
    0
            })
    2334
    0
        }
    2335
    }
    2336
    2337
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    2338
    impl PartialOrd for SubqueryAlias {
    2339
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    2340
    0
            match self.input.partial_cmp(&other.input) {
    2341
    0
                Some(Ordering::Equal) => self.alias.partial_cmp(&other.alias),
    2342
    0
                cmp => cmp,
    2343
            }
    2344
            // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    2345
    0
            .filter(|cmp| *cmp != Ordering::Equal || self == other)
    2346
    0
        }
    2347
    }
    2348
    2349
    /// Filters rows from its input that do not match an
    2350
    /// expression (essentially a WHERE clause with a predicate
    2351
    /// expression).
    2352
    ///
    2353
    /// Semantically, `<predicate>` is evaluated for each row of the input;
    2354
    /// If the value of `<predicate>` is true, the input row is passed to
    2355
    /// the output. If the value of `<predicate>` is false, the row is
    2356
    /// discarded.
    2357
    ///
    2358
    /// Filter should not be created directly but instead use `try_new()`
    2359
    /// and that these fields are only pub to support pattern matching
    2360
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    2361
    #[non_exhaustive]
    2362
    pub struct Filter {
    2363
        /// The predicate expression, which must have Boolean type.
    2364
        pub predicate: Expr,
    2365
        /// The incoming logical plan
    2366
        pub input: Arc<LogicalPlan>,
    2367
    }
    2368
    2369
    impl Filter {
    2370
        /// Create a new filter operator.
    2371
        ///
    2372
        /// Notes: as Aliases have no effect on the output of a filter operator,
    2373
        /// they are removed from the predicate expression.
    2374
    0
        pub fn try_new(predicate: Expr, input: Arc<LogicalPlan>) -> Result<Self> {
    2375
    0
            Self::try_new_internal(predicate, input)
    2376
    0
        }
    2377
    2378
        /// Create a new filter operator for a having clause.
    2379
        /// This is similar to a filter, but its having flag is set to true.
    2380
        #[deprecated(since = "48.0.0", note = "Use `try_new` instead")]
    2381
    0
        pub fn try_new_with_having(predicate: Expr, input: Arc<LogicalPlan>) -> Result<Self> {
    2382
    0
            Self::try_new_internal(predicate, input)
    2383
    0
        }
    2384
    2385
    0
        fn is_allowed_filter_type(data_type: &DataType) -> bool {
    2386
    0
            match data_type {
    2387
                // Interpret NULL as a missing boolean value.
    2388
    0
                DataType::Boolean | DataType::Null => true,
    2389
    0
                DataType::Dictionary(_, value_type) => {
    2390
    0
                    Filter::is_allowed_filter_type(value_type.as_ref())
    2391
                }
    2392
    0
                _ => false,
    2393
            }
    2394
    0
        }
    2395
    2396
    0
        fn try_new_internal(predicate: Expr, input: Arc<LogicalPlan>) -> Result<Self> {
    2397
            // Filter predicates must return a boolean value so we try and validate that here.
    2398
            // Note that it is not always possible to resolve the predicate expression during plan
    2399
            // construction (such as with correlated subqueries) so we make a best effort here and
    2400
            // ignore errors resolving the expression against the schema.
    2401
    0
            if let Ok(predicate_type) = predicate.get_type(input.schema()) {
    2402
    0
                if !Filter::is_allowed_filter_type(&predicate_type) {
    2403
    0
                    return plan_err!(
    2404
    0
                        "Cannot create filter with non-boolean predicate '{predicate}' returning {predicate_type}"
    2405
                    );
    2406
    0
                }
    2407
    0
            }
    2408
    2409
    0
            Ok(Self {
    2410
    0
                predicate: predicate.unalias_nested().data,
    2411
    0
                input,
    2412
    0
            })
    2413
    0
        }
    2414
    2415
        /// Is this filter guaranteed to return 0 or 1 row in a given instantiation?
    2416
        ///
    2417
        /// This function will return `true` if its predicate contains a conjunction of
    2418
        /// `col(a) = <expr>`, where its schema has a unique filter that is covered
    2419
        /// by this conjunction.
    2420
        ///
    2421
        /// For example, for the table:
    2422
        /// ```sql
    2423
        /// CREATE TABLE t (a INTEGER PRIMARY KEY, b INTEGER);
    2424
        /// ```
    2425
        /// `Filter(a = 2).is_scalar() == true`
    2426
        /// , whereas
    2427
        /// `Filter(b = 2).is_scalar() == false`
    2428
        /// and
    2429
        /// `Filter(a = 2 OR b = 2).is_scalar() == false`
    2430
    0
        fn is_scalar(&self) -> bool {
    2431
    0
            let schema = self.input.schema();
    2432
    2433
    0
            let functional_dependencies = self.input.schema().functional_dependencies();
    2434
    0
            let unique_keys = functional_dependencies.iter().filter(|dep| {
    2435
    0
                let nullable = dep.nullable
    2436
    0
                    && dep
    2437
    0
                        .source_indices
    2438
    0
                        .iter()
    2439
    0
                        .any(|&source| schema.field(source).is_nullable());
    2440
    0
                !nullable
    2441
    0
                    && dep.mode == Dependency::Single
    2442
    0
                    && dep.target_indices.len() == schema.fields().len()
    2443
    0
            });
    2444
    2445
    0
            let exprs = split_conjunction(&self.predicate);
    2446
    0
            let eq_pred_cols: HashSet<_> = exprs
    2447
    0
                .iter()
    2448
    0
                .filter_map(|expr| {
    2449
                    let Expr::BinaryExpr(BinaryExpr {
    2450
    0
                        left,
    2451
                        op: Operator::Eq,
    2452
    0
                        right,
    2453
    0
                    }) = expr
    2454
                    else {
    2455
    0
                        return None;
    2456
                    };
    2457
                    // This is a no-op filter expression
    2458
    0
                    if left == right {
    2459
    0
                        return None;
    2460
    0
                    }
    2461
    2462
    0
                    match (left.as_ref(), right.as_ref()) {
    2463
    0
                        (Expr::Column(_), Expr::Column(_)) => None,
    2464
    0
                        (Expr::Column(c), _) | (_, Expr::Column(c)) => {
    2465
    0
                            Some(schema.index_of_column(c).unwrap())
    2466
                        }
    2467
    0
                        _ => None,
    2468
                    }
    2469
    0
                })
    2470
    0
                .collect();
    2471
    2472
            // If we have a functional dependence that is a subset of our predicate,
    2473
            // this filter is scalar
    2474
    0
            for key in unique_keys {
    2475
    0
                if key.source_indices.iter().all(|c| eq_pred_cols.contains(c)) {
    2476
    0
                    return true;
    2477
    0
                }
    2478
            }
    2479
    0
            false
    2480
    0
        }
    2481
    }
    2482
    2483
    /// Window its input based on a set of window spec and window function (e.g. SUM or RANK)
    2484
    ///
    2485
    /// # Output Schema
    2486
    ///
    2487
    /// The output schema is the input schema followed by the window function
    2488
    /// expressions, in order.
    2489
    ///
    2490
    /// For example, given the input schema `"A", "B", "C"` and the window function
    2491
    /// `SUM(A) OVER (PARTITION BY B+1 ORDER BY C)`, the output schema will be `"A",
    2492
    /// "B", "C", "SUM(A) OVER ..."` where `"SUM(A) OVER ..."` is the name of the
    2493
    /// output column.
    2494
    ///
    2495
    /// Note that the `PARTITION BY` expression "B+1" is not produced in the output
    2496
    /// schema.
    2497
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    2498
    pub struct Window {
    2499
        /// The incoming logical plan
    2500
        pub input: Arc<LogicalPlan>,
    2501
        /// The window function expression
    2502
        pub window_expr: Vec<Expr>,
    2503
        /// The schema description of the window output
    2504
        pub schema: DFSchemaRef,
    2505
    }
    2506
    2507
    impl Window {
    2508
        /// Create a new window operator.
    2509
    0
        pub fn try_new(window_expr: Vec<Expr>, input: Arc<LogicalPlan>) -> Result<Self> {
    2510
    0
            let fields: Vec<(Option<TableReference>, Arc<Field>)> = input
    2511
    0
                .schema()
    2512
    0
                .iter()
    2513
    0
                .map(|(q, f)| (q.cloned(), Arc::clone(f)))
    2514
    0
                .collect();
    2515
    0
            let input_len = fields.len();
    2516
    0
            let mut window_fields = fields;
    2517
    0
            let expr_fields = exprlist_to_fields(window_expr.as_slice(), &input)?;
    2518
    0
            window_fields.extend_from_slice(expr_fields.as_slice());
    2519
    0
            let metadata = input.schema().metadata().clone();
    2520
    2521
            // Update functional dependencies for window:
    2522
    0
            let mut window_func_dependencies =
    2523
    0
                input.schema().functional_dependencies().clone();
    2524
    0
            window_func_dependencies.extend_target_indices(window_fields.len());
    2525
    2526
            // Since we know that ROW_NUMBER outputs will be unique (i.e. it consists
    2527
            // of consecutive numbers per partition), we can represent this fact with
    2528
            // functional dependencies.
    2529
    0
            let mut new_dependencies = window_expr
    2530
    0
                .iter()
    2531
    0
                .enumerate()
    2532
    0
                .filter_map(|(idx, expr)| {
    2533
    0
                    let Expr::WindowFunction(window_fun) = expr else {
    2534
    0
                        return None;
    2535
                    };
    2536
                    let WindowFunction {
    2537
    0
                        fun: WindowFunctionDefinition::WindowUDF(udwf),
    2538
    0
                        params: WindowFunctionParams { partition_by, .. },
    2539
    0
                    } = window_fun.as_ref()
    2540
                    else {
    2541
    0
                        return None;
    2542
                    };
    2543
                    // When there is no PARTITION BY, row number will be unique
    2544
                    // across the entire table.
    2545
    0
                    if udwf.name() == "row_number" && partition_by.is_empty() {
    2546
    0
                        Some(idx + input_len)
    2547
                    } else {
    2548
    0
                        None
    2549
                    }
    2550
    0
                })
    2551
    0
                .map(|idx| {
    2552
    0
                    FunctionalDependence::new(vec![idx], vec![], false)
    2553
    0
                        .with_mode(Dependency::Single)
    2554
    0
                })
    2555
    0
                .collect::<Vec<_>>();
    2556
    2557
    0
            if !new_dependencies.is_empty() {
    2558
    0
                for dependence in new_dependencies.iter_mut() {
    2559
    0
                    dependence.target_indices = (0..window_fields.len()).collect();
    2560
    0
                }
    2561
                // Add the dependency introduced because of ROW_NUMBER window function to the functional dependency
    2562
    0
                let new_deps = FunctionalDependencies::new(new_dependencies);
    2563
    0
                window_func_dependencies.extend(new_deps);
    2564
    0
            }
    2565
    2566
            // Validate that FILTER clauses are only used with aggregate window functions
    2567
    0
            if let Some(e) = window_expr.iter().find(|e| {
    2568
    0
                matches!(
    2569
    0
                    e,
    2570
    0
                    Expr::WindowFunction(wf)
    2571
    0
                        if !matches!(wf.fun, WindowFunctionDefinition::AggregateUDF(_))
    2572
    0
                            && wf.params.filter.is_some()
    2573
                )
    2574
    0
            }) {
    2575
    0
                return plan_err!(
    2576
    0
                    "FILTER clause can only be used with aggregate window functions. Found in '{e}'"
    2577
                );
    2578
    0
            }
    2579
    2580
    0
            Self::try_new_with_schema(
    2581
    0
                window_expr,
    2582
    0
                input,
    2583
    0
                Arc::new(
    2584
    0
                    DFSchema::new_with_metadata(window_fields, metadata)?
    2585
    0
                        .with_functional_dependencies(window_func_dependencies)?,
    2586
                ),
    2587
            )
    2588
    0
        }
    2589
    2590
        /// Create a new window function using the provided schema to avoid the overhead of
    2591
        /// building the schema again when the schema is already known.
    2592
        ///
    2593
        /// This method should only be called when you are absolutely sure that the schema being
    2594
        /// provided is correct for the window function. If in doubt, call [try_new](Self::try_new) instead.
    2595
    0
        pub fn try_new_with_schema(
    2596
    0
            window_expr: Vec<Expr>,
    2597
    0
            input: Arc<LogicalPlan>,
    2598
    0
            schema: DFSchemaRef,
    2599
    0
        ) -> Result<Self> {
    2600
    0
            let input_fields_count = input.schema().fields().len();
    2601
    0
            if schema.fields().len() != input_fields_count + window_expr.len() {
    2602
    0
                return plan_err!(
    2603
    0
                    "Window schema has wrong number of fields. Expected {} got {}",
    2604
    0
                    input_fields_count + window_expr.len(),
    2605
    0
                    schema.fields().len()
    2606
                );
    2607
    0
            }
    2608
    2609
    0
            Ok(Window {
    2610
    0
                input,
    2611
    0
                window_expr,
    2612
    0
                schema,
    2613
    0
            })
    2614
    0
        }
    2615
    }
    2616
    2617
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    2618
    impl PartialOrd for Window {
    2619
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    2620
    0
            match self.input.partial_cmp(&other.input)? {
    2621
    0
                Ordering::Equal => {} // continue
    2622
    0
                not_equal => return Some(not_equal),
    2623
            }
    2624
    2625
    0
            match self.window_expr.partial_cmp(&other.window_expr)? {
    2626
    0
                Ordering::Equal => {} // continue
    2627
    0
                not_equal => return Some(not_equal),
    2628
            }
    2629
    2630
            // Contract for PartialOrd and PartialEq consistency requires that
    2631
            // a == b if and only if partial_cmp(a, b) == Some(Equal).
    2632
    0
            if self == other {
    2633
    0
                Some(Ordering::Equal)
    2634
            } else {
    2635
    0
                None
    2636
            }
    2637
    0
        }
    2638
    }
    2639
    2640
    /// Produces rows from a table provider by reference or from the context
    2641
    #[derive(Clone)]
    2642
    pub struct TableScan {
    2643
        /// The name of the table
    2644
        pub table_name: TableReference,
    2645
        /// The source of the table
    2646
        pub source: Arc<dyn TableSource>,
    2647
        /// Optional column indices to use as a projection
    2648
        pub projection: Option<Vec<usize>>,
    2649
        /// The schema description of the output
    2650
        pub projected_schema: DFSchemaRef,
    2651
        /// Optional expressions to be used as filters by the table provider
    2652
        pub filters: Vec<Expr>,
    2653
        /// Optional number of rows to read
    2654
        pub fetch: Option<usize>,
    2655
    }
    2656
    2657
    impl Debug for TableScan {
    2658
    0
        fn fmt(&self, f: &mut Formatter) -> fmt::Result {
    2659
    0
            f.debug_struct("TableScan")
    2660
    0
                .field("table_name", &self.table_name)
    2661
    0
                .field("source", &"...")
    2662
    0
                .field("projection", &self.projection)
    2663
    0
                .field("projected_schema", &self.projected_schema)
    2664
    0
                .field("filters", &self.filters)
    2665
    0
                .field("fetch", &self.fetch)
    2666
    0
                .finish_non_exhaustive()
    2667
    0
        }
    2668
    }
    2669
    2670
    impl PartialEq for TableScan {
    2671
    0
        fn eq(&self, other: &Self) -> bool {
    2672
    0
            self.table_name == other.table_name
    2673
    0
                && self.projection == other.projection
    2674
    0
                && self.projected_schema == other.projected_schema
    2675
    0
                && self.filters == other.filters
    2676
    0
                && self.fetch == other.fetch
    2677
    0
        }
    2678
    }
    2679
    2680
    impl Eq for TableScan {}
    2681
    2682
    // Manual implementation needed because of `source` and `projected_schema` fields.
    2683
    // Comparison excludes these field.
    2684
    impl PartialOrd for TableScan {
    2685
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    2686
            #[derive(PartialEq, PartialOrd)]
    2687
            struct ComparableTableScan<'a> {
    2688
                /// The name of the table
    2689
                pub table_name: &'a TableReference,
    2690
                /// Optional column indices to use as a projection
    2691
                pub projection: &'a Option<Vec<usize>>,
    2692
                /// Optional expressions to be used as filters by the table provider
    2693
                pub filters: &'a Vec<Expr>,
    2694
                /// Optional number of rows to read
    2695
                pub fetch: &'a Option<usize>,
    2696
            }
    2697
    0
            let comparable_self = ComparableTableScan {
    2698
    0
                table_name: &self.table_name,
    2699
    0
                projection: &self.projection,
    2700
    0
                filters: &self.filters,
    2701
    0
                fetch: &self.fetch,
    2702
    0
            };
    2703
    0
            let comparable_other = ComparableTableScan {
    2704
    0
                table_name: &other.table_name,
    2705
    0
                projection: &other.projection,
    2706
    0
                filters: &other.filters,
    2707
    0
                fetch: &other.fetch,
    2708
    0
            };
    2709
    0
            comparable_self
    2710
    0
                .partial_cmp(&comparable_other)
    2711
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    2712
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    2713
    0
        }
    2714
    }
    2715
    2716
    impl Hash for TableScan {
    2717
    0
        fn hash<H: Hasher>(&self, state: &mut H) {
    2718
    0
            self.table_name.hash(state);
    2719
    0
            self.projection.hash(state);
    2720
    0
            self.projected_schema.hash(state);
    2721
    0
            self.filters.hash(state);
    2722
    0
            self.fetch.hash(state);
    2723
    0
        }
    2724
    }
    2725
    2726
    impl TableScan {
    2727
        /// Initialize TableScan with appropriate schema from the given
    2728
        /// arguments.
    2729
    0
        pub fn try_new(
    2730
    0
            table_name: impl Into<TableReference>,
    2731
    0
            table_source: Arc<dyn TableSource>,
    2732
    0
            projection: Option<Vec<usize>>,
    2733
    0
            filters: Vec<Expr>,
    2734
    0
            fetch: Option<usize>,
    2735
    0
        ) -> Result<Self> {
    2736
    0
            let table_name = table_name.into();
    2737
    2738
    0
            if table_name.table().is_empty() {
    2739
    0
                return plan_err!("table_name cannot be empty");
    2740
    0
            }
    2741
    0
            let schema = table_source.schema();
    2742
    0
            let func_dependencies = FunctionalDependencies::new_from_constraints(
    2743
    0
                table_source.constraints(),
    2744
    0
                schema.fields.len(),
    2745
            );
    2746
    0
            let projected_schema = projection
    2747
    0
                .as_ref()
    2748
    0
                .map(|p| {
    2749
    0
                    let projected_func_dependencies =
    2750
    0
                        func_dependencies.project_functional_dependencies(p, p.len());
    2751
    2752
    0
                    let df_schema = DFSchema::new_with_metadata(
    2753
    0
                        p.iter()
    2754
    0
                            .map(|i| {
    2755
    0
                                (Some(table_name.clone()), Arc::clone(&schema.fields()[*i]))
    2756
    0
                            })
    2757
    0
                            .collect(),
    2758
    0
                        schema.metadata.clone(),
    2759
    0
                    )?;
    2760
    0
                    df_schema.with_functional_dependencies(projected_func_dependencies)
    2761
    0
                })
    2762
    0
                .unwrap_or_else(|| {
    2763
    0
                    let df_schema =
    2764
    0
                        DFSchema::try_from_qualified_schema(table_name.clone(), &schema)?;
    2765
    0
                    df_schema.with_functional_dependencies(func_dependencies)
    2766
    0
                })?;
    2767
    0
            let projected_schema = Arc::new(projected_schema);
    2768
    2769
    0
            Ok(Self {
    2770
    0
                table_name,
    2771
    0
                source: table_source,
    2772
    0
                projection,
    2773
    0
                projected_schema,
    2774
    0
                filters,
    2775
    0
                fetch,
    2776
    0
            })
    2777
    0
        }
    2778
    }
    2779
    2780
    // Repartition the plan based on a partitioning scheme.
    2781
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    2782
    pub struct Repartition {
    2783
        /// The incoming logical plan
    2784
        pub input: Arc<LogicalPlan>,
    2785
        /// The partitioning scheme
    2786
        pub partitioning_scheme: Partitioning,
    2787
    }
    2788
    2789
    /// Union multiple inputs
    2790
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    2791
    pub struct Union {
    2792
        /// Inputs to merge
    2793
        pub inputs: Vec<Arc<LogicalPlan>>,
    2794
        /// Union schema. Should be the same for all inputs.
    2795
        pub schema: DFSchemaRef,
    2796
    }
    2797
    2798
    impl Union {
    2799
        /// Constructs new Union instance deriving schema from inputs.
    2800
        /// Schema data types must match exactly.
    2801
    0
        pub fn try_new(inputs: Vec<Arc<LogicalPlan>>) -> Result<Self> {
    2802
    0
            let schema = Self::derive_schema_from_inputs(&inputs, false, false)?;
    2803
    0
            Ok(Union { inputs, schema })
    2804
    0
        }
    2805
    2806
        /// Constructs new Union instance deriving schema from inputs.
    2807
        /// Inputs do not have to have matching types and produced schema will
    2808
        /// take type from the first input.
    2809
        // TODO (https://github.com/apache/datafusion/issues/14380): Avoid creating uncoerced union at all.
    2810
    0
        pub fn try_new_with_loose_types(inputs: Vec<Arc<LogicalPlan>>) -> Result<Self> {
    2811
    0
            let schema = Self::derive_schema_from_inputs(&inputs, true, false)?;
    2812
    0
            Ok(Union { inputs, schema })
    2813
    0
        }
    2814
    2815
        /// Constructs a new Union instance that combines rows from different tables by name,
    2816
        /// instead of by position. This means that the specified inputs need not have schemas
    2817
        /// that are all the same width.
    2818
    0
        pub fn try_new_by_name(inputs: Vec<Arc<LogicalPlan>>) -> Result<Self> {
    2819
    0
            let schema = Self::derive_schema_from_inputs(&inputs, true, true)?;
    2820
    0
            let inputs = Self::rewrite_inputs_from_schema(&schema, inputs)?;
    2821
    2822
    0
            Ok(Union { inputs, schema })
    2823
    0
        }
    2824
    2825
        /// When constructing a `UNION BY NAME`, we need to wrap inputs
    2826
        /// in an additional `Projection` to account for absence of columns
    2827
        /// in input schemas or differing projection orders.
    2828
    0
        fn rewrite_inputs_from_schema(
    2829
    0
            schema: &Arc<DFSchema>,
    2830
    0
            inputs: Vec<Arc<LogicalPlan>>,
    2831
    0
        ) -> Result<Vec<Arc<LogicalPlan>>> {
    2832
    0
            let schema_width = schema.iter().count();
    2833
    0
            let mut wrapped_inputs = Vec::with_capacity(inputs.len());
    2834
    0
            for input in inputs {
    2835
                // Any columns that exist within the derived schema but do not exist
    2836
                // within an input's schema should be replaced with `NULL` aliased
    2837
                // to the appropriate column in the derived schema.
    2838
    0
                let mut expr = Vec::with_capacity(schema_width);
    2839
    0
                for column in schema.columns() {
    2840
    0
                    if input
    2841
    0
                        .schema()
    2842
    0
                        .has_column_with_unqualified_name(column.name())
    2843
    0
                    {
    2844
    0
                        expr.push(Expr::Column(column));
    2845
    0
                    } else {
    2846
    0
                        expr.push(
    2847
    0
                            Expr::Literal(ScalarValue::Null, None).alias(column.name()),
    2848
    0
                        );
    2849
    0
                    }
    2850
                }
    2851
    0
                wrapped_inputs.push(Arc::new(LogicalPlan::Projection(
    2852
    0
                    Projection::try_new_with_schema(expr, input, Arc::clone(schema))?,
    2853
                )));
    2854
            }
    2855
    2856
    0
            Ok(wrapped_inputs)
    2857
    0
        }
    2858
    2859
        /// Constructs new Union instance deriving schema from inputs.
    2860
        ///
    2861
        /// If `loose_types` is true, inputs do not need to have matching types and
    2862
        /// the produced schema will use the type from the first input.
    2863
        /// TODO (<https://github.com/apache/datafusion/issues/14380>): This is not necessarily reasonable behavior.
    2864
        ///
    2865
        /// If `by_name` is `true`, input schemas need not be the same width. That is,
    2866
        /// the constructed schema follows `UNION BY NAME` semantics.
    2867
    0
        fn derive_schema_from_inputs(
    2868
    0
            inputs: &[Arc<LogicalPlan>],
    2869
    0
            loose_types: bool,
    2870
    0
            by_name: bool,
    2871
    0
        ) -> Result<DFSchemaRef> {
    2872
    0
            if inputs.len() < 2 {
    2873
    0
                return plan_err!("UNION requires at least two inputs");
    2874
    0
            }
    2875
    2876
    0
            if by_name {
    2877
    0
                Self::derive_schema_from_inputs_by_name(inputs, loose_types)
    2878
            } else {
    2879
    0
                Self::derive_schema_from_inputs_by_position(inputs, loose_types)
    2880
            }
    2881
    0
        }
    2882
    2883
    0
        fn derive_schema_from_inputs_by_name(
    2884
    0
            inputs: &[Arc<LogicalPlan>],
    2885
    0
            loose_types: bool,
    2886
    0
        ) -> Result<DFSchemaRef> {
    2887
            type FieldData<'a> =
    2888
                (&'a DataType, bool, Vec<&'a HashMap<String, String>>, usize);
    2889
    0
            let mut cols: Vec<(&str, FieldData)> = Vec::new();
    2890
    0
            for input in inputs.iter() {
    2891
    0
                for field in input.schema().fields() {
    2892
    0
                    if let Some((_, (data_type, is_nullable, metadata, occurrences))) =
    2893
    0
                        cols.iter_mut().find(|(name, _)| name == field.name())
    2894
                    {
    2895
    0
                        if !loose_types && *data_type != field.data_type() {
    2896
    0
                            return plan_err!(
    2897
    0
                                "Found different types for field {}",
    2898
    0
                                field.name()
    2899
                            );
    2900
    0
                        }
    2901
    2902
    0
                        metadata.push(field.metadata());
    2903
                        // If the field is nullable in any one of the inputs,
    2904
                        // then the field in the final schema is also nullable.
    2905
    0
                        *is_nullable |= field.is_nullable();
    2906
    0
                        *occurrences += 1;
    2907
    0
                    } else {
    2908
    0
                        cols.push((
    2909
    0
                            field.name(),
    2910
    0
                            (
    2911
    0
                                field.data_type(),
    2912
    0
                                field.is_nullable(),
    2913
    0
                                vec![field.metadata()],
    2914
    0
                                1,
    2915
    0
                            ),
    2916
    0
                        ));
    2917
    0
                    }
    2918
                }
    2919
            }
    2920
    2921
    0
            let union_fields = cols
    2922
    0
                .into_iter()
    2923
    0
                .map(
    2924
    0
                    |(name, (data_type, is_nullable, unmerged_metadata, occurrences))| {
    2925
                        // If the final number of occurrences of the field is less
    2926
                        // than the number of inputs (i.e. the field is missing from
    2927
                        // one or more inputs), then it must be treated as nullable.
    2928
    0
                        let final_is_nullable = if occurrences == inputs.len() {
    2929
    0
                            is_nullable
    2930
                        } else {
    2931
    0
                            true
    2932
                        };
    2933
    2934
    0
                        let mut field =
    2935
    0
                            Field::new(name, data_type.clone(), final_is_nullable);
    2936
    0
                        field.set_metadata(intersect_metadata_for_union(unmerged_metadata));
    2937
    2938
    0
                        (None, Arc::new(field))
    2939
    0
                    },
    2940
                )
    2941
    0
                .collect::<Vec<(Option<TableReference>, _)>>();
    2942
    2943
    0
            let union_schema_metadata = intersect_metadata_for_union(
    2944
    0
                inputs.iter().map(|input| input.schema().metadata()),
    2945
            );
    2946
    2947
            // Functional Dependencies are not preserved after UNION operation
    2948
    0
            let schema = DFSchema::new_with_metadata(union_fields, union_schema_metadata)?;
    2949
    0
            let schema = Arc::new(schema);
    2950
    2951
    0
            Ok(schema)
    2952
    0
        }
    2953
    2954
    0
        fn derive_schema_from_inputs_by_position(
    2955
    0
            inputs: &[Arc<LogicalPlan>],
    2956
    0
            loose_types: bool,
    2957
    0
        ) -> Result<DFSchemaRef> {
    2958
    0
            let first_schema = inputs[0].schema();
    2959
    0
            let fields_count = first_schema.fields().len();
    2960
    0
            for input in inputs.iter().skip(1) {
    2961
    0
                if fields_count != input.schema().fields().len() {
    2962
    0
                    return plan_err!(
    2963
    0
                        "UNION queries have different number of columns: \
    2964
    0
                        left has {} columns whereas right has {} columns",
    2965
                        fields_count,
    2966
    0
                        input.schema().fields().len()
    2967
                    );
    2968
    0
                }
    2969
            }
    2970
    2971
    0
            let mut name_counts: HashMap<String, usize> = HashMap::new();
    2972
    0
            let union_fields = (0..fields_count)
    2973
    0
                .map(|i| {
    2974
    0
                    let fields = inputs
    2975
    0
                        .iter()
    2976
    0
                        .map(|input| input.schema().field(i))
    2977
    0
                        .collect::<Vec<_>>();
    2978
    0
                    let first_field = fields[0];
    2979
    0
                    let base_name = first_field.name().to_string();
    2980
    2981
    0
                    let data_type = if loose_types {
    2982
                        // TODO apply type coercion here, or document why it's better to defer
    2983
                        // temporarily use the data type from the left input and later rely on the analyzer to
    2984
                        // coerce the two schemas into a common one.
    2985
    0
                        first_field.data_type()
    2986
                    } else {
    2987
    0
                        fields.iter().skip(1).try_fold(
    2988
    0
                            first_field.data_type(),
    2989
    0
                            |acc, field| {
    2990
    0
                                if acc != field.data_type() {
    2991
    0
                                    return plan_err!(
    2992
    0
                                        "UNION field {i} have different type in inputs: \
    2993
    0
                                        left has {} whereas right has {}",
    2994
    0
                                        first_field.data_type(),
    2995
    0
                                        field.data_type()
    2996
                                    );
    2997
    0
                                }
    2998
    0
                                Ok(acc)
    2999
    0
                            },
    3000
    0
                        )?
    3001
                    };
    3002
    0
                    let nullable = fields.iter().any(|field| field.is_nullable());
    3003
    3004
                    // Generate unique field name
    3005
    0
                    let name = if let Some(count) = name_counts.get_mut(&base_name) {
    3006
    0
                        *count += 1;
    3007
    0
                        format!("{base_name}_{count}")
    3008
                    } else {
    3009
    0
                        name_counts.insert(base_name.clone(), 0);
    3010
    0
                        base_name
    3011
                    };
    3012
    3013
    0
                    let mut field = Field::new(&name, data_type.clone(), nullable);
    3014
    0
                    let field_metadata = intersect_metadata_for_union(
    3015
    0
                        fields.iter().map(|field| field.metadata()),
    3016
                    );
    3017
    0
                    field.set_metadata(field_metadata);
    3018
    0
                    Ok((None, Arc::new(field)))
    3019
    0
                })
    3020
    0
                .collect::<Result<_>>()?;
    3021
    0
            let union_schema_metadata = intersect_metadata_for_union(
    3022
    0
                inputs.iter().map(|input| input.schema().metadata()),
    3023
            );
    3024
    3025
            // Functional Dependencies are not preserved after UNION operation
    3026
    0
            let schema = DFSchema::new_with_metadata(union_fields, union_schema_metadata)?;
    3027
    0
            let schema = Arc::new(schema);
    3028
    3029
    0
            Ok(schema)
    3030
    0
        }
    3031
    }
    3032
    3033
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    3034
    impl PartialOrd for Union {
    3035
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    3036
    0
            self.inputs
    3037
    0
                .partial_cmp(&other.inputs)
    3038
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    3039
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    3040
    0
        }
    3041
    }
    3042
    3043
    /// Describe the schema of table
    3044
    ///
    3045
    /// # Example output:
    3046
    ///
    3047
    /// ```sql
    3048
    /// > describe traces;
    3049
    /// +--------------------+-----------------------------+-------------+
    3050
    /// | column_name        | data_type                   | is_nullable |
    3051
    /// +--------------------+-----------------------------+-------------+
    3052
    /// | attributes         | Utf8                        | YES         |
    3053
    /// | duration_nano      | Int64                       | YES         |
    3054
    /// | end_time_unix_nano | Int64                       | YES         |
    3055
    /// | service.name       | Dictionary(Int32, Utf8)     | YES         |
    3056
    /// | span.kind          | Utf8                        | YES         |
    3057
    /// | span.name          | Utf8                        | YES         |
    3058
    /// | span_id            | Dictionary(Int32, Utf8)     | YES         |
    3059
    /// | time               | Timestamp(Nanosecond, None) | NO          |
    3060
    /// | trace_id           | Dictionary(Int32, Utf8)     | YES         |
    3061
    /// | otel.status_code   | Utf8                        | YES         |
    3062
    /// | parent_span_id     | Utf8                        | YES         |
    3063
    /// +--------------------+-----------------------------+-------------+
    3064
    /// ```
    3065
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    3066
    pub struct DescribeTable {
    3067
        /// Table schema
    3068
        pub schema: Arc<Schema>,
    3069
        /// schema of describe table output
    3070
        pub output_schema: DFSchemaRef,
    3071
    }
    3072
    3073
    // Manual implementation of `PartialOrd`, returning none since there are no comparable types in
    3074
    // `DescribeTable`. This allows `LogicalPlan` to derive `PartialOrd`.
    3075
    impl PartialOrd for DescribeTable {
    3076
    0
        fn partial_cmp(&self, _other: &Self) -> Option<Ordering> {
    3077
            // There is no relevant comparison for schemas
    3078
    0
            None
    3079
    0
        }
    3080
    }
    3081
    3082
    /// Options for EXPLAIN
    3083
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    3084
    pub struct ExplainOption {
    3085
        /// Include detailed debug info
    3086
        pub verbose: bool,
    3087
        /// Actually execute the plan and report metrics
    3088
        pub analyze: bool,
    3089
        /// Output syntax/format
    3090
        pub format: ExplainFormat,
    3091
    }
    3092
    3093
    impl Default for ExplainOption {
    3094
    0
        fn default() -> Self {
    3095
    0
            ExplainOption {
    3096
    0
                verbose: false,
    3097
    0
                analyze: false,
    3098
    0
                format: ExplainFormat::Indent,
    3099
    0
            }
    3100
    0
        }
    3101
    }
    3102
    3103
    impl ExplainOption {
    3104
        /// Builder‐style setter for `verbose`
    3105
    0
        pub fn with_verbose(mut self, verbose: bool) -> Self {
    3106
    0
            self.verbose = verbose;
    3107
    0
            self
    3108
    0
        }
    3109
    3110
        /// Builder‐style setter for `analyze`
    3111
    0
        pub fn with_analyze(mut self, analyze: bool) -> Self {
    3112
    0
            self.analyze = analyze;
    3113
    0
            self
    3114
    0
        }
    3115
    3116
        /// Builder‐style setter for `format`
    3117
    0
        pub fn with_format(mut self, format: ExplainFormat) -> Self {
    3118
    0
            self.format = format;
    3119
    0
            self
    3120
    0
        }
    3121
    }
    3122
    3123
    /// Produces a relation with string representations of
    3124
    /// various parts of the plan
    3125
    ///
    3126
    /// See [the documentation] for more information
    3127
    ///
    3128
    /// [the documentation]: https://datafusion.apache.org/user-guide/sql/explain.html
    3129
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    3130
    pub struct Explain {
    3131
        /// Should extra (detailed, intermediate plans) be included?
    3132
        pub verbose: bool,
    3133
        /// Output format for explain, if specified.
    3134
        /// If none, defaults to `text`
    3135
        pub explain_format: ExplainFormat,
    3136
        /// The logical plan that is being EXPLAIN'd
    3137
        pub plan: Arc<LogicalPlan>,
    3138
        /// Represent the various stages plans have gone through
    3139
        pub stringified_plans: Vec<StringifiedPlan>,
    3140
        /// The output schema of the explain (2 columns of text)
    3141
        pub schema: DFSchemaRef,
    3142
        /// Used by physical planner to check if should proceed with planning
    3143
        pub logical_optimization_succeeded: bool,
    3144
    }
    3145
    3146
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    3147
    impl PartialOrd for Explain {
    3148
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    3149
            #[derive(PartialEq, PartialOrd)]
    3150
            struct ComparableExplain<'a> {
    3151
                /// Should extra (detailed, intermediate plans) be included?
    3152
                pub verbose: &'a bool,
    3153
                /// The logical plan that is being EXPLAIN'd
    3154
                pub plan: &'a Arc<LogicalPlan>,
    3155
                /// Represent the various stages plans have gone through
    3156
                pub stringified_plans: &'a Vec<StringifiedPlan>,
    3157
                /// Used by physical planner to check if should proceed with planning
    3158
                pub logical_optimization_succeeded: &'a bool,
    3159
            }
    3160
    0
            let comparable_self = ComparableExplain {
    3161
    0
                verbose: &self.verbose,
    3162
    0
                plan: &self.plan,
    3163
    0
                stringified_plans: &self.stringified_plans,
    3164
    0
                logical_optimization_succeeded: &self.logical_optimization_succeeded,
    3165
    0
            };
    3166
    0
            let comparable_other = ComparableExplain {
    3167
    0
                verbose: &other.verbose,
    3168
    0
                plan: &other.plan,
    3169
    0
                stringified_plans: &other.stringified_plans,
    3170
    0
                logical_optimization_succeeded: &other.logical_optimization_succeeded,
    3171
    0
            };
    3172
    0
            comparable_self
    3173
    0
                .partial_cmp(&comparable_other)
    3174
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    3175
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    3176
    0
        }
    3177
    }
    3178
    3179
    /// Runs the actual plan, and then prints the physical plan with
    3180
    /// with execution metrics.
    3181
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    3182
    pub struct Analyze {
    3183
        /// Should extra detail be included?
    3184
        pub verbose: bool,
    3185
        /// The logical plan that is being EXPLAIN ANALYZE'd
    3186
        pub input: Arc<LogicalPlan>,
    3187
        /// The output schema of the explain (2 columns of text)
    3188
        pub schema: DFSchemaRef,
    3189
    }
    3190
    3191
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    3192
    impl PartialOrd for Analyze {
    3193
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    3194
    0
            match self.verbose.partial_cmp(&other.verbose) {
    3195
    0
                Some(Ordering::Equal) => self.input.partial_cmp(&other.input),
    3196
    0
                cmp => cmp,
    3197
            }
    3198
            // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    3199
    0
            .filter(|cmp| *cmp != Ordering::Equal || self == other)
    3200
    0
        }
    3201
    }
    3202
    3203
    /// Extension operator defined outside of DataFusion
    3204
    // TODO(clippy): This clippy `allow` should be removed if
    3205
    // the manual `PartialEq` is removed in favor of a derive.
    3206
    // (see `PartialEq` the impl for details.)
    3207
    #[allow(clippy::derived_hash_with_manual_eq)]
    3208
    #[derive(Debug, Clone, Eq, Hash)]
    3209
    pub struct Extension {
    3210
        /// The runtime extension operator
    3211
        pub node: Arc<dyn UserDefinedLogicalNode>,
    3212
    }
    3213
    3214
    // `PartialEq` cannot be derived for types containing `Arc<dyn Trait>`.
    3215
    // This manual implementation should be removed if
    3216
    // https://github.com/rust-lang/rust/issues/39128 is fixed.
    3217
    impl PartialEq for Extension {
    3218
    0
        fn eq(&self, other: &Self) -> bool {
    3219
    0
            self.node.eq(&other.node)
    3220
    0
        }
    3221
    }
    3222
    3223
    impl PartialOrd for Extension {
    3224
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    3225
    0
            self.node.partial_cmp(&other.node)
    3226
    0
        }
    3227
    }
    3228
    3229
    /// Produces the first `n` tuples from its input and discards the rest.
    3230
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    3231
    pub struct Limit {
    3232
        /// Number of rows to skip before fetch
    3233
        pub skip: Option<Box<Expr>>,
    3234
        /// Maximum number of rows to fetch,
    3235
        /// None means fetching all rows
    3236
        pub fetch: Option<Box<Expr>>,
    3237
        /// The logical plan
    3238
        pub input: Arc<LogicalPlan>,
    3239
    }
    3240
    3241
    /// Different types of skip expression in Limit plan.
    3242
    pub enum SkipType {
    3243
        /// The skip expression is a literal value.
    3244
        Literal(usize),
    3245
        /// Currently only supports expressions that can be folded into constants.
    3246
        UnsupportedExpr,
    3247
    }
    3248
    3249
    /// Different types of fetch expression in Limit plan.
    3250
    pub enum FetchType {
    3251
        /// The fetch expression is a literal value.
    3252
        /// `Literal(None)` means the fetch expression is not provided.
    3253
        Literal(Option<usize>),
    3254
        /// Currently only supports expressions that can be folded into constants.
    3255
        UnsupportedExpr,
    3256
    }
    3257
    3258
    impl Limit {
    3259
        /// Get the skip type from the limit plan.
    3260
    0
        pub fn get_skip_type(&self) -> Result<SkipType> {
    3261
    0
            match self.skip.as_deref() {
    3262
    0
                Some(expr) => match *expr {
    3263
    0
                    Expr::Literal(ScalarValue::Int64(s), _) => {
    3264
                        // `skip = NULL` is equivalent to `skip = 0`
    3265
    0
                        let s = s.unwrap_or(0);
    3266
    0
                        if s >= 0 {
    3267
    0
                            Ok(SkipType::Literal(s as usize))
    3268
                        } else {
    3269
    0
                            plan_err!("OFFSET must be >=0, '{}' was provided", s)
    3270
                        }
    3271
                    }
    3272
    0
                    _ => Ok(SkipType::UnsupportedExpr),
    3273
                },
    3274
                // `skip = None` is equivalent to `skip = 0`
    3275
    0
                None => Ok(SkipType::Literal(0)),
    3276
            }
    3277
    0
        }
    3278
    3279
        /// Get the fetch type from the limit plan.
    3280
    0
        pub fn get_fetch_type(&self) -> Result<FetchType> {
    3281
    0
            match self.fetch.as_deref() {
    3282
    0
                Some(expr) => match *expr {
    3283
    0
                    Expr::Literal(ScalarValue::Int64(Some(s)), _) => {
    3284
    0
                        if s >= 0 {
    3285
    0
                            Ok(FetchType::Literal(Some(s as usize)))
    3286
                        } else {
    3287
    0
                            plan_err!("LIMIT must be >= 0, '{}' was provided", s)
    3288
                        }
    3289
                    }
    3290
                    Expr::Literal(ScalarValue::Int64(None), _) => {
    3291
    0
                        Ok(FetchType::Literal(None))
    3292
                    }
    3293
    0
                    _ => Ok(FetchType::UnsupportedExpr),
    3294
                },
    3295
    0
                None => Ok(FetchType::Literal(None)),
    3296
            }
    3297
    0
        }
    3298
    }
    3299
    3300
    /// Removes duplicate rows from the input
    3301
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    3302
    pub enum Distinct {
    3303
        /// Plain `DISTINCT` referencing all selection expressions
    3304
        All(Arc<LogicalPlan>),
    3305
        /// The `Postgres` addition, allowing separate control over DISTINCT'd and selected columns
    3306
        On(DistinctOn),
    3307
    }
    3308
    3309
    impl Distinct {
    3310
        /// return a reference to the nodes input
    3311
    0
        pub fn input(&self) -> &Arc<LogicalPlan> {
    3312
    0
            match self {
    3313
    0
                Distinct::All(input) => input,
    3314
    0
                Distinct::On(DistinctOn { input, .. }) => input,
    3315
            }
    3316
    0
        }
    3317
    }
    3318
    3319
    /// Removes duplicate rows from the input
    3320
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    3321
    pub struct DistinctOn {
    3322
        /// The `DISTINCT ON` clause expression list
    3323
        pub on_expr: Vec<Expr>,
    3324
        /// The selected projection expression list
    3325
        pub select_expr: Vec<Expr>,
    3326
        /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause when
    3327
        /// present. Note that those matching expressions actually wrap the `ON` expressions with
    3328
        /// additional info pertaining to the sorting procedure (i.e. ASC/DESC, and NULLS FIRST/LAST).
    3329
        pub sort_expr: Option<Vec<SortExpr>>,
    3330
        /// The logical plan that is being DISTINCT'd
    3331
        pub input: Arc<LogicalPlan>,
    3332
        /// The schema description of the DISTINCT ON output
    3333
        pub schema: DFSchemaRef,
    3334
    }
    3335
    3336
    impl DistinctOn {
    3337
        /// Create a new `DistinctOn` struct.
    3338
    0
        pub fn try_new(
    3339
    0
            on_expr: Vec<Expr>,
    3340
    0
            select_expr: Vec<Expr>,
    3341
    0
            sort_expr: Option<Vec<SortExpr>>,
    3342
    0
            input: Arc<LogicalPlan>,
    3343
    0
        ) -> Result<Self> {
    3344
    0
            if on_expr.is_empty() {
    3345
    0
                return plan_err!("No `ON` expressions provided");
    3346
    0
            }
    3347
    3348
    0
            let on_expr = normalize_cols(on_expr, input.as_ref())?;
    3349
    0
            let qualified_fields = exprlist_to_fields(select_expr.as_slice(), &input)?
    3350
    0
                .into_iter()
    3351
    0
                .collect();
    3352
    3353
    0
            let dfschema = DFSchema::new_with_metadata(
    3354
    0
                qualified_fields,
    3355
    0
                input.schema().metadata().clone(),
    3356
    0
            )?;
    3357
    3358
    0
            let mut distinct_on = DistinctOn {
    3359
    0
                on_expr,
    3360
    0
                select_expr,
    3361
    0
                sort_expr: None,
    3362
    0
                input,
    3363
    0
                schema: Arc::new(dfschema),
    3364
    0
            };
    3365
    3366
    0
            if let Some(sort_expr) = sort_expr {
    3367
    0
                distinct_on = distinct_on.with_sort_expr(sort_expr)?;
    3368
    0
            }
    3369
    3370
    0
            Ok(distinct_on)
    3371
    0
        }
    3372
    3373
        /// Try to update `self` with a new sort expressions.
    3374
        ///
    3375
        /// Validates that the sort expressions are a super-set of the `ON` expressions.
    3376
    0
        pub fn with_sort_expr(mut self, sort_expr: Vec<SortExpr>) -> Result<Self> {
    3377
    0
            let sort_expr = normalize_sorts(sort_expr, self.input.as_ref())?;
    3378
    3379
            // Check that the left-most sort expressions are the same as the `ON` expressions.
    3380
    0
            let mut matched = true;
    3381
    0
            for (on, sort) in self.on_expr.iter().zip(sort_expr.iter()) {
    3382
    0
                if on != &sort.expr {
    3383
    0
                    matched = false;
    3384
    0
                    break;
    3385
    0
                }
    3386
            }
    3387
    3388
    0
            if self.on_expr.len() > sort_expr.len() || !matched {
    3389
    0
                return plan_err!(
    3390
    0
                    "SELECT DISTINCT ON expressions must match initial ORDER BY expressions"
    3391
                );
    3392
    0
            }
    3393
    3394
    0
            self.sort_expr = Some(sort_expr);
    3395
    0
            Ok(self)
    3396
    0
        }
    3397
    }
    3398
    3399
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    3400
    impl PartialOrd for DistinctOn {
    3401
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    3402
            #[derive(PartialEq, PartialOrd)]
    3403
            struct ComparableDistinctOn<'a> {
    3404
                /// The `DISTINCT ON` clause expression list
    3405
                pub on_expr: &'a Vec<Expr>,
    3406
                /// The selected projection expression list
    3407
                pub select_expr: &'a Vec<Expr>,
    3408
                /// The `ORDER BY` clause, whose initial expressions must match those of the `ON` clause when
    3409
                /// present. Note that those matching expressions actually wrap the `ON` expressions with
    3410
                /// additional info pertaining to the sorting procedure (i.e. ASC/DESC, and NULLS FIRST/LAST).
    3411
                pub sort_expr: &'a Option<Vec<SortExpr>>,
    3412
                /// The logical plan that is being DISTINCT'd
    3413
                pub input: &'a Arc<LogicalPlan>,
    3414
            }
    3415
    0
            let comparable_self = ComparableDistinctOn {
    3416
    0
                on_expr: &self.on_expr,
    3417
    0
                select_expr: &self.select_expr,
    3418
    0
                sort_expr: &self.sort_expr,
    3419
    0
                input: &self.input,
    3420
    0
            };
    3421
    0
            let comparable_other = ComparableDistinctOn {
    3422
    0
                on_expr: &other.on_expr,
    3423
    0
                select_expr: &other.select_expr,
    3424
    0
                sort_expr: &other.sort_expr,
    3425
    0
                input: &other.input,
    3426
    0
            };
    3427
    0
            comparable_self
    3428
    0
                .partial_cmp(&comparable_other)
    3429
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    3430
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    3431
    0
        }
    3432
    }
    3433
    3434
    /// Aggregates its input based on a set of grouping and aggregate
    3435
    /// expressions (e.g. SUM).
    3436
    ///
    3437
    /// # Output Schema
    3438
    ///
    3439
    /// The output schema is the group expressions followed by the aggregate
    3440
    /// expressions in order.
    3441
    ///
    3442
    /// For example, given the input schema `"A", "B", "C"` and the aggregate
    3443
    /// `SUM(A) GROUP BY C+B`, the output schema will be `"C+B", "SUM(A)"` where
    3444
    /// "C+B" and "SUM(A)" are the names of the output columns. Note that "C+B" is a
    3445
    /// single new column
    3446
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    3447
    // mark non_exhaustive to encourage use of try_new/new()
    3448
    #[non_exhaustive]
    3449
    pub struct Aggregate {
    3450
        /// The incoming logical plan
    3451
        pub input: Arc<LogicalPlan>,
    3452
        /// Grouping expressions
    3453
        pub group_expr: Vec<Expr>,
    3454
        /// Aggregate expressions
    3455
        pub aggr_expr: Vec<Expr>,
    3456
        /// The schema description of the aggregate output
    3457
        pub schema: DFSchemaRef,
    3458
    }
    3459
    3460
    impl Aggregate {
    3461
        /// Create a new aggregate operator.
    3462
    0
        pub fn try_new(
    3463
    0
            input: Arc<LogicalPlan>,
    3464
    0
            group_expr: Vec<Expr>,
    3465
    0
            aggr_expr: Vec<Expr>,
    3466
    0
        ) -> Result<Self> {
    3467
    0
            let group_expr = enumerate_grouping_sets(group_expr)?;
    3468
    3469
    0
            let is_grouping_set = matches!(group_expr.as_slice(), [Expr::GroupingSet(_)]);
    3470
    3471
    0
            let grouping_expr: Vec<&Expr> = grouping_set_to_exprlist(group_expr.as_slice())?;
    3472
    3473
    0
            let mut qualified_fields = exprlist_to_fields(grouping_expr, &input)?;
    3474
    3475
            // Even columns that cannot be null will become nullable when used in a grouping set.
    3476
    0
            if is_grouping_set {
    3477
    0
                qualified_fields = qualified_fields
    3478
    0
                    .into_iter()
    3479
    0
                    .map(|(q, f)| (q, f.as_ref().clone().with_nullable(true).into()))
    3480
    0
                    .collect::<Vec<_>>();
    3481
    0
                qualified_fields.push((
    3482
    0
                    None,
    3483
    0
                    Field::new(
    3484
    0
                        Self::INTERNAL_GROUPING_ID,
    3485
    0
                        Self::grouping_id_type(qualified_fields.len()),
    3486
    0
                        false,
    3487
    0
                    )
    3488
    0
                    .into(),
    3489
    0
                ));
    3490
    0
            }
    3491
    3492
    0
            qualified_fields.extend(exprlist_to_fields(aggr_expr.as_slice(), &input)?);
    3493
    3494
    0
            let schema = DFSchema::new_with_metadata(
    3495
    0
                qualified_fields,
    3496
    0
                input.schema().metadata().clone(),
    3497
    0
            )?;
    3498
    3499
    0
            Self::try_new_with_schema(input, group_expr, aggr_expr, Arc::new(schema))
    3500
    0
        }
    3501
    3502
        /// Create a new aggregate operator using the provided schema to avoid the overhead of
    3503
        /// building the schema again when the schema is already known.
    3504
        ///
    3505
        /// This method should only be called when you are absolutely sure that the schema being
    3506
        /// provided is correct for the aggregate. If in doubt, call [try_new](Self::try_new) instead.
    3507
        #[expect(clippy::needless_pass_by_value)]
    3508
    0
        pub fn try_new_with_schema(
    3509
    0
            input: Arc<LogicalPlan>,
    3510
    0
            group_expr: Vec<Expr>,
    3511
    0
            aggr_expr: Vec<Expr>,
    3512
    0
            schema: DFSchemaRef,
    3513
    0
        ) -> Result<Self> {
    3514
    0
            if group_expr.is_empty() && aggr_expr.is_empty() {
    3515
    0
                return plan_err!(
    3516
    0
                    "Aggregate requires at least one grouping or aggregate expression. \
    3517
    0
                    Aggregate without grouping expressions nor aggregate expressions is \
    3518
    0
                    logically equivalent to, but less efficient than, VALUES producing \
    3519
    0
                    single row. Please use VALUES instead."
    3520
                );
    3521
    0
            }
    3522
    0
            let group_expr_count = grouping_set_expr_count(&group_expr)?;
    3523
    0
            if schema.fields().len() != group_expr_count + aggr_expr.len() {
    3524
    0
                return plan_err!(
    3525
    0
                    "Aggregate schema has wrong number of fields. Expected {} got {}",
    3526
    0
                    group_expr_count + aggr_expr.len(),
    3527
    0
                    schema.fields().len()
    3528
                );
    3529
    0
            }
    3530
    3531
    0
            let aggregate_func_dependencies =
    3532
    0
                calc_func_dependencies_for_aggregate(&group_expr, &input, &schema)?;
    3533
    0
            let new_schema = schema.as_ref().clone();
    3534
    0
            let schema = Arc::new(
    3535
    0
                new_schema.with_functional_dependencies(aggregate_func_dependencies)?,
    3536
            );
    3537
    0
            Ok(Self {
    3538
    0
                input,
    3539
    0
                group_expr,
    3540
    0
                aggr_expr,
    3541
    0
                schema,
    3542
    0
            })
    3543
    0
        }
    3544
    3545
    0
        fn is_grouping_set(&self) -> bool {
    3546
    0
            matches!(self.group_expr.as_slice(), [Expr::GroupingSet(_)])
    3547
    0
        }
    3548
    3549
        /// Get the output expressions.
    3550
    0
        fn output_expressions(&self) -> Result<Vec<&Expr>> {
    3551
    0
            static INTERNAL_ID_EXPR: LazyLock<Expr> = LazyLock::new(|| {
    3552
    0
                Expr::Column(Column::from_name(Aggregate::INTERNAL_GROUPING_ID))
    3553
    0
            });
    3554
    0
            let mut exprs = grouping_set_to_exprlist(self.group_expr.as_slice())?;
    3555
    0
            if self.is_grouping_set() {
    3556
    0
                exprs.push(&INTERNAL_ID_EXPR);
    3557
    0
            }
    3558
    0
            exprs.extend(self.aggr_expr.iter());
    3559
    0
            debug_assert!(exprs.len() == self.schema.fields().len());
    3560
    0
            Ok(exprs)
    3561
    0
        }
    3562
    3563
        /// Get the length of the group by expression in the output schema
    3564
        /// This is not simply group by expression length. Expression may be
    3565
        /// GroupingSet, etc. In these case we need to get inner expression lengths.
    3566
    0
        pub fn group_expr_len(&self) -> Result<usize> {
    3567
    0
            grouping_set_expr_count(&self.group_expr)
    3568
    0
        }
    3569
    3570
        /// Returns the data type of the grouping id.
    3571
        /// The grouping ID value is a bitmask where each set bit
    3572
        /// indicates that the corresponding grouping expression is
    3573
        /// null
    3574
    0
        pub fn grouping_id_type(group_exprs: usize) -> DataType {
    3575
    0
            if group_exprs <= 8 {
    3576
    0
                DataType::UInt8
    3577
    0
            } else if group_exprs <= 16 {
    3578
    0
                DataType::UInt16
    3579
    0
            } else if group_exprs <= 32 {
    3580
    0
                DataType::UInt32
    3581
            } else {
    3582
    0
                DataType::UInt64
    3583
            }
    3584
    0
        }
    3585
    3586
        /// Internal column used when the aggregation is a grouping set.
    3587
        ///
    3588
        /// This column contains a bitmask where each bit represents a grouping
    3589
        /// expression. The least significant bit corresponds to the rightmost
    3590
        /// grouping expression. A bit value of 0 indicates that the corresponding
    3591
        /// column is included in the grouping set, while a value of 1 means it is excluded.
    3592
        ///
    3593
        /// For example, for the grouping expressions CUBE(a, b), the grouping ID
    3594
        /// column will have the following values:
    3595
        ///     0b00: Both `a` and `b` are included
    3596
        ///     0b01: `b` is excluded
    3597
        ///     0b10: `a` is excluded
    3598
        ///     0b11: Both `a` and `b` are excluded
    3599
        ///
    3600
        /// This internal column is necessary because excluded columns are replaced
    3601
        /// with `NULL` values. To handle these cases correctly, we must distinguish
    3602
        /// between an actual `NULL` value in a column and a column being excluded from the set.
    3603
        pub const INTERNAL_GROUPING_ID: &'static str = "__grouping_id";
    3604
    }
    3605
    3606
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    3607
    impl PartialOrd for Aggregate {
    3608
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    3609
    0
            match self.input.partial_cmp(&other.input) {
    3610
                Some(Ordering::Equal) => {
    3611
    0
                    match self.group_expr.partial_cmp(&other.group_expr) {
    3612
    0
                        Some(Ordering::Equal) => self.aggr_expr.partial_cmp(&other.aggr_expr),
    3613
    0
                        cmp => cmp,
    3614
                    }
    3615
                }
    3616
    0
                cmp => cmp,
    3617
            }
    3618
            // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    3619
    0
            .filter(|cmp| *cmp != Ordering::Equal || self == other)
    3620
    0
        }
    3621
    }
    3622
    3623
    /// Checks whether any expression in `group_expr` contains `Expr::GroupingSet`.
    3624
    0
    fn contains_grouping_set(group_expr: &[Expr]) -> bool {
    3625
    0
        group_expr
    3626
    0
            .iter()
    3627
    0
            .any(|expr| matches!(expr, Expr::GroupingSet(_)))
    3628
    0
    }
    3629
    3630
    /// Calculates functional dependencies for aggregate expressions.
    3631
    0
    fn calc_func_dependencies_for_aggregate(
    3632
    0
        // Expressions in the GROUP BY clause:
    3633
    0
        group_expr: &[Expr],
    3634
    0
        // Input plan of the aggregate:
    3635
    0
        input: &LogicalPlan,
    3636
    0
        // Aggregate schema
    3637
    0
        aggr_schema: &DFSchema,
    3638
    0
    ) -> Result<FunctionalDependencies> {
    3639
        // We can do a case analysis on how to propagate functional dependencies based on
    3640
        // whether the GROUP BY in question contains a grouping set expression:
    3641
        // - If so, the functional dependencies will be empty because we cannot guarantee
    3642
        //   that GROUP BY expression results will be unique.
    3643
        // - Otherwise, it may be possible to propagate functional dependencies.
    3644
    0
        if !contains_grouping_set(group_expr) {
    3645
    0
            let group_by_expr_names = group_expr
    3646
    0
                .iter()
    3647
    0
                .map(|item| item.schema_name().to_string())
    3648
    0
                .collect::<IndexSet<_>>()
    3649
    0
                .into_iter()
    3650
    0
                .collect::<Vec<_>>();
    3651
    0
            let aggregate_func_dependencies = aggregate_functional_dependencies(
    3652
    0
                input.schema(),
    3653
    0
                &group_by_expr_names,
    3654
    0
                aggr_schema,
    3655
            );
    3656
    0
            Ok(aggregate_func_dependencies)
    3657
        } else {
    3658
    0
            Ok(FunctionalDependencies::empty())
    3659
        }
    3660
    0
    }
    3661
    3662
    /// This function projects functional dependencies of the `input` plan according
    3663
    /// to projection expressions `exprs`.
    3664
    0
    fn calc_func_dependencies_for_project(
    3665
    0
        exprs: &[Expr],
    3666
    0
        input: &LogicalPlan,
    3667
    0
    ) -> Result<FunctionalDependencies> {
    3668
    0
        let input_fields = input.schema().field_names();
    3669
        // Calculate expression indices (if present) in the input schema.
    3670
    0
        let proj_indices = exprs
    3671
    0
            .iter()
    3672
    0
            .map(|expr| match expr {
    3673
                #[expect(deprecated)]
    3674
    0
                Expr::Wildcard { qualifier, options } => {
    3675
    0
                    let wildcard_fields = exprlist_to_fields(
    3676
    0
                        vec![&Expr::Wildcard {
    3677
    0
                            qualifier: qualifier.clone(),
    3678
    0
                            options: options.clone(),
    3679
    0
                        }],
    3680
    0
                        input,
    3681
    0
                    )?;
    3682
                    Ok::<_, DataFusionError>(
    3683
    0
                        wildcard_fields
    3684
    0
                            .into_iter()
    3685
    0
                            .filter_map(|(qualifier, f)| {
    3686
    0
                                let flat_name = qualifier
    3687
    0
                                    .map(|t| format!("{}.{}", t, f.name()))
    3688
    0
                                    .unwrap_or_else(|| f.name().clone());
    3689
    0
                                input_fields.iter().position(|item| *item == flat_name)
    3690
    0
                            })
    3691
    0
                            .collect::<Vec<_>>(),
    3692
                    )
    3693
                }
    3694
    0
                Expr::Alias(alias) => {
    3695
    0
                    let name = format!("{}", alias.expr);
    3696
    0
                    Ok(input_fields
    3697
    0
                        .iter()
    3698
    0
                        .position(|item| *item == name)
    3699
    0
                        .map(|i| vec![i])
    3700
    0
                        .unwrap_or(vec![]))
    3701
                }
    3702
                _ => {
    3703
    0
                    let name = format!("{expr}");
    3704
    0
                    Ok(input_fields
    3705
    0
                        .iter()
    3706
    0
                        .position(|item| *item == name)
    3707
    0
                        .map(|i| vec![i])
    3708
    0
                        .unwrap_or(vec![]))
    3709
                }
    3710
    0
            })
    3711
    0
            .collect::<Result<Vec<_>>>()?
    3712
    0
            .into_iter()
    3713
    0
            .flatten()
    3714
    0
            .collect::<Vec<_>>();
    3715
    3716
    0
        Ok(input
    3717
    0
            .schema()
    3718
    0
            .functional_dependencies()
    3719
    0
            .project_functional_dependencies(&proj_indices, exprs.len()))
    3720
    0
    }
    3721
    3722
    /// Sorts its input according to a list of sort expressions.
    3723
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    3724
    pub struct Sort {
    3725
        /// The sort expressions
    3726
        pub expr: Vec<SortExpr>,
    3727
        /// The incoming logical plan
    3728
        pub input: Arc<LogicalPlan>,
    3729
        /// Optional fetch limit
    3730
        pub fetch: Option<usize>,
    3731
    }
    3732
    3733
    /// Join two logical plans on one or more join columns
    3734
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    3735
    pub struct Join {
    3736
        /// Left input
    3737
        pub left: Arc<LogicalPlan>,
    3738
        /// Right input
    3739
        pub right: Arc<LogicalPlan>,
    3740
        /// Equijoin clause expressed as pairs of (left, right) join expressions
    3741
        pub on: Vec<(Expr, Expr)>,
    3742
        /// Filters applied during join (non-equi conditions)
    3743
        pub filter: Option<Expr>,
    3744
        /// Join type
    3745
        pub join_type: JoinType,
    3746
        /// Join constraint
    3747
        pub join_constraint: JoinConstraint,
    3748
        /// The output schema, containing fields from the left and right inputs
    3749
        pub schema: DFSchemaRef,
    3750
        /// Defines the null equality for the join.
    3751
        pub null_equality: NullEquality,
    3752
    }
    3753
    3754
    impl Join {
    3755
        /// Creates a new Join operator with automatically computed schema.
    3756
        ///
    3757
        /// This constructor computes the schema based on the join type and inputs,
    3758
        /// removing the need to manually specify the schema or call `recompute_schema`.
    3759
        ///
    3760
        /// # Arguments
    3761
        ///
    3762
        /// * `left` - Left input plan
    3763
        /// * `right` - Right input plan
    3764
        /// * `on` - Join condition as a vector of (left_expr, right_expr) pairs
    3765
        /// * `filter` - Optional filter expression (for non-equijoin conditions)
    3766
        /// * `join_type` - Type of join (Inner, Left, Right, etc.)
    3767
        /// * `join_constraint` - Join constraint (On, Using)
    3768
        /// * `null_equality` - How to handle nulls in join comparisons
    3769
        ///
    3770
        /// # Returns
    3771
        ///
    3772
        /// A new Join operator with the computed schema
    3773
    0
        pub fn try_new(
    3774
    0
            left: Arc<LogicalPlan>,
    3775
    0
            right: Arc<LogicalPlan>,
    3776
    0
            on: Vec<(Expr, Expr)>,
    3777
    0
            filter: Option<Expr>,
    3778
    0
            join_type: JoinType,
    3779
    0
            join_constraint: JoinConstraint,
    3780
    0
            null_equality: NullEquality,
    3781
    0
        ) -> Result<Self> {
    3782
    0
            let join_schema = build_join_schema(left.schema(), right.schema(), &join_type)?;
    3783
    3784
    0
            Ok(Join {
    3785
    0
                left,
    3786
    0
                right,
    3787
    0
                on,
    3788
    0
                filter,
    3789
    0
                join_type,
    3790
    0
                join_constraint,
    3791
    0
                schema: Arc::new(join_schema),
    3792
    0
                null_equality,
    3793
    0
            })
    3794
    0
        }
    3795
    3796
        /// Create Join with input which wrapped with projection, this method is used in physical planning only to help
    3797
        /// create the physical join.
    3798
    0
        pub fn try_new_with_project_input(
    3799
    0
            original: &LogicalPlan,
    3800
    0
            left: Arc<LogicalPlan>,
    3801
    0
            right: Arc<LogicalPlan>,
    3802
    0
            column_on: (Vec<Column>, Vec<Column>),
    3803
    0
        ) -> Result<(Self, bool)> {
    3804
    0
            let original_join = match original {
    3805
    0
                LogicalPlan::Join(join) => join,
    3806
    0
                _ => return plan_err!("Could not create join with project input"),
    3807
            };
    3808
    3809
    0
            let mut left_sch = LogicalPlanBuilder::from(Arc::clone(&left));
    3810
    0
            let mut right_sch = LogicalPlanBuilder::from(Arc::clone(&right));
    3811
    3812
    0
            let mut requalified = false;
    3813
    3814
            // By definition, the resulting schema of an inner/left/right & full join will have first the left side fields and then the right,
    3815
            // potentially having duplicate field names. Note this will only qualify fields if they have not been qualified before.
    3816
    0
            if original_join.join_type == JoinType::Inner
    3817
    0
                || original_join.join_type == JoinType::Left
    3818
    0
                || original_join.join_type == JoinType::Right
    3819
    0
                || original_join.join_type == JoinType::Full
    3820
            {
    3821
    0
                (left_sch, right_sch, requalified) =
    3822
    0
                    requalify_sides_if_needed(left_sch.clone(), right_sch.clone())?;
    3823
    0
            }
    3824
    3825
    0
            let on: Vec<(Expr, Expr)> = column_on
    3826
    0
                .0
    3827
    0
                .into_iter()
    3828
    0
                .zip(column_on.1)
    3829
    0
                .map(|(l, r)| (Expr::Column(l), Expr::Column(r)))
    3830
    0
                .collect();
    3831
    3832
    0
            let join_schema = build_join_schema(
    3833
    0
                left_sch.schema(),
    3834
    0
                right_sch.schema(),
    3835
    0
                &original_join.join_type,
    3836
    0
            )?;
    3837
    3838
    0
            Ok((
    3839
    0
                Join {
    3840
    0
                    left,
    3841
    0
                    right,
    3842
    0
                    on,
    3843
    0
                    filter: original_join.filter.clone(),
    3844
    0
                    join_type: original_join.join_type,
    3845
    0
                    join_constraint: original_join.join_constraint,
    3846
    0
                    schema: Arc::new(join_schema),
    3847
    0
                    null_equality: original_join.null_equality,
    3848
    0
                },
    3849
    0
                requalified,
    3850
    0
            ))
    3851
    0
        }
    3852
    }
    3853
    3854
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    3855
    impl PartialOrd for Join {
    3856
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    3857
            #[derive(PartialEq, PartialOrd)]
    3858
            struct ComparableJoin<'a> {
    3859
                /// Left input
    3860
                pub left: &'a Arc<LogicalPlan>,
    3861
                /// Right input
    3862
                pub right: &'a Arc<LogicalPlan>,
    3863
                /// Equijoin clause expressed as pairs of (left, right) join expressions
    3864
                pub on: &'a Vec<(Expr, Expr)>,
    3865
                /// Filters applied during join (non-equi conditions)
    3866
                pub filter: &'a Option<Expr>,
    3867
                /// Join type
    3868
                pub join_type: &'a JoinType,
    3869
                /// Join constraint
    3870
                pub join_constraint: &'a JoinConstraint,
    3871
                /// The null handling behavior for equalities
    3872
                pub null_equality: &'a NullEquality,
    3873
            }
    3874
    0
            let comparable_self = ComparableJoin {
    3875
    0
                left: &self.left,
    3876
    0
                right: &self.right,
    3877
    0
                on: &self.on,
    3878
    0
                filter: &self.filter,
    3879
    0
                join_type: &self.join_type,
    3880
    0
                join_constraint: &self.join_constraint,
    3881
    0
                null_equality: &self.null_equality,
    3882
    0
            };
    3883
    0
            let comparable_other = ComparableJoin {
    3884
    0
                left: &other.left,
    3885
    0
                right: &other.right,
    3886
    0
                on: &other.on,
    3887
    0
                filter: &other.filter,
    3888
    0
                join_type: &other.join_type,
    3889
    0
                join_constraint: &other.join_constraint,
    3890
    0
                null_equality: &other.null_equality,
    3891
    0
            };
    3892
    0
            comparable_self
    3893
    0
                .partial_cmp(&comparable_other)
    3894
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    3895
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    3896
    0
        }
    3897
    }
    3898
    3899
    /// Subquery
    3900
    #[derive(Clone, PartialEq, Eq, PartialOrd, Hash)]
    3901
    pub struct Subquery {
    3902
        /// The subquery
    3903
        pub subquery: Arc<LogicalPlan>,
    3904
        /// The outer references used in the subquery
    3905
        pub outer_ref_columns: Vec<Expr>,
    3906
        /// Span information for subquery projection columns
    3907
        pub spans: Spans,
    3908
    }
    3909
    3910
    impl Normalizeable for Subquery {
    3911
    0
        fn can_normalize(&self) -> bool {
    3912
    0
            false
    3913
    0
        }
    3914
    }
    3915
    3916
    impl NormalizeEq for Subquery {
    3917
    0
        fn normalize_eq(&self, other: &Self) -> bool {
    3918
            // TODO: may be implement NormalizeEq for LogicalPlan?
    3919
    0
            *self.subquery == *other.subquery
    3920
    0
                && self.outer_ref_columns.len() == other.outer_ref_columns.len()
    3921
    0
                && self
    3922
    0
                    .outer_ref_columns
    3923
    0
                    .iter()
    3924
    0
                    .zip(other.outer_ref_columns.iter())
    3925
    0
                    .all(|(a, b)| a.normalize_eq(b))
    3926
    0
        }
    3927
    }
    3928
    3929
    impl Subquery {
    3930
    0
        pub fn try_from_expr(plan: &Expr) -> Result<&Subquery> {
    3931
    0
            match plan {
    3932
    0
                Expr::ScalarSubquery(it) => Ok(it),
    3933
    0
                Expr::Cast(cast) => Subquery::try_from_expr(cast.expr.as_ref()),
    3934
    0
                _ => plan_err!("Could not coerce into ScalarSubquery!"),
    3935
            }
    3936
    0
        }
    3937
    3938
    0
        pub fn with_plan(&self, plan: Arc<LogicalPlan>) -> Subquery {
    3939
    0
            Subquery {
    3940
    0
                subquery: plan,
    3941
    0
                outer_ref_columns: self.outer_ref_columns.clone(),
    3942
    0
                spans: Spans::new(),
    3943
    0
            }
    3944
    0
        }
    3945
    }
    3946
    3947
    impl Debug for Subquery {
    3948
    0
        fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
    3949
    0
            write!(f, "<subquery>")
    3950
    0
        }
    3951
    }
    3952
    3953
    /// Logical partitioning schemes supported by [`LogicalPlan::Repartition`]
    3954
    ///
    3955
    /// See [`Partitioning`] for more details on partitioning
    3956
    ///
    3957
    /// [`Partitioning`]: https://docs.rs/datafusion/latest/datafusion/physical_expr/enum.Partitioning.html#
    3958
    #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
    3959
    pub enum Partitioning {
    3960
        /// Allocate batches using a round-robin algorithm and the specified number of partitions
    3961
        RoundRobinBatch(usize),
    3962
        /// Allocate rows based on a hash of one of more expressions and the specified number
    3963
        /// of partitions.
    3964
        Hash(Vec<Expr>, usize),
    3965
        /// The DISTRIBUTE BY clause is used to repartition the data based on the input expressions
    3966
        DistributeBy(Vec<Expr>),
    3967
    }
    3968
    3969
    /// Represent the unnesting operation on a list column, such as the recursion depth and
    3970
    /// the output column name after unnesting
    3971
    ///
    3972
    /// Example: given `ColumnUnnestList { output_column: "output_name", depth: 2 }`
    3973
    ///
    3974
    /// ```text
    3975
    ///   input             output_name
    3976
    ///  ┌─────────┐      ┌─────────┐
    3977
    ///  │{{1,2}}  │      │ 1       │
    3978
    ///  ├─────────┼─────►├─────────┤
    3979
    ///  │{{3}}    │      │ 2       │
    3980
    ///  ├─────────┤      ├─────────┤
    3981
    ///  │{{4},{5}}│      │ 3       │
    3982
    ///  └─────────┘      ├─────────┤
    3983
    ///                   │ 4       │
    3984
    ///                   ├─────────┤
    3985
    ///                   │ 5       │
    3986
    ///                   └─────────┘
    3987
    /// ```
    3988
    #[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd)]
    3989
    pub struct ColumnUnnestList {
    3990
        pub output_column: Column,
    3991
        pub depth: usize,
    3992
    }
    3993
    3994
    impl Display for ColumnUnnestList {
    3995
    0
        fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
    3996
    0
            write!(f, "{}|depth={}", self.output_column, self.depth)
    3997
    0
        }
    3998
    }
    3999
    4000
    /// Unnest a column that contains a nested list type. See
    4001
    /// [`UnnestOptions`] for more details.
    4002
    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
    4003
    pub struct Unnest {
    4004
        /// The incoming logical plan
    4005
        pub input: Arc<LogicalPlan>,
    4006
        /// Columns to run unnest on, can be a list of (List/Struct) columns
    4007
        pub exec_columns: Vec<Column>,
    4008
        /// refer to the indices(in the input schema) of columns
    4009
        /// that have type list to run unnest on
    4010
        pub list_type_columns: Vec<(usize, ColumnUnnestList)>,
    4011
        /// refer to the indices (in the input schema) of columns
    4012
        /// that have type struct to run unnest on
    4013
        pub struct_type_columns: Vec<usize>,
    4014
        /// Having items aligned with the output columns
    4015
        /// representing which column in the input schema each output column depends on
    4016
        pub dependency_indices: Vec<usize>,
    4017
        /// The output schema, containing the unnested field column.
    4018
        pub schema: DFSchemaRef,
    4019
        /// Options
    4020
        pub options: UnnestOptions,
    4021
    }
    4022
    4023
    // Manual implementation needed because of `schema` field. Comparison excludes this field.
    4024
    impl PartialOrd for Unnest {
    4025
    0
        fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
    4026
            #[derive(PartialEq, PartialOrd)]
    4027
            struct ComparableUnnest<'a> {
    4028
                /// The incoming logical plan
    4029
                pub input: &'a Arc<LogicalPlan>,
    4030
                /// Columns to run unnest on, can be a list of (List/Struct) columns
    4031
                pub exec_columns: &'a Vec<Column>,
    4032
                /// refer to the indices(in the input schema) of columns
    4033
                /// that have type list to run unnest on
    4034
                pub list_type_columns: &'a Vec<(usize, ColumnUnnestList)>,
    4035
                /// refer to the indices (in the input schema) of columns
    4036
                /// that have type struct to run unnest on
    4037
                pub struct_type_columns: &'a Vec<usize>,
    4038
                /// Having items aligned with the output columns
    4039
                /// representing which column in the input schema each output column depends on
    4040
                pub dependency_indices: &'a Vec<usize>,
    4041
                /// Options
    4042
                pub options: &'a UnnestOptions,
    4043
            }
    4044
    0
            let comparable_self = ComparableUnnest {
    4045
    0
                input: &self.input,
    4046
    0
                exec_columns: &self.exec_columns,
    4047
    0
                list_type_columns: &self.list_type_columns,
    4048
    0
                struct_type_columns: &self.struct_type_columns,
    4049
    0
                dependency_indices: &self.dependency_indices,
    4050
    0
                options: &self.options,
    4051
    0
            };
    4052
    0
            let comparable_other = ComparableUnnest {
    4053
    0
                input: &other.input,
    4054
    0
                exec_columns: &other.exec_columns,
    4055
    0
                list_type_columns: &other.list_type_columns,
    4056
    0
                struct_type_columns: &other.struct_type_columns,
    4057
    0
                dependency_indices: &other.dependency_indices,
    4058
    0
                options: &other.options,
    4059
    0
            };
    4060
    0
            comparable_self
    4061
    0
                .partial_cmp(&comparable_other)
    4062
                // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields
    4063
    0
                .filter(|cmp| *cmp != Ordering::Equal || self == other)
    4064
    0
        }
    4065
    }
    4066
    4067
    impl Unnest {
    4068
    0
        pub fn try_new(
    4069
    0
            input: Arc<LogicalPlan>,
    4070
    0
            exec_columns: Vec<Column>,
    4071
    0
            options: UnnestOptions,
    4072
    0
        ) -> Result<Self> {
    4073
    0
            if exec_columns.is_empty() {
    4074
    0
                return plan_err!("unnest plan requires at least 1 column to unnest");
    4075
    0
            }
    4076
    4077
    0
            let mut list_columns: Vec<(usize, ColumnUnnestList)> = vec![];
    4078
    0
            let mut struct_columns = vec![];
    4079
    0
            let indices_to_unnest = exec_columns
    4080
    0
                .iter()
    4081
    0
                .map(|c| Ok((input.schema().index_of_column(c)?, c)))
    4082
    0
                .collect::<Result<HashMap<usize, &Column>>>()?;
    4083
    4084
    0
            let input_schema = input.schema();
    4085
    4086
    0
            let mut dependency_indices = vec![];
    4087
            // Transform input schema into new schema
    4088
            // Given this comprehensive example
    4089
            //
    4090
            // input schema:
    4091
            // 1.col1_unnest_placeholder: list[list[int]],
    4092
            // 2.col1: list[list[int]]
    4093
            // 3.col2: list[int]
    4094
            // with unnest on unnest(col1,depth=2), unnest(col1,depth=1) and unnest(col2,depth=1)
    4095
            // output schema:
    4096
            // 1.unnest_col1_depth_2: int
    4097
            // 2.unnest_col1_depth_1: list[int]
    4098
            // 3.col1: list[list[int]]
    4099
            // 4.unnest_col2_depth_1: int
    4100
            // Meaning the placeholder column will be replaced by its unnested variation(s), note
    4101
            // the plural.
    4102
    0
            let fields = input_schema
    4103
    0
                .iter()
    4104
    0
                .enumerate()
    4105
    0
                .map(|(index, (original_qualifier, original_field))| {
    4106
    0
                    match indices_to_unnest.get(&index) {
    4107
    0
                        Some(column_to_unnest) => {
    4108
    0
                            let recursions_on_column = options
    4109
    0
                                .recursions
    4110
    0
                                .iter()
    4111
    0
                                .filter(|p| -> bool { &p.input_column == *column_to_unnest })
    4112
    0
                                .collect::<Vec<_>>();
    4113
    0
                            let mut transformed_columns = recursions_on_column
    4114
    0
                                .iter()
    4115
    0
                                .map(|r| {
    4116
    0
                                    list_columns.push((
    4117
    0
                                        index,
    4118
    0
                                        ColumnUnnestList {
    4119
    0
                                            output_column: r.output_column.clone(),
    4120
    0
                                            depth: r.depth,
    4121
    0
                                        },
    4122
    0
                                    ));
    4123
    0
                                    Ok(get_unnested_columns(
    4124
    0
                                        &r.output_column.name,
    4125
    0
                                        original_field.data_type(),
    4126
    0
                                        r.depth,
    4127
    0
                                    )?
    4128
    0
                                    .into_iter()
    4129
    0
                                    .next()
    4130
    0
                                    .unwrap()) // because unnesting a list column always result into one result
    4131
    0
                                })
    4132
    0
                                .collect::<Result<Vec<(Column, Arc<Field>)>>>()?;
    4133
    0
                            if transformed_columns.is_empty() {
    4134
    0
                                transformed_columns = get_unnested_columns(
    4135
    0
                                    &column_to_unnest.name,
    4136
    0
                                    original_field.data_type(),
    4137
                                    1,
    4138
    0
                                )?;
    4139
    0
                                match original_field.data_type() {
    4140
    0
                                    DataType::Struct(_) => {
    4141
    0
                                        struct_columns.push(index);
    4142
    0
                                    }
    4143
                                    DataType::List(_)
    4144
                                    | DataType::FixedSizeList(_, _)
    4145
    0
                                    | DataType::LargeList(_) => {
    4146
    0
                                        list_columns.push((
    4147
    0
                                            index,
    4148
    0
                                            ColumnUnnestList {
    4149
    0
                                                output_column: Column::from_name(
    4150
    0
                                                    &column_to_unnest.name,
    4151
    0
                                                ),
    4152
    0
                                                depth: 1,
    4153
    0
                                            },
    4154
    0
                                        ));
    4155
    0
                                    }
    4156
    0
                                    _ => {}
    4157
                                };
    4158
    0
                            }
    4159
    4160
                            // new columns dependent on the same original index
    4161
    0
                            dependency_indices.extend(std::iter::repeat_n(
    4162
    0
                                index,
    4163
    0
                                transformed_columns.len(),
    4164
                            ));
    4165
    0
                            Ok(transformed_columns
    4166
    0
                                .iter()
    4167
    0
                                .map(|(col, field)| {
    4168
    0
                                    (col.relation.to_owned(), field.to_owned())
    4169
    0
                                })
    4170
    0
                                .collect())
    4171
                        }
    4172
                        None => {
    4173
    0
                            dependency_indices.push(index);
    4174
    0
                            Ok(vec![(
    4175
    0
                                original_qualifier.cloned(),
    4176
    0
                                Arc::clone(original_field),
    4177
    0
                            )])
    4178
                        }
    4179
                    }
    4180
    0
                })
    4181
    0
                .collect::<Result<Vec<_>>>()?
    4182
    0
                .into_iter()
    4183
    0
                .flatten()
    4184
    0
                .collect::<Vec<_>>();
    4185
    4186
    0
            let metadata = input_schema.metadata().clone();
    4187
    0
            let df_schema = DFSchema::new_with_metadata(fields, metadata)?;
    4188
            // We can use the existing functional dependencies:
    4189
    0
            let deps = input_schema.functional_dependencies().clone();
    4190
    0
            let schema = Arc::new(df_schema.with_functional_dependencies(deps)?);
    4191
    4192
    0
            Ok(Unnest {
    4193
    0
                input,
    4194
    0
                exec_columns,
    4195
    0
                list_type_columns: list_columns,
    4196
    0
                struct_type_columns: struct_columns,
    4197
    0
                dependency_indices,
    4198
    0
                schema,
    4199
    0
                options,
    4200
    0
            })
    4201
    0
        }
    4202
    }
    4203
    4204
    // Based on data type, either struct or a variant of list
    4205
    // return a set of columns as the result of unnesting
    4206
    // the input columns.
    4207
    // For example, given a column with name "a",
    4208
    // - List(Element) returns ["a"] with data type Element
    4209
    // - Struct(field1, field2) returns ["a.field1","a.field2"]
    4210
    // For list data type, an argument depth is used to specify
    4211
    // the recursion level
    4212
    0
    fn get_unnested_columns(
    4213
    0
        col_name: &String,
    4214
    0
        data_type: &DataType,
    4215
    0
        depth: usize,
    4216
    0
    ) -> Result<Vec<(Column, Arc<Field>)>> {
    4217
    0
        let mut qualified_columns = Vec::with_capacity(1);
    4218
    4219
    0
        match data_type {
    4220
            DataType::List(_) | DataType::FixedSizeList(_, _) | DataType::LargeList(_) => {
    4221
    0
                let data_type = get_unnested_list_datatype_recursive(data_type, depth)?;
    4222
    0
                let new_field = Arc::new(Field::new(
    4223
    0
                    col_name, data_type,
    4224
                    // Unnesting may produce NULLs even if the list is not null.
    4225
                    // For example: unnest([1], []) -> 1, null
    4226
                    true,
    4227
                ));
    4228
    0
                let column = Column::from_name(col_name);
    4229
                // let column = Column::from((None, &new_field));
    4230
    0
                qualified_columns.push((column, new_field));
    4231
            }
    4232
    0
            DataType::Struct(fields) => {
    4233
    0
                qualified_columns.extend(fields.iter().map(|f| {
    4234
    0
                    let new_name = format!("{}.{}", col_name, f.name());
    4235
    0
                    let column = Column::from_name(&new_name);
    4236
    0
                    let new_field = f.as_ref().clone().with_name(new_name);
    4237
                    // let column = Column::from((None, &f));
    4238
    0
                    (column, Arc::new(new_field))
    4239
    0
                }))
    4240
            }
    4241
            _ => {
    4242
    0
                return internal_err!("trying to unnest on invalid data type {data_type}");
    4243
            }
    4244
        };
    4245
    0
        Ok(qualified_columns)
    4246
    0
    }
    4247
    4248
    // Get the data type of a multi-dimensional type after unnesting it
    4249
    // with a given depth
    4250
    0
    fn get_unnested_list_datatype_recursive(
    4251
    0
        data_type: &DataType,
    4252
    0
        depth: usize,
    4253
    0
    ) -> Result<DataType> {
    4254
    0
        match data_type {
    4255
    0
            DataType::List(field)
    4256
    0
            | DataType::FixedSizeList(field, _)
    4257
    0
            | DataType::LargeList(field) => {
    4258
    0
                if depth == 1 {
    4259
    0
                    return Ok(field.data_type().clone());
    4260
    0
                }
    4261
    0
                return get_unnested_list_datatype_recursive(field.data_type(), depth - 1);
    4262
            }
    4263
    0
            _ => {}
    4264
        };
    4265
    4266
    0
        internal_err!("trying to unnest on invalid data type {data_type}")
    4267
    0
    }
    4268
    4269
    #[cfg(test)]
    4270
    mod tests {
    4271
        use super::*;
    4272
        use crate::builder::LogicalTableSource;
    4273
        use crate::logical_plan::table_scan;
    4274
        use crate::select_expr::SelectExpr;
    4275
        use crate::test::function_stub::{count, count_udaf};
    4276
        use crate::{
    4277
            binary_expr, col, exists, in_subquery, lit, placeholder, scalar_subquery,
    4278
            GroupingSet,
    4279
        };
    4280
        use datafusion_common::metadata::ScalarAndMetadata;
    4281
        use datafusion_common::tree_node::{
    4282
            TransformedResult, TreeNodeRewriter, TreeNodeVisitor,
    4283
        };
    4284
        use datafusion_common::{not_impl_err, Constraint, ScalarValue};
    4285
        use insta::{assert_debug_snapshot, assert_snapshot};
    4286
        use std::hash::DefaultHasher;
    4287
    4288
        fn employee_schema() -> Schema {
    4289
            Schema::new(vec![
    4290
                Field::new("id", DataType::Int32, false),
    4291
                Field::new("first_name", DataType::Utf8, false),
    4292
                Field::new("last_name", DataType::Utf8, false),
    4293
                Field::new("state", DataType::Utf8, false),
    4294
                Field::new("salary", DataType::Int32, false),
    4295
            ])
    4296
        }
    4297
    4298
        fn display_plan() -> Result<LogicalPlan> {
    4299
            let plan1 = table_scan(Some("employee_csv"), &employee_schema(), Some(vec![3]))?
    4300
                .build()?;
    4301
    4302
            table_scan(Some("employee_csv"), &employee_schema(), Some(vec![0, 3]))?
    4303
                .filter(in_subquery(col("state"), Arc::new(plan1)))?
    4304
                .project(vec![col("id")])?
    4305
                .build()
    4306
        }
    4307
    4308
        #[test]
    4309
        fn test_display_indent() -> Result<()> {
    4310
            let plan = display_plan()?;
    4311
    4312
            assert_snapshot!(plan.display_indent(), @r"
    4313
            Projection: employee_csv.id
    4314
              Filter: employee_csv.state IN (<subquery>)
    4315
                Subquery:
    4316
                  TableScan: employee_csv projection=[state]
    4317
                TableScan: employee_csv projection=[id, state]
    4318
            ");
    4319
            Ok(())
    4320
        }
    4321
    4322
        #[test]
    4323
        fn test_display_indent_schema() -> Result<()> {
    4324
            let plan = display_plan()?;
    4325
    4326
            assert_snapshot!(plan.display_indent_schema(), @r"
    4327
            Projection: employee_csv.id [id:Int32]
    4328
              Filter: employee_csv.state IN (<subquery>) [id:Int32, state:Utf8]
    4329
                Subquery: [state:Utf8]
    4330
                  TableScan: employee_csv projection=[state] [state:Utf8]
    4331
                TableScan: employee_csv projection=[id, state] [id:Int32, state:Utf8]
    4332
            ");
    4333
            Ok(())
    4334
        }
    4335
    4336
        #[test]
    4337
        fn test_display_subquery_alias() -> Result<()> {
    4338
            let plan1 = table_scan(Some("employee_csv"), &employee_schema(), Some(vec![3]))?
    4339
                .build()?;
    4340
            let plan1 = Arc::new(plan1);
    4341
    4342
            let plan =
    4343
                table_scan(Some("employee_csv"), &employee_schema(), Some(vec![0, 3]))?
    4344
                    .project(vec![col("id"), exists(plan1).alias("exists")])?
    4345
                    .build();
    4346
    4347
            assert_snapshot!(plan?.display_indent(), @r"
    4348
            Projection: employee_csv.id, EXISTS (<subquery>) AS exists
    4349
              Subquery:
    4350
                TableScan: employee_csv projection=[state]
    4351
              TableScan: employee_csv projection=[id, state]
    4352
            ");
    4353
            Ok(())
    4354
        }
    4355
    4356
        #[test]
    4357
        fn test_display_graphviz() -> Result<()> {
    4358
            let plan = display_plan()?;
    4359
    4360
            // just test for a few key lines in the output rather than the
    4361
            // whole thing to make test maintenance easier.
    4362
            assert_snapshot!(plan.display_graphviz(), @r#"
    4363
            // Begin DataFusion GraphViz Plan,
    4364
            // display it online here: https://dreampuf.github.io/GraphvizOnline
    4365
    4366
            digraph {
    4367
              subgraph cluster_1
    4368
              {
    4369
                graph[label="LogicalPlan"]
    4370
                2[shape=box label="Projection: employee_csv.id"]
    4371
                3[shape=box label="Filter: employee_csv.state IN (<subquery>)"]
    4372
                2 -> 3 [arrowhead=none, arrowtail=normal, dir=back]
    4373
                4[shape=box label="Subquery:"]
    4374
                3 -> 4 [arrowhead=none, arrowtail=normal, dir=back]
    4375
                5[shape=box label="TableScan: employee_csv projection=[state]"]
    4376
                4 -> 5 [arrowhead=none, arrowtail=normal, dir=back]
    4377
                6[shape=box label="TableScan: employee_csv projection=[id, state]"]
    4378
                3 -> 6 [arrowhead=none, arrowtail=normal, dir=back]
    4379
              }
    4380
              subgraph cluster_7
    4381
              {
    4382
                graph[label="Detailed LogicalPlan"]
    4383
                8[shape=box label="Projection: employee_csv.id\nSchema: [id:Int32]"]
    4384
                9[shape=box label="Filter: employee_csv.state IN (<subquery>)\nSchema: [id:Int32, state:Utf8]"]
    4385
                8 -> 9 [arrowhead=none, arrowtail=normal, dir=back]
    4386
                10[shape=box label="Subquery:\nSchema: [state:Utf8]"]
    4387
                9 -> 10 [arrowhead=none, arrowtail=normal, dir=back]
    4388
                11[shape=box label="TableScan: employee_csv projection=[state]\nSchema: [state:Utf8]"]
    4389
                10 -> 11 [arrowhead=none, arrowtail=normal, dir=back]
    4390
                12[shape=box label="TableScan: employee_csv projection=[id, state]\nSchema: [id:Int32, state:Utf8]"]
    4391
                9 -> 12 [arrowhead=none, arrowtail=normal, dir=back]
    4392
              }
    4393
            }
    4394
            // End DataFusion GraphViz Plan
    4395
            "#);
    4396
            Ok(())
    4397
        }
    4398
    4399
        #[test]
    4400
        fn test_display_pg_json() -> Result<()> {
    4401
            let plan = display_plan()?;
    4402
    4403
            assert_snapshot!(plan.display_pg_json(), @r#"
    4404
            [
    4405
              {
    4406
                "Plan": {
    4407
                  "Expressions": [
    4408
                    "employee_csv.id"
    4409
                  ],
    4410
                  "Node Type": "Projection",
    4411
                  "Output": [
    4412
                    "id"
    4413
                  ],
    4414
                  "Plans": [
    4415
                    {
    4416
                      "Condition": "employee_csv.state IN (<subquery>)",
    4417
                      "Node Type": "Filter",
    4418
                      "Output": [
    4419
                        "id",
    4420
                        "state"
    4421
                      ],
    4422
                      "Plans": [
    4423
                        {
    4424
                          "Node Type": "Subquery",
    4425
                          "Output": [
    4426
                            "state"
    4427
                          ],
    4428
                          "Plans": [
    4429
                            {
    4430
                              "Node Type": "TableScan",
    4431
                              "Output": [
    4432
                                "state"
    4433
                              ],
    4434
                              "Plans": [],
    4435
                              "Relation Name": "employee_csv"
    4436
                            }
    4437
                          ]
    4438
                        },
    4439
                        {
    4440
                          "Node Type": "TableScan",
    4441
                          "Output": [
    4442
                            "id",
    4443
                            "state"
    4444
                          ],
    4445
                          "Plans": [],
    4446
                          "Relation Name": "employee_csv"
    4447
                        }
    4448
                      ]
    4449
                    }
    4450
                  ]
    4451
                }
    4452
              }
    4453
            ]
    4454
            "#);
    4455
            Ok(())
    4456
        }
    4457
    4458
        /// Tests for the Visitor trait and walking logical plan nodes
    4459
        #[derive(Debug, Default)]
    4460
        struct OkVisitor {
    4461
            strings: Vec<String>,
    4462
        }
    4463
    4464
        impl<'n> TreeNodeVisitor<'n> for OkVisitor {
    4465
            type Node = LogicalPlan;
    4466
    4467
            fn f_down(&mut self, plan: &'n LogicalPlan) -> Result<TreeNodeRecursion> {
    4468
                let s = match plan {
    4469
                    LogicalPlan::Projection { .. } => "pre_visit Projection",
    4470
                    LogicalPlan::Filter { .. } => "pre_visit Filter",
    4471
                    LogicalPlan::TableScan { .. } => "pre_visit TableScan",
    4472
                    _ => {
    4473
                        return not_impl_err!("unknown plan type");
    4474
                    }
    4475
                };
    4476
    4477
                self.strings.push(s.into());
    4478
                Ok(TreeNodeRecursion::Continue)
    4479
            }
    4480
    4481
            fn f_up(&mut self, plan: &'n LogicalPlan) -> Result<TreeNodeRecursion> {
    4482
                let s = match plan {
    4483
                    LogicalPlan::Projection { .. } => "post_visit Projection",
    4484
                    LogicalPlan::Filter { .. } => "post_visit Filter",
    4485
                    LogicalPlan::TableScan { .. } => "post_visit TableScan",
    4486
                    _ => {
    4487
                        return not_impl_err!("unknown plan type");
    4488
                    }
    4489
                };
    4490
    4491
                self.strings.push(s.into());
    4492
                Ok(TreeNodeRecursion::Continue)
    4493
            }
    4494
        }
    4495
    4496
        #[test]
    4497
        fn visit_order() {
    4498
            let mut visitor = OkVisitor::default();
    4499
            let plan = test_plan();
    4500
            let res = plan.visit_with_subqueries(&mut visitor);
    4501
            assert!(res.is_ok());
    4502
    4503
            assert_debug_snapshot!(visitor.strings, @r#"
    4504
            [
    4505
                "pre_visit Projection",
    4506
                "pre_visit Filter",
    4507
                "pre_visit TableScan",
    4508
                "post_visit TableScan",
    4509
                "post_visit Filter",
    4510
                "post_visit Projection",
    4511
            ]
    4512
            "#);
    4513
        }
    4514
    4515
        #[derive(Debug, Default)]
    4516
        /// Counter than counts to zero and returns true when it gets there
    4517
        struct OptionalCounter {
    4518
            val: Option<usize>,
    4519
        }
    4520
    4521
        impl OptionalCounter {
    4522
            fn new(val: usize) -> Self {
    4523
                Self { val: Some(val) }
    4524
            }
    4525
            // Decrements the counter by 1, if any, returning true if it hits zero
    4526
            fn dec(&mut self) -> bool {
    4527
                if Some(0) == self.val {
    4528
                    true
    4529
                } else {
    4530
                    self.val = self.val.take().map(|i| i - 1);
    4531
                    false
    4532
                }
    4533
            }
    4534
        }
    4535
    4536
        #[derive(Debug, Default)]
    4537
        /// Visitor that returns false after some number of visits
    4538
        struct StoppingVisitor {
    4539
            inner: OkVisitor,
    4540
            /// When Some(0) returns false from pre_visit
    4541
            return_false_from_pre_in: OptionalCounter,
    4542
            /// When Some(0) returns false from post_visit
    4543
            return_false_from_post_in: OptionalCounter,
    4544
        }
    4545
    4546
        impl<'n> TreeNodeVisitor<'n> for StoppingVisitor {
    4547
            type Node = LogicalPlan;
    4548
    4549
            fn f_down(&mut self, plan: &'n LogicalPlan) -> Result<TreeNodeRecursion> {
    4550
                if self.return_false_from_pre_in.dec() {
    4551
                    return Ok(TreeNodeRecursion::Stop);
    4552
                }
    4553
                self.inner.f_down(plan)?;
    4554
    4555
                Ok(TreeNodeRecursion::Continue)
    4556
            }
    4557
    4558
            fn f_up(&mut self, plan: &'n LogicalPlan) -> Result<TreeNodeRecursion> {
    4559
                if self.return_false_from_post_in.dec() {
    4560
                    return Ok(TreeNodeRecursion::Stop);
    4561
                }
    4562
    4563
                self.inner.f_up(plan)
    4564
            }
    4565
        }
    4566
    4567
        /// test early stopping in pre-visit
    4568
        #[test]
    4569
        fn early_stopping_pre_visit() {
    4570
            let mut visitor = StoppingVisitor {
    4571
                return_false_from_pre_in: OptionalCounter::new(2),
    4572
                ..Default::default()
    4573
            };
    4574
            let plan = test_plan();
    4575
            let res = plan.visit_with_subqueries(&mut visitor);
    4576
            assert!(res.is_ok());
    4577
    4578
            assert_debug_snapshot!(
    4579
                visitor.inner.strings,
    4580
                @r#"
    4581
            [
    4582
                "pre_visit Projection",
    4583
                "pre_visit Filter",
    4584
            ]
    4585
            "#
    4586
            );
    4587
        }
    4588
    4589
        #[test]
    4590
        fn early_stopping_post_visit() {
    4591
            let mut visitor = StoppingVisitor {
    4592
                return_false_from_post_in: OptionalCounter::new(1),
    4593
                ..Default::default()
    4594
            };
    4595
            let plan = test_plan();
    4596
            let res = plan.visit_with_subqueries(&mut visitor);
    4597
            assert!(res.is_ok());
    4598
    4599
            assert_debug_snapshot!(
    4600
                visitor.inner.strings,
    4601
                @r#"
    4602
            [
    4603
                "pre_visit Projection",
    4604
                "pre_visit Filter",
    4605
                "pre_visit TableScan",
    4606
                "post_visit TableScan",
    4607
            ]
    4608
            "#
    4609
            );
    4610
        }
    4611
    4612
        #[derive(Debug, Default)]
    4613
        /// Visitor that returns an error after some number of visits
    4614
        struct ErrorVisitor {
    4615
            inner: OkVisitor,
    4616
            /// When Some(0) returns false from pre_visit
    4617
            return_error_from_pre_in: OptionalCounter,
    4618
            /// When Some(0) returns false from post_visit
    4619
            return_error_from_post_in: OptionalCounter,
    4620
        }
    4621
    4622
        impl<'n> TreeNodeVisitor<'n> for ErrorVisitor {
    4623
            type Node = LogicalPlan;
    4624
    4625
            fn f_down(&mut self, plan: &'n LogicalPlan) -> Result<TreeNodeRecursion> {
    4626
                if self.return_error_from_pre_in.dec() {
    4627
                    return not_impl_err!("Error in pre_visit");
    4628
                }
    4629
    4630
                self.inner.f_down(plan)
    4631
            }
    4632
    4633
            fn f_up(&mut self, plan: &'n LogicalPlan) -> Result<TreeNodeRecursion> {
    4634
                if self.return_error_from_post_in.dec() {
    4635
                    return not_impl_err!("Error in post_visit");
    4636
                }
    4637
    4638
                self.inner.f_up(plan)
    4639
            }
    4640
        }
    4641
    4642
        #[test]
    4643
        fn error_pre_visit() {
    4644
            let mut visitor = ErrorVisitor {
    4645
                return_error_from_pre_in: OptionalCounter::new(2),
    4646
                ..Default::default()
    4647
            };
    4648
            let plan = test_plan();
    4649
            let res = plan.visit_with_subqueries(&mut visitor).unwrap_err();
    4650
            assert_snapshot!(
    4651
                res.strip_backtrace(),
    4652
                @"This feature is not implemented: Error in pre_visit"
    4653
            );
    4654
            assert_debug_snapshot!(
    4655
                visitor.inner.strings,
    4656
                @r#"
    4657
            [
    4658
                "pre_visit Projection",
    4659
                "pre_visit Filter",
    4660
            ]
    4661
            "#
    4662
            );
    4663
        }
    4664
    4665
        #[test]
    4666
        fn error_post_visit() {
    4667
            let mut visitor = ErrorVisitor {
    4668
                return_error_from_post_in: OptionalCounter::new(1),
    4669
                ..Default::default()
    4670
            };
    4671
            let plan = test_plan();
    4672
            let res = plan.visit_with_subqueries(&mut visitor).unwrap_err();
    4673
            assert_snapshot!(
    4674
                res.strip_backtrace(),
    4675
                @"This feature is not implemented: Error in post_visit"
    4676
            );
    4677
            assert_debug_snapshot!(
    4678
                visitor.inner.strings,
    4679
                @r#"
    4680
            [
    4681
                "pre_visit Projection",
    4682
                "pre_visit Filter",
    4683
                "pre_visit TableScan",
    4684
                "post_visit TableScan",
    4685
            ]
    4686
            "#
    4687
            );
    4688
        }
    4689
    4690
        #[test]
    4691
        fn test_partial_eq_hash_and_partial_ord() {
    4692
            let empty_values = Arc::new(LogicalPlan::EmptyRelation(EmptyRelation {
    4693
                produce_one_row: true,
    4694
                schema: Arc::new(DFSchema::empty()),
    4695
            }));
    4696
    4697
            let count_window_function = |schema| {
    4698
                Window::try_new_with_schema(
    4699
                    vec![Expr::WindowFunction(Box::new(WindowFunction::new(
    4700
                        WindowFunctionDefinition::AggregateUDF(count_udaf()),
    4701
                        vec![],
    4702
                    )))],
    4703
                    Arc::clone(&empty_values),
    4704
                    Arc::new(schema),
    4705
                )
    4706
                .unwrap()
    4707
            };
    4708
    4709
            let schema_without_metadata = || {
    4710
                DFSchema::from_unqualified_fields(
    4711
                    vec![Field::new("count", DataType::Int64, false)].into(),
    4712
                    HashMap::new(),
    4713
                )
    4714
                .unwrap()
    4715
            };
    4716
    4717
            let schema_with_metadata = || {
    4718
                DFSchema::from_unqualified_fields(
    4719
                    vec![Field::new("count", DataType::Int64, false)].into(),
    4720
                    [("key".to_string(), "value".to_string())].into(),
    4721
                )
    4722
                .unwrap()
    4723
            };
    4724
    4725
            // A Window
    4726
            let f = count_window_function(schema_without_metadata());
    4727
    4728
            // Same like `f`, different instance
    4729
            let f2 = count_window_function(schema_without_metadata());
    4730
            assert_eq!(f, f2);
    4731
            assert_eq!(hash(&f), hash(&f2));
    4732
            assert_eq!(f.partial_cmp(&f2), Some(Ordering::Equal));
    4733
    4734
            // Same like `f`, except for schema metadata
    4735
            let o = count_window_function(schema_with_metadata());
    4736
            assert_ne!(f, o);
    4737
            assert_ne!(hash(&f), hash(&o)); // hash can collide for different values but does not collide in this test
    4738
            assert_eq!(f.partial_cmp(&o), None);
    4739
        }
    4740
    4741
        fn hash<T: Hash>(value: &T) -> u64 {
    4742
            let hasher = &mut DefaultHasher::new();
    4743
            value.hash(hasher);
    4744
            hasher.finish()
    4745
        }
    4746
    4747
        #[test]
    4748
        fn projection_expr_schema_mismatch() -> Result<()> {
    4749
            let empty_schema = Arc::new(DFSchema::empty());
    4750
            let p = Projection::try_new_with_schema(
    4751
                vec![col("a")],
    4752
                Arc::new(LogicalPlan::EmptyRelation(EmptyRelation {
    4753
                    produce_one_row: false,
    4754
                    schema: Arc::clone(&empty_schema),
    4755
                })),
    4756
                empty_schema,
    4757
            );
    4758
            assert_snapshot!(p.unwrap_err().strip_backtrace(), @"Error during planning: Projection has mismatch between number of expressions (1) and number of fields in schema (0)");
    4759
            Ok(())
    4760
        }
    4761
    4762
        fn test_plan() -> LogicalPlan {
    4763
            let schema = Schema::new(vec![
    4764
                Field::new("id", DataType::Int32, false),
    4765
                Field::new("state", DataType::Utf8, false),
    4766
            ]);
    4767
    4768
            table_scan(TableReference::none(), &schema, Some(vec![0, 1]))
    4769
                .unwrap()
    4770
                .filter(col("state").eq(lit("CO")))
    4771
                .unwrap()
    4772
                .project(vec![col("id")])
    4773
                .unwrap()
    4774
                .build()
    4775
                .unwrap()
    4776
        }
    4777
    4778
        #[test]
    4779
        fn test_replace_invalid_placeholder() {
    4780
            // test empty placeholder
    4781
            let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    4782
    4783
            let plan = table_scan(TableReference::none(), &schema, None)
    4784
                .unwrap()
    4785
                .filter(col("id").eq(placeholder("")))
    4786
                .unwrap()
    4787
                .build()
    4788
                .unwrap();
    4789
    4790
            let param_values = vec![ScalarValue::Int32(Some(42))];
    4791
            plan.replace_params_with_values(&param_values.clone().into())
    4792
                .expect_err("unexpectedly succeeded to replace an invalid placeholder");
    4793
    4794
            // test $0 placeholder
    4795
            let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    4796
    4797
            let plan = table_scan(TableReference::none(), &schema, None)
    4798
                .unwrap()
    4799
                .filter(col("id").eq(placeholder("$0")))
    4800
                .unwrap()
    4801
                .build()
    4802
                .unwrap();
    4803
    4804
            plan.replace_params_with_values(&param_values.clone().into())
    4805
                .expect_err("unexpectedly succeeded to replace an invalid placeholder");
    4806
    4807
            // test $00 placeholder
    4808
            let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    4809
    4810
            let plan = table_scan(TableReference::none(), &schema, None)
    4811
                .unwrap()
    4812
                .filter(col("id").eq(placeholder("$00")))
    4813
                .unwrap()
    4814
                .build()
    4815
                .unwrap();
    4816
    4817
            plan.replace_params_with_values(&param_values.into())
    4818
                .expect_err("unexpectedly succeeded to replace an invalid placeholder");
    4819
        }
    4820
    4821
        #[test]
    4822
        fn test_replace_placeholder_mismatched_metadata() {
    4823
            let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    4824
    4825
            // Create a prepared statement with explicit fields that do not have metadata
    4826
            let plan = table_scan(TableReference::none(), &schema, None)
    4827
                .unwrap()
    4828
                .filter(col("id").eq(placeholder("$1")))
    4829
                .unwrap()
    4830
                .build()
    4831
                .unwrap();
    4832
            let prepared_builder = LogicalPlanBuilder::new(plan)
    4833
                .prepare(
    4834
                    "".to_string(),
    4835
                    vec![Field::new("", DataType::Int32, true).into()],
    4836
                )
    4837
                .unwrap();
    4838
    4839
            // Attempt to bind a parameter with metadata
    4840
            let mut scalar_meta = HashMap::new();
    4841
            scalar_meta.insert("some_key".to_string(), "some_value".to_string());
    4842
            let param_values = ParamValues::List(vec![ScalarAndMetadata::new(
    4843
                ScalarValue::Int32(Some(42)),
    4844
                Some(scalar_meta.into()),
    4845
            )]);
    4846
            prepared_builder
    4847
                .plan()
    4848
                .clone()
    4849
                .with_param_values(param_values)
    4850
                .expect_err("prepared field metadata mismatch unexpectedly succeeded");
    4851
        }
    4852
    4853
        #[test]
    4854
        fn test_replace_placeholder_empty_relation_valid_schema() {
    4855
            // SELECT $1, $2;
    4856
            let plan = LogicalPlanBuilder::empty(false)
    4857
                .project(vec![
    4858
                    SelectExpr::from(placeholder("$1")),
    4859
                    SelectExpr::from(placeholder("$2")),
    4860
                ])
    4861
                .unwrap()
    4862
                .build()
    4863
                .unwrap();
    4864
    4865
            // original
    4866
            assert_snapshot!(plan.display_indent_schema(), @r"
    4867
            Projection: $1, $2 [$1:Null;N, $2:Null;N]
    4868
              EmptyRelation: rows=0 []
    4869
            ");
    4870
    4871
            let plan = plan
    4872
                .with_param_values(vec![ScalarValue::from(1i32), ScalarValue::from("s")])
    4873
                .unwrap();
    4874
    4875
            // replaced
    4876
            assert_snapshot!(plan.display_indent_schema(), @r#"
    4877
            Projection: Int32(1) AS $1, Utf8("s") AS $2 [$1:Int32, $2:Utf8]
    4878
              EmptyRelation: rows=0 []
    4879
            "#);
    4880
        }
    4881
    4882
        #[test]
    4883
        fn test_nullable_schema_after_grouping_set() {
    4884
            let schema = Schema::new(vec![
    4885
                Field::new("foo", DataType::Int32, false),
    4886
                Field::new("bar", DataType::Int32, false),
    4887
            ]);
    4888
    4889
            let plan = table_scan(TableReference::none(), &schema, None)
    4890
                .unwrap()
    4891
                .aggregate(
    4892
                    vec![Expr::GroupingSet(GroupingSet::GroupingSets(vec![
    4893
                        vec![col("foo")],
    4894
                        vec![col("bar")],
    4895
                    ]))],
    4896
                    vec![count(lit(true))],
    4897
                )
    4898
                .unwrap()
    4899
                .build()
    4900
                .unwrap();
    4901
    4902
            let output_schema = plan.schema();
    4903
    4904
            assert!(output_schema
    4905
                .field_with_name(None, "foo")
    4906
                .unwrap()
    4907
                .is_nullable(),);
    4908
            assert!(output_schema
    4909
                .field_with_name(None, "bar")
    4910
                .unwrap()
    4911
                .is_nullable());
    4912
        }
    4913
    4914
        #[test]
    4915
        fn test_filter_is_scalar() {
    4916
            // test empty placeholder
    4917
            let schema =
    4918
                Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
    4919
    4920
            let source = Arc::new(LogicalTableSource::new(schema));
    4921
            let schema = Arc::new(
    4922
                DFSchema::try_from_qualified_schema(
    4923
                    TableReference::bare("tab"),
    4924
                    &source.schema(),
    4925
                )
    4926
                .unwrap(),
    4927
            );
    4928
            let scan = Arc::new(LogicalPlan::TableScan(TableScan {
    4929
                table_name: TableReference::bare("tab"),
    4930
                source: Arc::clone(&source) as Arc<dyn TableSource>,
    4931
                projection: None,
    4932
                projected_schema: Arc::clone(&schema),
    4933
                filters: vec![],
    4934
                fetch: None,
    4935
            }));
    4936
            let col = schema.field_names()[0].clone();
    4937
    4938
            let filter = Filter::try_new(
    4939
                Expr::Column(col.into()).eq(Expr::Literal(ScalarValue::Int32(Some(1)), None)),
    4940
                scan,
    4941
            )
    4942
            .unwrap();
    4943
            assert!(!filter.is_scalar());
    4944
            let unique_schema = Arc::new(
    4945
                schema
    4946
                    .as_ref()
    4947
                    .clone()
    4948
                    .with_functional_dependencies(
    4949
                        FunctionalDependencies::new_from_constraints(
    4950
                            Some(&Constraints::new_unverified(vec![Constraint::Unique(
    4951
                                vec![0],
    4952
                            )])),
    4953
                            1,
    4954
                        ),
    4955
                    )
    4956
                    .unwrap(),
    4957
            );
    4958
            let scan = Arc::new(LogicalPlan::TableScan(TableScan {
    4959
                table_name: TableReference::bare("tab"),
    4960
                source,
    4961
                projection: None,
    4962
                projected_schema: Arc::clone(&unique_schema),
    4963
                filters: vec![],
    4964
                fetch: None,
    4965
            }));
    4966
            let col = schema.field_names()[0].clone();
    4967
    4968
            let filter =
    4969
                Filter::try_new(Expr::Column(col.into()).eq(lit(1i32)), scan).unwrap();
    4970
            assert!(filter.is_scalar());
    4971
        }
    4972
    4973
        #[test]
    4974
        fn test_transform_explain() {
    4975
            let schema = Schema::new(vec![
    4976
                Field::new("foo", DataType::Int32, false),
    4977
                Field::new("bar", DataType::Int32, false),
    4978
            ]);
    4979
    4980
            let plan = table_scan(TableReference::none(), &schema, None)
    4981
                .unwrap()
    4982
                .explain(false, false)
    4983
                .unwrap()
    4984
                .build()
    4985
                .unwrap();
    4986
    4987
            let external_filter = col("foo").eq(lit(true));
    4988
    4989
            // after transformation, because plan is not the same anymore,
    4990
            // the parent plan is built again with call to LogicalPlan::with_new_inputs -> with_new_exprs
    4991
            let plan = plan
    4992
                .transform(|plan| match plan {
    4993
                    LogicalPlan::TableScan(table) => {
    4994
                        let filter = Filter::try_new(
    4995
                            external_filter.clone(),
    4996
                            Arc::new(LogicalPlan::TableScan(table)),
    4997
                        )
    4998
                        .unwrap();
    4999
                        Ok(Transformed::yes(LogicalPlan::Filter(filter)))
    5000
                    }
    5001
                    x => Ok(Transformed::no(x)),
    5002
                })
    5003
                .data()
    5004
                .unwrap();
    5005
    5006
            let actual = format!("{}", plan.display_indent());
    5007
            assert_snapshot!(actual, @r"
    5008
            Explain
    5009
              Filter: foo = Boolean(true)
    5010
                TableScan: ?table?
    5011
            ")
    5012
        }
    5013
    5014
        #[test]
    5015
        fn test_plan_partial_ord() {
    5016
            let empty_relation = LogicalPlan::EmptyRelation(EmptyRelation {
    5017
                produce_one_row: false,
    5018
                schema: Arc::new(DFSchema::empty()),
    5019
            });
    5020
    5021
            let describe_table = LogicalPlan::DescribeTable(DescribeTable {
    5022
                schema: Arc::new(Schema::new(vec![Field::new(
    5023
                    "foo",
    5024
                    DataType::Int32,
    5025
                    false,
    5026
                )])),
    5027
                output_schema: DFSchemaRef::new(DFSchema::empty()),
    5028
            });
    5029
    5030
            let describe_table_clone = LogicalPlan::DescribeTable(DescribeTable {
    5031
                schema: Arc::new(Schema::new(vec![Field::new(
    5032
                    "foo",
    5033
                    DataType::Int32,
    5034
                    false,
    5035
                )])),
    5036
                output_schema: DFSchemaRef::new(DFSchema::empty()),
    5037
            });
    5038
    5039
            assert_eq!(
    5040
                empty_relation.partial_cmp(&describe_table),
    5041
                Some(Ordering::Less)
    5042
            );
    5043
            assert_eq!(
    5044
                describe_table.partial_cmp(&empty_relation),
    5045
                Some(Ordering::Greater)
    5046
            );
    5047
            assert_eq!(describe_table.partial_cmp(&describe_table_clone), None);
    5048
        }
    5049
    5050
        #[test]
    5051
        fn test_limit_with_new_children() {
    5052
            let input = Arc::new(LogicalPlan::Values(Values {
    5053
                schema: Arc::new(DFSchema::empty()),
    5054
                values: vec![vec![]],
    5055
            }));
    5056
            let cases = [
    5057
                LogicalPlan::Limit(Limit {
    5058
                    skip: None,
    5059
                    fetch: None,
    5060
                    input: Arc::clone(&input),
    5061
                }),
    5062
                LogicalPlan::Limit(Limit {
    5063
                    skip: None,
    5064
                    fetch: Some(Box::new(Expr::Literal(
    5065
                        ScalarValue::new_ten(&DataType::UInt32).unwrap(),
    5066
                        None,
    5067
                    ))),
    5068
                    input: Arc::clone(&input),
    5069
                }),
    5070
                LogicalPlan::Limit(Limit {
    5071
                    skip: Some(Box::new(Expr::Literal(
    5072
                        ScalarValue::new_ten(&DataType::UInt32).unwrap(),
    5073
                        None,
    5074
                    ))),
    5075
                    fetch: None,
    5076
                    input: Arc::clone(&input),
    5077
                }),
    5078
                LogicalPlan::Limit(Limit {
    5079
                    skip: Some(Box::new(Expr::Literal(
    5080
                        ScalarValue::new_one(&DataType::UInt32).unwrap(),
    5081
                        None,
    5082
                    ))),
    5083
                    fetch: Some(Box::new(Expr::Literal(
    5084
                        ScalarValue::new_ten(&DataType::UInt32).unwrap(),
    5085
                        None,
    5086
                    ))),
    5087
                    input,
    5088
                }),
    5089
            ];
    5090
    5091
            for limit in cases {
    5092
                let new_limit = limit
    5093
                    .with_new_exprs(
    5094
                        limit.expressions(),
    5095
                        limit.inputs().into_iter().cloned().collect(),
    5096
                    )
    5097
                    .unwrap();
    5098
                assert_eq!(limit, new_limit);
    5099
            }
    5100
        }
    5101
    5102
        #[test]
    5103
        fn test_with_subqueries_jump() {
    5104
            // The test plan contains a `Project` node above a `Filter` node, and the
    5105
            // `Project` node contains a subquery plan with a `Filter` root node, so returning
    5106
            // `TreeNodeRecursion::Jump` on `Project` should cause not visiting any of the
    5107
            // `Filter`s.
    5108
            let subquery_schema =
    5109
                Schema::new(vec![Field::new("sub_id", DataType::Int32, false)]);
    5110
    5111
            let subquery_plan =
    5112
                table_scan(TableReference::none(), &subquery_schema, Some(vec![0]))
    5113
                    .unwrap()
    5114
                    .filter(col("sub_id").eq(lit(0)))
    5115
                    .unwrap()
    5116
                    .build()
    5117
                    .unwrap();
    5118
    5119
            let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
    5120
    5121
            let plan = table_scan(TableReference::none(), &schema, Some(vec![0]))
    5122
                .unwrap()
    5123
                .filter(col("id").eq(lit(0)))
    5124
                .unwrap()
    5125
                .project(vec![col("id"), scalar_subquery(Arc::new(subquery_plan))])
    5126
                .unwrap()
    5127
                .build()
    5128
                .unwrap();
    5129
    5130
            let mut filter_found = false;
    5131
            plan.apply_with_subqueries(|plan| {
    5132
                match plan {
    5133
                    LogicalPlan::Projection(..) => return Ok(TreeNodeRecursion::Jump),
    5134
                    LogicalPlan::Filter(..) => filter_found = true,
    5135
                    _ => {}
    5136
                }
    5137
                Ok(TreeNodeRecursion::Continue)
    5138
            })
    5139
            .unwrap();
    5140
            assert!(!filter_found);
    5141
    5142
            struct ProjectJumpVisitor {
    5143
                filter_found: bool,
    5144
            }
    5145
    5146
            impl ProjectJumpVisitor {
    5147
                fn new() -> Self {
    5148
                    Self {
    5149
                        filter_found: false,
    5150
                    }
    5151
                }
    5152
            }
    5153
    5154
            impl<'n> TreeNodeVisitor<'n> for ProjectJumpVisitor {
    5155
                type Node = LogicalPlan;
    5156
    5157
                fn f_down(&mut self, node: &'n Self::Node) -> Result<TreeNodeRecursion> {
    5158
                    match node {
    5159
                        LogicalPlan::Projection(..) => return Ok(TreeNodeRecursion::Jump),
    5160
                        LogicalPlan::Filter(..) => self.filter_found = true,
    5161
                        _ => {}
    5162
                    }
    5163
                    Ok(TreeNodeRecursion::Continue)
    5164
                }
    5165
            }
    5166
    5167
            let mut visitor = ProjectJumpVisitor::new();
    5168
            plan.visit_with_subqueries(&mut visitor).unwrap();
    5169
            assert!(!visitor.filter_found);
    5170
    5171
            let mut filter_found = false;
    5172
            plan.clone()
    5173
                .transform_down_with_subqueries(|plan| {
    5174
                    match plan {
    5175
                        LogicalPlan::Projection(..) => {
    5176
                            return Ok(Transformed::new(plan, false, TreeNodeRecursion::Jump))
    5177
                        }
    5178
                        LogicalPlan::Filter(..) => filter_found = true,
    5179
                        _ => {}
    5180
                    }
    5181
                    Ok(Transformed::no(plan))
    5182
                })
    5183
                .unwrap();
    5184
            assert!(!filter_found);
    5185
    5186
            let mut filter_found = false;
    5187
            plan.clone()
    5188
                .transform_down_up_with_subqueries(
    5189
                    |plan| {
    5190
                        match plan {
    5191
                            LogicalPlan::Projection(..) => {
    5192
                                return Ok(Transformed::new(
    5193
                                    plan,
    5194
                                    false,
    5195
                                    TreeNodeRecursion::Jump,
    5196
                                ))
    5197
                            }
    5198
                            LogicalPlan::Filter(..) => filter_found = true,
    5199
                            _ => {}
    5200
                        }
    5201
                        Ok(Transformed::no(plan))
    5202
                    },
    5203
                    |plan| Ok(Transformed::no(plan)),
    5204
                )
    5205
                .unwrap();
    5206
            assert!(!filter_found);
    5207
    5208
            struct ProjectJumpRewriter {
    5209
                filter_found: bool,
    5210
            }
    5211
    5212
            impl ProjectJumpRewriter {
    5213
                fn new() -> Self {
    5214
                    Self {
    5215
                        filter_found: false,
    5216
                    }
    5217
                }
    5218
            }
    5219
    5220
            impl TreeNodeRewriter for ProjectJumpRewriter {
    5221
                type Node = LogicalPlan;
    5222
    5223
                fn f_down(&mut self, node: Self::Node) -> Result<Transformed<Self::Node>> {
    5224
                    match node {
    5225
                        LogicalPlan::Projection(..) => {
    5226
                            return Ok(Transformed::new(node, false, TreeNodeRecursion::Jump))
    5227
                        }
    5228
                        LogicalPlan::Filter(..) => self.filter_found = true,
    5229
                        _ => {}
    5230
                    }
    5231
                    Ok(Transformed::no(node))
    5232
                }
    5233
            }
    5234
    5235
            let mut rewriter = ProjectJumpRewriter::new();
    5236
            plan.rewrite_with_subqueries(&mut rewriter).unwrap();
    5237
            assert!(!rewriter.filter_found);
    5238
        }
    5239
    5240
        #[test]
    5241
        fn test_with_unresolved_placeholders() {
    5242
            let field_name = "id";
    5243
            let placeholder_value = "$1";
    5244
            let schema = Schema::new(vec![Field::new(field_name, DataType::Int32, false)]);
    5245
    5246
            let plan = table_scan(TableReference::none(), &schema, None)
    5247
                .unwrap()
    5248
                .filter(col(field_name).eq(placeholder(placeholder_value)))
    5249
                .unwrap()
    5250
                .build()
    5251
                .unwrap();
    5252
    5253
            // Check that the placeholder parameters have not received a DataType.
    5254
            let params = plan.get_parameter_fields().unwrap();
    5255
            assert_eq!(params.len(), 1);
    5256
    5257
            let parameter_type = params.clone().get(placeholder_value).unwrap().clone();
    5258
            assert_eq!(parameter_type, None);
    5259
        }
    5260
    5261
        #[test]
    5262
        fn test_join_with_new_exprs() -> Result<()> {
    5263
            fn create_test_join(
    5264
                on: Vec<(Expr, Expr)>,
    5265
                filter: Option<Expr>,
    5266
            ) -> Result<LogicalPlan> {
    5267
                let schema = Schema::new(vec![
    5268
                    Field::new("a", DataType::Int32, false),
    5269
                    Field::new("b", DataType::Int32, false),
    5270
                ]);
    5271
    5272
                let left_schema = DFSchema::try_from_qualified_schema("t1", &schema)?;
    5273
                let right_schema = DFSchema::try_from_qualified_schema("t2", &schema)?;
    5274
    5275
                Ok(LogicalPlan::Join(Join {
    5276
                    left: Arc::new(
    5277
                        table_scan(Some("t1"), left_schema.as_arrow(), None)?.build()?,
    5278
                    ),
    5279
                    right: Arc::new(
    5280
                        table_scan(Some("t2"), right_schema.as_arrow(), None)?.build()?,
    5281
                    ),
    5282
                    on,
    5283
                    filter,
    5284
                    join_type: JoinType::Inner,
    5285
                    join_constraint: JoinConstraint::On,
    5286
                    schema: Arc::new(left_schema.join(&right_schema)?),
    5287
                    null_equality: NullEquality::NullEqualsNothing,
    5288
                }))
    5289
            }
    5290
    5291
            {
    5292
                let join = create_test_join(vec![(col("t1.a"), (col("t2.a")))], None)?;
    5293
                let LogicalPlan::Join(join) = join.with_new_exprs(
    5294
                    join.expressions(),
    5295
                    join.inputs().into_iter().cloned().collect(),
    5296
                )?
    5297
                else {
    5298
                    unreachable!()
    5299
                };
    5300
                assert_eq!(join.on, vec![(col("t1.a"), (col("t2.a")))]);
    5301
                assert_eq!(join.filter, None);
    5302
            }
    5303
    5304
            {
    5305
                let join = create_test_join(vec![], Some(col("t1.a").gt(col("t2.a"))))?;
    5306
                let LogicalPlan::Join(join) = join.with_new_exprs(
    5307
                    join.expressions(),
    5308
                    join.inputs().into_iter().cloned().collect(),
    5309
                )?
    5310
                else {
    5311
                    unreachable!()
    5312
                };
    5313
                assert_eq!(join.on, vec![]);
    5314
                assert_eq!(join.filter, Some(col("t1.a").gt(col("t2.a"))));
    5315
            }
    5316
    5317
            {
    5318
                let join = create_test_join(
    5319
                    vec![(col("t1.a"), (col("t2.a")))],
    5320
                    Some(col("t1.b").gt(col("t2.b"))),
    5321
                )?;
    5322
                let LogicalPlan::Join(join) = join.with_new_exprs(
    5323
                    join.expressions(),
    5324
                    join.inputs().into_iter().cloned().collect(),
    5325
                )?
    5326
                else {
    5327
                    unreachable!()
    5328
                };
    5329
                assert_eq!(join.on, vec![(col("t1.a"), (col("t2.a")))]);
    5330
                assert_eq!(join.filter, Some(col("t1.b").gt(col("t2.b"))));
    5331
            }
    5332
    5333
            {
    5334
                let join = create_test_join(
    5335
                    vec![(col("t1.a"), (col("t2.a"))), (col("t1.b"), (col("t2.b")))],
    5336
                    None,
    5337
                )?;
    5338
                let LogicalPlan::Join(join) = join.with_new_exprs(
    5339
                    vec![
    5340
                        binary_expr(col("t1.a"), Operator::Plus, lit(1)),
    5341
                        binary_expr(col("t2.a"), Operator::Plus, lit(2)),
    5342
                        col("t1.b"),
    5343
                        col("t2.b"),
    5344
                        lit(true),
    5345
                    ],
    5346
                    join.inputs().into_iter().cloned().collect(),
    5347
                )?
    5348
                else {
    5349
                    unreachable!()
    5350
                };
    5351
                assert_eq!(
    5352
                    join.on,
    5353
                    vec![
    5354
                        (
    5355
                            binary_expr(col("t1.a"), Operator::Plus, lit(1)),
    5356
                            binary_expr(col("t2.a"), Operator::Plus, lit(2))
    5357
                        ),
    5358
                        (col("t1.b"), (col("t2.b")))
    5359
                    ]
    5360
                );
    5361
                assert_eq!(join.filter, Some(lit(true)));
    5362
            }
    5363
    5364
            Ok(())
    5365
        }
    5366
    5367
        #[test]
    5368
        fn test_join_try_new() -> Result<()> {
    5369
            let schema = Schema::new(vec![
    5370
                Field::new("a", DataType::Int32, false),
    5371
                Field::new("b", DataType::Int32, false),
    5372
            ]);
    5373
    5374
            let left_scan = table_scan(Some("t1"), &schema, None)?.build()?;
    5375
    5376
            let right_scan = table_scan(Some("t2"), &schema, None)?.build()?;
    5377
    5378
            let join_types = vec![
    5379
                JoinType::Inner,
    5380
                JoinType::Left,
    5381
                JoinType::Right,
    5382
                JoinType::Full,
    5383
                JoinType::LeftSemi,
    5384
                JoinType::LeftAnti,
    5385
                JoinType::RightSemi,
    5386
                JoinType::RightAnti,
    5387
                JoinType::LeftMark,
    5388
            ];
    5389
    5390
            for join_type in join_types {
    5391
                let join = Join::try_new(
    5392
                    Arc::new(left_scan.clone()),
    5393
                    Arc::new(right_scan.clone()),
    5394
                    vec![(col("t1.a"), col("t2.a"))],
    5395
                    Some(col("t1.b").gt(col("t2.b"))),
    5396
                    join_type,
    5397
                    JoinConstraint::On,
    5398
                    NullEquality::NullEqualsNothing,
    5399
                )?;
    5400
    5401
                match join_type {
    5402
                    JoinType::LeftSemi | JoinType::LeftAnti => {
    5403
                        assert_eq!(join.schema.fields().len(), 2);
    5404
    5405
                        let fields = join.schema.fields();
    5406
                        assert_eq!(
    5407
                            fields[0].name(),
    5408
                            "a",
    5409
                            "First field should be 'a' from left table"
    5410
                        );
    5411
                        assert_eq!(
    5412
                            fields[1].name(),
    5413
                            "b",
    5414
                            "Second field should be 'b' from left table"
    5415
                        );
    5416
                    }
    5417
                    JoinType::RightSemi | JoinType::RightAnti => {
    5418
                        assert_eq!(join.schema.fields().len(), 2);
    5419
    5420
                        let fields = join.schema.fields();
    5421
                        assert_eq!(
    5422
                            fields[0].name(),
    5423
                            "a",
    5424
                            "First field should be 'a' from right table"
    5425
                        );
    5426
                        assert_eq!(
    5427
                            fields[1].name(),
    5428
                            "b",
    5429
                            "Second field should be 'b' from right table"
    5430
                        );
    5431
                    }
    5432
                    JoinType::LeftMark => {
    5433
                        assert_eq!(join.schema.fields().len(), 3);
    5434
    5435
                        let fields = join.schema.fields();
    5436
                        assert_eq!(
    5437
                            fields[0].name(),
    5438
                            "a",
    5439
                            "First field should be 'a' from left table"
    5440
                        );
    5441
                        assert_eq!(
    5442
                            fields[1].name(),
    5443
                            "b",
    5444
                            "Second field should be 'b' from left table"
    5445
                        );
    5446
                        assert_eq!(
    5447
                            fields[2].name(),
    5448
                            "mark",
    5449
                            "Third field should be the mark column"
    5450
                        );
    5451
    5452
                        assert!(!fields[0].is_nullable());
    5453
                        assert!(!fields[1].is_nullable());
    5454
                        assert!(!fields[2].is_nullable());
    5455
                    }
    5456
                    _ => {
    5457
                        assert_eq!(join.schema.fields().len(), 4);
    5458
    5459
                        let fields = join.schema.fields();
    5460
                        assert_eq!(
    5461
                            fields[0].name(),
    5462
                            "a",
    5463
                            "First field should be 'a' from left table"
    5464
                        );
    5465
                        assert_eq!(
    5466
                            fields[1].name(),
    5467
                            "b",
    5468
                            "Second field should be 'b' from left table"
    5469
                        );
    5470
                        assert_eq!(
    5471
                            fields[2].name(),
    5472
                            "a",
    5473
                            "Third field should be 'a' from right table"
    5474
                        );
    5475
                        assert_eq!(
    5476
                            fields[3].name(),
    5477
                            "b",
    5478
                            "Fourth field should be 'b' from right table"
    5479
                        );
    5480
    5481
                        if join_type == JoinType::Left {
    5482
                            // Left side fields (first two) shouldn't be nullable
    5483
                            assert!(!fields[0].is_nullable());
    5484
                            assert!(!fields[1].is_nullable());
    5485
                            // Right side fields (third and fourth) should be nullable
    5486
                            assert!(fields[2].is_nullable());
    5487
                            assert!(fields[3].is_nullable());
    5488
                        } else if join_type == JoinType::Right {
    5489
                            // Left side fields (first two) should be nullable
    5490
                            assert!(fields[0].is_nullable());
    5491
                            assert!(fields[1].is_nullable());
    5492
                            // Right side fields (third and fourth) shouldn't be nullable
    5493
                            assert!(!fields[2].is_nullable());
    5494
                            assert!(!fields[3].is_nullable());
    5495
                        } else if join_type == JoinType::Full {
    5496
                            assert!(fields[0].is_nullable());
    5497
                            assert!(fields[1].is_nullable());
    5498
                            assert!(fields[2].is_nullable());
    5499
                            assert!(fields[3].is_nullable());
    5500
                        }
    5501
                    }
    5502
                }
    5503
    5504
                assert_eq!(join.on, vec![(col("t1.a"), col("t2.a"))]);
    5505
                assert_eq!(join.filter, Some(col("t1.b").gt(col("t2.b"))));
    5506
                assert_eq!(join.join_type, join_type);
    5507
                assert_eq!(join.join_constraint, JoinConstraint::On);
    5508
                assert_eq!(join.null_equality, NullEquality::NullEqualsNothing);
    5509
            }
    5510
    5511
            Ok(())
    5512
        }
    5513
    5514
        #[test]
    5515
        fn test_join_try_new_with_using_constraint_and_overlapping_columns() -> Result<()> {
    5516
            let left_schema = Schema::new(vec![
    5517
                Field::new("id", DataType::Int32, false), // Common column in both tables
    5518
                Field::new("name", DataType::Utf8, false), // Unique to left
    5519
                Field::new("value", DataType::Int32, false), // Common column, different meaning
    5520
            ]);
    5521
    5522
            let right_schema = Schema::new(vec![
    5523
                Field::new("id", DataType::Int32, false), // Common column in both tables
    5524
                Field::new("category", DataType::Utf8, false), // Unique to right
    5525
                Field::new("value", DataType::Float64, true), // Common column, different meaning
    5526
            ]);
    5527
    5528
            let left_plan = table_scan(Some("t1"), &left_schema, None)?.build()?;
    5529
    5530
            let right_plan = table_scan(Some("t2"), &right_schema, None)?.build()?;
    5531
    5532
            // Test 1: USING constraint with a common column
    5533
            {
    5534
                // In the logical plan, both copies of the `id` column are preserved
    5535
                // The USING constraint is handled later during physical execution, where the common column appears once
    5536
                let join = Join::try_new(
    5537
                    Arc::new(left_plan.clone()),
    5538
                    Arc::new(right_plan.clone()),
    5539
                    vec![(col("t1.id"), col("t2.id"))],
    5540
                    None,
    5541
                    JoinType::Inner,
    5542
                    JoinConstraint::Using,
    5543
                    NullEquality::NullEqualsNothing,
    5544
                )?;
    5545
    5546
                let fields = join.schema.fields();
    5547
    5548
                assert_eq!(fields.len(), 6);
    5549
    5550
                assert_eq!(
    5551
                    fields[0].name(),
    5552
                    "id",
    5553
                    "First field should be 'id' from left table"
    5554
                );
    5555
                assert_eq!(
    5556
                    fields[1].name(),
    5557
                    "name",
    5558
                    "Second field should be 'name' from left table"
    5559
                );
    5560
                assert_eq!(
    5561
                    fields[2].name(),
    5562
                    "value",
    5563
                    "Third field should be 'value' from left table"
    5564
                );
    5565
                assert_eq!(
    5566
                    fields[3].name(),
    5567
                    "id",
    5568
                    "Fourth field should be 'id' from right table"
    5569
                );
    5570
                assert_eq!(
    5571
                    fields[4].name(),
    5572
                    "category",
    5573
                    "Fifth field should be 'category' from right table"
    5574
                );
    5575
                assert_eq!(
    5576
                    fields[5].name(),
    5577
                    "value",
    5578
                    "Sixth field should be 'value' from right table"
    5579
                );
    5580
    5581
                assert_eq!(join.join_constraint, JoinConstraint::Using);
    5582
            }
    5583
    5584
            // Test 2: Complex join condition with expressions
    5585
            {
    5586
                // Complex condition: join on id equality AND where left.value < right.value
    5587
                let join = Join::try_new(
    5588
                    Arc::new(left_plan.clone()),
    5589
                    Arc::new(right_plan.clone()),
    5590
                    vec![(col("t1.id"), col("t2.id"))], // Equijoin condition
    5591
                    Some(col("t1.value").lt(col("t2.value"))), // Non-equi filter condition
    5592
                    JoinType::Inner,
    5593
                    JoinConstraint::On,
    5594
                    NullEquality::NullEqualsNothing,
    5595
                )?;
    5596
    5597
                let fields = join.schema.fields();
    5598
                assert_eq!(fields.len(), 6);
    5599
    5600
                assert_eq!(
    5601
                    fields[0].name(),
    5602
                    "id",
    5603
                    "First field should be 'id' from left table"
    5604
                );
    5605
                assert_eq!(
    5606
                    fields[1].name(),
    5607
                    "name",
    5608
                    "Second field should be 'name' from left table"
    5609
                );
    5610
                assert_eq!(
    5611
                    fields[2].name(),
    5612
                    "value",
    5613
                    "Third field should be 'value' from left table"
    5614
                );
    5615
                assert_eq!(
    5616
                    fields[3].name(),
    5617
                    "id",
    5618
                    "Fourth field should be 'id' from right table"
    5619
                );
    5620
                assert_eq!(
    5621
                    fields[4].name(),
    5622
                    "category",
    5623
                    "Fifth field should be 'category' from right table"
    5624
                );
    5625
                assert_eq!(
    5626
                    fields[5].name(),
    5627
                    "value",
    5628
                    "Sixth field should be 'value' from right table"
    5629
                );
    5630
    5631
                assert_eq!(join.filter, Some(col("t1.value").lt(col("t2.value"))));
    5632
            }
    5633
    5634
            // Test 3: Join with null equality behavior set to true
    5635
            {
    5636
                let join = Join::try_new(
    5637
                    Arc::new(left_plan.clone()),
    5638
                    Arc::new(right_plan.clone()),
    5639
                    vec![(col("t1.id"), col("t2.id"))],
    5640
                    None,
    5641
                    JoinType::Inner,
    5642
                    JoinConstraint::On,
    5643
                    NullEquality::NullEqualsNull,
    5644
                )?;
    5645
    5646
                assert_eq!(join.null_equality, NullEquality::NullEqualsNull);
    5647
            }
    5648
    5649
            Ok(())
    5650
        }
    5651
    5652
        #[test]
    5653
        fn test_join_try_new_schema_validation() -> Result<()> {
    5654
            let left_schema = Schema::new(vec![
    5655
                Field::new("id", DataType::Int32, false),
    5656
                Field::new("name", DataType::Utf8, false),
    5657
                Field::new("value", DataType::Float64, true),
    5658
            ]);
    5659
    5660
            let right_schema = Schema::new(vec![
    5661
                Field::new("id", DataType::Int32, false),
    5662
                Field::new("category", DataType::Utf8, true),
    5663
                Field::new("code", DataType::Int16, false),
    5664
            ]);
    5665
    5666
            let left_plan = table_scan(Some("t1"), &left_schema, None)?.build()?;
    5667
    5668
            let right_plan = table_scan(Some("t2"), &right_schema, None)?.build()?;
    5669
    5670
            let join_types = vec![
    5671
                JoinType::Inner,
    5672
                JoinType::Left,
    5673
                JoinType::Right,
    5674
                JoinType::Full,
    5675
            ];
    5676
    5677
            for join_type in join_types {
    5678
                let join = Join::try_new(
    5679
                    Arc::new(left_plan.clone()),
    5680
                    Arc::new(right_plan.clone()),
    5681
                    vec![(col("t1.id"), col("t2.id"))],
    5682
                    Some(col("t1.value").gt(lit(5.0))),
    5683
                    join_type,
    5684
                    JoinConstraint::On,
    5685
                    NullEquality::NullEqualsNothing,
    5686
                )?;
    5687
    5688
                let fields = join.schema.fields();
    5689
                assert_eq!(fields.len(), 6, "Expected 6 fields for {join_type} join");
    5690
    5691
                for (i, field) in fields.iter().enumerate() {
    5692
                    let expected_nullable = match (i, &join_type) {
    5693
                        // Left table fields (indices 0, 1, 2)
    5694
                        (0, JoinType::Right | JoinType::Full) => true, // id becomes nullable in RIGHT/FULL
    5695
                        (1, JoinType::Right | JoinType::Full) => true, // name becomes nullable in RIGHT/FULL
    5696
                        (2, _) => true, // value is already nullable
    5697
    5698
                        // Right table fields (indices 3, 4, 5)
    5699
                        (3, JoinType::Left | JoinType::Full) => true, // id becomes nullable in LEFT/FULL
    5700
                        (4, _) => true, // category is already nullable
    5701
                        (5, JoinType::Left | JoinType::Full) => true, // code becomes nullable in LEFT/FULL
    5702
    5703
                        _ => false,
    5704
                    };
    5705
    5706
                    assert_eq!(
    5707
                        field.is_nullable(),
    5708
                        expected_nullable,
    5709
                        "Field {} ({}) nullability incorrect for {:?} join",
    5710
                        i,
    5711
                        field.name(),
    5712
                        join_type
    5713
                    );
    5714
                }
    5715
            }
    5716
    5717
            let using_join = Join::try_new(
    5718
                Arc::new(left_plan.clone()),
    5719
                Arc::new(right_plan.clone()),
    5720
                vec![(col("t1.id"), col("t2.id"))],
    5721
                None,
    5722
                JoinType::Inner,
    5723
                JoinConstraint::Using,
    5724
                NullEquality::NullEqualsNothing,
    5725
            )?;
    5726
    5727
            assert_eq!(
    5728
                using_join.schema.fields().len(),
    5729
                6,
    5730
                "USING join should have all fields"
    5731
            );
    5732
            assert_eq!(using_join.join_constraint, JoinConstraint::Using);
    5733
    5734
            Ok(())
    5735
        }
    5736
    }