## 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/expr_schema.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
    use super::{predicate_bounds, Between, Expr, Like};
    19
    use crate::expr::{
    20
        AggregateFunction, AggregateFunctionParams, Alias, BinaryExpr, Cast, InList,
    21
        InSubquery, Placeholder, ScalarFunction, TryCast, Unnest, WindowFunction,
    22
        WindowFunctionParams,
    23
    };
    24
    use crate::expr_rewriter::rewrite_with_guarantees;
    25
    use crate::type_coercion::functions::{
    26
        data_types_with_scalar_udf, fields_with_aggregate_udf, fields_with_window_udf,
    27
    };
    28
    use crate::udf::ReturnFieldArgs;
    29
    use crate::{utils, LogicalPlan, Projection, Subquery, WindowFunctionDefinition};
    30
    use arrow::compute::can_cast_types;
    31
    use arrow::datatypes::{DataType, Field, FieldRef};
    32
    use datafusion_common::metadata::FieldMetadata;
    33
    use datafusion_common::{
    34
        not_impl_err, plan_datafusion_err, plan_err, Column, DataFusionError, ExprSchema,
    35
        Result, ScalarValue, Spans, TableReference,
    36
    };
    37
    use datafusion_expr_common::interval_arithmetic::NullableInterval;
    38
    use datafusion_expr_common::type_coercion::binary::BinaryTypeCoercer;
    39
    use datafusion_functions_window_common::field::WindowUDFFieldArgs;
    40
    use std::sync::Arc;
    41
    42
    /// Trait to allow expr to typable with respect to a schema
    43
    pub trait ExprSchemable {
    44
        /// Given a schema, return the type of the expr
    45
        fn get_type(&self, schema: &dyn ExprSchema) -> Result<DataType>;
    46
    47
        /// Given a schema, return the nullability of the expr
    48
        fn nullable(&self, input_schema: &dyn ExprSchema) -> Result<bool>;
    49
    50
        /// Given a schema, return the expr's optional metadata
    51
        fn metadata(&self, schema: &dyn ExprSchema) -> Result<FieldMetadata>;
    52
    53
        /// Convert to a field with respect to a schema
    54
        fn to_field(
    55
            &self,
    56
            input_schema: &dyn ExprSchema,
    57
        ) -> Result<(Option<TableReference>, Arc<Field>)>;
    58
    59
        /// Cast to a type with respect to a schema
    60
        fn cast_to(self, cast_to_type: &DataType, schema: &dyn ExprSchema) -> Result<Expr>;
    61
    62
        /// Given a schema, return the type and nullability of the expr
    63
        fn data_type_and_nullable(&self, schema: &dyn ExprSchema)
    64
            -> Result<(DataType, bool)>;
    65
    }
    66
    67
    impl ExprSchemable for Expr {
    68
        /// Returns the [arrow::datatypes::DataType] of the expression
    69
        /// based on [ExprSchema]
    70
        ///
    71
        /// Note: [`DFSchema`] implements [ExprSchema].
    72
        ///
    73
        /// [`DFSchema`]: datafusion_common::DFSchema
    74
        ///
    75
        /// # Examples
    76
        ///
    77
        /// Get the type of an expression that adds 2 columns. Adding an Int32
    78
        /// and Float32 results in Float32 type
    79
        ///
    80
        /// ```
    81
        /// # use arrow::datatypes::{DataType, Field};
    82
        /// # use datafusion_common::DFSchema;
    83
        /// # use datafusion_expr::{col, ExprSchemable};
    84
        /// # use std::collections::HashMap;
    85
        ///
    86
        /// fn main() {
    87
        ///     let expr = col("c1") + col("c2");
    88
        ///     let schema = DFSchema::from_unqualified_fields(
    89
        ///         vec![
    90
        ///             Field::new("c1", DataType::Int32, true),
    91
        ///             Field::new("c2", DataType::Float32, true),
    92
        ///         ]
    93
        ///         .into(),
    94
        ///         HashMap::new(),
    95
        ///     )
    96
        ///     .unwrap();
    97
        ///     assert_eq!("Float32", format!("{}", expr.get_type(&schema).unwrap()));
    98
        /// }
    99
        /// ```
    100
        ///
    101
        /// # Errors
    102
        ///
    103
        /// This function errors when it is not possible to compute its
    104
        /// [arrow::datatypes::DataType].  This happens when e.g. the
    105
        /// expression refers to a column that does not exist in the
    106
        /// schema, or when the expression is incorrectly typed
    107
        /// (e.g. `[utf8] + [bool]`).
    108
        #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
    109
    0
        fn get_type(&self, schema: &dyn ExprSchema) -> Result<DataType> {
    110
    0
            match self {
    111
    0
                Expr::Alias(Alias { expr, name, .. }) => match &**expr {
    112
    0
                    Expr::Placeholder(Placeholder { field, .. }) => match &field {
    113
    0
                        None => schema.data_type(&Column::from_name(name)).cloned(),
    114
    0
                        Some(field) => Ok(field.data_type().clone()),
    115
                    },
    116
    0
                    _ => expr.get_type(schema),
    117
                },
    118
    0
                Expr::Negative(expr) => expr.get_type(schema),
    119
    0
                Expr::Column(c) => Ok(schema.data_type(c)?.clone()),
    120
    0
                Expr::OuterReferenceColumn(field, _) => Ok(field.data_type().clone()),
    121
    0
                Expr::ScalarVariable(ty, _) => Ok(ty.clone()),
    122
    0
                Expr::Literal(l, _) => Ok(l.data_type()),
    123
    0
                Expr::Case(case) => {
    124
    0
                    for (_, then_expr) in &case.when_then_expr {
    125
    0
                        let then_type = then_expr.get_type(schema)?;
    126
    0
                        if !then_type.is_null() {
    127
    0
                            return Ok(then_type);
    128
    0
                        }
    129
                    }
    130
    0
                    case.else_expr
    131
    0
                        .as_ref()
    132
    0
                        .map_or(Ok(DataType::Null), |e| e.get_type(schema))
    133
                }
    134
    0
                Expr::Cast(Cast { data_type, .. })
    135
    0
                | Expr::TryCast(TryCast { data_type, .. }) => Ok(data_type.clone()),
    136
    0
                Expr::Unnest(Unnest { expr }) => {
    137
    0
                    let arg_data_type = expr.get_type(schema)?;
    138
                    // Unnest's output type is the inner type of the list
    139
    0
                    match arg_data_type {
    140
    0
                        DataType::List(field)
    141
    0
                        | DataType::LargeList(field)
    142
    0
                        | DataType::FixedSizeList(field, _) => Ok(field.data_type().clone()),
    143
    0
                        DataType::Struct(_) => Ok(arg_data_type),
    144
                        DataType::Null => {
    145
    0
                            not_impl_err!("unnest() does not support null yet")
    146
                        }
    147
                        _ => {
    148
    0
                            plan_err!(
    149
    0
                                "unnest() can only be applied to array, struct and null"
    150
                            )
    151
                        }
    152
                    }
    153
                }
    154
    0
                Expr::ScalarFunction(_func) => {
    155
    0
                    let (return_type, _) = self.data_type_and_nullable(schema)?;
    156
    0
                    Ok(return_type)
    157
                }
    158
    0
                Expr::WindowFunction(window_function) => self
    159
    0
                    .data_type_and_nullable_with_window_function(schema, window_function)
    160
    0
                    .map(|(return_type, _)| return_type),
    161
                Expr::AggregateFunction(AggregateFunction {
    162
    0
                    func,
    163
    0
                    params: AggregateFunctionParams { args, .. },
    164
                }) => {
    165
    0
                    let fields = args
    166
    0
                        .iter()
    167
    0
                        .map(|e| e.to_field(schema).map(|(_, f)| f))
    168
    0
                        .collect::<Result<Vec<_>>>()?;
    169
    0
                    let new_fields = fields_with_aggregate_udf(&fields, func)
    170
    0
                        .map_err(|err| {
    171
    0
                            let data_types = fields
    172
    0
                                .iter()
    173
    0
                                .map(|f| f.data_type().clone())
    174
    0
                                .collect::<Vec<_>>();
    175
    0
                            plan_datafusion_err!(
    176
    0
                                "{} {}",
    177
    0
                                match err {
    178
    0
                                    DataFusionError::Plan(msg) => msg,
    179
    0
                                    err => err.to_string(),
    180
                                },
    181
    0
                                utils::generate_signature_error_msg(
    182
    0
                                    func.name(),
    183
    0
                                    func.signature().clone(),
    184
    0
                                    &data_types
    185
                                )
    186
                            )
    187
    0
                        })?
    188
    0
                        .into_iter()
    189
    0
                        .collect::<Vec<_>>();
    190
    0
                    Ok(func.return_field(&new_fields)?.data_type().clone())
    191
                }
    192
                Expr::Not(_)
    193
                | Expr::IsNull(_)
    194
                | Expr::Exists { .. }
    195
                | Expr::InSubquery(_)
    196
                | Expr::Between { .. }
    197
                | Expr::InList { .. }
    198
                | Expr::IsNotNull(_)
    199
                | Expr::IsTrue(_)
    200
                | Expr::IsFalse(_)
    201
                | Expr::IsUnknown(_)
    202
                | Expr::IsNotTrue(_)
    203
                | Expr::IsNotFalse(_)
    204
    0
                | Expr::IsNotUnknown(_) => Ok(DataType::Boolean),
    205
    0
                Expr::ScalarSubquery(subquery) => {
    206
    0
                    Ok(subquery.subquery.schema().field(0).data_type().clone())
    207
                }
    208
                Expr::BinaryExpr(BinaryExpr {
    209
    0
                    ref left,
    210
    0
                    ref right,
    211
    0
                    ref op,
    212
    0
                }) => BinaryTypeCoercer::new(
    213
    0
                    &left.get_type(schema)?,
    214
    0
                    op,
    215
    0
                    &right.get_type(schema)?,
    216
                )
    217
    0
                .get_result_type(),
    218
    0
                Expr::Like { .. } | Expr::SimilarTo { .. } => Ok(DataType::Boolean),
    219
    0
                Expr::Placeholder(Placeholder { field, .. }) => {
    220
    0
                    if let Some(field) = field {
    221
    0
                        Ok(field.data_type().clone())
    222
                    } else {
    223
                        // If the placeholder's type hasn't been specified, treat it as
    224
                        // null (unspecified placeholders generate an error during planning)
    225
    0
                        Ok(DataType::Null)
    226
                    }
    227
                }
    228
                #[expect(deprecated)]
    229
    0
                Expr::Wildcard { .. } => Ok(DataType::Null),
    230
                Expr::GroupingSet(_) => {
    231
                    // Grouping sets do not really have a type and do not appear in projections
    232
    0
                    Ok(DataType::Null)
    233
                }
    234
            }
    235
    0
        }
    236
    237
        /// Returns the nullability of the expression based on [ExprSchema].
    238
        ///
    239
        /// Note: [`DFSchema`] implements [ExprSchema].
    240
        ///
    241
        /// [`DFSchema`]: datafusion_common::DFSchema
    242
        ///
    243
        /// # Errors
    244
        ///
    245
        /// This function errors when it is not possible to compute its
    246
        /// nullability.  This happens when the expression refers to a
    247
        /// column that does not exist in the schema.
    248
    0
        fn nullable(&self, input_schema: &dyn ExprSchema) -> Result<bool> {
    249
    0
            match self {
    250
    0
                Expr::Alias(Alias { expr, .. }) | Expr::Not(expr) | Expr::Negative(expr) => {
    251
    0
                    expr.nullable(input_schema)
    252
                }
    253
    254
    0
                Expr::InList(InList { expr, list, .. }) => {
    255
                    // Avoid inspecting too many expressions.
    256
                    const MAX_INSPECT_LIMIT: usize = 6;
    257
                    // Stop if a nullable expression is found or an error occurs.
    258
    0
                    let has_nullable = std::iter::once(expr.as_ref())
    259
    0
                        .chain(list)
    260
    0
                        .take(MAX_INSPECT_LIMIT)
    261
    0
                        .find_map(|e| {
    262
    0
                            e.nullable(input_schema)
    263
    0
                                .map(|nullable| if nullable { Some(()) } else { None })
    264
    0
                                .transpose()
    265
    0
                        })
    266
    0
                        .transpose()?;
    267
    0
                    Ok(match has_nullable {
    268
                        // If a nullable subexpression is found, the result may also be nullable.
    269
    0
                        Some(_) => true,
    270
                        // If the list is too long, we assume it is nullable.
    271
    0
                        None if list.len() + 1 > MAX_INSPECT_LIMIT => true,
    272
                        // All the subexpressions are non-nullable, so the result must be non-nullable.
    273
    0
                        _ => false,
    274
                    })
    275
                }
    276
    277
                Expr::Between(Between {
    278
    0
                    expr, low, high, ..
    279
    0
                }) => Ok(expr.nullable(input_schema)?
    280
    0
                    || low.nullable(input_schema)?
    281
    0
                    || high.nullable(input_schema)?),
    282
    283
    0
                Expr::Column(c) => input_schema.nullable(c),
    284
    0
                Expr::OuterReferenceColumn(field, _) => Ok(field.is_nullable()),
    285
    0
                Expr::Literal(value, _) => Ok(value.is_null()),
    286
    0
                Expr::Case(case) => {
    287
    0
                    let nullable_then = case
    288
    0
                        .when_then_expr
    289
    0
                        .iter()
    290
    0
                        .filter_map(|(w, t)| {
    291
    0
                            let is_nullable = match t.nullable(input_schema) {
    292
    0
                                Err(e) => return Some(Err(e)),
    293
    0
                                Ok(n) => n,
    294
                            };
    295
    296
                            // Branches with a then expression that is not nullable do not impact the
    297
                            // nullability of the case expression.
    298
    0
                            if !is_nullable {
    299
    0
                                return None;
    300
    0
                            }
    301
    302
                            // For case-with-expression assume all 'then' expressions are reachable
    303
    0
                            if case.expr.is_some() {
    304
    0
                                return Some(Ok(()));
    305
    0
                            }
    306
    307
                            // For branches with a nullable 'then' expression, try to determine
    308
                            // if the 'then' expression is ever reachable in the situation where
    309
                            // it would evaluate to null.
    310
    311
                            // First, derive a variant of the 'when' expression, where all occurrences
    312
                            // of the 'then' expression have been replaced by 'NULL'.
    313
    0
                            let certainly_null_expr = unwrap_certainly_null_expr(t).clone();
    314
    0
                            let certainly_null_type =
    315
    0
                                match certainly_null_expr.get_type(input_schema) {
    316
    0
                                    Err(e) => return Some(Err(e)),
    317
    0
                                    Ok(datatype) => datatype,
    318
                                };
    319
    0
                            let null_interval = NullableInterval::Null {
    320
    0
                                datatype: certainly_null_type,
    321
    0
                            };
    322
    0
                            let guarantees = vec![(certainly_null_expr, null_interval)];
    323
    0
                            let when_with_null =
    324
    0
                                match rewrite_with_guarantees(*w.clone(), &guarantees) {
    325
    0
                                    Err(e) => return Some(Err(e)),
    326
    0
                                    Ok(e) => e.data,
    327
                                };
    328
    329
                            // Next, determine the bounds of the derived 'when' expression to see if it
    330
                            // can ever evaluate to true.
    331
    0
                            let bounds = match predicate_bounds::evaluate_bounds(
    332
    0
                                &when_with_null,
    333
    0
                                input_schema,
    334
    0
                            ) {
    335
    0
                                Err(e) => return Some(Err(e)),
    336
    0
                                Ok(b) => b,
    337
                            };
    338
    339
    0
                            let can_be_true = match bounds
    340
    0
                                .contains_value(ScalarValue::Boolean(Some(true)))
    341
                            {
    342
    0
                                Err(e) => return Some(Err(e)),
    343
    0
                                Ok(b) => b,
    344
                            };
    345
    346
    0
                            if !can_be_true {
    347
                                // If the derived 'when' expression can never evaluate to true, the
    348
                                // 'then' expression is not reachable when it would evaluate to NULL.
    349
                                // The most common pattern for this is `WHEN x IS NOT NULL THEN x`.
    350
    0
                                None
    351
                            } else {
    352
                                // The branch might be taken
    353
    0
                                Some(Ok(()))
    354
                            }
    355
    0
                        })
    356
    0
                        .next();
    357
    358
    0
                    if let Some(nullable_then) = nullable_then {
    359
                        // There is at least one reachable nullable 'then' expression, so the case
    360
                        // expression itself is nullable.
    361
                        // Use `Result::map` to propagate the error from `nullable_then` if there is one.
    362
    0
                        nullable_then.map(|_| true)
    363
    0
                    } else if let Some(e) = &case.else_expr {
    364
                        // There are no reachable nullable 'then' expressions, so all we still need to
    365
                        // check is the 'else' expression's nullability.
    366
    0
                        e.nullable(input_schema)
    367
                    } else {
    368
                        // CASE produces NULL if there is no `else` expr
    369
                        // (aka when none of the `when_then_exprs` match)
    370
    0
                        Ok(true)
    371
                    }
    372
                }
    373
    0
                Expr::Cast(Cast { expr, .. }) => expr.nullable(input_schema),
    374
    0
                Expr::ScalarFunction(_func) => {
    375
    0
                    let (_, nullable) = self.data_type_and_nullable(input_schema)?;
    376
    0
                    Ok(nullable)
    377
                }
    378
    0
                Expr::AggregateFunction(AggregateFunction { func, .. }) => {
    379
    0
                    Ok(func.is_nullable())
    380
                }
    381
    0
                Expr::WindowFunction(window_function) => self
    382
    0
                    .data_type_and_nullable_with_window_function(
    383
    0
                        input_schema,
    384
    0
                        window_function,
    385
                    )
    386
    0
                    .map(|(_, nullable)| nullable),
    387
    0
                Expr::Placeholder(Placeholder { id: _, field }) => {
    388
    0
                    Ok(field.as_ref().map(|f| f.is_nullable()).unwrap_or(true))
    389
                }
    390
                Expr::ScalarVariable(_, _) | Expr::TryCast { .. } | Expr::Unnest(_) => {
    391
    0
                    Ok(true)
    392
                }
    393
                Expr::IsNull(_)
    394
                | Expr::IsNotNull(_)
    395
                | Expr::IsTrue(_)
    396
                | Expr::IsFalse(_)
    397
                | Expr::IsUnknown(_)
    398
                | Expr::IsNotTrue(_)
    399
                | Expr::IsNotFalse(_)
    400
                | Expr::IsNotUnknown(_)
    401
    0
                | Expr::Exists { .. } => Ok(false),
    402
    0
                Expr::InSubquery(InSubquery { expr, .. }) => expr.nullable(input_schema),
    403
    0
                Expr::ScalarSubquery(subquery) => {
    404
    0
                    Ok(subquery.subquery.schema().field(0).is_nullable())
    405
                }
    406
                Expr::BinaryExpr(BinaryExpr {
    407
    0
                    ref left,
    408
    0
                    ref right,
    409
                    ..
    410
    0
                }) => Ok(left.nullable(input_schema)? || right.nullable(input_schema)?),
    411
    0
                Expr::Like(Like { expr, pattern, .. })
    412
    0
                | Expr::SimilarTo(Like { expr, pattern, .. }) => {
    413
    0
                    Ok(expr.nullable(input_schema)? || pattern.nullable(input_schema)?)
    414
                }
    415
                #[expect(deprecated)]
    416
    0
                Expr::Wildcard { .. } => Ok(false),
    417
                Expr::GroupingSet(_) => {
    418
                    // Grouping sets do not really have the concept of nullable and do not appear
    419
                    // in projections
    420
    0
                    Ok(true)
    421
                }
    422
            }
    423
    0
        }
    424
    425
    0
        fn metadata(&self, schema: &dyn ExprSchema) -> Result<FieldMetadata> {
    426
    0
            self.to_field(schema)
    427
    0
                .map(|(_, field)| FieldMetadata::from(field.metadata()))
    428
    0
        }
    429
    430
        /// Returns the datatype and nullability of the expression based on [ExprSchema].
    431
        ///
    432
        /// Note: [`DFSchema`] implements [ExprSchema].
    433
        ///
    434
        /// [`DFSchema`]: datafusion_common::DFSchema
    435
        ///
    436
        /// # Errors
    437
        ///
    438
        /// This function errors when it is not possible to compute its
    439
        /// datatype or nullability.
    440
    0
        fn data_type_and_nullable(
    441
    0
            &self,
    442
    0
            schema: &dyn ExprSchema,
    443
    0
        ) -> Result<(DataType, bool)> {
    444
    0
            let field = self.to_field(schema)?.1;
    445
    446
    0
            Ok((field.data_type().clone(), field.is_nullable()))
    447
    0
        }
    448
    449
        /// Returns a [arrow::datatypes::Field] compatible with this expression.
    450
        ///
    451
        /// This function converts an expression into a field with appropriate metadata
    452
        /// and nullability based on the expression type and context. It is the primary
    453
        /// mechanism for determining field-level schemas.
    454
        ///
    455
        /// # Field Property Resolution
    456
        ///
    457
        /// For each expression, the following properties are determined:
    458
        ///
    459
        /// ## Data Type Resolution
    460
        /// - **Column references**: Data type from input schema field
    461
        /// - **Literals**: Data type inferred from literal value
    462
        /// - **Aliases**: Data type inherited from the underlying expression (the aliased expression)
    463
        /// - **Binary expressions**: Result type from type coercion rules
    464
        /// - **Boolean expressions**: Always a boolean type
    465
        /// - **Cast expressions**: Target data type from cast operation
    466
        /// - **Function calls**: Return type based on function signature and argument types
    467
        ///
    468
        /// ## Nullability Determination
    469
        /// - **Column references**: Inherit nullability from input schema field
    470
        /// - **Literals**: Nullable only if literal value is NULL
    471
        /// - **Aliases**: Inherit nullability from the underlying expression (the aliased expression)
    472
        /// - **Binary expressions**: Nullable if either operand is nullable
    473
        /// - **Boolean expressions**: Always non-nullable (IS NULL, EXISTS, etc.)
    474
        /// - **Cast expressions**: determined by the input expression's nullability rules
    475
        /// - **Function calls**: Based on function nullability rules and input nullability
    476
        ///
    477
        /// ## Metadata Handling
    478
        /// - **Column references**: Preserve original field metadata from input schema
    479
        /// - **Literals**: Use explicitly provided metadata, otherwise empty
    480
        /// - **Aliases**: Merge underlying expr metadata with alias-specific metadata, preferring the alias metadata
    481
        /// - **Binary expressions**: field metadata is empty
    482
        /// - **Boolean expressions**: field metadata is empty
    483
        /// - **Cast expressions**: determined by the input expression's field metadata handling
    484
        /// - **Scalar functions**: Generate metadata via function's [`return_field_from_args`] method,
    485
        ///   with the default implementation returning empty field metadata
    486
        /// - **Aggregate functions**: Generate metadata via function's [`return_field`] method,
    487
        ///   with the default implementation returning empty field metadata
    488
        /// - **Window functions**: field metadata is empty
    489
        ///
    490
        /// ## Table Reference Scoping
    491
        /// - Establishes proper qualified field references when columns belong to specific tables
    492
        /// - Maintains table context for accurate field resolution in multi-table scenarios
    493
        ///
    494
        /// So for example, a projected expression `col(c1) + col(c2)` is
    495
        /// placed in an output field **named** col("c1 + c2")
    496
        ///
    497
        /// [`return_field_from_args`]: crate::ScalarUDF::return_field_from_args
    498
        /// [`return_field`]: crate::AggregateUDF::return_field
    499
    0
        fn to_field(
    500
    0
            &self,
    501
    0
            schema: &dyn ExprSchema,
    502
    0
        ) -> Result<(Option<TableReference>, Arc<Field>)> {
    503
    0
            let (relation, schema_name) = self.qualified_name();
    504
            #[allow(deprecated)]
    505
    0
            let field = match self {
    506
                Expr::Alias(Alias {
    507
    0
                    expr,
    508
                    name: _,
    509
    0
                    metadata,
    510
                    ..
    511
                }) => {
    512
    0
                    let field = expr.to_field(schema).map(|(_, f)| f.as_ref().clone())?;
    513
    514
    0
                    let mut combined_metadata = expr.metadata(schema)?;
    515
    0
                    if let Some(metadata) = metadata {
    516
    0
                        combined_metadata.extend(metadata.clone());
    517
    0
                    }
    518
    519
    0
                    Ok(Arc::new(combined_metadata.add_to_field(field)))
    520
                }
    521
    0
                Expr::Negative(expr) => expr.to_field(schema).map(|(_, f)| f),
    522
    0
                Expr::Column(c) => schema.field_from_column(c).map(|f| Arc::new(f.clone())),
    523
    0
                Expr::OuterReferenceColumn(field, _) => {
    524
    0
                    Ok(Arc::new(field.as_ref().clone().with_name(&schema_name)))
    525
                }
    526
    0
                Expr::ScalarVariable(ty, _) => {
    527
    0
                    Ok(Arc::new(Field::new(&schema_name, ty.clone(), true)))
    528
                }
    529
    0
                Expr::Literal(l, metadata) => {
    530
    0
                    let mut field = Field::new(&schema_name, l.data_type(), l.is_null());
    531
    0
                    if let Some(metadata) = metadata {
    532
    0
                        field = metadata.add_to_field(field);
    533
    0
                    }
    534
    0
                    Ok(Arc::new(field))
    535
                }
    536
                Expr::IsNull(_)
    537
                | Expr::IsNotNull(_)
    538
                | Expr::IsTrue(_)
    539
                | Expr::IsFalse(_)
    540
                | Expr::IsUnknown(_)
    541
                | Expr::IsNotTrue(_)
    542
                | Expr::IsNotFalse(_)
    543
                | Expr::IsNotUnknown(_)
    544
                | Expr::Exists { .. } => {
    545
    0
                    Ok(Arc::new(Field::new(&schema_name, DataType::Boolean, false)))
    546
                }
    547
    0
                Expr::ScalarSubquery(subquery) => {
    548
    0
                    Ok(Arc::clone(&subquery.subquery.schema().fields()[0]))
    549
                }
    550
                Expr::BinaryExpr(BinaryExpr {
    551
    0
                    ref left,
    552
    0
                    ref right,
    553
    0
                    ref op,
    554
                }) => {
    555
    0
                    let (lhs_type, lhs_nullable) = left.data_type_and_nullable(schema)?;
    556
    0
                    let (rhs_type, rhs_nullable) = right.data_type_and_nullable(schema)?;
    557
    0
                    let mut coercer = BinaryTypeCoercer::new(&lhs_type, op, &rhs_type);
    558
    0
                    coercer.set_lhs_spans(left.spans().cloned().unwrap_or_default());
    559
    0
                    coercer.set_rhs_spans(right.spans().cloned().unwrap_or_default());
    560
    0
                    Ok(Arc::new(Field::new(
    561
    0
                        &schema_name,
    562
    0
                        coercer.get_result_type()?,
    563
    0
                        lhs_nullable || rhs_nullable,
    564
                    )))
    565
                }
    566
    0
                Expr::WindowFunction(window_function) => {
    567
    0
                    let (dt, nullable) = self.data_type_and_nullable_with_window_function(
    568
    0
                        schema,
    569
    0
                        window_function,
    570
    0
                    )?;
    571
    0
                    Ok(Arc::new(Field::new(&schema_name, dt, nullable)))
    572
                }
    573
    0
                Expr::AggregateFunction(aggregate_function) => {
    574
                    let AggregateFunction {
    575
    0
                        func,
    576
    0
                        params: AggregateFunctionParams { args, .. },
    577
                        ..
    578
    0
                    } = aggregate_function;
    579
    580
    0
                    let fields = args
    581
    0
                        .iter()
    582
    0
                        .map(|e| e.to_field(schema).map(|(_, f)| f))
    583
    0
                        .collect::<Result<Vec<_>>>()?;
    584
                    // Verify that function is invoked with correct number and type of arguments as defined in `TypeSignature`
    585
    0
                    let new_fields = fields_with_aggregate_udf(&fields, func)
    586
    0
                        .map_err(|err| {
    587
    0
                            let arg_types = fields
    588
    0
                                .iter()
    589
    0
                                .map(|f| f.data_type())
    590
    0
                                .cloned()
    591
    0
                                .collect::<Vec<_>>();
    592
    0
                            plan_datafusion_err!(
    593
    0
                                "{} {}",
    594
    0
                                match err {
    595
    0
                                    DataFusionError::Plan(msg) => msg,
    596
    0
                                    err => err.to_string(),
    597
                                },
    598
    0
                                utils::generate_signature_error_msg(
    599
    0
                                    func.name(),
    600
    0
                                    func.signature().clone(),
    601
    0
                                    &arg_types,
    602
                                )
    603
                            )
    604
    0
                        })?
    605
    0
                        .into_iter()
    606
    0
                        .collect::<Vec<_>>();
    607
    608
    0
                    func.return_field(&new_fields)
    609
                }
    610
    0
                Expr::ScalarFunction(ScalarFunction { func, args }) => {
    611
    0
                    let (arg_types, fields): (Vec<DataType>, Vec<Arc<Field>>) = args
    612
    0
                        .iter()
    613
    0
                        .map(|e| e.to_field(schema).map(|(_, f)| f))
    614
    0
                        .collect::<Result<Vec<_>>>()?
    615
    0
                        .into_iter()
    616
    0
                        .map(|f| (f.data_type().clone(), f))
    617
    0
                        .unzip();
    618
                    // Verify that function is invoked with correct number and type of arguments as defined in `TypeSignature`
    619
    0
                    let new_data_types = data_types_with_scalar_udf(&arg_types, func)
    620
    0
                        .map_err(|err| {
    621
    0
                            plan_datafusion_err!(
    622
    0
                                "{} {}",
    623
    0
                                match err {
    624
    0
                                    DataFusionError::Plan(msg) => msg,
    625
    0
                                    err => err.to_string(),
    626
                                },
    627
    0
                                utils::generate_signature_error_msg(
    628
    0
                                    func.name(),
    629
    0
                                    func.signature().clone(),
    630
    0
                                    &arg_types,
    631
                                )
    632
                            )
    633
    0
                        })?;
    634
    0
                    let new_fields = fields
    635
    0
                        .into_iter()
    636
    0
                        .zip(new_data_types)
    637
    0
                        .map(|(f, d)| f.as_ref().clone().with_data_type(d))
    638
    0
                        .map(Arc::new)
    639
    0
                        .collect::<Vec<FieldRef>>();
    640
    641
    0
                    let arguments = args
    642
    0
                        .iter()
    643
    0
                        .map(|e| match e {
    644
    0
                            Expr::Literal(sv, _) => Some(sv),
    645
    0
                            _ => None,
    646
    0
                        })
    647
    0
                        .collect::<Vec<_>>();
    648
    0
                    let args = ReturnFieldArgs {
    649
    0
                        arg_fields: &new_fields,
    650
    0
                        scalar_arguments: &arguments,
    651
    0
                    };
    652
    653
    0
                    func.return_field_from_args(args)
    654
                }
    655
                // _ => Ok((self.get_type(schema)?, self.nullable(schema)?)),
    656
    0
                Expr::Cast(Cast { expr, data_type }) => expr
    657
    0
                    .to_field(schema)
    658
    0
                    .map(|(_, f)| f.as_ref().clone().with_data_type(data_type.clone()))
    659
    0
                    .map(Arc::new),
    660
                Expr::Placeholder(Placeholder {
    661
                    id: _,
    662
    0
                    field: Some(field),
    663
    0
                }) => Ok(field.as_ref().clone().with_name(&schema_name).into()),
    664
                Expr::Like(_)
    665
                | Expr::SimilarTo(_)
    666
                | Expr::Not(_)
    667
                | Expr::Between(_)
    668
                | Expr::Case(_)
    669
                | Expr::TryCast(_)
    670
                | Expr::InList(_)
    671
                | Expr::InSubquery(_)
    672
                | Expr::Wildcard { .. }
    673
                | Expr::GroupingSet(_)
    674
                | Expr::Placeholder(_)
    675
    0
                | Expr::Unnest(_) => Ok(Arc::new(Field::new(
    676
    0
                    &schema_name,
    677
    0
                    self.get_type(schema)?,
    678
    0
                    self.nullable(schema)?,
    679
                ))),
    680
    0
            }?;
    681
    682
    0
            Ok((
    683
    0
                relation,
    684
    0
                Arc::new(field.as_ref().clone().with_name(schema_name)),
    685
    0
            ))
    686
    0
        }
    687
    688
        /// Wraps this expression in a cast to a target [arrow::datatypes::DataType].
    689
        ///
    690
        /// # Errors
    691
        ///
    692
        /// This function errors when it is impossible to cast the
    693
        /// expression to the target [arrow::datatypes::DataType].
    694
    0
        fn cast_to(self, cast_to_type: &DataType, schema: &dyn ExprSchema) -> Result<Expr> {
    695
    0
            let this_type = self.get_type(schema)?;
    696
    0
            if this_type == *cast_to_type {
    697
    0
                return Ok(self);
    698
    0
            }
    699
    700
            // TODO(kszucs): Most of the operations do not validate the type correctness
    701
            // like all of the binary expressions below. Perhaps Expr should track the
    702
            // type of the expression?
    703
    704
    0
            if can_cast_types(&this_type, cast_to_type) {
    705
    0
                match self {
    706
    0
                    Expr::ScalarSubquery(subquery) => {
    707
    0
                        Ok(Expr::ScalarSubquery(cast_subquery(subquery, cast_to_type)?))
    708
                    }
    709
    0
                    _ => Ok(Expr::Cast(Cast::new(Box::new(self), cast_to_type.clone()))),
    710
                }
    711
            } else {
    712
    0
                plan_err!("Cannot automatically convert {this_type} to {cast_to_type}")
    713
            }
    714
    0
        }
    715
    }
    716
    717
    /// Returns the innermost [Expr] that is provably null if `expr` is null.
    718
    0
    fn unwrap_certainly_null_expr(expr: &Expr) -> &Expr {
    719
    0
        match expr {
    720
    0
            Expr::Not(e) => unwrap_certainly_null_expr(e),
    721
    0
            Expr::Negative(e) => unwrap_certainly_null_expr(e),
    722
    0
            Expr::Cast(e) => unwrap_certainly_null_expr(e.expr.as_ref()),
    723
    0
            _ => expr,
    724
        }
    725
    0
    }
    726
    727
    impl Expr {
    728
        /// Common method for window functions that applies type coercion
    729
        /// to all arguments of the window function to check if it matches
    730
        /// its signature.
    731
        ///
    732
        /// If successful, this method returns the data type and
    733
        /// nullability of the window function's result.
    734
        ///
    735
        /// Otherwise, returns an error if there's a type mismatch between
    736
        /// the window function's signature and the provided arguments.
    737
    0
        fn data_type_and_nullable_with_window_function(
    738
    0
            &self,
    739
    0
            schema: &dyn ExprSchema,
    740
    0
            window_function: &WindowFunction,
    741
    0
        ) -> Result<(DataType, bool)> {
    742
            let WindowFunction {
    743
    0
                fun,
    744
    0
                params: WindowFunctionParams { args, .. },
    745
                ..
    746
    0
            } = window_function;
    747
    748
    0
            let fields = args
    749
    0
                .iter()
    750
    0
                .map(|e| e.to_field(schema).map(|(_, f)| f))
    751
    0
                .collect::<Result<Vec<_>>>()?;
    752
    0
            match fun {
    753
    0
                WindowFunctionDefinition::AggregateUDF(udaf) => {
    754
    0
                    let data_types = fields
    755
    0
                        .iter()
    756
    0
                        .map(|f| f.data_type())
    757
    0
                        .cloned()
    758
    0
                        .collect::<Vec<_>>();
    759
    0
                    let new_fields = fields_with_aggregate_udf(&fields, udaf)
    760
    0
                        .map_err(|err| {
    761
    0
                            plan_datafusion_err!(
    762
    0
                                "{} {}",
    763
    0
                                match err {
    764
    0
                                    DataFusionError::Plan(msg) => msg,
    765
    0
                                    err => err.to_string(),
    766
                                },
    767
    0
                                utils::generate_signature_error_msg(
    768
    0
                                    fun.name(),
    769
    0
                                    fun.signature(),
    770
    0
                                    &data_types
    771
                                )
    772
                            )
    773
    0
                        })?
    774
    0
                        .into_iter()
    775
    0
                        .collect::<Vec<_>>();
    776
    777
    0
                    let return_field = udaf.return_field(&new_fields)?;
    778
    779
    0
                    Ok((return_field.data_type().clone(), return_field.is_nullable()))
    780
                }
    781
    0
                WindowFunctionDefinition::WindowUDF(udwf) => {
    782
    0
                    let data_types = fields
    783
    0
                        .iter()
    784
    0
                        .map(|f| f.data_type())
    785
    0
                        .cloned()
    786
    0
                        .collect::<Vec<_>>();
    787
    0
                    let new_fields = fields_with_window_udf(&fields, udwf)
    788
    0
                        .map_err(|err| {
    789
    0
                            plan_datafusion_err!(
    790
    0
                                "{} {}",
    791
    0
                                match err {
    792
    0
                                    DataFusionError::Plan(msg) => msg,
    793
    0
                                    err => err.to_string(),
    794
                                },
    795
    0
                                utils::generate_signature_error_msg(
    796
    0
                                    fun.name(),
    797
    0
                                    fun.signature(),
    798
    0
                                    &data_types
    799
                                )
    800
                            )
    801
    0
                        })?
    802
    0
                        .into_iter()
    803
    0
                        .collect::<Vec<_>>();
    804
    0
                    let (_, function_name) = self.qualified_name();
    805
    0
                    let field_args = WindowUDFFieldArgs::new(&new_fields, &function_name);
    806
    807
    0
                    udwf.field(field_args)
    808
    0
                        .map(|field| (field.data_type().clone(), field.is_nullable()))
    809
                }
    810
            }
    811
    0
        }
    812
    }
    813
    814
    /// Cast subquery in InSubquery/ScalarSubquery to a given type.
    815
    ///
    816
    /// 1. **Projection plan**: If the subquery is a projection (i.e. a SELECT statement with specific
    817
    ///    columns), it casts the first expression in the projection to the target type and creates a
    818
    ///    new projection with the casted expression.
    819
    /// 2. **Non-projection plan**: If the subquery isn't a projection, it adds a projection to the plan
    820
    ///    with the casted first column.
    821
    0
    pub fn cast_subquery(subquery: Subquery, cast_to_type: &DataType) -> Result<Subquery> {
    822
    0
        if subquery.subquery.schema().field(0).data_type() == cast_to_type {
    823
    0
            return Ok(subquery);
    824
    0
        }
    825
    826
    0
        let plan = subquery.subquery.as_ref();
    827
    0
        let new_plan = match plan {
    828
    0
            LogicalPlan::Projection(projection) => {
    829
    0
                let cast_expr = projection.expr[0]
    830
    0
                    .clone()
    831
    0
                    .cast_to(cast_to_type, projection.input.schema())?;
    832
    0
                LogicalPlan::Projection(Projection::try_new(
    833
    0
                    vec![cast_expr],
    834
    0
                    Arc::clone(&projection.input),
    835
    0
                )?)
    836
            }
    837
            _ => {
    838
    0
                let cast_expr = Expr::Column(Column::from(plan.schema().qualified_field(0)))
    839
    0
                    .cast_to(cast_to_type, subquery.subquery.schema())?;
    840
    0
                LogicalPlan::Projection(Projection::try_new(
    841
    0
                    vec![cast_expr],
    842
    0
                    subquery.subquery,
    843
    0
                )?)
    844
            }
    845
        };
    846
    0
        Ok(Subquery {
    847
    0
            subquery: Arc::new(new_plan),
    848
    0
            outer_ref_columns: subquery.outer_ref_columns,
    849
    0
            spans: Spans::new(),
    850
    0
        })
    851
    0
    }
    852
    853
    #[cfg(test)]
    854
    mod tests {
    855
        use std::collections::HashMap;
    856
    857
        use super::*;
    858
        use crate::{and, col, lit, not, or, out_ref_col_with_metadata, when};
    859
    860
        use datafusion_common::{
    861
            assert_or_internal_err, DFSchema, DataFusionError, ScalarValue,
    862
        };
    863
    864
        macro_rules! test_is_expr_nullable {
    865
            ($EXPR_TYPE:ident) => {{
    866
                let expr = lit(ScalarValue::Null).$EXPR_TYPE();
    867
                assert!(!expr.nullable(&MockExprSchema::new()).unwrap());
    868
            }};
    869
        }
    870
    871
        #[test]
    872
        fn expr_schema_nullability() {
    873
            let expr = col("foo").eq(lit(1));
    874
            assert!(!expr.nullable(&MockExprSchema::new()).unwrap());
    875
            assert!(expr
    876
                .nullable(&MockExprSchema::new().with_nullable(true))
    877
                .unwrap());
    878
    879
            test_is_expr_nullable!(is_null);
    880
            test_is_expr_nullable!(is_not_null);
    881
            test_is_expr_nullable!(is_true);
    882
            test_is_expr_nullable!(is_not_true);
    883
            test_is_expr_nullable!(is_false);
    884
            test_is_expr_nullable!(is_not_false);
    885
            test_is_expr_nullable!(is_unknown);
    886
            test_is_expr_nullable!(is_not_unknown);
    887
        }
    888
    889
        #[test]
    890
        fn test_between_nullability() {
    891
            let get_schema = |nullable| {
    892
                MockExprSchema::new()
    893
                    .with_data_type(DataType::Int32)
    894
                    .with_nullable(nullable)
    895
            };
    896
    897
            let expr = col("foo").between(lit(1), lit(2));
    898
            assert!(!expr.nullable(&get_schema(false)).unwrap());
    899
            assert!(expr.nullable(&get_schema(true)).unwrap());
    900
    901
            let null = lit(ScalarValue::Int32(None));
    902
    903
            let expr = col("foo").between(null.clone(), lit(2));
    904
            assert!(expr.nullable(&get_schema(false)).unwrap());
    905
    906
            let expr = col("foo").between(lit(1), null.clone());
    907
            assert!(expr.nullable(&get_schema(false)).unwrap());
    908
    909
            let expr = col("foo").between(null.clone(), null);
    910
            assert!(expr.nullable(&get_schema(false)).unwrap());
    911
        }
    912
    913
        fn assert_nullability(expr: &Expr, schema: &dyn ExprSchema, expected: bool) {
    914
            assert_eq!(
    915
                expr.nullable(schema).unwrap(),
    916
                expected,
    917
                "Nullability of '{expr}' should be {expected}"
    918
            );
    919
        }
    920
    921
        fn assert_not_nullable(expr: &Expr, schema: &dyn ExprSchema) {
    922
            assert_nullability(expr, schema, false);
    923
        }
    924
    925
        fn assert_nullable(expr: &Expr, schema: &dyn ExprSchema) {
    926
            assert_nullability(expr, schema, true);
    927
        }
    928
    929
        #[test]
    930
        fn test_case_expression_nullability() -> Result<()> {
    931
            let nullable_schema = MockExprSchema::new()
    932
                .with_data_type(DataType::Int32)
    933
                .with_nullable(true);
    934
    935
            let not_nullable_schema = MockExprSchema::new()
    936
                .with_data_type(DataType::Int32)
    937
                .with_nullable(false);
    938
    939
            // CASE WHEN x IS NOT NULL THEN x ELSE 0
    940
            let e = when(col("x").is_not_null(), col("x")).otherwise(lit(0))?;
    941
            assert_not_nullable(&e, &nullable_schema);
    942
            assert_not_nullable(&e, &not_nullable_schema);
    943
    944
            // CASE WHEN NOT x IS NULL THEN x ELSE 0
    945
            let e = when(not(col("x").is_null()), col("x")).otherwise(lit(0))?;
    946
            assert_not_nullable(&e, &nullable_schema);
    947
            assert_not_nullable(&e, &not_nullable_schema);
    948
    949
            // CASE WHEN X = 5 THEN x ELSE 0
    950
            let e = when(col("x").eq(lit(5)), col("x")).otherwise(lit(0))?;
    951
            assert_not_nullable(&e, &nullable_schema);
    952
            assert_not_nullable(&e, &not_nullable_schema);
    953
    954
            // CASE WHEN x IS NOT NULL AND x = 5 THEN x ELSE 0
    955
            let e = when(and(col("x").is_not_null(), col("x").eq(lit(5))), col("x"))
    956
                .otherwise(lit(0))?;
    957
            assert_not_nullable(&e, &nullable_schema);
    958
            assert_not_nullable(&e, &not_nullable_schema);
    959
    960
            // CASE WHEN x = 5 AND x IS NOT NULL THEN x ELSE 0
    961
            let e = when(and(col("x").eq(lit(5)), col("x").is_not_null()), col("x"))
    962
                .otherwise(lit(0))?;
    963
            assert_not_nullable(&e, &nullable_schema);
    964
            assert_not_nullable(&e, &not_nullable_schema);
    965
    966
            // CASE WHEN x IS NOT NULL OR x = 5 THEN x ELSE 0
    967
            let e = when(or(col("x").is_not_null(), col("x").eq(lit(5))), col("x"))
    968
                .otherwise(lit(0))?;
    969
            assert_not_nullable(&e, &nullable_schema);
    970
            assert_not_nullable(&e, &not_nullable_schema);
    971
    972
            // CASE WHEN x = 5 OR x IS NOT NULL THEN x ELSE 0
    973
            let e = when(or(col("x").eq(lit(5)), col("x").is_not_null()), col("x"))
    974
                .otherwise(lit(0))?;
    975
            assert_not_nullable(&e, &nullable_schema);
    976
            assert_not_nullable(&e, &not_nullable_schema);
    977
    978
            // CASE WHEN (x = 5 AND x IS NOT NULL) OR (x = bar AND x IS NOT NULL) THEN x ELSE 0
    979
            let e = when(
    980
                or(
    981
                    and(col("x").eq(lit(5)), col("x").is_not_null()),
    982
                    and(col("x").eq(col("bar")), col("x").is_not_null()),
    983
                ),
    984
                col("x"),
    985
            )
    986
            .otherwise(lit(0))?;
    987
            assert_not_nullable(&e, &nullable_schema);
    988
            assert_not_nullable(&e, &not_nullable_schema);
    989
    990
            // CASE WHEN x = 5 OR x IS NULL THEN x ELSE 0
    991
            let e = when(or(col("x").eq(lit(5)), col("x").is_null()), col("x"))
    992
                .otherwise(lit(0))?;
    993
            assert_nullable(&e, &nullable_schema);
    994
            assert_not_nullable(&e, &not_nullable_schema);
    995
    996
            // CASE WHEN x IS TRUE THEN x ELSE 0
    997
            let e = when(col("x").is_true(), col("x")).otherwise(lit(0))?;
    998
            assert_not_nullable(&e, &nullable_schema);
    999
            assert_not_nullable(&e, &not_nullable_schema);
    1000
    1001
            // CASE WHEN x IS NOT TRUE THEN x ELSE 0
    1002
            let e = when(col("x").is_not_true(), col("x")).otherwise(lit(0))?;
    1003
            assert_nullable(&e, &nullable_schema);
    1004
            assert_not_nullable(&e, &not_nullable_schema);
    1005
    1006
            // CASE WHEN x IS FALSE THEN x ELSE 0
    1007
            let e = when(col("x").is_false(), col("x")).otherwise(lit(0))?;
    1008
            assert_not_nullable(&e, &nullable_schema);
    1009
            assert_not_nullable(&e, &not_nullable_schema);
    1010
    1011
            // CASE WHEN x IS NOT FALSE THEN x ELSE 0
    1012
            let e = when(col("x").is_not_false(), col("x")).otherwise(lit(0))?;
    1013
            assert_nullable(&e, &nullable_schema);
    1014
            assert_not_nullable(&e, &not_nullable_schema);
    1015
    1016
            // CASE WHEN x IS UNKNOWN THEN x ELSE 0
    1017
            let e = when(col("x").is_unknown(), col("x")).otherwise(lit(0))?;
    1018
            assert_nullable(&e, &nullable_schema);
    1019
            assert_not_nullable(&e, &not_nullable_schema);
    1020
    1021
            // CASE WHEN x IS NOT UNKNOWN THEN x ELSE 0
    1022
            let e = when(col("x").is_not_unknown(), col("x")).otherwise(lit(0))?;
    1023
            assert_not_nullable(&e, &nullable_schema);
    1024
            assert_not_nullable(&e, &not_nullable_schema);
    1025
    1026
            // CASE WHEN x LIKE 'x' THEN x ELSE 0
    1027
            let e = when(col("x").like(lit("x")), col("x")).otherwise(lit(0))?;
    1028
            assert_not_nullable(&e, &nullable_schema);
    1029
            assert_not_nullable(&e, &not_nullable_schema);
    1030
    1031
            // CASE WHEN 0 THEN x ELSE 0
    1032
            let e = when(lit(0), col("x")).otherwise(lit(0))?;
    1033
            assert_not_nullable(&e, &nullable_schema);
    1034
            assert_not_nullable(&e, &not_nullable_schema);
    1035
    1036
            // CASE WHEN 1 THEN x ELSE 0
    1037
            let e = when(lit(1), col("x")).otherwise(lit(0))?;
    1038
            assert_nullable(&e, &nullable_schema);
    1039
            assert_not_nullable(&e, &not_nullable_schema);
    1040
    1041
            Ok(())
    1042
        }
    1043
    1044
        #[test]
    1045
        fn test_inlist_nullability() {
    1046
            let get_schema = |nullable| {
    1047
                MockExprSchema::new()
    1048
                    .with_data_type(DataType::Int32)
    1049
                    .with_nullable(nullable)
    1050
            };
    1051
    1052
            let expr = col("foo").in_list(vec![lit(1); 5], false);
    1053
            assert!(!expr.nullable(&get_schema(false)).unwrap());
    1054
            assert!(expr.nullable(&get_schema(true)).unwrap());
    1055
            // Testing nullable() returns an error.
    1056
            assert!(expr
    1057
                .nullable(&get_schema(false).with_error_on_nullable(true))
    1058
                .is_err());
    1059
    1060
            let null = lit(ScalarValue::Int32(None));
    1061
            let expr = col("foo").in_list(vec![null, lit(1)], false);
    1062
            assert!(expr.nullable(&get_schema(false)).unwrap());
    1063
    1064
            // Testing on long list
    1065
            let expr = col("foo").in_list(vec![lit(1); 6], false);
    1066
            assert!(expr.nullable(&get_schema(false)).unwrap());
    1067
        }
    1068
    1069
        #[test]
    1070
        fn test_like_nullability() {
    1071
            let get_schema = |nullable| {
    1072
                MockExprSchema::new()
    1073
                    .with_data_type(DataType::Utf8)
    1074
                    .with_nullable(nullable)
    1075
            };
    1076
    1077
            let expr = col("foo").like(lit("bar"));
    1078
            assert!(!expr.nullable(&get_schema(false)).unwrap());
    1079
            assert!(expr.nullable(&get_schema(true)).unwrap());
    1080
    1081
            let expr = col("foo").like(lit(ScalarValue::Utf8(None)));
    1082
            assert!(expr.nullable(&get_schema(false)).unwrap());
    1083
        }
    1084
    1085
        #[test]
    1086
        fn expr_schema_data_type() {
    1087
            let expr = col("foo");
    1088
            assert_eq!(
    1089
                DataType::Utf8,
    1090
                expr.get_type(&MockExprSchema::new().with_data_type(DataType::Utf8))
    1091
                    .unwrap()
    1092
            );
    1093
        }
    1094
    1095
        #[test]
    1096
        fn test_expr_metadata() {
    1097
            let mut meta = HashMap::new();
    1098
            meta.insert("bar".to_string(), "buzz".to_string());
    1099
            let meta = FieldMetadata::from(meta);
    1100
            let expr = col("foo");
    1101
            let schema = MockExprSchema::new()
    1102
                .with_data_type(DataType::Int32)
    1103
                .with_metadata(meta.clone());
    1104
    1105
            // col, alias, and cast should be metadata-preserving
    1106
            assert_eq!(meta, expr.metadata(&schema).unwrap());
    1107
            assert_eq!(meta, expr.clone().alias("bar").metadata(&schema).unwrap());
    1108
            assert_eq!(
    1109
                meta,
    1110
                expr.clone()
    1111
                    .cast_to(&DataType::Int64, &schema)
    1112
                    .unwrap()
    1113
                    .metadata(&schema)
    1114
                    .unwrap()
    1115
            );
    1116
    1117
            let schema = DFSchema::from_unqualified_fields(
    1118
                vec![meta.add_to_field(Field::new("foo", DataType::Int32, true))].into(),
    1119
                HashMap::new(),
    1120
            )
    1121
            .unwrap();
    1122
    1123
            // verify to_field method populates metadata
    1124
            assert_eq!(meta, expr.metadata(&schema).unwrap());
    1125
    1126
            // outer ref constructed by `out_ref_col_with_metadata` should be metadata-preserving
    1127
            let outer_ref = out_ref_col_with_metadata(
    1128
                DataType::Int32,
    1129
                meta.to_hashmap(),
    1130
                Column::from_name("foo"),
    1131
            );
    1132
            assert_eq!(meta, outer_ref.metadata(&schema).unwrap());
    1133
        }
    1134
    1135
        #[test]
    1136
        fn test_expr_placeholder() {
    1137
            let schema = MockExprSchema::new();
    1138
    1139
            let mut placeholder_meta = HashMap::new();
    1140
            placeholder_meta.insert("bar".to_string(), "buzz".to_string());
    1141
            let placeholder_meta = FieldMetadata::from(placeholder_meta);
    1142
    1143
            let expr = Expr::Placeholder(Placeholder::new_with_field(
    1144
                "".to_string(),
    1145
                Some(
    1146
                    Field::new("", DataType::Utf8, true)
    1147
                        .with_metadata(placeholder_meta.to_hashmap())
    1148
                        .into(),
    1149
                ),
    1150
            ));
    1151
    1152
            assert_eq!(
    1153
                expr.data_type_and_nullable(&schema).unwrap(),
    1154
                (DataType::Utf8, true)
    1155
            );
    1156
            assert_eq!(placeholder_meta, expr.metadata(&schema).unwrap());
    1157
    1158
            let expr_alias = expr.alias("a placeholder by any other name");
    1159
            assert_eq!(
    1160
                expr_alias.data_type_and_nullable(&schema).unwrap(),
    1161
                (DataType::Utf8, true)
    1162
            );
    1163
            assert_eq!(placeholder_meta, expr_alias.metadata(&schema).unwrap());
    1164
    1165
            // Non-nullable placeholder field should remain non-nullable
    1166
            let expr = Expr::Placeholder(Placeholder::new_with_field(
    1167
                "".to_string(),
    1168
                Some(Field::new("", DataType::Utf8, false).into()),
    1169
            ));
    1170
            assert_eq!(
    1171
                expr.data_type_and_nullable(&schema).unwrap(),
    1172
                (DataType::Utf8, false)
    1173
            );
    1174
            let expr_alias = expr.alias("a placeholder by any other name");
    1175
            assert_eq!(
    1176
                expr_alias.data_type_and_nullable(&schema).unwrap(),
    1177
                (DataType::Utf8, false)
    1178
            );
    1179
        }
    1180
    1181
        #[derive(Debug)]
    1182
        struct MockExprSchema {
    1183
            field: Field,
    1184
            error_on_nullable: bool,
    1185
        }
    1186
    1187
        impl MockExprSchema {
    1188
            fn new() -> Self {
    1189
                Self {
    1190
                    field: Field::new("mock_field", DataType::Null, false),
    1191
                    error_on_nullable: false,
    1192
                }
    1193
            }
    1194
    1195
            fn with_nullable(mut self, nullable: bool) -> Self {
    1196
                self.field = self.field.with_nullable(nullable);
    1197
                self
    1198
            }
    1199
    1200
            fn with_data_type(mut self, data_type: DataType) -> Self {
    1201
                self.field = self.field.with_data_type(data_type);
    1202
                self
    1203
            }
    1204
    1205
            fn with_error_on_nullable(mut self, error_on_nullable: bool) -> Self {
    1206
                self.error_on_nullable = error_on_nullable;
    1207
                self
    1208
            }
    1209
    1210
            fn with_metadata(mut self, metadata: FieldMetadata) -> Self {
    1211
                self.field = metadata.add_to_field(self.field);
    1212
                self
    1213
            }
    1214
        }
    1215
    1216
        impl ExprSchema for MockExprSchema {
    1217
            fn nullable(&self, _col: &Column) -> Result<bool> {
    1218
                assert_or_internal_err!(!self.error_on_nullable, "nullable error");
    1219
                Ok(self.field.is_nullable())
    1220
            }
    1221
    1222
            fn field_from_column(&self, _col: &Column) -> Result<&Field> {
    1223
                Ok(&self.field)
    1224
            }
    1225
        }
    1226
    }