feat: deprecate AggregateUDF: :is_nullable

datafusion/expr/src/expr_schema.rs

@@ -353,8 +353,9 @@ impl ExprSchemable for Expr {
                 let (_, nullable) = self.data_type_and_nullable(input_schema)?;
                 Ok(nullable)
             }
-            Expr::AggregateFunction(AggregateFunction { func, .. }) => {
-                Ok(func.is_nullable())
+            Expr::AggregateFunction(_) => {
+                let (_, nullable) = self.data_type_and_nullable(input_schema)?;
+                Ok(nullable)
             }
             Expr::WindowFunction(window_function) => self
                 .data_type_and_nullable_with_window_function(

datafusion/expr/src/udaf.rs

@@ -208,7 +208,12 @@ impl AggregateUDF {
         self.inner.window_function_display_name(params)
     }
 
+    #[deprecated(
+        since = "52.0.0",
+        note = "Use `return_field` instead with return_field.is_nullable()."
+    )]
     pub fn is_nullable(&self) -> bool {
+        #[expect(deprecated)]
         self.inner.is_nullable()
     }
 
@@ -532,6 +537,10 @@ pub trait AggregateUDFImpl: Debug + DynEq + DynHash + Send + Sync {
     /// For example, aggregate functions like `COUNT` always return a non null value
     /// but others like `MIN` will return `NULL` if there is nullable input.
     /// Note that if the function is declared as *not* nullable, make sure the [`AggregateUDFImpl::default_value`] is `non-null`
+    #[deprecated(
+        since = "52.0.0",
+        note = "Use `return_field` instead with return_field.is_nullable()."
+    )]
     fn is_nullable(&self) -> bool {
         true
     }
@@ -1100,7 +1109,10 @@ pub fn udaf_default_return_field<F: AggregateUDFImpl + ?Sized>(
     Ok(Arc::new(Field::new(
         func.name(),
         data_type,
-        func.is_nullable(),
+        #[expect(deprecated)]
+        {
+            func.is_nullable()
+        },
     )))
 }
 
@@ -1247,6 +1259,7 @@ impl AggregateUDFImpl for AliasedAggregateUDFImpl {
     }
 
     fn is_nullable(&self) -> bool {
+        #[expect(deprecated)]
         self.inner.is_nullable()
     }
 

datafusion/ffi/src/udaf/mod.rs

@@ -344,6 +344,7 @@ impl From<Arc<AggregateUDF>> for FFI_AggregateUDF {
     fn from(udaf: Arc<AggregateUDF>) -> Self {
         let name = udaf.name().into();
         let aliases = udaf.aliases().iter().map(|a| a.to_owned().into()).collect();
+        #[expect(deprecated)]
         let is_nullable = udaf.is_nullable();
         let volatility = udaf.signature().volatility.into();
 

datafusion/physical-expr/src/aggregate.rs

@@ -222,7 +222,7 @@ impl AggregateExprBuilder {
         )?;
 
         let return_field = fun.return_field(&input_exprs_fields)?;
-        let is_nullable = fun.is_nullable();
+        let is_nullable = return_field.is_nullable();
         let name = match alias {
             None => {
                 return internal_err!(