datafusion_physical_expr/
planner.rs

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
18use std::sync::Arc;
19
20use crate::scalar_subquery::ScalarSubqueryExpr;
21use crate::{HigherOrderFunctionExpr, ScalarFunctionExpr};
22use crate::{
23    PhysicalExpr,
24    expressions::{self, Column, Literal, binary, like, similar_to},
25};
26
27use arrow::datatypes::Schema;
28use datafusion_common::config::ConfigOptions;
29use datafusion_common::datatype::FieldExt;
30use datafusion_common::metadata::{FieldMetadata, format_type_and_metadata};
31use datafusion_common::{
32    DFSchema, Result, ScalarValue, TableReference, ToDFSchema, exec_err,
33    internal_datafusion_err, not_impl_err, plan_datafusion_err, plan_err,
34};
35use datafusion_expr::execution_props::ExecutionProps;
36use datafusion_expr::expr::{
37    Alias, Cast, HigherOrderFunction, InList, Lambda, LambdaVariable, Placeholder,
38    ScalarFunction,
39};
40use datafusion_expr::physical_planning_context::PhysicalPlanningContext;
41use datafusion_expr::var_provider::VarType;
42use datafusion_expr::var_provider::is_system_variables;
43use datafusion_expr::{
44    Between, BinaryExpr, Expr, ExprSchemable, Like, Operator, TryCast, binary_expr, lit,
45};
46
47/// [PhysicalExpr] evaluate DataFusion expressions such as `A + 1`, or `CAST(c1
48/// AS int)`.
49///
50/// [PhysicalExpr] are the physical counterpart to [Expr] used in logical
51/// planning, and can be evaluated directly on a [RecordBatch]. They are
52/// normally created from [Expr] by a [PhysicalPlanner] and can be created
53/// directly using [create_physical_expr].
54///
55/// A Physical expression knows its type, nullability and how to evaluate itself.
56///
57/// [PhysicalPlanner]: https://docs.rs/datafusion/latest/datafusion/physical_planner/trait.PhysicalPlanner.html
58/// [RecordBatch]: https://docs.rs/arrow/latest/arrow/record_batch/struct.RecordBatch.html
59///
60/// # Example: Create `PhysicalExpr` from `Expr`
61/// ```
62/// # use arrow::datatypes::{DataType, Field, Schema};
63/// # use datafusion_common::DFSchema;
64/// # use datafusion_expr::{Expr, col, lit};
65/// # use datafusion_physical_expr::create_physical_expr;
66/// # use datafusion_expr::execution_props::ExecutionProps;
67/// # use datafusion_expr::physical_planning_context::PhysicalPlanningContext;
68/// // For a logical expression `a = 1`, we can create a physical expression
69/// let expr = col("a").eq(lit(1));
70/// // To create a PhysicalExpr we need 1. a schema
71/// let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
72/// let df_schema = DFSchema::try_from(schema).unwrap();
73/// // 2. ExecutionProps
74/// let props = ExecutionProps::new();
75/// // We can now create a PhysicalExpr. Expressions with no scalar
76/// // subqueries use an empty `PhysicalPlanningContext`:
77/// let physical_expr =
78///     create_physical_expr(&expr, &df_schema, &props, &PhysicalPlanningContext::default())
79///         .unwrap();
80/// ```
81///
82/// # Example: Executing a PhysicalExpr to obtain [ColumnarValue]
83/// ```
84/// # use std::sync::Arc;
85/// # use arrow::array::{cast::AsArray, BooleanArray, Int32Array, RecordBatch};
86/// # use arrow::datatypes::{DataType, Field, Schema};
87/// # use datafusion_common::{assert_batches_eq, DFSchema};
88/// # use datafusion_expr::{Expr, col, lit, ColumnarValue};
89/// # use datafusion_physical_expr::create_physical_expr;
90/// # use datafusion_expr::execution_props::ExecutionProps;
91/// # use datafusion_expr::physical_planning_context::PhysicalPlanningContext;
92/// # let expr = col("a").eq(lit(1));
93/// # let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
94/// # let df_schema = DFSchema::try_from(schema.clone()).unwrap();
95/// # let props = ExecutionProps::new();
96/// // Given a PhysicalExpr, for `a = 1` we can evaluate it against a RecordBatch like this:
97/// let physical_expr =
98///     create_physical_expr(&expr, &df_schema, &props, &PhysicalPlanningContext::default())
99///         .unwrap();
100/// // Input of [1,2,3]
101/// let input_batch = RecordBatch::try_from_iter(vec![
102///   ("a", Arc::new(Int32Array::from(vec![1, 2, 3])) as _)
103/// ]).unwrap();
104/// // The result is a ColumnarValue (either an Array or a Scalar)
105/// let result = physical_expr.evaluate(&input_batch).unwrap();
106/// // In this case, a BooleanArray with the result of the comparison
107/// let ColumnarValue::Array(arr) = result else {
108///  panic!("Expected an array")
109/// };
110/// assert_eq!(arr.as_boolean(), &BooleanArray::from(vec![true, false, false]));
111/// ```
112///
113/// [ColumnarValue]: datafusion_expr::ColumnarValue
114///
115/// Create a physical expression from a logical expression ([Expr]).
116///
117/// # Arguments
118///
119/// * `e` - The logical expression
120/// * `input_dfschema` - The DataFusion schema for the input, used to resolve `Column` references
121///   to qualified or unqualified fields by name.
122/// * `execution_props` - Per-execution properties such as the query start time.
123/// * `planning_ctx` - The [`PhysicalPlanningContext`] used to resolve
124///   `Expr::ScalarSubquery` and `Expr::LambdaVariable` nodes. The physical
125///   planner threads the subquery index map and shared results container from
126///   its `ScalarSubqueryExec` construction into calls to
127///   `create_physical_expr`; the lambda variable qualifiers are added by this
128///   function itself as it descends into lambda bodies. Callers creating
129///   physical expressions outside of physical planning should pass
130///   `&PhysicalPlanningContext::default()`; converting a scalar subquery then returns a
131///   planning error.
132#[cfg_attr(feature = "recursive_protection", recursive::recursive)]
133pub fn create_physical_expr(
134    e: &Expr,
135    input_dfschema: &DFSchema,
136    execution_props: &ExecutionProps,
137    planning_ctx: &PhysicalPlanningContext,
138) -> Result<Arc<dyn PhysicalExpr>> {
139    let input_schema = input_dfschema.as_arrow();
140
141    match e {
142        Expr::Alias(Alias { expr, metadata, .. }) => {
143            if let Expr::Literal(v, prior_metadata) = expr.as_ref() {
144                let new_metadata = FieldMetadata::merge_options(
145                    prior_metadata.as_ref(),
146                    metadata.as_ref(),
147                );
148                Ok(Arc::new(Literal::new_with_metadata(
149                    v.clone(),
150                    new_metadata,
151                )))
152            } else {
153                Ok(create_physical_expr(
154                    expr,
155                    input_dfschema,
156                    execution_props,
157                    planning_ctx,
158                )?)
159            }
160        }
161        Expr::Column(c) => {
162            let idx = input_dfschema.index_of_column(c)?;
163            Ok(Arc::new(Column::new(&c.name, idx)))
164        }
165        Expr::Literal(value, metadata) => Ok(Arc::new(Literal::new_with_metadata(
166            value.clone(),
167            metadata.clone(),
168        ))),
169        Expr::ScalarVariable(_, variable_names) => {
170            if is_system_variables(variable_names) {
171                match execution_props.get_var_provider(VarType::System) {
172                    Some(provider) => {
173                        let scalar_value = provider.get_value(variable_names.clone())?;
174                        Ok(Arc::new(Literal::new(scalar_value)))
175                    }
176                    _ => plan_err!("No system variable provider found"),
177                }
178            } else {
179                match execution_props.get_var_provider(VarType::UserDefined) {
180                    Some(provider) => {
181                        let scalar_value = provider.get_value(variable_names.clone())?;
182                        Ok(Arc::new(Literal::new(scalar_value)))
183                    }
184                    _ => plan_err!("No user defined variable provider found"),
185                }
186            }
187        }
188        Expr::IsTrue(expr) => {
189            let binary_op = binary_expr(
190                expr.as_ref().clone(),
191                Operator::IsNotDistinctFrom,
192                lit(true),
193            );
194            create_physical_expr(
195                &binary_op,
196                input_dfschema,
197                execution_props,
198                planning_ctx,
199            )
200        }
201        Expr::IsNotTrue(expr) => {
202            let binary_op =
203                binary_expr(expr.as_ref().clone(), Operator::IsDistinctFrom, lit(true));
204            create_physical_expr(
205                &binary_op,
206                input_dfschema,
207                execution_props,
208                planning_ctx,
209            )
210        }
211        Expr::IsFalse(expr) => {
212            let binary_op = binary_expr(
213                expr.as_ref().clone(),
214                Operator::IsNotDistinctFrom,
215                lit(false),
216            );
217            create_physical_expr(
218                &binary_op,
219                input_dfschema,
220                execution_props,
221                planning_ctx,
222            )
223        }
224        Expr::IsNotFalse(expr) => {
225            let binary_op =
226                binary_expr(expr.as_ref().clone(), Operator::IsDistinctFrom, lit(false));
227            create_physical_expr(
228                &binary_op,
229                input_dfschema,
230                execution_props,
231                planning_ctx,
232            )
233        }
234        Expr::IsUnknown(expr) => {
235            let binary_op = binary_expr(
236                expr.as_ref().clone(),
237                Operator::IsNotDistinctFrom,
238                Expr::Literal(ScalarValue::Boolean(None), None),
239            );
240            create_physical_expr(
241                &binary_op,
242                input_dfschema,
243                execution_props,
244                planning_ctx,
245            )
246        }
247        Expr::IsNotUnknown(expr) => {
248            let binary_op = binary_expr(
249                expr.as_ref().clone(),
250                Operator::IsDistinctFrom,
251                Expr::Literal(ScalarValue::Boolean(None), None),
252            );
253            create_physical_expr(
254                &binary_op,
255                input_dfschema,
256                execution_props,
257                planning_ctx,
258            )
259        }
260        Expr::BinaryExpr(BinaryExpr { left, op, right }) => {
261            // Create physical expressions for left and right operands
262            let lhs = create_physical_expr(
263                left,
264                input_dfschema,
265                execution_props,
266                planning_ctx,
267            )?;
268            let rhs = create_physical_expr(
269                right,
270                input_dfschema,
271                execution_props,
272                planning_ctx,
273            )?;
274            // Note that the logical planner is responsible
275            // for type coercion on the arguments (e.g. if one
276            // argument was originally Int32 and one was
277            // Int64 they will both be coerced to Int64).
278            //
279            // There should be no coercion during physical
280            // planning.
281            binary(lhs, *op, rhs, input_schema)
282        }
283        Expr::Like(Like {
284            negated,
285            expr,
286            pattern,
287            escape_char,
288            case_insensitive,
289        }) => {
290            // `\` is the implicit escape, see https://github.com/apache/datafusion/issues/13291
291            if escape_char.unwrap_or('\\') != '\\' {
292                return exec_err!(
293                    "LIKE does not support escape_char other than the backslash (\\)"
294                );
295            }
296            let physical_expr = create_physical_expr(
297                expr,
298                input_dfschema,
299                execution_props,
300                planning_ctx,
301            )?;
302            let physical_pattern = create_physical_expr(
303                pattern,
304                input_dfschema,
305                execution_props,
306                planning_ctx,
307            )?;
308            like(
309                *negated,
310                *case_insensitive,
311                physical_expr,
312                physical_pattern,
313                input_schema,
314            )
315        }
316        Expr::SimilarTo(Like {
317            negated,
318            expr,
319            pattern,
320            escape_char,
321            case_insensitive,
322        }) => {
323            if escape_char.is_some() {
324                return exec_err!("SIMILAR TO does not support escape_char yet");
325            }
326            let physical_expr = create_physical_expr(
327                expr,
328                input_dfschema,
329                execution_props,
330                planning_ctx,
331            )?;
332            let physical_pattern = create_physical_expr(
333                pattern,
334                input_dfschema,
335                execution_props,
336                planning_ctx,
337            )?;
338            similar_to(*negated, *case_insensitive, physical_expr, physical_pattern)
339        }
340        Expr::Case(case) => {
341            let expr: Option<Arc<dyn PhysicalExpr>> = if let Some(e) = &case.expr {
342                Some(create_physical_expr(
343                    e.as_ref(),
344                    input_dfschema,
345                    execution_props,
346                    planning_ctx,
347                )?)
348            } else {
349                None
350            };
351            let (when_expr, then_expr): (Vec<&Expr>, Vec<&Expr>) = case
352                .when_then_expr
353                .iter()
354                .map(|(w, t)| (w.as_ref(), t.as_ref()))
355                .unzip();
356            let when_expr = create_physical_exprs(
357                when_expr,
358                input_dfschema,
359                execution_props,
360                planning_ctx,
361            )?;
362            let then_expr = create_physical_exprs(
363                then_expr,
364                input_dfschema,
365                execution_props,
366                planning_ctx,
367            )?;
368            let when_then_expr: Vec<(Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)> =
369                when_expr
370                    .iter()
371                    .zip(then_expr.iter())
372                    .map(|(w, t)| (Arc::clone(w), Arc::clone(t)))
373                    .collect();
374            let else_expr: Option<Arc<dyn PhysicalExpr>> =
375                if let Some(e) = &case.else_expr {
376                    Some(create_physical_expr(
377                        e.as_ref(),
378                        input_dfschema,
379                        execution_props,
380                        planning_ctx,
381                    )?)
382                } else {
383                    None
384                };
385            Ok(expressions::case(expr, when_then_expr, else_expr)?)
386        }
387        Expr::Cast(Cast { expr, field }) => expressions::cast_with_target_field(
388            create_physical_expr(expr, input_dfschema, execution_props, planning_ctx)?,
389            input_schema,
390            Arc::clone(field),
391            None,
392        ),
393        Expr::TryCast(TryCast { expr, field }) => {
394            if !field.metadata().is_empty() {
395                let (_, src_field) = expr.to_field(input_dfschema)?;
396                return plan_err!(
397                    "TryCast from {} to {} is not supported",
398                    format_type_and_metadata(
399                        src_field.data_type(),
400                        Some(src_field.metadata()),
401                    ),
402                    format_type_and_metadata(field.data_type(), Some(field.metadata()))
403                );
404            }
405
406            expressions::try_cast(
407                create_physical_expr(
408                    expr,
409                    input_dfschema,
410                    execution_props,
411                    planning_ctx,
412                )?,
413                input_schema,
414                field.data_type().clone(),
415            )
416        }
417        Expr::Not(expr) => expressions::not(create_physical_expr(
418            expr,
419            input_dfschema,
420            execution_props,
421            planning_ctx,
422        )?),
423        Expr::Negative(expr) => expressions::negative(
424            create_physical_expr(expr, input_dfschema, execution_props, planning_ctx)?,
425            input_schema,
426        ),
427        Expr::IsNull(expr) => expressions::is_null(create_physical_expr(
428            expr,
429            input_dfschema,
430            execution_props,
431            planning_ctx,
432        )?),
433        Expr::IsNotNull(expr) => expressions::is_not_null(create_physical_expr(
434            expr,
435            input_dfschema,
436            execution_props,
437            planning_ctx,
438        )?),
439        Expr::ScalarFunction(ScalarFunction { func, args }) => {
440            let physical_args = create_physical_exprs(
441                args,
442                input_dfschema,
443                execution_props,
444                planning_ctx,
445            )?;
446            let config_options = match execution_props.config_options.as_ref() {
447                Some(config_options) => Arc::clone(config_options),
448                None => Arc::new(ConfigOptions::default()),
449            };
450
451            Ok(Arc::new(ScalarFunctionExpr::try_new(
452                Arc::clone(func),
453                physical_args,
454                input_schema,
455                config_options,
456            )?))
457        }
458        Expr::Between(Between {
459            expr,
460            negated,
461            low,
462            high,
463        }) => {
464            let value_expr = create_physical_expr(
465                expr,
466                input_dfschema,
467                execution_props,
468                planning_ctx,
469            )?;
470            let low_expr =
471                create_physical_expr(low, input_dfschema, execution_props, planning_ctx)?;
472            let high_expr = create_physical_expr(
473                high,
474                input_dfschema,
475                execution_props,
476                planning_ctx,
477            )?;
478
479            // rewrite the between into the two binary operators
480            let binary_expr = binary(
481                binary(
482                    Arc::clone(&value_expr),
483                    Operator::GtEq,
484                    low_expr,
485                    input_schema,
486                )?,
487                Operator::And,
488                binary(
489                    Arc::clone(&value_expr),
490                    Operator::LtEq,
491                    high_expr,
492                    input_schema,
493                )?,
494                input_schema,
495            );
496
497            if *negated {
498                expressions::not(binary_expr?)
499            } else {
500                binary_expr
501            }
502        }
503        Expr::InList(InList {
504            expr,
505            list,
506            negated,
507        }) => match expr.as_ref() {
508            Expr::Literal(ScalarValue::Utf8(None), _) => {
509                Ok(expressions::lit(ScalarValue::Boolean(None)))
510            }
511            _ => {
512                let value_expr = create_physical_expr(
513                    expr,
514                    input_dfschema,
515                    execution_props,
516                    planning_ctx,
517                )?;
518
519                let list_exprs = create_physical_exprs(
520                    list,
521                    input_dfschema,
522                    execution_props,
523                    planning_ctx,
524                )?;
525                expressions::in_list(value_expr, list_exprs, negated, input_schema)
526            }
527        },
528        Expr::ScalarSubquery(sq) => {
529            match planning_ctx.index_of(sq) {
530                Some(index) => {
531                    let schema = sq.subquery.schema();
532                    if schema.fields().len() != 1 {
533                        return plan_err!(
534                            "Scalar subquery must return exactly one column, got {}",
535                            schema.fields().len()
536                        );
537                    }
538                    let dt = schema.field(0).data_type().clone();
539                    let nullable = schema.field(0).is_nullable();
540                    Ok(Arc::new(ScalarSubqueryExpr::new(
541                        dt,
542                        nullable,
543                        index,
544                        planning_ctx.results().clone(),
545                    )))
546                }
547                None => {
548                    // Not found: either a correlated subquery that wasn't
549                    // rewritten to a join, or an uncorrelated one that wasn't
550                    // registered by the physical planner.
551                    not_impl_err!(
552                        "Physical plan does not support logical expression {e:?}"
553                    )
554                }
555            }
556        }
557        Expr::Placeholder(Placeholder { id, .. }) => {
558            exec_err!("Placeholder '{id}' was not provided a value for execution.")
559        }
560        Expr::HigherOrderFunction(invocation @ HigherOrderFunction { func, args }) => {
561            let num_lambdas = args
562                .iter()
563                .filter(|arg| matches!(arg, Expr::Lambda(_)))
564                .count();
565
566            let mut lambda_parameters =
567                invocation.lambda_parameters(input_dfschema)?.into_iter();
568
569            if num_lambdas > lambda_parameters.len() {
570                return plan_err!(
571                    "{} lambda_parameters returned only {} values for {num_lambdas} lambdas",
572                    func.name(),
573                    lambda_parameters.len()
574                );
575            }
576
577            let lambda_qualifier = 1 + input_dfschema
578                .iter()
579                .filter_map(|(qualifier, _field)| {
580                    qualifier.and_then(|tbl| {
581                        tbl.table().strip_prefix("lambda_")?.parse::<usize>().ok()
582                    })
583                })
584                .max()
585                .unwrap_or_default();
586
587            let qualifier = TableReference::bare(format!("lambda_{lambda_qualifier}"));
588
589            let physical_args = args
590                .iter()
591                .map(|arg| match arg {
592                    Expr::Lambda(lambda) => {
593                        let lambda_parameters = lambda_parameters
594                            .next()
595                            .ok_or_else(|| {
596                                internal_datafusion_err!(
597                                    "lambda_parameters len should have been checked above"
598                                )
599                            })?
600                            .into_iter()
601                            .zip(&lambda.params)
602                            .map(|(field, name)| {
603                                (Some(qualifier.clone()), field.renamed(name.as_str()))
604                            });
605
606                        let new_fields = input_dfschema
607                            .iter()
608                            .map(|(tbl, field)| (tbl.cloned(), Arc::clone(field)))
609                            .chain(lambda_parameters)
610                            .collect();
611
612                        let lambda_schema = DFSchema::new_with_metadata(
613                            new_fields,
614                            input_dfschema.metadata().clone(),
615                        )?;
616
617                        let planning_ctx = planning_ctx
618                            .clone()
619                            .with_qualified_lambda_variables(&qualifier, &lambda.params);
620
621                        create_physical_expr(
622                            arg,
623                            &lambda_schema,
624                            execution_props,
625                            &planning_ctx,
626                        )
627                    }
628                    _ => create_physical_expr(
629                        arg,
630                        input_dfschema,
631                        execution_props,
632                        planning_ctx,
633                    ),
634                })
635                .collect::<Result<_>>()?;
636
637            let config_options = match execution_props.config_options.as_ref() {
638                Some(config_options) => Arc::clone(config_options),
639                None => Arc::new(ConfigOptions::default()),
640            };
641
642            Ok(Arc::new(HigherOrderFunctionExpr::try_new_with_schema(
643                Arc::clone(func),
644                physical_args,
645                input_schema,
646                config_options,
647            )?))
648        }
649        Expr::Lambda(Lambda { params, body }) => expressions::lambda(
650            params,
651            create_physical_expr(body, input_dfschema, execution_props, planning_ctx)?,
652        ),
653        Expr::LambdaVariable(LambdaVariable {
654            name,
655            field,
656            spans: _,
657        }) => {
658            let field = field.as_ref().ok_or_else(|| {
659                plan_datafusion_err!("unresolved LambdaVariable {name}")
660            })?;
661
662            let qualifier =
663                planning_ctx
664                    .lambda_variable_qualifier(name)
665                    .ok_or_else(|| {
666                        plan_datafusion_err!(
667                            "qualifier for lambda variable {name} not found"
668                        )
669                    })?;
670
671            let index = input_dfschema
672                .index_of_column_by_name(Some(qualifier), name)
673                .ok_or_else(|| {
674                    plan_datafusion_err!(
675                        "lambda variable {qualifier}.{name} not found in planning schema"
676                    )
677                })?;
678
679            let schema_field = input_dfschema.field(index);
680
681            // LambdaVariable.field will be made optional as in Expr::Placeholder
682            // and only LambdaVariable.name used, and field.name ignored,
683            // so they're not enforced to match for logical expressions
684            // Rename the field to match the schema one and use it's PartialEq impl instead
685            // of checking property by property and fail if new properties get's added to it.
686            // While not necessary, the sql planner does create lambda vars with matching names,
687            // so this shouldn't allocate with a lambda var from it
688            let renamed_field = Arc::clone(field).renamed(name);
689
690            if &renamed_field != schema_field {
691                return plan_err!(
692                    "LambdaVariable field and schema field mismatch {} != {}",
693                    renamed_field,
694                    schema_field
695                );
696            }
697
698            Ok(Arc::new(expressions::LambdaVariable::new(
699                index,
700                Arc::clone(schema_field),
701            )))
702        }
703        other => {
704            not_impl_err!("Physical plan does not support logical expression {other:?}")
705        }
706    }
707}
708
709/// Create vector of Physical Expression from a vector of logical expression
710///
711/// See [`create_physical_expr`] for details on the `planning_ctx` argument.
712pub fn create_physical_exprs<'a, I>(
713    exprs: I,
714    input_dfschema: &DFSchema,
715    execution_props: &ExecutionProps,
716    planning_ctx: &PhysicalPlanningContext,
717) -> Result<Vec<Arc<dyn PhysicalExpr>>>
718where
719    I: IntoIterator<Item = &'a Expr>,
720{
721    exprs
722        .into_iter()
723        .map(|expr| {
724            create_physical_expr(expr, input_dfschema, execution_props, planning_ctx)
725        })
726        .collect()
727}
728
729/// Convert a logical expression to a physical expression (without any simplification, etc)
730pub fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
731    // TODO this makes a deep copy of the Schema. Should take SchemaRef instead and avoid deep copy
732    let df_schema = schema.clone().to_dfschema().unwrap();
733    let execution_props = ExecutionProps::new();
734    create_physical_expr(
735        expr,
736        &df_schema,
737        &execution_props,
738        &PhysicalPlanningContext::default(),
739    )
740    .unwrap()
741}
742
743#[cfg(test)]
744mod tests {
745    use arrow::array::{ArrayRef, BooleanArray, RecordBatch, StringArray};
746    use arrow::datatypes::{DataType, Field};
747    use datafusion_expr::col;
748
749    use super::*;
750
751    fn test_cast_schema() -> Schema {
752        Schema::new(vec![Field::new("a", DataType::Int32, false)])
753    }
754
755    fn lower_cast_expr(expr: &Expr, schema: &Schema) -> Result<Arc<dyn PhysicalExpr>> {
756        let df_schema = DFSchema::try_from(schema.clone())?;
757        create_physical_expr(
758            expr,
759            &df_schema,
760            &ExecutionProps::new(),
761            &PhysicalPlanningContext::default(),
762        )
763    }
764
765    fn as_planner_cast(physical: &Arc<dyn PhysicalExpr>) -> &expressions::CastExpr {
766        physical
767            .downcast_ref::<expressions::CastExpr>()
768            .expect("planner should lower logical CAST to CastExpr")
769    }
770
771    #[test]
772    fn test_create_physical_expr_scalar_input_output() -> Result<()> {
773        let expr = col("letter").eq(lit("A"));
774
775        let schema = Schema::new(vec![Field::new("letter", DataType::Utf8, false)]);
776        let df_schema = DFSchema::try_from_qualified_schema("data", &schema)?;
777        let p = create_physical_expr(
778            &expr,
779            &df_schema,
780            &ExecutionProps::new(),
781            &PhysicalPlanningContext::default(),
782        )?;
783
784        let batch = RecordBatch::try_new(
785            Arc::new(schema),
786            vec![Arc::new(StringArray::from_iter_values(vec![
787                "A", "B", "C", "D",
788            ]))],
789        )?;
790        let result = p.evaluate(&batch)?;
791        let result = result.into_array(4).expect("Failed to convert to array");
792
793        assert_eq!(
794            &result,
795            &(Arc::new(BooleanArray::from(vec![true, false, false, false,])) as ArrayRef)
796        );
797
798        Ok(())
799    }
800
801    #[test]
802    fn test_cast_lowering_preserves_target_field_metadata() -> Result<()> {
803        let schema = test_cast_schema();
804        let target_field = Arc::new(
805            Field::new("cast_target", DataType::Int64, true)
806                .with_metadata([("target_meta".to_string(), "1".to_string())].into()),
807        );
808        let cast_expr = Expr::Cast(Cast::new_from_field(
809            Box::new(col("a")),
810            Arc::clone(&target_field),
811        ));
812
813        let physical = lower_cast_expr(&cast_expr, &schema)?;
814        let cast = as_planner_cast(&physical);
815
816        assert_eq!(cast.target_field(), &target_field);
817        assert_eq!(physical.return_field(&schema)?, target_field);
818        assert!(physical.nullable(&schema)?);
819
820        Ok(())
821    }
822
823    #[test]
824    fn test_cast_lowering_preserves_standard_cast_semantics() -> Result<()> {
825        let schema = test_cast_schema();
826        let cast_expr = Expr::Cast(Cast::new(Box::new(col("a")), DataType::Int64));
827
828        let physical = lower_cast_expr(&cast_expr, &schema)?;
829        let cast = as_planner_cast(&physical);
830        let returned_field = physical.return_field(&schema)?;
831
832        assert_eq!(cast.cast_type(), &DataType::Int64);
833        assert_eq!(returned_field.name(), "a");
834        assert_eq!(returned_field.data_type(), &DataType::Int64);
835        assert!(!physical.nullable(&schema)?);
836
837        Ok(())
838    }
839
840    #[test]
841    fn test_cast_lowering_preserves_same_type_field_semantics() -> Result<()> {
842        let schema = test_cast_schema();
843        let target_field = Arc::new(
844            Field::new("same_type_cast", DataType::Int32, true).with_metadata(
845                [("target_meta".to_string(), "same-type".to_string())].into(),
846            ),
847        );
848        let cast_expr = Expr::Cast(Cast::new_from_field(
849            Box::new(col("a")),
850            Arc::clone(&target_field),
851        ));
852
853        let physical = lower_cast_expr(&cast_expr, &schema)?;
854        let cast = as_planner_cast(&physical);
855
856        assert_eq!(cast.target_field(), &target_field);
857        assert_eq!(physical.return_field(&schema)?, target_field);
858        assert!(physical.nullable(&schema)?);
859
860        Ok(())
861    }
862
863    /// Test that deeply nested expressions do not cause a stack overflow.
864    ///
865    /// This test only runs when the `recursive_protection` feature is enabled,
866    /// as it would overflow the stack otherwise.
867    #[test]
868    #[cfg_attr(not(feature = "recursive_protection"), ignore)]
869    fn test_deeply_nested_binary_expr() -> Result<()> {
870        // Create a deeply nested binary expression tree: ((((a + a) + a) + a) + ... )
871        // With 1000 levels of nesting, this would overflow the stack without recursion protection.
872        let depth = 1000;
873
874        let mut expr = col("a");
875        for _ in 0..depth {
876            expr = Expr::BinaryExpr(BinaryExpr {
877                left: Box::new(expr),
878                op: Operator::Plus,
879                right: Box::new(col("a")),
880            });
881        }
882
883        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
884        let df_schema = DFSchema::try_from(schema)?;
885
886        // This should not stack overflow
887        let _physical_expr = create_physical_expr(
888            &expr,
889            &df_schema,
890            &ExecutionProps::new(),
891            &PhysicalPlanningContext::default(),
892        )?;
893
894        Ok(())
895    }
896}
datafusion_physical_expr/planner.rs

datafusion_physical_expr/
planner.rs
