fix(spark): ANSI-aware custom nullability for make_interval

datafusion/catalog/src/information_schema.rs

@@ -436,6 +436,7 @@ fn get_udf_args_and_return_types(
                     .return_field_from_args(ReturnFieldArgs {
                         arg_fields: &arg_fields,
                         scalar_arguments: &scalar_arguments,
+                        config_options: &ConfigOptions::default(),
                     })
                     .map(|f| {
                         remove_native_type_prefix(&NativeType::from(

datafusion/expr/src/expr_schema.rs

@@ -31,6 +31,7 @@ use crate::{LogicalPlan, Projection, Subquery, WindowFunctionDefinition, utils};
 use arrow::compute::can_cast_types;
 use arrow::datatypes::FieldRef;
 use arrow::datatypes::{DataType, Field};
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::datatype::FieldExt;
 use datafusion_common::{
     Column, DataFusionError, ExprSchema, Result, ScalarValue, Spans, TableReference,
@@ -590,6 +591,7 @@ impl ExprSchemable for Expr {
                 let args = ReturnFieldArgs {
                     arg_fields: &new_fields,
                     scalar_arguments: &arguments,
+                    config_options: &ConfigOptions::default(),
                 };
 
                 func.return_field_from_args(args)

datafusion/expr/src/udf.rs

@@ -451,6 +451,17 @@ pub struct ReturnFieldArgs<'a> {
     /// For example, if a function is called like `my_function(column_a, 5)`
     /// this field will be `[None, Some(ScalarValue::Int32(Some(5)))]`
     pub scalar_arguments: &'a [Option<&'a ScalarValue>],
+    /// Session configuration options at planning time.
+    ///
+    /// UDFs whose return type or nullability depends on session settings (e.g.
+    /// `spark.sql.ansi.enabled`) should read from this field rather than
+    /// storing config as a struct field and relying on [`ScalarUDFImpl::with_updated_config`].
+    ///
+    /// When called from logical planning (e.g. [`ExprSchema::to_field`]) this
+    /// may reflect a default / unset configuration rather than the actual
+    /// session config.  The physical-planning call site always provides the
+    /// real session config.
+    pub config_options: &'a ConfigOptions,
 }
 
 /// Trait for implementing user defined scalar functions.

datafusion/ffi/src/udf/return_type_args.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use arrow_schema::FieldRef;
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::scalar::ScalarValue;
 use datafusion_common::{DataFusionError, ffi_datafusion_err};
 use datafusion_expr::ReturnFieldArgs;
@@ -74,11 +75,14 @@ impl TryFrom<ReturnFieldArgs<'_>> for FFI_ReturnFieldArgs {
 pub struct ForeignReturnFieldArgsOwned {
     arg_fields: Vec<FieldRef>,
     scalar_arguments: Vec<Option<ScalarValue>>,
+    // ConfigOptions cannot cross the FFI boundary; always holds the default.
+    config_options: ConfigOptions,
 }
 
 pub struct ForeignReturnFieldArgs<'a> {
     arg_fields: &'a [FieldRef],
     scalar_arguments: Vec<Option<&'a ScalarValue>>,
+    config_options: &'a ConfigOptions,
 }
 
 impl TryFrom<&FFI_ReturnFieldArgs> for ForeignReturnFieldArgsOwned {
@@ -105,6 +109,7 @@ impl TryFrom<&FFI_ReturnFieldArgs> for ForeignReturnFieldArgsOwned {
         Ok(Self {
             arg_fields,
             scalar_arguments,
+            config_options: ConfigOptions::default(),
         })
     }
 }
@@ -118,6 +123,7 @@ impl<'a> From<&'a ForeignReturnFieldArgsOwned> for ForeignReturnFieldArgs<'a> {
                 .iter()
                 .map(|opt| opt.as_ref())
                 .collect(),
+            config_options: &value.config_options,
         }
     }
 }
@@ -127,6 +133,7 @@ impl<'a> From<&'a ForeignReturnFieldArgs<'a>> for ReturnFieldArgs<'a> {
         ReturnFieldArgs {
             arg_fields: value.arg_fields,
             scalar_arguments: &value.scalar_arguments,
+            config_options: value.config_options,
         }
     }
 }

datafusion/functions-nested/src/map_values.rs

@@ -149,6 +149,7 @@ mod tests {
     use crate::map_values::MapValuesFunc;
     use arrow::datatypes::{DataType, Field, FieldRef};
     use datafusion_common::ScalarValue;
+    use datafusion_common::config::ConfigOptions;
     use datafusion_expr::ScalarUDFImpl;
     use std::sync::Arc;
 
@@ -196,6 +197,7 @@ mod tests {
             let args = datafusion_expr::ReturnFieldArgs {
                 arg_fields: &[field],
                 scalar_arguments: &[None::<&ScalarValue>],
+                config_options: &ConfigOptions::default(),
             };
 
             func.return_field_from_args(args).unwrap()

datafusion/functions-nested/src/remove.rs

@@ -476,6 +476,7 @@ mod tests {
     };
     use arrow::datatypes::{DataType, Field, Int32Type};
     use datafusion_common::ScalarValue;
+    use datafusion_common::config::ConfigOptions;
     use datafusion_expr::{ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl};
     use datafusion_expr_common::columnar_value::ColumnarValue;
     use std::ops::Deref;
@@ -500,6 +501,7 @@ mod tests {
                     .return_field_from_args(ReturnFieldArgs {
                         arg_fields: &args_fields,
                         scalar_arguments: &scalar_args,
+                        config_options: &ConfigOptions::default(),
                     })
                     .unwrap();
 
@@ -532,6 +534,7 @@ mod tests {
                     .return_field_from_args(ReturnFieldArgs {
                         arg_fields: &args_fields,
                         scalar_arguments: &scalar_args,
+                        config_options: &ConfigOptions::default(),
                     })
                     .unwrap();
 
@@ -553,6 +556,7 @@ mod tests {
                     .return_field_from_args(ReturnFieldArgs {
                         arg_fields: &[Arc::clone(&input_field)],
                         scalar_arguments: &[None],
+                        config_options: &ConfigOptions::default(),
                     })
                     .unwrap();
 
@@ -656,6 +660,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &args_fields,
                 scalar_arguments: &scalar_args,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
 
@@ -763,6 +768,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &args_fields,
                 scalar_arguments: &scalar_args,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
 
@@ -867,6 +873,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &args_fields,
                 scalar_arguments: &scalar_args,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
 

datafusion/functions/benches/date_trunc.rs

@@ -60,6 +60,7 @@ fn criterion_benchmark(c: &mut Criterion) {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &arg_fields,
                 scalar_arguments: &scalar_arguments,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
         let config_options = Arc::new(ConfigOptions::default());

datafusion/functions/src/core/with_metadata.rs

@@ -159,6 +159,7 @@ mod tests {
     use super::*;
     use arrow::datatypes::Field;
     use datafusion_common::ScalarValue;
+    use datafusion_common::config::ConfigOptions;
     use std::sync::Arc;
 
     fn field(name: &str, dt: DataType, nullable: bool) -> FieldRef {
@@ -185,6 +186,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &fields,
                 scalar_arguments: &scalars,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
         assert_eq!(ret.name(), "my_col");
@@ -218,6 +220,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &fields,
                 scalar_arguments: &scalars,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
         assert_eq!(ret.name(), "x");
@@ -246,6 +249,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &fields,
                 scalar_arguments: &scalars,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
         assert_eq!(ret.metadata().get("a").map(String::as_str), Some("1"));
@@ -271,6 +275,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &fields,
                 scalar_arguments: &scalars,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap_err();
         assert!(err.to_string().contains("odd number"));
@@ -287,6 +292,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &fields,
                 scalar_arguments: &scalars,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap_err();
         assert!(err.to_string().contains("at least one"));
@@ -308,6 +314,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &fields,
                 scalar_arguments: &scalars,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
         assert_eq!(ret.metadata().get("unit").map(String::as_str), Some(""));
@@ -328,6 +335,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &fields,
                 scalar_arguments: &scalars,
+                config_options: &ConfigOptions::default(),
             })
             .unwrap_err();
         assert!(err.to_string().contains("non-empty constant string"));

datafusion/functions/src/datetime/now.rs

@@ -174,13 +174,15 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &empty_fields,
                 scalar_arguments: &empty_scalars,
+                config_options: &ConfigOptions::default(),
             })
             .expect("legacy now() return field");
 
         let configured_field = configured_now
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &empty_fields,
                 scalar_arguments: &empty_scalars,
+                config_options: &ConfigOptions::default(),
             })
             .expect("configured now() return field");
 

datafusion/functions/src/unicode/strpos.rs

@@ -336,6 +336,7 @@ mod tests {
     use arrow::datatypes::DataType::{Int32, Int64};
 
     use arrow::datatypes::{DataType, Field};
+    use datafusion_common::config::ConfigOptions;
     use datafusion_common::{Result, ScalarValue};
     use datafusion_expr::{ColumnarValue, ScalarUDFImpl};
 
@@ -434,6 +435,7 @@ mod tests {
                     Field::new("f2", DataType::Utf8, substring_nullable).into(),
                 ],
                 scalar_arguments: &[None::<&ScalarValue>, None::<&ScalarValue>],
+                config_options: &ConfigOptions::default(),
             };
 
             strpos.return_field_from_args(args).unwrap().is_nullable()

datafusion/functions/src/utils.rs

@@ -291,6 +291,7 @@ pub mod test {
         let return_field = func.return_field_from_args(datafusion_expr::ReturnFieldArgs {
             arg_fields: &field_array,
             scalar_arguments: &scalar_arguments_refs,
+        config_options: &ConfigOptions::default(),
         });
             let arg_fields = $ARGS.iter()
             .enumerate()

datafusion/physical-expr/src/scalar_function.rs

@@ -109,6 +109,7 @@ impl ScalarFunctionExpr {
         let ret_args = ReturnFieldArgs {
             arg_fields: &arg_fields,
             scalar_arguments: &arguments,
+            config_options: &config_options,
         };
         let return_field = fun.return_field_from_args(ret_args)?;
         Ok(Self {

datafusion/spark/src/function/array/shuffle.rs

@@ -270,6 +270,7 @@ fn fixed_size_array_shuffle(
 mod tests {
     use super::*;
     use arrow::datatypes::Field;
+    use datafusion_common::config::ConfigOptions;
     use datafusion_expr::ReturnFieldArgs;
 
     #[test]
@@ -287,6 +288,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &[Arc::clone(&non_nullable_field)],
                 scalar_arguments: &[None],
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
 
@@ -305,6 +307,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &[Arc::clone(&nullable_field)],
                 scalar_arguments: &[None],
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
 

datafusion/spark/src/function/array/slice.rs

@@ -186,6 +186,7 @@ mod tests {
     use arrow::datatypes::Field;
     use datafusion_common::ScalarValue;
     use datafusion_common::cast::as_list_array;
+    use datafusion_common::config::ConfigOptions;
     use datafusion_expr::ReturnFieldArgs;
 
     #[test]
@@ -200,6 +201,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &arg_fields,
                 scalar_arguments: &[],
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
         assert_eq!(

datafusion/spark/src/function/bitmap/bitmap_count.rs

@@ -250,6 +250,7 @@ mod tests {
         let result = bitmap_count.return_field_from_args(ReturnFieldArgs {
             arg_fields: &[Arc::clone(&non_nullable_field)],
             scalar_arguments: &[None],
+            config_options: &ConfigOptions::default(),
         })?;
 
         // The result should not be nullable (same as input)
@@ -262,6 +263,7 @@ mod tests {
         let result = bitmap_count.return_field_from_args(ReturnFieldArgs {
             arg_fields: &[Arc::clone(&nullable_field)],
             scalar_arguments: &[None],
+            config_options: &ConfigOptions::default(),
         })?;
 
         // The result should be nullable (same as input)

datafusion/spark/src/function/bitwise/bit_count.rs

@@ -172,6 +172,7 @@ mod tests {
         UInt16Array, UInt32Array, UInt64Array,
     };
     use arrow::datatypes::Field;
+    use datafusion_common::config::ConfigOptions;
 
     #[test]
     fn test_bit_count_basic() {
@@ -357,6 +358,7 @@ mod tests {
         let result = bit_count.return_field_from_args(ReturnFieldArgs {
             arg_fields: &[Arc::clone(&non_nullable_field)],
             scalar_arguments: &[None],
+            config_options: &ConfigOptions::default(),
         })?;
 
         // The result should not be nullable (same as input)
@@ -369,6 +371,7 @@ mod tests {
         let result = bit_count.return_field_from_args(ReturnFieldArgs {
             arg_fields: &[Arc::clone(&nullable_field)],
             scalar_arguments: &[None],
+            config_options: &ConfigOptions::default(),
         })?;
 
         // The result should be nullable (same as input)

datafusion/spark/src/function/bitwise/bit_get.rs

@@ -123,6 +123,7 @@ fn spark_bit_get(args: &[ArrayRef]) -> Result<ArrayRef> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use datafusion_common::config::ConfigOptions;
 
     #[test]
     fn test_bit_get_nullability_non_nullable_inputs() {
@@ -134,6 +135,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &[value_field, pos_field],
                 scalar_arguments: &[None, None],
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();
 
@@ -151,6 +153,7 @@ mod tests {
             .return_field_from_args(ReturnFieldArgs {
                 arg_fields: &[value_field, pos_field],
                 scalar_arguments: &[None, None],
+                config_options: &ConfigOptions::default(),
             })
             .unwrap();