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

datafusion_expr/
expr_schema.rs
