## 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/utils.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
    //! Expression utilities
    19
    20
    use std::cmp::Ordering;
    21
    use std::collections::{BTreeSet, HashSet};
    22
    use std::sync::Arc;
    23
    24
    use crate::expr::{Alias, Sort, WildcardOptions, WindowFunctionParams};
    25
    use crate::expr_rewriter::strip_outer_reference;
    26
    use crate::{
    27
        and, BinaryExpr, Expr, ExprSchemable, Filter, GroupingSet, LogicalPlan, Operator,
    28
    };
    29
    use datafusion_expr_common::signature::{Signature, TypeSignature};
    30
    31
    use arrow::datatypes::{DataType, Field, Schema};
    32
    use datafusion_common::tree_node::{
    33
        Transformed, TransformedResult, TreeNode, TreeNodeRecursion,
    34
    };
    35
    use datafusion_common::utils::get_at_indices;
    36
    use datafusion_common::{
    37
        internal_err, plan_datafusion_err, plan_err, Column, DFSchema, DFSchemaRef, HashMap,
    38
        Result, TableReference,
    39
    };
    40
    41
    #[cfg(not(feature = "sql"))]
    42
    use crate::expr::{ExceptSelectItem, ExcludeSelectItem};
    43
    use indexmap::IndexSet;
    44
    #[cfg(feature = "sql")]
    45
    use sqlparser::ast::{ExceptSelectItem, ExcludeSelectItem};
    46
    47
    pub use datafusion_functions_aggregate_common::order::AggregateOrderSensitivity;
    48
    49
    ///  The value to which `COUNT(*)` is expanded to in
    50
    ///  `COUNT(<constant>)` expressions
    51
    pub use datafusion_common::utils::expr::COUNT_STAR_EXPANSION;
    52
    53
    /// Count the number of distinct exprs in a list of group by expressions. If the
    54
    /// first element is a `GroupingSet` expression then it must be the only expr.
    55
    0
    pub fn grouping_set_expr_count(group_expr: &[Expr]) -> Result<usize> {
    56
    0
        if let Some(Expr::GroupingSet(grouping_set)) = group_expr.first() {
    57
    0
            if group_expr.len() > 1 {
    58
    0
                return plan_err!(
    59
    0
                    "Invalid group by expressions, GroupingSet must be the only expression"
    60
                );
    61
    0
            }
    62
            // Groupings sets have an additional integral column for the grouping id
    63
    0
            Ok(grouping_set.distinct_expr().len() + 1)
    64
        } else {
    65
    0
            grouping_set_to_exprlist(group_expr).map(|exprs| exprs.len())
    66
        }
    67
    0
    }
    68
    69
    /// The [power set] (or powerset) of a set S is the set of all subsets of S, \
    70
    /// including the empty set and S itself.
    71
    ///
    72
    /// Example:
    73
    ///
    74
    /// If S is the set {x, y, z}, then all the subsets of S are \
    75
    ///  {} \
    76
    ///  {x} \
    77
    ///  {y} \
    78
    ///  {z} \
    79
    ///  {x, y} \
    80
    ///  {x, z} \
    81
    ///  {y, z} \
    82
    ///  {x, y, z} \
    83
    ///  and hence the power set of S is {{}, {x}, {y}, {z}, {x, y}, {x, z}, {y, z}, {x, y, z}}.
    84
    ///
    85
    /// [power set]: https://en.wikipedia.org/wiki/Power_set
    86
    0
    fn powerset<T>(slice: &[T]) -> Result<Vec<Vec<&T>>, String> {
    87
    0
        if slice.len() >= 64 {
    88
    0
            return Err("The size of the set must be less than 64.".into());
    89
    0
        }
    90
    91
    0
        let mut v = Vec::new();
    92
    0
        for mask in 0..(1 << slice.len()) {
    93
    0
            let mut ss = vec![];
    94
    0
            let mut bitset = mask;
    95
    0
            while bitset > 0 {
    96
    0
                let rightmost: u64 = bitset & !(bitset - 1);
    97
    0
                let idx = rightmost.trailing_zeros();
    98
    0
                let item = slice.get(idx as usize).unwrap();
    99
    0
                ss.push(item);
    100
    0
                // zero the trailing bit
    101
    0
                bitset &= bitset - 1;
    102
    0
            }
    103
    0
            v.push(ss);
    104
        }
    105
    0
        Ok(v)
    106
    0
    }
    107
    108
    /// check the number of expressions contained in the grouping_set
    109
    0
    fn check_grouping_set_size_limit(size: usize) -> Result<()> {
    110
    0
        let max_grouping_set_size = 65535;
    111
    0
        if size > max_grouping_set_size {
    112
    0
            return plan_err!("The number of group_expression in grouping_set exceeds the maximum limit {max_grouping_set_size}, found {size}");
    113
    0
        }
    114
    115
    0
        Ok(())
    116
    0
    }
    117
    118
    /// check the number of grouping_set contained in the grouping sets
    119
    0
    fn check_grouping_sets_size_limit(size: usize) -> Result<()> {
    120
    0
        let max_grouping_sets_size = 4096;
    121
    0
        if size > max_grouping_sets_size {
    122
    0
            return plan_err!("The number of grouping_set in grouping_sets exceeds the maximum limit {max_grouping_sets_size}, found {size}");
    123
    0
        }
    124
    125
    0
        Ok(())
    126
    0
    }
    127
    128
    /// Merge two grouping_set
    129
    ///
    130
    /// # Example
    131
    /// ```text
    132
    /// (A, B), (C, D) -> (A, B, C, D)
    133
    /// ```
    134
    ///
    135
    /// # Error
    136
    /// - [`DataFusionError`]: The number of group_expression in grouping_set exceeds the maximum limit
    137
    ///
    138
    /// [`DataFusionError`]: datafusion_common::DataFusionError
    139
    0
    fn merge_grouping_set<T: Clone>(left: &[T], right: &[T]) -> Result<Vec<T>> {
    140
    0
        check_grouping_set_size_limit(left.len() + right.len())?;
    141
    0
        Ok(left.iter().chain(right.iter()).cloned().collect())
    142
    0
    }
    143
    144
    /// Compute the cross product of two grouping_sets
    145
    ///
    146
    /// # Example
    147
    /// ```text
    148
    /// [(A, B), (C, D)], [(E), (F)] -> [(A, B, E), (A, B, F), (C, D, E), (C, D, F)]
    149
    /// ```
    150
    ///
    151
    /// # Error
    152
    /// - [`DataFusionError`]: The number of group_expression in grouping_set exceeds the maximum limit
    153
    /// - [`DataFusionError`]: The number of grouping_set in grouping_sets exceeds the maximum limit
    154
    ///
    155
    /// [`DataFusionError`]: datafusion_common::DataFusionError
    156
    0
    fn cross_join_grouping_sets<T: Clone>(
    157
    0
        left: &[Vec<T>],
    158
    0
        right: &[Vec<T>],
    159
    0
    ) -> Result<Vec<Vec<T>>> {
    160
    0
        let grouping_sets_size = left.len() * right.len();
    161
    162
    0
        check_grouping_sets_size_limit(grouping_sets_size)?;
    163
    164
    0
        let mut result = Vec::with_capacity(grouping_sets_size);
    165
    0
        for le in left {
    166
    0
            for re in right {
    167
    0
                result.push(merge_grouping_set(le, re)?);
    168
            }
    169
        }
    170
    0
        Ok(result)
    171
    0
    }
    172
    173
    /// Convert multiple grouping expressions into one [`GroupingSet::GroupingSets`],\
    174
    /// if the grouping expression does not contain [`Expr::GroupingSet`] or only has one expression,\
    175
    /// no conversion will be performed.
    176
    ///
    177
    /// e.g.
    178
    ///
    179
    /// person.id,\
    180
    /// GROUPING SETS ((person.age, person.salary),(person.age)),\
    181
    /// ROLLUP(person.state, person.birth_date)
    182
    ///
    183
    /// =>
    184
    ///
    185
    /// GROUPING SETS (\
    186
    ///   (person.id, person.age, person.salary),\
    187
    ///   (person.id, person.age, person.salary, person.state),\
    188
    ///   (person.id, person.age, person.salary, person.state, person.birth_date),\
    189
    ///   (person.id, person.age),\
    190
    ///   (person.id, person.age, person.state),\
    191
    ///   (person.id, person.age, person.state, person.birth_date)\
    192
    /// )
    193
    0
    pub fn enumerate_grouping_sets(group_expr: Vec<Expr>) -> Result<Vec<Expr>> {
    194
    0
        let has_grouping_set = group_expr
    195
    0
            .iter()
    196
    0
            .any(|expr| matches!(expr, Expr::GroupingSet(_)));
    197
    0
        if !has_grouping_set || group_expr.len() == 1 {
    198
    0
            return Ok(group_expr);
    199
    0
        }
    200
        // Only process mix grouping sets
    201
    0
        let partial_sets = group_expr
    202
    0
            .iter()
    203
    0
            .map(|expr| {
    204
    0
                let exprs = match expr {
    205
    0
                    Expr::GroupingSet(GroupingSet::GroupingSets(grouping_sets)) => {
    206
    0
                        check_grouping_sets_size_limit(grouping_sets.len())?;
    207
    0
                        grouping_sets.iter().map(|e| e.iter().collect()).collect()
    208
                    }
    209
    0
                    Expr::GroupingSet(GroupingSet::Cube(group_exprs)) => {
    210
    0
                        let grouping_sets = powerset(group_exprs)
    211
    0
                            .map_err(|e| plan_datafusion_err!("{}", e))?;
    212
    0
                        check_grouping_sets_size_limit(grouping_sets.len())?;
    213
    0
                        grouping_sets
    214
                    }
    215
    0
                    Expr::GroupingSet(GroupingSet::Rollup(group_exprs)) => {
    216
    0
                        let size = group_exprs.len();
    217
    0
                        let slice = group_exprs.as_slice();
    218
    0
                        check_grouping_sets_size_limit(size * (size + 1) / 2 + 1)?;
    219
    0
                        (0..(size + 1))
    220
    0
                            .map(|i| slice[0..i].iter().collect())
    221
    0
                            .collect()
    222
                    }
    223
    0
                    expr => vec![vec![expr]],
    224
                };
    225
    0
                Ok(exprs)
    226
    0
            })
    227
    0
            .collect::<Result<Vec<_>>>()?;
    228
    229
        // Cross Join
    230
    0
        let grouping_sets = partial_sets
    231
    0
            .into_iter()
    232
    0
            .map(Ok)
    233
    0
            .reduce(|l, r| cross_join_grouping_sets(&l?, &r?))
    234
    0
            .transpose()?
    235
    0
            .map(|e| {
    236
    0
                e.into_iter()
    237
    0
                    .map(|e| e.into_iter().cloned().collect())
    238
    0
                    .collect()
    239
    0
            })
    240
    0
            .unwrap_or_default();
    241
    242
    0
        Ok(vec![Expr::GroupingSet(GroupingSet::GroupingSets(
    243
    0
            grouping_sets,
    244
    0
        ))])
    245
    0
    }
    246
    247
    /// Find all distinct exprs in a list of group by expressions. If the
    248
    /// first element is a `GroupingSet` expression then it must be the only expr.
    249
    0
    pub fn grouping_set_to_exprlist(group_expr: &[Expr]) -> Result<Vec<&Expr>> {
    250
    0
        if let Some(Expr::GroupingSet(grouping_set)) = group_expr.first() {
    251
    0
            if group_expr.len() > 1 {
    252
    0
                return plan_err!(
    253
    0
                    "Invalid group by expressions, GroupingSet must be the only expression"
    254
                );
    255
    0
            }
    256
    0
            Ok(grouping_set.distinct_expr())
    257
        } else {
    258
    0
            Ok(group_expr
    259
    0
                .iter()
    260
    0
                .collect::<IndexSet<_>>()
    261
    0
                .into_iter()
    262
    0
                .collect())
    263
        }
    264
    0
    }
    265
    266
    /// Recursively walk an expression tree, collecting the unique set of columns
    267
    /// referenced in the expression
    268
    0
    pub fn expr_to_columns(expr: &Expr, accum: &mut HashSet<Column>) -> Result<()> {
    269
    0
        expr.apply(|expr| {
    270
    0
            match expr {
    271
    0
                Expr::Column(qc) => {
    272
    0
                    accum.insert(qc.clone());
    273
    0
                }
    274
                // Use explicit pattern match instead of a default
    275
                // implementation, so that in the future if someone adds
    276
                // new Expr types, they will check here as well
    277
                // TODO: remove the next line after `Expr::Wildcard` is removed
    278
                #[expect(deprecated)]
    279
                Expr::Unnest(_)
    280
                | Expr::ScalarVariable(_, _)
    281
                | Expr::Alias(_)
    282
                | Expr::Literal(_, _)
    283
                | Expr::BinaryExpr { .. }
    284
                | Expr::Like { .. }
    285
                | Expr::SimilarTo { .. }
    286
                | Expr::Not(_)
    287
                | Expr::IsNotNull(_)
    288
                | Expr::IsNull(_)
    289
                | Expr::IsTrue(_)
    290
                | Expr::IsFalse(_)
    291
                | Expr::IsUnknown(_)
    292
                | Expr::IsNotTrue(_)
    293
                | Expr::IsNotFalse(_)
    294
                | Expr::IsNotUnknown(_)
    295
                | Expr::Negative(_)
    296
                | Expr::Between { .. }
    297
                | Expr::Case { .. }
    298
                | Expr::Cast { .. }
    299
                | Expr::TryCast { .. }
    300
                | Expr::ScalarFunction(..)
    301
                | Expr::WindowFunction { .. }
    302
                | Expr::AggregateFunction { .. }
    303
                | Expr::GroupingSet(_)
    304
                | Expr::InList { .. }
    305
                | Expr::Exists { .. }
    306
                | Expr::InSubquery(_)
    307
                | Expr::ScalarSubquery(_)
    308
                | Expr::Wildcard { .. }
    309
                | Expr::Placeholder(_)
    310
    0
                | Expr::OuterReferenceColumn { .. } => {}
    311
            }
    312
    0
            Ok(TreeNodeRecursion::Continue)
    313
    0
        })
    314
    0
        .map(|_| ())
    315
    0
    }
    316
    317
    /// Find excluded columns in the schema, if any
    318
    /// SELECT * EXCLUDE(col1, col2), would return `vec![col1, col2]`
    319
    0
    fn get_excluded_columns(
    320
    0
        opt_exclude: Option<&ExcludeSelectItem>,
    321
    0
        opt_except: Option<&ExceptSelectItem>,
    322
    0
        schema: &DFSchema,
    323
    0
        qualifier: Option<&TableReference>,
    324
    0
    ) -> Result<Vec<Column>> {
    325
    0
        let mut idents = vec![];
    326
    0
        if let Some(excepts) = opt_except {
    327
    0
            idents.push(&excepts.first_element);
    328
    0
            idents.extend(&excepts.additional_elements);
    329
    0
        }
    330
    0
        if let Some(exclude) = opt_exclude {
    331
    0
            match exclude {
    332
    0
                ExcludeSelectItem::Single(ident) => idents.push(ident),
    333
    0
                ExcludeSelectItem::Multiple(idents_inner) => idents.extend(idents_inner),
    334
            }
    335
    0
        }
    336
        // Excluded columns should be unique
    337
    0
        let n_elem = idents.len();
    338
    0
        let unique_idents = idents.into_iter().collect::<HashSet<_>>();
    339
        // If HashSet size, and vector length are different, this means that some of the excluded columns
    340
        // are not unique. In this case return error.
    341
    0
        if n_elem != unique_idents.len() {
    342
    0
            return plan_err!("EXCLUDE or EXCEPT contains duplicate column names");
    343
    0
        }
    344
    345
    0
        let mut result = vec![];
    346
    0
        for ident in unique_idents.into_iter() {
    347
    0
            let col_name = ident.value.as_str();
    348
    0
            let (qualifier, field) = schema.qualified_field_with_name(qualifier, col_name)?;
    349
    0
            result.push(Column::from((qualifier, field)));
    350
        }
    351
    0
        Ok(result)
    352
    0
    }
    353
    354
    /// Returns all `Expr`s in the schema, except the `Column`s in the `columns_to_skip`
    355
    0
    fn get_exprs_except_skipped(
    356
    0
        schema: &DFSchema,
    357
    0
        columns_to_skip: &HashSet<Column>,
    358
    0
    ) -> Vec<Expr> {
    359
    0
        if columns_to_skip.is_empty() {
    360
    0
            schema.iter().map(Expr::from).collect::<Vec<Expr>>()
    361
        } else {
    362
    0
            schema
    363
    0
                .columns()
    364
    0
                .iter()
    365
    0
                .filter_map(|c| {
    366
    0
                    if !columns_to_skip.contains(c) {
    367
    0
                        Some(Expr::Column(c.clone()))
    368
                    } else {
    369
    0
                        None
    370
                    }
    371
    0
                })
    372
    0
                .collect::<Vec<Expr>>()
    373
        }
    374
    0
    }
    375
    376
    /// For each column specified in the USING JOIN condition, the JOIN plan outputs it twice
    377
    /// (once for each join side), but an unqualified wildcard should include it only once.
    378
    /// This function returns the columns that should be excluded.
    379
    0
    fn exclude_using_columns(plan: &LogicalPlan) -> Result<HashSet<Column>> {
    380
    0
        let using_columns = plan.using_columns()?;
    381
    0
        let excluded = using_columns
    382
    0
            .into_iter()
    383
            // For each USING JOIN condition, only expand to one of each join column in projection
    384
    0
            .flat_map(|cols| {
    385
    0
                let mut cols = cols.into_iter().collect::<Vec<_>>();
    386
                // sort join columns to make sure we consistently keep the same
    387
                // qualified column
    388
    0
                cols.sort();
    389
    0
                let mut out_column_names: HashSet<String> = HashSet::new();
    390
    0
                cols.into_iter().filter_map(move |c| {
    391
    0
                    if out_column_names.contains(&c.name) {
    392
    0
                        Some(c)
    393
                    } else {
    394
    0
                        out_column_names.insert(c.name);
    395
    0
                        None
    396
                    }
    397
    0
                })
    398
    0
            })
    399
    0
            .collect::<HashSet<_>>();
    400
    0
        Ok(excluded)
    401
    0
    }
    402
    403
    /// Resolves an `Expr::Wildcard` to a collection of `Expr::Column`'s.
    404
    0
    pub fn expand_wildcard(
    405
    0
        schema: &DFSchema,
    406
    0
        plan: &LogicalPlan,
    407
    0
        wildcard_options: Option<&WildcardOptions>,
    408
    0
    ) -> Result<Vec<Expr>> {
    409
    0
        let mut columns_to_skip = exclude_using_columns(plan)?;
    410
    0
        let excluded_columns = if let Some(WildcardOptions {
    411
    0
            exclude: opt_exclude,
    412
    0
            except: opt_except,
    413
            ..
    414
    0
        }) = wildcard_options
    415
        {
    416
    0
            get_excluded_columns(opt_exclude.as_ref(), opt_except.as_ref(), schema, None)?
    417
        } else {
    418
    0
            vec![]
    419
        };
    420
        // Add each excluded `Column` to columns_to_skip
    421
    0
        columns_to_skip.extend(excluded_columns);
    422
    0
        Ok(get_exprs_except_skipped(schema, &columns_to_skip))
    423
    0
    }
    424
    425
    /// Resolves an `Expr::Wildcard` to a collection of qualified `Expr::Column`'s.
    426
    0
    pub fn expand_qualified_wildcard(
    427
    0
        qualifier: &TableReference,
    428
    0
        schema: &DFSchema,
    429
    0
        wildcard_options: Option<&WildcardOptions>,
    430
    0
    ) -> Result<Vec<Expr>> {
    431
    0
        let qualified_indices = schema.fields_indices_with_qualified(qualifier);
    432
    0
        let projected_func_dependencies = schema
    433
    0
            .functional_dependencies()
    434
    0
            .project_functional_dependencies(&qualified_indices, qualified_indices.len());
    435
    0
        let fields_with_qualified = get_at_indices(schema.fields(), &qualified_indices)?;
    436
    0
        if fields_with_qualified.is_empty() {
    437
    0
            return plan_err!("Invalid qualifier {qualifier}");
    438
    0
        }
    439
    440
    0
        let qualified_schema = Arc::new(Schema::new_with_metadata(
    441
    0
            fields_with_qualified,
    442
    0
            schema.metadata().clone(),
    443
        ));
    444
    0
        let qualified_dfschema =
    445
    0
            DFSchema::try_from_qualified_schema(qualifier.clone(), &qualified_schema)?
    446
    0
                .with_functional_dependencies(projected_func_dependencies)?;
    447
    0
        let excluded_columns = if let Some(WildcardOptions {
    448
    0
            exclude: opt_exclude,
    449
    0
            except: opt_except,
    450
            ..
    451
    0
        }) = wildcard_options
    452
        {
    453
    0
            get_excluded_columns(
    454
    0
                opt_exclude.as_ref(),
    455
    0
                opt_except.as_ref(),
    456
    0
                schema,
    457
    0
                Some(qualifier),
    458
    0
            )?
    459
        } else {
    460
    0
            vec![]
    461
        };
    462
        // Add each excluded `Column` to columns_to_skip
    463
    0
        let mut columns_to_skip = HashSet::new();
    464
    0
        columns_to_skip.extend(excluded_columns);
    465
    0
        Ok(get_exprs_except_skipped(
    466
    0
            &qualified_dfschema,
    467
    0
            &columns_to_skip,
    468
    0
        ))
    469
    0
    }
    470
    471
    /// (expr, "is the SortExpr for window (either comes from PARTITION BY or ORDER BY columns)")
    472
    /// If bool is true SortExpr comes from `PARTITION BY` column, if false comes from `ORDER BY` column
    473
    type WindowSortKey = Vec<(Sort, bool)>;
    474
    475
    /// Generate a sort key for a given window expr's partition_by and order_by expr
    476
    0
    pub fn generate_sort_key(
    477
    0
        partition_by: &[Expr],
    478
    0
        order_by: &[Sort],
    479
    0
    ) -> Result<WindowSortKey> {
    480
    0
        let normalized_order_by_keys = order_by
    481
    0
            .iter()
    482
    0
            .map(|e| {
    483
    0
                let Sort { expr, .. } = e;
    484
    0
                Sort::new(expr.clone(), true, false)
    485
    0
            })
    486
    0
            .collect::<Vec<_>>();
    487
    488
    0
        let mut final_sort_keys = vec![];
    489
    0
        let mut is_partition_flag = vec![];
    490
    0
        partition_by.iter().for_each(|e| {
    491
            // By default, create sort key with ASC is true and NULLS LAST to be consistent with
    492
            // PostgreSQL's rule: https://www.postgresql.org/docs/current/queries-order.html
    493
    0
            let e = e.clone().sort(true, false);
    494
    0
            if let Some(pos) = normalized_order_by_keys.iter().position(|key| key.eq(&e)) {
    495
    0
                let order_by_key = &order_by[pos];
    496
    0
                if !final_sort_keys.contains(order_by_key) {
    497
    0
                    final_sort_keys.push(order_by_key.clone());
    498
    0
                    is_partition_flag.push(true);
    499
    0
                }
    500
    0
            } else if !final_sort_keys.contains(&e) {
    501
    0
                final_sort_keys.push(e);
    502
    0
                is_partition_flag.push(true);
    503
    0
            }
    504
    0
        });
    505
    506
    0
        order_by.iter().for_each(|e| {
    507
    0
            if !final_sort_keys.contains(e) {
    508
    0
                final_sort_keys.push(e.clone());
    509
    0
                is_partition_flag.push(false);
    510
    0
            }
    511
    0
        });
    512
    0
        let res = final_sort_keys
    513
    0
            .into_iter()
    514
    0
            .zip(is_partition_flag)
    515
    0
            .collect::<Vec<_>>();
    516
    0
        Ok(res)
    517
    0
    }
    518
    519
    /// Compare the sort expr as PostgreSQL's common_prefix_cmp():
    520
    /// <https://github.com/postgres/postgres/blob/master/src/backend/optimizer/plan/planner.c>
    521
    0
    pub fn compare_sort_expr(
    522
    0
        sort_expr_a: &Sort,
    523
    0
        sort_expr_b: &Sort,
    524
    0
        schema: &DFSchemaRef,
    525
    0
    ) -> Ordering {
    526
        let Sort {
    527
    0
            expr: expr_a,
    528
    0
            asc: asc_a,
    529
    0
            nulls_first: nulls_first_a,
    530
    0
        } = sort_expr_a;
    531
    532
        let Sort {
    533
    0
            expr: expr_b,
    534
    0
            asc: asc_b,
    535
    0
            nulls_first: nulls_first_b,
    536
    0
        } = sort_expr_b;
    537
    538
    0
        let ref_indexes_a = find_column_indexes_referenced_by_expr(expr_a, schema);
    539
    0
        let ref_indexes_b = find_column_indexes_referenced_by_expr(expr_b, schema);
    540
    0
        for (idx_a, idx_b) in ref_indexes_a.iter().zip(ref_indexes_b.iter()) {
    541
    0
            match idx_a.cmp(idx_b) {
    542
                Ordering::Less => {
    543
    0
                    return Ordering::Less;
    544
                }
    545
                Ordering::Greater => {
    546
    0
                    return Ordering::Greater;
    547
                }
    548
    0
                Ordering::Equal => {}
    549
            }
    550
        }
    551
    0
        match ref_indexes_a.len().cmp(&ref_indexes_b.len()) {
    552
    0
            Ordering::Less => return Ordering::Greater,
    553
            Ordering::Greater => {
    554
    0
                return Ordering::Less;
    555
            }
    556
    0
            Ordering::Equal => {}
    557
        }
    558
    0
        match (asc_a, asc_b) {
    559
            (true, false) => {
    560
    0
                return Ordering::Greater;
    561
            }
    562
            (false, true) => {
    563
    0
                return Ordering::Less;
    564
            }
    565
    0
            _ => {}
    566
        }
    567
    0
        match (nulls_first_a, nulls_first_b) {
    568
            (true, false) => {
    569
    0
                return Ordering::Less;
    570
            }
    571
            (false, true) => {
    572
    0
                return Ordering::Greater;
    573
            }
    574
    0
            _ => {}
    575
        }
    576
    0
        Ordering::Equal
    577
    0
    }
    578
    579
    /// Group a slice of window expression expr by their order by expressions
    580
    0
    pub fn group_window_expr_by_sort_keys(
    581
    0
        window_expr: impl IntoIterator<Item = Expr>,
    582
    0
    ) -> Result<Vec<(WindowSortKey, Vec<Expr>)>> {
    583
    0
        let mut result = vec![];
    584
    0
        window_expr.into_iter().try_for_each(|expr| match &expr {
    585
    0
            Expr::WindowFunction(window_fun) => {
    586
    0
                let WindowFunctionParams{ partition_by, order_by, ..} = &window_fun.as_ref().params;
    587
    0
                let sort_key = generate_sort_key(partition_by, order_by)?;
    588
    0
                if let Some((_, values)) = result.iter_mut().find(
    589
    0
                    |group: &&mut (WindowSortKey, Vec<Expr>)| matches!(group, (key, _) if *key == sort_key),
    590
    0
                ) {
    591
    0
                    values.push(expr);
    592
    0
                } else {
    593
    0
                    result.push((sort_key, vec![expr]))
    594
                }
    595
    0
                Ok(())
    596
            }
    597
    0
            other => internal_err!(
    598
    0
                "Impossibly got non-window expr {other:?}"
    599
            ),
    600
    0
        })?;
    601
    0
        Ok(result)
    602
    0
    }
    603
    604
    /// Collect all deeply nested `Expr::AggregateFunction`.
    605
    /// They are returned in order of occurrence (depth
    606
    /// first), with duplicates omitted.
    607
    0
    pub fn find_aggregate_exprs<'a>(exprs: impl IntoIterator<Item = &'a Expr>) -> Vec<Expr> {
    608
    0
        find_exprs_in_exprs(exprs, &|nested_expr| {
    609
    0
            matches!(nested_expr, Expr::AggregateFunction { .. })
    610
    0
        })
    611
    0
    }
    612
    613
    /// Collect all deeply nested `Expr::WindowFunction`. They are returned in order of occurrence
    614
    /// (depth first), with duplicates omitted.
    615
    0
    pub fn find_window_exprs<'a>(exprs: impl IntoIterator<Item = &'a Expr>) -> Vec<Expr> {
    616
    0
        find_exprs_in_exprs(exprs, &|nested_expr| {
    617
    0
            matches!(nested_expr, Expr::WindowFunction { .. })
    618
    0
        })
    619
    0
    }
    620
    621
    /// Collect all deeply nested `Expr::OuterReferenceColumn`. They are returned in order of occurrence
    622
    /// (depth first), with duplicates omitted.
    623
    0
    pub fn find_out_reference_exprs(expr: &Expr) -> Vec<Expr> {
    624
    0
        find_exprs_in_expr(expr, &|nested_expr| {
    625
    0
            matches!(nested_expr, Expr::OuterReferenceColumn { .. })
    626
    0
        })
    627
    0
    }
    628
    629
    /// Search the provided `Expr`'s, and all of their nested `Expr`, for any that
    630
    /// pass the provided test. The returned `Expr`'s are deduplicated and returned
    631
    /// in order of appearance (depth first).
    632
    0
    fn find_exprs_in_exprs<'a, F>(
    633
    0
        exprs: impl IntoIterator<Item = &'a Expr>,
    634
    0
        test_fn: &F,
    635
    0
    ) -> Vec<Expr>
    636
    0
    where
    637
    0
        F: Fn(&Expr) -> bool,
    638
    {
    639
    0
        exprs
    640
    0
            .into_iter()
    641
    0
            .flat_map(|expr| find_exprs_in_expr(expr, test_fn))
    642
    0
            .fold(vec![], |mut acc, expr| {
    643
    0
                if !acc.contains(&expr) {
    644
    0
                    acc.push(expr)
    645
    0
                }
    646
    0
                acc
    647
    0
            })
    648
    0
    }
    649
    650
    /// Search an `Expr`, and all of its nested `Expr`'s, for any that pass the
    651
    /// provided test. The returned `Expr`'s are deduplicated and returned in order
    652
    /// of appearance (depth first).
    653
    0
    fn find_exprs_in_expr<F>(expr: &Expr, test_fn: &F) -> Vec<Expr>
    654
    0
    where
    655
    0
        F: Fn(&Expr) -> bool,
    656
    {
    657
    0
        let mut exprs = vec![];
    658
    0
        expr.apply(|expr| {
    659
    0
            if test_fn(expr) {
    660
    0
                if !(exprs.contains(expr)) {
    661
    0
                    exprs.push(expr.clone())
    662
    0
                }
    663
                // Stop recursing down this expr once we find a match
    664
    0
                return Ok(TreeNodeRecursion::Jump);
    665
    0
            }
    666
    667
    0
            Ok(TreeNodeRecursion::Continue)
    668
    0
        })
    669
        // pre_visit always returns OK, so this will always too
    670
    0
        .expect("no way to return error during recursion");
    671
    0
        exprs
    672
    0
    }
    673
    674
    /// Recursively inspect an [`Expr`] and all its children.
    675
    0
    pub fn inspect_expr_pre<F, E>(expr: &Expr, mut f: F) -> Result<(), E>
    676
    0
    where
    677
    0
        F: FnMut(&Expr) -> Result<(), E>,
    678
    {
    679
    0
        let mut err = Ok(());
    680
    0
        expr.apply(|expr| {
    681
    0
            if let Err(e) = f(expr) {
    682
                // Save the error for later (it may not be a DataFusionError)
    683
    0
                err = Err(e);
    684
    0
                Ok(TreeNodeRecursion::Stop)
    685
            } else {
    686
                // keep going
    687
    0
                Ok(TreeNodeRecursion::Continue)
    688
            }
    689
    0
        })
    690
        // The closure always returns OK, so this will always too
    691
    0
        .expect("no way to return error during recursion");
    692
    693
    0
        err
    694
    0
    }
    695
    696
    /// Create schema fields from an expression list, for use in result set schema construction
    697
    ///
    698
    /// This function converts a list of expressions into a list of complete schema fields,
    699
    /// making comprehensive determinations about each field's properties including:
    700
    /// - **Data type**: Resolved based on expression type and input schema context
    701
    /// - **Nullability**: Determined by expression-specific nullability rules
    702
    /// - **Metadata**: Computed based on expression type (preserving, merging, or generating new metadata)
    703
    /// - **Table reference scoping**: Establishing proper qualified field references
    704
    ///
    705
    /// Each expression is converted to a field by calling [`Expr::to_field`], which performs
    706
    /// the complete field resolution process for all field properties.
    707
    ///
    708
    /// # Returns
    709
    ///
    710
    /// A `Result` containing a vector of `(Option<TableReference>, Arc<Field>)` tuples,
    711
    /// where each Field contains complete schema information (type, nullability, metadata)
    712
    /// and proper table reference scoping for the corresponding expression.
    713
    0
    pub fn exprlist_to_fields<'a>(
    714
    0
        exprs: impl IntoIterator<Item = &'a Expr>,
    715
    0
        plan: &LogicalPlan,
    716
    0
    ) -> Result<Vec<(Option<TableReference>, Arc<Field>)>> {
    717
        // Look for exact match in plan's output schema
    718
    0
        let input_schema = plan.schema();
    719
    0
        exprs
    720
    0
            .into_iter()
    721
    0
            .map(|e| e.to_field(input_schema))
    722
    0
            .collect()
    723
    0
    }
    724
    725
    /// Convert an expression into Column expression if it's already provided as input plan.
    726
    ///
    727
    /// For example, it rewrites:
    728
    ///
    729
    /// ```text
    730
    /// .aggregate(vec![col("c1")], vec![sum(col("c2"))])?
    731
    /// .project(vec![col("c1"), sum(col("c2"))?
    732
    /// ```
    733
    ///
    734
    /// Into:
    735
    ///
    736
    /// ```text
    737
    /// .aggregate(vec![col("c1")], vec![sum(col("c2"))])?
    738
    /// .project(vec![col("c1"), col("SUM(c2)")?
    739
    /// ```
    740
    0
    pub fn columnize_expr(e: Expr, input: &LogicalPlan) -> Result<Expr> {
    741
    0
        let output_exprs = match input.columnized_output_exprs() {
    742
    0
            Ok(exprs) if !exprs.is_empty() => exprs,
    743
    0
            _ => return Ok(e),
    744
        };
    745
    0
        let exprs_map: HashMap<&Expr, Column> = output_exprs.into_iter().collect();
    746
    0
        e.transform_down(|node: Expr| match exprs_map.get(&node) {
    747
    0
            Some(column) => Ok(Transformed::new(
    748
    0
                Expr::Column(column.clone()),
    749
    0
                true,
    750
    0
                TreeNodeRecursion::Jump,
    751
    0
            )),
    752
    0
            None => Ok(Transformed::no(node)),
    753
    0
        })
    754
    0
        .data()
    755
    0
    }
    756
    757
    /// Collect all deeply nested `Expr::Column`'s. They are returned in order of
    758
    /// appearance (depth first), and may contain duplicates.
    759
    0
    pub fn find_column_exprs(exprs: &[Expr]) -> Vec<Expr> {
    760
    0
        exprs
    761
    0
            .iter()
    762
    0
            .flat_map(find_columns_referenced_by_expr)
    763
    0
            .map(Expr::Column)
    764
    0
            .collect()
    765
    0
    }
    766
    767
    0
    pub(crate) fn find_columns_referenced_by_expr(e: &Expr) -> Vec<Column> {
    768
    0
        let mut exprs = vec![];
    769
    0
        e.apply(|expr| {
    770
    0
            if let Expr::Column(c) = expr {
    771
    0
                exprs.push(c.clone())
    772
    0
            }
    773
    0
            Ok(TreeNodeRecursion::Continue)
    774
    0
        })
    775
        // As the closure always returns Ok, this "can't" error
    776
    0
        .expect("Unexpected error");
    777
    0
        exprs
    778
    0
    }
    779
    780
    /// Convert any `Expr` to an `Expr::Column`.
    781
    0
    pub fn expr_as_column_expr(expr: &Expr, plan: &LogicalPlan) -> Result<Expr> {
    782
    0
        match expr {
    783
    0
            Expr::Column(col) => {
    784
    0
                let (qualifier, field) = plan.schema().qualified_field_from_column(col)?;
    785
    0
                Ok(Expr::from(Column::from((qualifier, field))))
    786
            }
    787
    0
            _ => Ok(Expr::Column(Column::from_name(
    788
    0
                expr.schema_name().to_string(),
    789
    0
            ))),
    790
        }
    791
    0
    }
    792
    793
    /// Recursively walk an expression tree, collecting the column indexes
    794
    /// referenced in the expression
    795
    0
    pub(crate) fn find_column_indexes_referenced_by_expr(
    796
    0
        e: &Expr,
    797
    0
        schema: &DFSchemaRef,
    798
    0
    ) -> Vec<usize> {
    799
    0
        let mut indexes = vec![];
    800
    0
        e.apply(|expr| {
    801
    0
            match expr {
    802
    0
                Expr::Column(qc) => {
    803
    0
                    if let Ok(idx) = schema.index_of_column(qc) {
    804
    0
                        indexes.push(idx);
    805
    0
                    }
    806
                }
    807
    0
                Expr::Literal(_, _) => {
    808
    0
                    indexes.push(usize::MAX);
    809
    0
                }
    810
    0
                _ => {}
    811
            }
    812
    0
            Ok(TreeNodeRecursion::Continue)
    813
    0
        })
    814
    0
        .unwrap();
    815
    0
        indexes
    816
    0
    }
    817
    818
    /// Can this data type be used in hash join equal conditions??
    819
    /// Data types here come from function 'equal_rows', if more data types are supported
    820
    /// in create_hashes, add those data types here to generate join logical plan.
    821
    0
    pub fn can_hash(data_type: &DataType) -> bool {
    822
    0
        match data_type {
    823
    0
            DataType::Null => true,
    824
    0
            DataType::Boolean => true,
    825
    0
            DataType::Int8 => true,
    826
    0
            DataType::Int16 => true,
    827
    0
            DataType::Int32 => true,
    828
    0
            DataType::Int64 => true,
    829
    0
            DataType::UInt8 => true,
    830
    0
            DataType::UInt16 => true,
    831
    0
            DataType::UInt32 => true,
    832
    0
            DataType::UInt64 => true,
    833
    0
            DataType::Float16 => true,
    834
    0
            DataType::Float32 => true,
    835
    0
            DataType::Float64 => true,
    836
    0
            DataType::Decimal32(_, _) => true,
    837
    0
            DataType::Decimal64(_, _) => true,
    838
    0
            DataType::Decimal128(_, _) => true,
    839
    0
            DataType::Decimal256(_, _) => true,
    840
    0
            DataType::Timestamp(_, _) => true,
    841
    0
            DataType::Utf8 => true,
    842
    0
            DataType::LargeUtf8 => true,
    843
    0
            DataType::Utf8View => true,
    844
    0
            DataType::Binary => true,
    845
    0
            DataType::LargeBinary => true,
    846
    0
            DataType::BinaryView => true,
    847
    0
            DataType::Date32 => true,
    848
    0
            DataType::Date64 => true,
    849
    0
            DataType::Time32(_) => true,
    850
    0
            DataType::Time64(_) => true,
    851
    0
            DataType::Duration(_) => true,
    852
    0
            DataType::Interval(_) => true,
    853
    0
            DataType::FixedSizeBinary(_) => true,
    854
    0
            DataType::Dictionary(key_type, value_type) => {
    855
    0
                DataType::is_dictionary_key_type(key_type) && can_hash(value_type)
    856
            }
    857
    0
            DataType::List(value_type) => can_hash(value_type.data_type()),
    858
    0
            DataType::LargeList(value_type) => can_hash(value_type.data_type()),
    859
    0
            DataType::FixedSizeList(value_type, _) => can_hash(value_type.data_type()),
    860
    0
            DataType::Map(map_struct, true | false) => can_hash(map_struct.data_type()),
    861
    0
            DataType::Struct(fields) => fields.iter().all(|f| can_hash(f.data_type())),
    862
    863
            DataType::ListView(_)
    864
            | DataType::LargeListView(_)
    865
            | DataType::Union(_, _)
    866
    0
            | DataType::RunEndEncoded(_, _) => false,
    867
        }
    868
    0
    }
    869
    870
    /// Check whether all columns are from the schema.
    871
    0
    pub fn check_all_columns_from_schema(
    872
    0
        columns: &HashSet<&Column>,
    873
    0
        schema: &DFSchema,
    874
    0
    ) -> Result<bool> {
    875
    0
        for col in columns.iter() {
    876
    0
            let exist = schema.is_column_from_schema(col);
    877
    0
            if !exist {
    878
    0
                return Ok(false);
    879
    0
            }
    880
        }
    881
    882
    0
        Ok(true)
    883
    0
    }
    884
    885
    /// Give two sides of the equijoin predicate, return a valid join key pair.
    886
    /// If there is no valid join key pair, return None.
    887
    ///
    888
    /// A valid join means:
    889
    /// 1. All referenced column of the left side is from the left schema, and
    890
    ///    all referenced column of the right side is from the right schema.
    891
    /// 2. Or opposite. All referenced column of the left side is from the right schema,
    892
    ///    and the right side is from the left schema.
    893
    0
    pub fn find_valid_equijoin_key_pair(
    894
    0
        left_key: &Expr,
    895
    0
        right_key: &Expr,
    896
    0
        left_schema: &DFSchema,
    897
    0
        right_schema: &DFSchema,
    898
    0
    ) -> Result<Option<(Expr, Expr)>> {
    899
    0
        let left_using_columns = left_key.column_refs();
    900
    0
        let right_using_columns = right_key.column_refs();
    901
    902
        // Conditions like a = 10, will be added to non-equijoin.
    903
    0
        if left_using_columns.is_empty() || right_using_columns.is_empty() {
    904
    0
            return Ok(None);
    905
    0
        }
    906
    907
    0
        if check_all_columns_from_schema(&left_using_columns, left_schema)?
    908
    0
            && check_all_columns_from_schema(&right_using_columns, right_schema)?
    909
        {
    910
    0
            return Ok(Some((left_key.clone(), right_key.clone())));
    911
    0
        } else if check_all_columns_from_schema(&right_using_columns, left_schema)?
    912
    0
            && check_all_columns_from_schema(&left_using_columns, right_schema)?
    913
        {
    914
    0
            return Ok(Some((right_key.clone(), left_key.clone())));
    915
    0
        }
    916
    917
    0
        Ok(None)
    918
    0
    }
    919
    920
    /// Creates a detailed error message for a function with wrong signature.
    921
    ///
    922
    /// For example, a query like `select round(3.14, 1.1);` would yield:
    923
    /// ```text
    924
    /// Error during planning: No function matches 'round(Float64, Float64)'. You might need to add explicit type casts.
    925
    ///     Candidate functions:
    926
    ///     round(Float64, Int64)
    927
    ///     round(Float32, Int64)
    928
    ///     round(Float64)
    929
    ///     round(Float32)
    930
    /// ```
    931
    #[expect(clippy::needless_pass_by_value)]
    932
    0
    pub fn generate_signature_error_msg(
    933
    0
        func_name: &str,
    934
    0
        func_signature: Signature,
    935
    0
        input_expr_types: &[DataType],
    936
    0
    ) -> String {
    937
    0
        let candidate_signatures = func_signature
    938
    0
            .type_signature
    939
    0
            .to_string_repr_with_names(func_signature.parameter_names.as_deref())
    940
    0
            .iter()
    941
    0
            .map(|args_str| format!("\t{func_name}({args_str})"))
    942
    0
            .collect::<Vec<String>>()
    943
    0
            .join("\n");
    944
    945
    0
        format!(
    946
    0
                "No function matches the given name and argument types '{}({})'. You might need to add explicit type casts.\n\tCandidate functions:\n{}",
    947
    0
                func_name, TypeSignature::join_types(input_expr_types, ", "), candidate_signatures
    948
            )
    949
    0
    }
    950
    951
    /// Splits a conjunctive [`Expr`] such as `A AND B AND C` => `[A, B, C]`
    952
    ///
    953
    /// See [`split_conjunction_owned`] for more details and an example.
    954
    0
    pub fn split_conjunction(expr: &Expr) -> Vec<&Expr> {
    955
    0
        split_conjunction_impl(expr, vec![])
    956
    0
    }
    957
    958
    0
    fn split_conjunction_impl<'a>(expr: &'a Expr, mut exprs: Vec<&'a Expr>) -> Vec<&'a Expr> {
    959
    0
        match expr {
    960
            Expr::BinaryExpr(BinaryExpr {
    961
    0
                right,
    962
                op: Operator::And,
    963
    0
                left,
    964
            }) => {
    965
    0
                let exprs = split_conjunction_impl(left, exprs);
    966
    0
                split_conjunction_impl(right, exprs)
    967
            }
    968
    0
            Expr::Alias(Alias { expr, .. }) => split_conjunction_impl(expr, exprs),
    969
    0
            other => {
    970
    0
                exprs.push(other);
    971
    0
                exprs
    972
            }
    973
        }
    974
    0
    }
    975
    976
    /// Iterate parts in a conjunctive [`Expr`] such as `A AND B AND C` => `[A, B, C]`
    977
    ///
    978
    /// See [`split_conjunction_owned`] for more details and an example.
    979
    0
    pub fn iter_conjunction(expr: &Expr) -> impl Iterator<Item = &Expr> {
    980
    0
        let mut stack = vec![expr];
    981
    0
        std::iter::from_fn(move || {
    982
    0
            while let Some(expr) = stack.pop() {
    983
    0
                match expr {
    984
                    Expr::BinaryExpr(BinaryExpr {
    985
    0
                        right,
    986
                        op: Operator::And,
    987
    0
                        left,
    988
    0
                    }) => {
    989
    0
                        stack.push(right);
    990
    0
                        stack.push(left);
    991
    0
                    }
    992
    0
                    Expr::Alias(Alias { expr, .. }) => stack.push(expr),
    993
    0
                    other => return Some(other),
    994
                }
    995
            }
    996
    0
            None
    997
    0
        })
    998
    0
    }
    999
    1000
    /// Iterate parts in a conjunctive [`Expr`] such as `A AND B AND C` => `[A, B, C]`
    1001
    ///
    1002
    /// See [`split_conjunction_owned`] for more details and an example.
    1003
    0
    pub fn iter_conjunction_owned(expr: Expr) -> impl Iterator<Item = Expr> {
    1004
    0
        let mut stack = vec![expr];
    1005
    0
        std::iter::from_fn(move || {
    1006
    0
            while let Some(expr) = stack.pop() {
    1007
    0
                match expr {
    1008
                    Expr::BinaryExpr(BinaryExpr {
    1009
    0
                        right,
    1010
                        op: Operator::And,
    1011
    0
                        left,
    1012
    0
                    }) => {
    1013
    0
                        stack.push(*right);
    1014
    0
                        stack.push(*left);
    1015
    0
                    }
    1016
    0
                    Expr::Alias(Alias { expr, .. }) => stack.push(*expr),
    1017
    0
                    other => return Some(other),
    1018
                }
    1019
            }
    1020
    0
            None
    1021
    0
        })
    1022
    0
    }
    1023
    1024
    /// Splits an owned conjunctive [`Expr`] such as `A AND B AND C` => `[A, B, C]`
    1025
    ///
    1026
    /// This is often used to "split" filter expressions such as `col1 = 5
    1027
    /// AND col2 = 10` into [`col1 = 5`, `col2 = 10`];
    1028
    ///
    1029
    /// # Example
    1030
    /// ```
    1031
    /// # use datafusion_expr::{col, lit};
    1032
    /// # use datafusion_expr::utils::split_conjunction_owned;
    1033
    /// // a=1 AND b=2
    1034
    /// let expr = col("a").eq(lit(1)).and(col("b").eq(lit(2)));
    1035
    ///
    1036
    /// // [a=1, b=2]
    1037
    /// let split = vec![col("a").eq(lit(1)), col("b").eq(lit(2))];
    1038
    ///
    1039
    /// // use split_conjunction_owned to split them
    1040
    /// assert_eq!(split_conjunction_owned(expr), split);
    1041
    /// ```
    1042
    0
    pub fn split_conjunction_owned(expr: Expr) -> Vec<Expr> {
    1043
    0
        split_binary_owned(expr, Operator::And)
    1044
    0
    }
    1045
    1046
    /// Splits an owned binary operator tree [`Expr`] such as `A <OP> B <OP> C` => `[A, B, C]`
    1047
    ///
    1048
    /// This is often used to "split" expressions such as `col1 = 5
    1049
    /// AND col2 = 10` into [`col1 = 5`, `col2 = 10`];
    1050
    ///
    1051
    /// # Example
    1052
    /// ```
    1053
    /// # use datafusion_expr::{col, lit, Operator};
    1054
    /// # use datafusion_expr::utils::split_binary_owned;
    1055
    /// # use std::ops::Add;
    1056
    /// // a=1 + b=2
    1057
    /// let expr = col("a").eq(lit(1)).add(col("b").eq(lit(2)));
    1058
    ///
    1059
    /// // [a=1, b=2]
    1060
    /// let split = vec![col("a").eq(lit(1)), col("b").eq(lit(2))];
    1061
    ///
    1062
    /// // use split_binary_owned to split them
    1063
    /// assert_eq!(split_binary_owned(expr, Operator::Plus), split);
    1064
    /// ```
    1065
    0
    pub fn split_binary_owned(expr: Expr, op: Operator) -> Vec<Expr> {
    1066
    0
        split_binary_owned_impl(expr, op, vec![])
    1067
    0
    }
    1068
    1069
    0
    fn split_binary_owned_impl(
    1070
    0
        expr: Expr,
    1071
    0
        operator: Operator,
    1072
    0
        mut exprs: Vec<Expr>,
    1073
    0
    ) -> Vec<Expr> {
    1074
    0
        match expr {
    1075
    0
            Expr::BinaryExpr(BinaryExpr { right, op, left }) if op == operator => {
    1076
    0
                let exprs = split_binary_owned_impl(*left, operator, exprs);
    1077
    0
                split_binary_owned_impl(*right, operator, exprs)
    1078
            }
    1079
    0
            Expr::Alias(Alias { expr, .. }) => {
    1080
    0
                split_binary_owned_impl(*expr, operator, exprs)
    1081
            }
    1082
    0
            other => {
    1083
    0
                exprs.push(other);
    1084
    0
                exprs
    1085
            }
    1086
        }
    1087
    0
    }
    1088
    1089
    /// Splits an binary operator tree [`Expr`] such as `A <OP> B <OP> C` => `[A, B, C]`
    1090
    ///
    1091
    /// See [`split_binary_owned`] for more details and an example.
    1092
    0
    pub fn split_binary(expr: &Expr, op: Operator) -> Vec<&Expr> {
    1093
    0
        split_binary_impl(expr, op, vec![])
    1094
    0
    }
    1095
    1096
    0
    fn split_binary_impl<'a>(
    1097
    0
        expr: &'a Expr,
    1098
    0
        operator: Operator,
    1099
    0
        mut exprs: Vec<&'a Expr>,
    1100
    0
    ) -> Vec<&'a Expr> {
    1101
    0
        match expr {
    1102
    0
            Expr::BinaryExpr(BinaryExpr { right, op, left }) if *op == operator => {
    1103
    0
                let exprs = split_binary_impl(left, operator, exprs);
    1104
    0
                split_binary_impl(right, operator, exprs)
    1105
            }
    1106
    0
            Expr::Alias(Alias { expr, .. }) => split_binary_impl(expr, operator, exprs),
    1107
    0
            other => {
    1108
    0
                exprs.push(other);
    1109
    0
                exprs
    1110
            }
    1111
        }
    1112
    0
    }
    1113
    1114
    /// Combines an array of filter expressions into a single filter
    1115
    /// expression consisting of the input filter expressions joined with
    1116
    /// logical AND.
    1117
    ///
    1118
    /// Returns None if the filters array is empty.
    1119
    ///
    1120
    /// # Example
    1121
    /// ```
    1122
    /// # use datafusion_expr::{col, lit};
    1123
    /// # use datafusion_expr::utils::conjunction;
    1124
    /// // a=1 AND b=2
    1125
    /// let expr = col("a").eq(lit(1)).and(col("b").eq(lit(2)));
    1126
    ///
    1127
    /// // [a=1, b=2]
    1128
    /// let split = vec![col("a").eq(lit(1)), col("b").eq(lit(2))];
    1129
    ///
    1130
    /// // use conjunction to join them together with `AND`
    1131
    /// assert_eq!(conjunction(split), Some(expr));
    1132
    /// ```
    1133
    0
    pub fn conjunction(filters: impl IntoIterator<Item = Expr>) -> Option<Expr> {
    1134
    0
        filters.into_iter().reduce(Expr::and)
    1135
    0
    }
    1136
    1137
    /// Combines an array of filter expressions into a single filter
    1138
    /// expression consisting of the input filter expressions joined with
    1139
    /// logical OR.
    1140
    ///
    1141
    /// Returns None if the filters array is empty.
    1142
    ///
    1143
    /// # Example
    1144
    /// ```
    1145
    /// # use datafusion_expr::{col, lit};
    1146
    /// # use datafusion_expr::utils::disjunction;
    1147
    /// // a=1 OR b=2
    1148
    /// let expr = col("a").eq(lit(1)).or(col("b").eq(lit(2)));
    1149
    ///
    1150
    /// // [a=1, b=2]
    1151
    /// let split = vec![col("a").eq(lit(1)), col("b").eq(lit(2))];
    1152
    ///
    1153
    /// // use disjunction to join them together with `OR`
    1154
    /// assert_eq!(disjunction(split), Some(expr));
    1155
    /// ```
    1156
    0
    pub fn disjunction(filters: impl IntoIterator<Item = Expr>) -> Option<Expr> {
    1157
    0
        filters.into_iter().reduce(Expr::or)
    1158
    0
    }
    1159
    1160
    /// Returns a new [LogicalPlan] that filters the output of  `plan` with a
    1161
    /// [LogicalPlan::Filter] with all `predicates` ANDed.
    1162
    ///
    1163
    /// # Example
    1164
    /// Before:
    1165
    /// ```text
    1166
    /// plan
    1167
    /// ```
    1168
    ///
    1169
    /// After:
    1170
    /// ```text
    1171
    /// Filter(predicate)
    1172
    ///   plan
    1173
    /// ```
    1174
    0
    pub fn add_filter(plan: LogicalPlan, predicates: &[&Expr]) -> Result<LogicalPlan> {
    1175
        // reduce filters to a single filter with an AND
    1176
    0
        let predicate = predicates
    1177
    0
            .iter()
    1178
    0
            .skip(1)
    1179
    0
            .fold(predicates[0].clone(), |acc, predicate| {
    1180
    0
                and(acc, (*predicate).to_owned())
    1181
    0
            });
    1182
    1183
    0
        Ok(LogicalPlan::Filter(Filter::try_new(
    1184
    0
            predicate,
    1185
    0
            Arc::new(plan),
    1186
    0
        )?))
    1187
    0
    }
    1188
    1189
    /// Looks for correlating expressions: for example, a binary expression with one field from the subquery, and
    1190
    /// one not in the subquery (closed upon from outer scope)
    1191
    ///
    1192
    /// # Arguments
    1193
    ///
    1194
    /// * `exprs` - List of expressions that may or may not be joins
    1195
    ///
    1196
    /// # Return value
    1197
    ///
    1198
    /// Tuple of (expressions containing joins, remaining non-join expressions)
    1199
    0
    pub fn find_join_exprs(exprs: Vec<&Expr>) -> Result<(Vec<Expr>, Vec<Expr>)> {
    1200
    0
        let mut joins = vec![];
    1201
    0
        let mut others = vec![];
    1202
    0
        for filter in exprs.into_iter() {
    1203
            // If the expression contains correlated predicates, add it to join filters
    1204
    0
            if filter.contains_outer() {
    1205
    0
                if !matches!(filter, Expr::BinaryExpr(BinaryExpr{ left, op: Operator::Eq, right }) if left.eq(right))
    1206
    0
                {
    1207
    0
                    joins.push(strip_outer_reference((*filter).clone()));
    1208
    0
                }
    1209
    0
            } else {
    1210
    0
                others.push((*filter).clone());
    1211
    0
            }
    1212
        }
    1213
    1214
    0
        Ok((joins, others))
    1215
    0
    }
    1216
    1217
    /// Returns the first (and only) element in a slice, or an error
    1218
    ///
    1219
    /// # Arguments
    1220
    ///
    1221
    /// * `slice` - The slice to extract from
    1222
    ///
    1223
    /// # Return value
    1224
    ///
    1225
    /// The first element, or an error
    1226
    0
    pub fn only_or_err<T>(slice: &[T]) -> Result<&T> {
    1227
    0
        match slice {
    1228
    0
            [it] => Ok(it),
    1229
    0
            [] => plan_err!("No items found!"),
    1230
    0
            _ => plan_err!("More than one item found!"),
    1231
        }
    1232
    0
    }
    1233
    1234
    /// merge inputs schema into a single schema.
    1235
    ///
    1236
    /// This function merges schemas from multiple logical plan inputs using [`DFSchema::merge`].
    1237
    /// Refer to that documentation for details on precedence and metadata handling.
    1238
    0
    pub fn merge_schema(inputs: &[&LogicalPlan]) -> DFSchema {
    1239
    0
        if inputs.len() == 1 {
    1240
    0
            inputs[0].schema().as_ref().clone()
    1241
        } else {
    1242
    0
            inputs.iter().map(|input| input.schema()).fold(
    1243
    0
                DFSchema::empty(),
    1244
    0
                |mut lhs, rhs| {
    1245
    0
                    lhs.merge(rhs);
    1246
    0
                    lhs
    1247
    0
                },
    1248
            )
    1249
        }
    1250
    0
    }
    1251
    1252
    /// Build state name. State is the intermediate state of the aggregate function.
    1253
    0
    pub fn format_state_name(name: &str, state_name: &str) -> String {
    1254
    0
        format!("{name}[{state_name}]")
    1255
    0
    }
    1256
    1257
    /// Determine the set of [`Column`]s produced by the subquery.
    1258
    0
    pub fn collect_subquery_cols(
    1259
    0
        exprs: &[Expr],
    1260
    0
        subquery_schema: &DFSchema,
    1261
    0
    ) -> Result<BTreeSet<Column>> {
    1262
    0
        exprs.iter().try_fold(BTreeSet::new(), |mut cols, expr| {
    1263
    0
            let mut using_cols: Vec<Column> = vec![];
    1264
    0
            for col in expr.column_refs().into_iter() {
    1265
    0
                if subquery_schema.has_column(col) {
    1266
    0
                    using_cols.push(col.clone());
    1267
    0
                }
    1268
            }
    1269
    1270
    0
            cols.extend(using_cols);
    1271
    0
            Result::<_>::Ok(cols)
    1272
    0
        })
    1273
    0
    }
    1274
    1275
    #[cfg(test)]
    1276
    mod tests {
    1277
        use super::*;
    1278
        use crate::{
    1279
            col, cube,
    1280
            expr::WindowFunction,
    1281
            expr_vec_fmt, grouping_set, lit, rollup,
    1282
            test::function_stub::{max_udaf, min_udaf, sum_udaf},
    1283
            Cast, ExprFunctionExt, WindowFunctionDefinition,
    1284
        };
    1285
        use arrow::datatypes::{UnionFields, UnionMode};
    1286
        use datafusion_expr_common::signature::{TypeSignature, Volatility};
    1287
    1288
        #[test]
    1289
        fn test_group_window_expr_by_sort_keys_empty_case() -> Result<()> {
    1290
            let result = group_window_expr_by_sort_keys(vec![])?;
    1291
            let expected: Vec<(WindowSortKey, Vec<Expr>)> = vec![];
    1292
            assert_eq!(expected, result);
    1293
            Ok(())
    1294
        }
    1295
    1296
        #[test]
    1297
        fn test_group_window_expr_by_sort_keys_empty_window() -> Result<()> {
    1298
            let max1 = Expr::from(WindowFunction::new(
    1299
                WindowFunctionDefinition::AggregateUDF(max_udaf()),
    1300
                vec![col("name")],
    1301
            ));
    1302
            let max2 = Expr::from(WindowFunction::new(
    1303
                WindowFunctionDefinition::AggregateUDF(max_udaf()),
    1304
                vec![col("name")],
    1305
            ));
    1306
            let min3 = Expr::from(WindowFunction::new(
    1307
                WindowFunctionDefinition::AggregateUDF(min_udaf()),
    1308
                vec![col("name")],
    1309
            ));
    1310
            let sum4 = Expr::from(WindowFunction::new(
    1311
                WindowFunctionDefinition::AggregateUDF(sum_udaf()),
    1312
                vec![col("age")],
    1313
            ));
    1314
            let exprs = &[max1.clone(), max2.clone(), min3.clone(), sum4.clone()];
    1315
            let result = group_window_expr_by_sort_keys(exprs.to_vec())?;
    1316
            let key = vec![];
    1317
            let expected: Vec<(WindowSortKey, Vec<Expr>)> =
    1318
                vec![(key, vec![max1, max2, min3, sum4])];
    1319
            assert_eq!(expected, result);
    1320
            Ok(())
    1321
        }
    1322
    1323
        #[test]
    1324
        fn test_group_window_expr_by_sort_keys() -> Result<()> {
    1325
            let age_asc = Sort::new(col("age"), true, true);
    1326
            let name_desc = Sort::new(col("name"), false, true);
    1327
            let created_at_desc = Sort::new(col("created_at"), false, true);
    1328
            let max1 = Expr::from(WindowFunction::new(
    1329
                WindowFunctionDefinition::AggregateUDF(max_udaf()),
    1330
                vec![col("name")],
    1331
            ))
    1332
            .order_by(vec![age_asc.clone(), name_desc.clone()])
    1333
            .build()
    1334
            .unwrap();
    1335
            let max2 = Expr::from(WindowFunction::new(
    1336
                WindowFunctionDefinition::AggregateUDF(max_udaf()),
    1337
                vec![col("name")],
    1338
            ));
    1339
            let min3 = Expr::from(WindowFunction::new(
    1340
                WindowFunctionDefinition::AggregateUDF(min_udaf()),
    1341
                vec![col("name")],
    1342
            ))
    1343
            .order_by(vec![age_asc.clone(), name_desc.clone()])
    1344
            .build()
    1345
            .unwrap();
    1346
            let sum4 = Expr::from(WindowFunction::new(
    1347
                WindowFunctionDefinition::AggregateUDF(sum_udaf()),
    1348
                vec![col("age")],
    1349
            ))
    1350
            .order_by(vec![
    1351
                name_desc.clone(),
    1352
                age_asc.clone(),
    1353
                created_at_desc.clone(),
    1354
            ])
    1355
            .build()
    1356
            .unwrap();
    1357
            // FIXME use as_ref
    1358
            let exprs = &[max1.clone(), max2.clone(), min3.clone(), sum4.clone()];
    1359
            let result = group_window_expr_by_sort_keys(exprs.to_vec())?;
    1360
    1361
            let key1 = vec![(age_asc.clone(), false), (name_desc.clone(), false)];
    1362
            let key2 = vec![];
    1363
            let key3 = vec![
    1364
                (name_desc, false),
    1365
                (age_asc, false),
    1366
                (created_at_desc, false),
    1367
            ];
    1368
    1369
            let expected: Vec<(WindowSortKey, Vec<Expr>)> = vec![
    1370
                (key1, vec![max1, min3]),
    1371
                (key2, vec![max2]),
    1372
                (key3, vec![sum4]),
    1373
            ];
    1374
            assert_eq!(expected, result);
    1375
            Ok(())
    1376
        }
    1377
    1378
        #[test]
    1379
        fn avoid_generate_duplicate_sort_keys() -> Result<()> {
    1380
            let asc_or_desc = [true, false];
    1381
            let nulls_first_or_last = [true, false];
    1382
            let partition_by = &[col("age"), col("name"), col("created_at")];
    1383
            for asc_ in asc_or_desc {
    1384
                for nulls_first_ in nulls_first_or_last {
    1385
                    let order_by = &[
    1386
                        Sort {
    1387
                            expr: col("age"),
    1388
                            asc: asc_,
    1389
                            nulls_first: nulls_first_,
    1390
                        },
    1391
                        Sort {
    1392
                            expr: col("name"),
    1393
                            asc: asc_,
    1394
                            nulls_first: nulls_first_,
    1395
                        },
    1396
                    ];
    1397
    1398
                    let expected = vec![
    1399
                        (
    1400
                            Sort {
    1401
                                expr: col("age"),
    1402
                                asc: asc_,
    1403
                                nulls_first: nulls_first_,
    1404
                            },
    1405
                            true,
    1406
                        ),
    1407
                        (
    1408
                            Sort {
    1409
                                expr: col("name"),
    1410
                                asc: asc_,
    1411
                                nulls_first: nulls_first_,
    1412
                            },
    1413
                            true,
    1414
                        ),
    1415
                        (
    1416
                            Sort {
    1417
                                expr: col("created_at"),
    1418
                                asc: true,
    1419
                                nulls_first: false,
    1420
                            },
    1421
                            true,
    1422
                        ),
    1423
                    ];
    1424
                    let result = generate_sort_key(partition_by, order_by)?;
    1425
                    assert_eq!(expected, result);
    1426
                }
    1427
            }
    1428
            Ok(())
    1429
        }
    1430
    1431
        #[test]
    1432
        fn test_enumerate_grouping_sets() -> Result<()> {
    1433
            let multi_cols = vec![col("col1"), col("col2"), col("col3")];
    1434
            let simple_col = col("simple_col");
    1435
            let cube = cube(multi_cols.clone());
    1436
            let rollup = rollup(multi_cols.clone());
    1437
            let grouping_set = grouping_set(vec![multi_cols]);
    1438
    1439
            // 1. col
    1440
            let sets = enumerate_grouping_sets(vec![simple_col.clone()])?;
    1441
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1442
            assert_eq!("[simple_col]", &result);
    1443
    1444
            // 2. cube
    1445
            let sets = enumerate_grouping_sets(vec![cube.clone()])?;
    1446
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1447
            assert_eq!("[CUBE (col1, col2, col3)]", &result);
    1448
    1449
            // 3. rollup
    1450
            let sets = enumerate_grouping_sets(vec![rollup.clone()])?;
    1451
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1452
            assert_eq!("[ROLLUP (col1, col2, col3)]", &result);
    1453
    1454
            // 4. col + cube
    1455
            let sets = enumerate_grouping_sets(vec![simple_col.clone(), cube.clone()])?;
    1456
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1457
            assert_eq!(
    1458
                "[GROUPING SETS (\
    1459
                (simple_col), \
    1460
                (simple_col, col1), \
    1461
                (simple_col, col2), \
    1462
                (simple_col, col1, col2), \
    1463
                (simple_col, col3), \
    1464
                (simple_col, col1, col3), \
    1465
                (simple_col, col2, col3), \
    1466
                (simple_col, col1, col2, col3))]",
    1467
                &result
    1468
            );
    1469
    1470
            // 5. col + rollup
    1471
            let sets = enumerate_grouping_sets(vec![simple_col.clone(), rollup.clone()])?;
    1472
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1473
            assert_eq!(
    1474
                "[GROUPING SETS (\
    1475
                (simple_col), \
    1476
                (simple_col, col1), \
    1477
                (simple_col, col1, col2), \
    1478
                (simple_col, col1, col2, col3))]",
    1479
                &result
    1480
            );
    1481
    1482
            // 6. col + grouping_set
    1483
            let sets =
    1484
                enumerate_grouping_sets(vec![simple_col.clone(), grouping_set.clone()])?;
    1485
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1486
            assert_eq!(
    1487
                "[GROUPING SETS (\
    1488
                (simple_col, col1, col2, col3))]",
    1489
                &result
    1490
            );
    1491
    1492
            // 7. col + grouping_set + rollup
    1493
            let sets = enumerate_grouping_sets(vec![
    1494
                simple_col.clone(),
    1495
                grouping_set,
    1496
                rollup.clone(),
    1497
            ])?;
    1498
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1499
            assert_eq!(
    1500
                "[GROUPING SETS (\
    1501
                (simple_col, col1, col2, col3), \
    1502
                (simple_col, col1, col2, col3, col1), \
    1503
                (simple_col, col1, col2, col3, col1, col2), \
    1504
                (simple_col, col1, col2, col3, col1, col2, col3))]",
    1505
                &result
    1506
            );
    1507
    1508
            // 8. col + cube + rollup
    1509
            let sets = enumerate_grouping_sets(vec![simple_col, cube, rollup])?;
    1510
            let result = format!("[{}]", expr_vec_fmt!(sets));
    1511
            assert_eq!(
    1512
                "[GROUPING SETS (\
    1513
                (simple_col), \
    1514
                (simple_col, col1), \
    1515
                (simple_col, col1, col2), \
    1516
                (simple_col, col1, col2, col3), \
    1517
                (simple_col, col1), \
    1518
                (simple_col, col1, col1), \
    1519
                (simple_col, col1, col1, col2), \
    1520
                (simple_col, col1, col1, col2, col3), \
    1521
                (simple_col, col2), \
    1522
                (simple_col, col2, col1), \
    1523
                (simple_col, col2, col1, col2), \
    1524
                (simple_col, col2, col1, col2, col3), \
    1525
                (simple_col, col1, col2), \
    1526
                (simple_col, col1, col2, col1), \
    1527
                (simple_col, col1, col2, col1, col2), \
    1528
                (simple_col, col1, col2, col1, col2, col3), \
    1529
                (simple_col, col3), \
    1530
                (simple_col, col3, col1), \
    1531
                (simple_col, col3, col1, col2), \
    1532
                (simple_col, col3, col1, col2, col3), \
    1533
                (simple_col, col1, col3), \
    1534
                (simple_col, col1, col3, col1), \
    1535
                (simple_col, col1, col3, col1, col2), \
    1536
                (simple_col, col1, col3, col1, col2, col3), \
    1537
                (simple_col, col2, col3), \
    1538
                (simple_col, col2, col3, col1), \
    1539
                (simple_col, col2, col3, col1, col2), \
    1540
                (simple_col, col2, col3, col1, col2, col3), \
    1541
                (simple_col, col1, col2, col3), \
    1542
                (simple_col, col1, col2, col3, col1), \
    1543
                (simple_col, col1, col2, col3, col1, col2), \
    1544
                (simple_col, col1, col2, col3, col1, col2, col3))]",
    1545
                &result
    1546
            );
    1547
    1548
            Ok(())
    1549
        }
    1550
        #[test]
    1551
        fn test_split_conjunction() {
    1552
            let expr = col("a");
    1553
            let result = split_conjunction(&expr);
    1554
            assert_eq!(result, vec![&expr]);
    1555
        }
    1556
    1557
        #[test]
    1558
        fn test_split_conjunction_two() {
    1559
            let expr = col("a").eq(lit(5)).and(col("b"));
    1560
            let expr1 = col("a").eq(lit(5));
    1561
            let expr2 = col("b");
    1562
    1563
            let result = split_conjunction(&expr);
    1564
            assert_eq!(result, vec![&expr1, &expr2]);
    1565
        }
    1566
    1567
        #[test]
    1568
        fn test_split_conjunction_alias() {
    1569
            let expr = col("a").eq(lit(5)).and(col("b").alias("the_alias"));
    1570
            let expr1 = col("a").eq(lit(5));
    1571
            let expr2 = col("b"); // has no alias
    1572
    1573
            let result = split_conjunction(&expr);
    1574
            assert_eq!(result, vec![&expr1, &expr2]);
    1575
        }
    1576
    1577
        #[test]
    1578
        fn test_split_conjunction_or() {
    1579
            let expr = col("a").eq(lit(5)).or(col("b"));
    1580
            let result = split_conjunction(&expr);
    1581
            assert_eq!(result, vec![&expr]);
    1582
        }
    1583
    1584
        #[test]
    1585
        fn test_split_binary_owned() {
    1586
            let expr = col("a");
    1587
            assert_eq!(split_binary_owned(expr.clone(), Operator::And), vec![expr]);
    1588
        }
    1589
    1590
        #[test]
    1591
        fn test_split_binary_owned_two() {
    1592
            assert_eq!(
    1593
                split_binary_owned(col("a").eq(lit(5)).and(col("b")), Operator::And),
    1594
                vec![col("a").eq(lit(5)), col("b")]
    1595
            );
    1596
        }
    1597
    1598
        #[test]
    1599
        fn test_split_binary_owned_different_op() {
    1600
            let expr = col("a").eq(lit(5)).or(col("b"));
    1601
            assert_eq!(
    1602
                // expr is connected by OR, but pass in AND
    1603
                split_binary_owned(expr.clone(), Operator::And),
    1604
                vec![expr]
    1605
            );
    1606
        }
    1607
    1608
        #[test]
    1609
        fn test_split_conjunction_owned() {
    1610
            let expr = col("a");
    1611
            assert_eq!(split_conjunction_owned(expr.clone()), vec![expr]);
    1612
        }
    1613
    1614
        #[test]
    1615
        fn test_split_conjunction_owned_two() {
    1616
            assert_eq!(
    1617
                split_conjunction_owned(col("a").eq(lit(5)).and(col("b"))),
    1618
                vec![col("a").eq(lit(5)), col("b")]
    1619
            );
    1620
        }
    1621
    1622
        #[test]
    1623
        fn test_split_conjunction_owned_alias() {
    1624
            assert_eq!(
    1625
                split_conjunction_owned(col("a").eq(lit(5)).and(col("b").alias("the_alias"))),
    1626
                vec![
    1627
                    col("a").eq(lit(5)),
    1628
                    // no alias on b
    1629
                    col("b"),
    1630
                ]
    1631
            );
    1632
        }
    1633
    1634
        #[test]
    1635
        fn test_conjunction_empty() {
    1636
            assert_eq!(conjunction(vec![]), None);
    1637
        }
    1638
    1639
        #[test]
    1640
        fn test_conjunction() {
    1641
            // `[A, B, C]`
    1642
            let expr = conjunction(vec![col("a"), col("b"), col("c")]);
    1643
    1644
            // --> `(A AND B) AND C`
    1645
            assert_eq!(expr, Some(col("a").and(col("b")).and(col("c"))));
    1646
    1647
            // which is different than `A AND (B AND C)`
    1648
            assert_ne!(expr, Some(col("a").and(col("b").and(col("c")))));
    1649
        }
    1650
    1651
        #[test]
    1652
        fn test_disjunction_empty() {
    1653
            assert_eq!(disjunction(vec![]), None);
    1654
        }
    1655
    1656
        #[test]
    1657
        fn test_disjunction() {
    1658
            // `[A, B, C]`
    1659
            let expr = disjunction(vec![col("a"), col("b"), col("c")]);
    1660
    1661
            // --> `(A OR B) OR C`
    1662
            assert_eq!(expr, Some(col("a").or(col("b")).or(col("c"))));
    1663
    1664
            // which is different than `A OR (B OR C)`
    1665
            assert_ne!(expr, Some(col("a").or(col("b").or(col("c")))));
    1666
        }
    1667
    1668
        #[test]
    1669
        fn test_split_conjunction_owned_or() {
    1670
            let expr = col("a").eq(lit(5)).or(col("b"));
    1671
            assert_eq!(split_conjunction_owned(expr.clone()), vec![expr]);
    1672
        }
    1673
    1674
        #[test]
    1675
        fn test_collect_expr() -> Result<()> {
    1676
            let mut accum: HashSet<Column> = HashSet::new();
    1677
            expr_to_columns(
    1678
                &Expr::Cast(Cast::new(Box::new(col("a")), DataType::Float64)),
    1679
                &mut accum,
    1680
            )?;
    1681
            expr_to_columns(
    1682
                &Expr::Cast(Cast::new(Box::new(col("a")), DataType::Float64)),
    1683
                &mut accum,
    1684
            )?;
    1685
            assert_eq!(1, accum.len());
    1686
            assert!(accum.contains(&Column::from_name("a")));
    1687
            Ok(())
    1688
        }
    1689
    1690
        #[test]
    1691
        fn test_can_hash() {
    1692
            let union_fields: UnionFields = [
    1693
                (0, Arc::new(Field::new("A", DataType::Int32, true))),
    1694
                (1, Arc::new(Field::new("B", DataType::Float64, true))),
    1695
            ]
    1696
            .into_iter()
    1697
            .collect();
    1698
    1699
            let union_type = DataType::Union(union_fields, UnionMode::Sparse);
    1700
            assert!(!can_hash(&union_type));
    1701
    1702
            let list_union_type =
    1703
                DataType::List(Arc::new(Field::new("my_union", union_type, true)));
    1704
            assert!(!can_hash(&list_union_type));
    1705
        }
    1706
    1707
        #[test]
    1708
        fn test_generate_signature_error_msg_with_parameter_names() {
    1709
            let sig = Signature::one_of(
    1710
                vec![
    1711
                    TypeSignature::Exact(vec![DataType::Utf8, DataType::Int64]),
    1712
                    TypeSignature::Exact(vec![
    1713
                        DataType::Utf8,
    1714
                        DataType::Int64,
    1715
                        DataType::Int64,
    1716
                    ]),
    1717
                ],
    1718
                Volatility::Immutable,
    1719
            )
    1720
            .with_parameter_names(vec![
    1721
                "str".to_string(),
    1722
                "start_pos".to_string(),
    1723
                "length".to_string(),
    1724
            ])
    1725
            .expect("valid parameter names");
    1726
    1727
            // Generate error message with only 1 argument provided
    1728
            let error_msg = generate_signature_error_msg("substr", sig, &[DataType::Utf8]);
    1729
    1730
            assert!(
    1731
                error_msg.contains("str: Utf8, start_pos: Int64"),
    1732
                "Expected 'str: Utf8, start_pos: Int64' in error message, got: {error_msg}"
    1733
            );
    1734
            assert!(
    1735
                error_msg.contains("str: Utf8, start_pos: Int64, length: Int64"),
    1736
                "Expected 'str: Utf8, start_pos: Int64, length: Int64' in error message, got: {error_msg}"
    1737
            );
    1738
        }
    1739
    1740
        #[test]
    1741
        fn test_generate_signature_error_msg_without_parameter_names() {
    1742
            let sig = Signature::one_of(
    1743
                vec![TypeSignature::Any(2), TypeSignature::Any(3)],
    1744
                Volatility::Immutable,
    1745
            );
    1746
    1747
            let error_msg = generate_signature_error_msg("my_func", sig, &[DataType::Int32]);
    1748
    1749
            assert!(
    1750
                error_msg.contains("Any, Any"),
    1751
                "Expected 'Any, Any' without parameter names, got: {error_msg}"
    1752
            );
    1753
        }
    1754
    }