## 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/physical-expr/src/planner.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 std::sync::Arc;
    19
    20
    use crate::ScalarFunctionExpr;
    21
    use crate::{
    22
        expressions::{self, binary, like, similar_to, Column, Literal},
    23
        PhysicalExpr,
    24
    };
    25
    26
    use arrow::datatypes::Schema;
    27
    use datafusion_common::config::ConfigOptions;
    28
    use datafusion_common::metadata::FieldMetadata;
    29
    use datafusion_common::{
    30
        exec_err, not_impl_err, plan_err, DFSchema, Result, ScalarValue, ToDFSchema,
    31
    };
    32
    use datafusion_expr::execution_props::ExecutionProps;
    33
    use datafusion_expr::expr::{Alias, Cast, InList, Placeholder, ScalarFunction};
    34
    use datafusion_expr::var_provider::is_system_variables;
    35
    use datafusion_expr::var_provider::VarType;
    36
    use datafusion_expr::{
    37
        binary_expr, lit, Between, BinaryExpr, Expr, Like, Operator, TryCast,
    38
    };
    39
    40
    /// [PhysicalExpr] evaluate DataFusion expressions such as `A + 1`, or `CAST(c1
    41
    /// AS int)`.
    42
    ///
    43
    /// [PhysicalExpr] are the physical counterpart to [Expr] used in logical
    44
    /// planning, and can be evaluated directly on a [RecordBatch]. They are
    45
    /// normally created from [Expr] by a [PhysicalPlanner] and can be created
    46
    /// directly using [create_physical_expr].
    47
    ///
    48
    /// A Physical expression knows its type, nullability and how to evaluate itself.
    49
    ///
    50
    /// [PhysicalPlanner]: https://docs.rs/datafusion/latest/datafusion/physical_planner/trait.PhysicalPlanner.html
    51
    /// [RecordBatch]: https://docs.rs/arrow/latest/arrow/record_batch/struct.RecordBatch.html
    52
    ///
    53
    /// # Example: Create `PhysicalExpr` from `Expr`
    54
    /// ```
    55
    /// # use arrow::datatypes::{DataType, Field, Schema};
    56
    /// # use datafusion_common::DFSchema;
    57
    /// # use datafusion_expr::{Expr, col, lit};
    58
    /// # use datafusion_physical_expr::create_physical_expr;
    59
    /// # use datafusion_expr::execution_props::ExecutionProps;
    60
    /// // For a logical expression `a = 1`, we can create a physical expression
    61
    /// let expr = col("a").eq(lit(1));
    62
    /// // To create a PhysicalExpr we need 1. a schema
    63
    /// let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
    64
    /// let df_schema = DFSchema::try_from(schema).unwrap();
    65
    /// // 2. ExecutionProps
    66
    /// let props = ExecutionProps::new();
    67
    /// // We can now create a PhysicalExpr:
    68
    /// let physical_expr = create_physical_expr(&expr, &df_schema, &props).unwrap();
    69
    /// ```
    70
    ///
    71
    /// # Example: Executing a PhysicalExpr to obtain [ColumnarValue]
    72
    /// ```
    73
    /// # use std::sync::Arc;
    74
    /// # use arrow::array::{cast::AsArray, BooleanArray, Int32Array, RecordBatch};
    75
    /// # use arrow::datatypes::{DataType, Field, Schema};
    76
    /// # use datafusion_common::{assert_batches_eq, DFSchema};
    77
    /// # use datafusion_expr::{Expr, col, lit, ColumnarValue};
    78
    /// # use datafusion_physical_expr::create_physical_expr;
    79
    /// # use datafusion_expr::execution_props::ExecutionProps;
    80
    /// # let expr = col("a").eq(lit(1));
    81
    /// # let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
    82
    /// # let df_schema = DFSchema::try_from(schema.clone()).unwrap();
    83
    /// # let props = ExecutionProps::new();
    84
    /// // Given a PhysicalExpr, for `a = 1` we can evaluate it against a RecordBatch like this:
    85
    /// let physical_expr = create_physical_expr(&expr, &df_schema, &props).unwrap();
    86
    /// // Input of [1,2,3]
    87
    /// let input_batch = RecordBatch::try_from_iter(vec![
    88
    ///   ("a", Arc::new(Int32Array::from(vec![1, 2, 3])) as _)
    89
    /// ]).unwrap();
    90
    /// // The result is a ColumnarValue (either an Array or a Scalar)
    91
    /// let result = physical_expr.evaluate(&input_batch).unwrap();
    92
    /// // In this case, a BooleanArray with the result of the comparison
    93
    /// let ColumnarValue::Array(arr) = result else {
    94
    ///  panic!("Expected an array")
    95
    /// };
    96
    /// assert_eq!(arr.as_boolean(), &BooleanArray::from(vec![true, false, false]));
    97
    /// ```
    98
    ///
    99
    /// [ColumnarValue]: datafusion_expr::ColumnarValue
    100
    ///
    101
    /// Create a physical expression from a logical expression ([Expr]).
    102
    ///
    103
    /// # Arguments
    104
    ///
    105
    /// * `e` - The logical expression
    106
    /// * `input_dfschema` - The DataFusion schema for the input, used to resolve `Column` references
    107
    ///   to qualified or unqualified fields by name.
    108
    0
    pub fn create_physical_expr(
    109
    0
        e: &Expr,
    110
    0
        input_dfschema: &DFSchema,
    111
    0
        execution_props: &ExecutionProps,
    112
    0
    ) -> Result<Arc<dyn PhysicalExpr>> {
    113
    0
        let input_schema = input_dfschema.as_arrow();
    114
    115
    0
        match e {
    116
    0
            Expr::Alias(Alias { expr, metadata, .. }) => {
    117
    0
                if let Expr::Literal(v, prior_metadata) = expr.as_ref() {
    118
    0
                    let new_metadata = FieldMetadata::merge_options(
    119
    0
                        prior_metadata.as_ref(),
    120
    0
                        metadata.as_ref(),
    121
                    );
    122
    0
                    Ok(Arc::new(Literal::new_with_metadata(
    123
    0
                        v.clone(),
    124
    0
                        new_metadata,
    125
    0
                    )))
    126
                } else {
    127
    0
                    Ok(create_physical_expr(expr, input_dfschema, execution_props)?)
    128
                }
    129
            }
    130
    0
            Expr::Column(c) => {
    131
    0
                let idx = input_dfschema.index_of_column(c)?;
    132
    0
                Ok(Arc::new(Column::new(&c.name, idx)))
    133
            }
    134
    0
            Expr::Literal(value, metadata) => Ok(Arc::new(Literal::new_with_metadata(
    135
    0
                value.clone(),
    136
    0
                metadata.clone(),
    137
    0
            ))),
    138
    0
            Expr::ScalarVariable(_, variable_names) => {
    139
    0
                if is_system_variables(variable_names) {
    140
    0
                    match execution_props.get_var_provider(VarType::System) {
    141
    0
                        Some(provider) => {
    142
    0
                            let scalar_value = provider.get_value(variable_names.clone())?;
    143
    0
                            Ok(Arc::new(Literal::new(scalar_value)))
    144
                        }
    145
    0
                        _ => plan_err!("No system variable provider found"),
    146
                    }
    147
                } else {
    148
    0
                    match execution_props.get_var_provider(VarType::UserDefined) {
    149
    0
                        Some(provider) => {
    150
    0
                            let scalar_value = provider.get_value(variable_names.clone())?;
    151
    0
                            Ok(Arc::new(Literal::new(scalar_value)))
    152
                        }
    153
    0
                        _ => plan_err!("No user defined variable provider found"),
    154
                    }
    155
                }
    156
            }
    157
    0
            Expr::IsTrue(expr) => {
    158
    0
                let binary_op = binary_expr(
    159
    0
                    expr.as_ref().clone(),
    160
    0
                    Operator::IsNotDistinctFrom,
    161
    0
                    lit(true),
    162
                );
    163
    0
                create_physical_expr(&binary_op, input_dfschema, execution_props)
    164
            }
    165
    0
            Expr::IsNotTrue(expr) => {
    166
    0
                let binary_op =
    167
    0
                    binary_expr(expr.as_ref().clone(), Operator::IsDistinctFrom, lit(true));
    168
    0
                create_physical_expr(&binary_op, input_dfschema, execution_props)
    169
            }
    170
    0
            Expr::IsFalse(expr) => {
    171
    0
                let binary_op = binary_expr(
    172
    0
                    expr.as_ref().clone(),
    173
    0
                    Operator::IsNotDistinctFrom,
    174
    0
                    lit(false),
    175
                );
    176
    0
                create_physical_expr(&binary_op, input_dfschema, execution_props)
    177
            }
    178
    0
            Expr::IsNotFalse(expr) => {
    179
    0
                let binary_op =
    180
    0
                    binary_expr(expr.as_ref().clone(), Operator::IsDistinctFrom, lit(false));
    181
    0
                create_physical_expr(&binary_op, input_dfschema, execution_props)
    182
            }
    183
    0
            Expr::IsUnknown(expr) => {
    184
    0
                let binary_op = binary_expr(
    185
    0
                    expr.as_ref().clone(),
    186
    0
                    Operator::IsNotDistinctFrom,
    187
    0
                    Expr::Literal(ScalarValue::Boolean(None), None),
    188
                );
    189
    0
                create_physical_expr(&binary_op, input_dfschema, execution_props)
    190
            }
    191
    0
            Expr::IsNotUnknown(expr) => {
    192
    0
                let binary_op = binary_expr(
    193
    0
                    expr.as_ref().clone(),
    194
    0
                    Operator::IsDistinctFrom,
    195
    0
                    Expr::Literal(ScalarValue::Boolean(None), None),
    196
                );
    197
    0
                create_physical_expr(&binary_op, input_dfschema, execution_props)
    198
            }
    199
    0
            Expr::BinaryExpr(BinaryExpr { left, op, right }) => {
    200
                // Create physical expressions for left and right operands
    201
    0
                let lhs = create_physical_expr(left, input_dfschema, execution_props)?;
    202
    0
                let rhs = create_physical_expr(right, input_dfschema, execution_props)?;
    203
                // Note that the logical planner is responsible
    204
                // for type coercion on the arguments (e.g. if one
    205
                // argument was originally Int32 and one was
    206
                // Int64 they will both be coerced to Int64).
    207
                //
    208
                // There should be no coercion during physical
    209
                // planning.
    210
    0
                binary(lhs, *op, rhs, input_schema)
    211
            }
    212
            Expr::Like(Like {
    213
    0
                negated,
    214
    0
                expr,
    215
    0
                pattern,
    216
    0
                escape_char,
    217
    0
                case_insensitive,
    218
            }) => {
    219
                // `\` is the implicit escape, see https://github.com/apache/datafusion/issues/13291
    220
    0
                if escape_char.unwrap_or('\\') != '\\' {
    221
    0
                    return exec_err!(
    222
    0
                        "LIKE does not support escape_char other than the backslash (\\)"
    223
                    );
    224
    0
                }
    225
    0
                let physical_expr =
    226
    0
                    create_physical_expr(expr, input_dfschema, execution_props)?;
    227
    0
                let physical_pattern =
    228
    0
                    create_physical_expr(pattern, input_dfschema, execution_props)?;
    229
    0
                like(
    230
    0
                    *negated,
    231
    0
                    *case_insensitive,
    232
    0
                    physical_expr,
    233
    0
                    physical_pattern,
    234
    0
                    input_schema,
    235
                )
    236
            }
    237
            Expr::SimilarTo(Like {
    238
    0
                negated,
    239
    0
                expr,
    240
    0
                pattern,
    241
    0
                escape_char,
    242
    0
                case_insensitive,
    243
            }) => {
    244
    0
                if escape_char.is_some() {
    245
    0
                    return exec_err!("SIMILAR TO does not support escape_char yet");
    246
    0
                }
    247
    0
                let physical_expr =
    248
    0
                    create_physical_expr(expr, input_dfschema, execution_props)?;
    249
    0
                let physical_pattern =
    250
    0
                    create_physical_expr(pattern, input_dfschema, execution_props)?;
    251
    0
                similar_to(*negated, *case_insensitive, physical_expr, physical_pattern)
    252
            }
    253
    0
            Expr::Case(case) => {
    254
    0
                let expr: Option<Arc<dyn PhysicalExpr>> = if let Some(e) = &case.expr {
    255
    0
                    Some(create_physical_expr(
    256
    0
                        e.as_ref(),
    257
    0
                        input_dfschema,
    258
    0
                        execution_props,
    259
    0
                    )?)
    260
                } else {
    261
    0
                    None
    262
                };
    263
    0
                let (when_expr, then_expr): (Vec<&Expr>, Vec<&Expr>) = case
    264
    0
                    .when_then_expr
    265
    0
                    .iter()
    266
    0
                    .map(|(w, t)| (w.as_ref(), t.as_ref()))
    267
    0
                    .unzip();
    268
    0
                let when_expr =
    269
    0
                    create_physical_exprs(when_expr, input_dfschema, execution_props)?;
    270
    0
                let then_expr =
    271
    0
                    create_physical_exprs(then_expr, input_dfschema, execution_props)?;
    272
    0
                let when_then_expr: Vec<(Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)> =
    273
    0
                    when_expr
    274
    0
                        .iter()
    275
    0
                        .zip(then_expr.iter())
    276
    0
                        .map(|(w, t)| (Arc::clone(w), Arc::clone(t)))
    277
    0
                        .collect();
    278
    0
                let else_expr: Option<Arc<dyn PhysicalExpr>> =
    279
    0
                    if let Some(e) = &case.else_expr {
    280
    0
                        Some(create_physical_expr(
    281
    0
                            e.as_ref(),
    282
    0
                            input_dfschema,
    283
    0
                            execution_props,
    284
    0
                        )?)
    285
                    } else {
    286
    0
                        None
    287
                    };
    288
    0
                Ok(expressions::case(expr, when_then_expr, else_expr)?)
    289
            }
    290
    0
            Expr::Cast(Cast { expr, data_type }) => expressions::cast(
    291
    0
                create_physical_expr(expr, input_dfschema, execution_props)?,
    292
    0
                input_schema,
    293
    0
                data_type.clone(),
    294
            ),
    295
    0
            Expr::TryCast(TryCast { expr, data_type }) => expressions::try_cast(
    296
    0
                create_physical_expr(expr, input_dfschema, execution_props)?,
    297
    0
                input_schema,
    298
    0
                data_type.clone(),
    299
            ),
    300
    0
            Expr::Not(expr) => {
    301
    0
                expressions::not(create_physical_expr(expr, input_dfschema, execution_props)?)
    302
            }
    303
    0
            Expr::Negative(expr) => expressions::negative(
    304
    0
                create_physical_expr(expr, input_dfschema, execution_props)?,
    305
    0
                input_schema,
    306
            ),
    307
    0
            Expr::IsNull(expr) => expressions::is_null(create_physical_expr(
    308
    0
                expr,
    309
    0
                input_dfschema,
    310
    0
                execution_props,
    311
    0
            )?),
    312
    0
            Expr::IsNotNull(expr) => expressions::is_not_null(create_physical_expr(
    313
    0
                expr,
    314
    0
                input_dfschema,
    315
    0
                execution_props,
    316
    0
            )?),
    317
    0
            Expr::ScalarFunction(ScalarFunction { func, args }) => {
    318
    0
                let physical_args =
    319
    0
                    create_physical_exprs(args, input_dfschema, execution_props)?;
    320
    0
                let config_options = match execution_props.config_options.as_ref() {
    321
    0
                    Some(config_options) => Arc::clone(config_options),
    322
    0
                    None => Arc::new(ConfigOptions::default()),
    323
                };
    324
    325
    0
                Ok(Arc::new(ScalarFunctionExpr::try_new(
    326
    0
                    Arc::clone(func),
    327
    0
                    physical_args,
    328
    0
                    input_schema,
    329
    0
                    config_options,
    330
    0
                )?))
    331
            }
    332
            Expr::Between(Between {
    333
    0
                expr,
    334
    0
                negated,
    335
    0
                low,
    336
    0
                high,
    337
            }) => {
    338
    0
                let value_expr = create_physical_expr(expr, input_dfschema, execution_props)?;
    339
    0
                let low_expr = create_physical_expr(low, input_dfschema, execution_props)?;
    340
    0
                let high_expr = create_physical_expr(high, input_dfschema, execution_props)?;
    341
    342
                // rewrite the between into the two binary operators
    343
    0
                let binary_expr = binary(
    344
    0
                    binary(
    345
    0
                        Arc::clone(&value_expr),
    346
    0
                        Operator::GtEq,
    347
    0
                        low_expr,
    348
    0
                        input_schema,
    349
    0
                    )?,
    350
    0
                    Operator::And,
    351
    0
                    binary(
    352
    0
                        Arc::clone(&value_expr),
    353
    0
                        Operator::LtEq,
    354
    0
                        high_expr,
    355
    0
                        input_schema,
    356
    0
                    )?,
    357
    0
                    input_schema,
    358
                );
    359
    360
    0
                if *negated {
    361
    0
                    expressions::not(binary_expr?)
    362
                } else {
    363
    0
                    binary_expr
    364
                }
    365
            }
    366
            Expr::InList(InList {
    367
    0
                expr,
    368
    0
                list,
    369
    0
                negated,
    370
    0
            }) => match expr.as_ref() {
    371
                Expr::Literal(ScalarValue::Utf8(None), _) => {
    372
    0
                    Ok(expressions::lit(ScalarValue::Boolean(None)))
    373
                }
    374
                _ => {
    375
    0
                    let value_expr =
    376
    0
                        create_physical_expr(expr, input_dfschema, execution_props)?;
    377
    378
    0
                    let list_exprs =
    379
    0
                        create_physical_exprs(list, input_dfschema, execution_props)?;
    380
    0
                    expressions::in_list(value_expr, list_exprs, negated, input_schema)
    381
                }
    382
            },
    383
    0
            Expr::Placeholder(Placeholder { id, .. }) => {
    384
    0
                exec_err!("Placeholder '{id}' was not provided a value for execution.")
    385
            }
    386
    0
            other => {
    387
    0
                not_impl_err!("Physical plan does not support logical expression {other:?}")
    388
            }
    389
        }
    390
    0
    }
    391
    392
    /// Create vector of Physical Expression from a vector of logical expression
    393
    0
    pub fn create_physical_exprs<'a, I>(
    394
    0
        exprs: I,
    395
    0
        input_dfschema: &DFSchema,
    396
    0
        execution_props: &ExecutionProps,
    397
    0
    ) -> Result<Vec<Arc<dyn PhysicalExpr>>>
    398
    0
    where
    399
    0
        I: IntoIterator<Item = &'a Expr>,
    400
    {
    401
    0
        exprs
    402
    0
            .into_iter()
    403
    0
            .map(|expr| create_physical_expr(expr, input_dfschema, execution_props))
    404
    0
            .collect()
    405
    0
    }
    406
    407
    /// Convert a logical expression to a physical expression (without any simplification, etc)
    408
    0
    pub fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
    409
        // TODO this makes a deep copy of the Schema. Should take SchemaRef instead and avoid deep copy
    410
    0
        let df_schema = schema.clone().to_dfschema().unwrap();
    411
    0
        let execution_props = ExecutionProps::new();
    412
    0
        create_physical_expr(expr, &df_schema, &execution_props).unwrap()
    413
    0
    }
    414
    415
    #[cfg(test)]
    416
    mod tests {
    417
        use arrow::array::{ArrayRef, BooleanArray, RecordBatch, StringArray};
    418
        use arrow::datatypes::{DataType, Field};
    419
    420
        use datafusion_expr::{col, lit};
    421
    422
        use super::*;
    423
    424
        #[test]
    425
    0
        fn test_create_physical_expr_scalar_input_output() -> Result<()> {
    426
    0
            let expr = col("letter").eq(lit("A"));
    427
    428
    0
            let schema = Schema::new(vec![Field::new("letter", DataType::Utf8, false)]);
    429
    0
            let df_schema = DFSchema::try_from_qualified_schema("data", &schema)?;
    430
    0
            let p = create_physical_expr(&expr, &df_schema, &ExecutionProps::new())?;
    431
    432
    0
            let batch = RecordBatch::try_new(
    433
    0
                Arc::new(schema),
    434
    0
                vec![Arc::new(StringArray::from_iter_values(vec![
    435
    0
                    "A", "B", "C", "D",
    436
    0
                ]))],
    437
    0
            )?;
    438
    0
            let result = p.evaluate(&batch)?;
    439
    0
            let result = result.into_array(4).expect("Failed to convert to array");
    440
    441
    0
            assert_eq!(
    442
    0
                &result,
    443
    0
                &(Arc::new(BooleanArray::from(vec![true, false, false, false,])) as ArrayRef)
    444
            );
    445
    446
    0
            Ok(())
    447
    0
        }
    448
    }