fix: spark sha1 nullability reporting

datafusion/spark/src/function/hash/sha1.rs

@@ -20,7 +20,7 @@ use std::fmt::Write;
 use std::sync::Arc;
 
 use arrow::array::{ArrayRef, StringArray};
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, Field, FieldRef};
 use datafusion_common::cast::{
     as_binary_array, as_binary_view_array, as_fixed_size_binary_array,
     as_large_binary_array,
@@ -29,8 +29,8 @@ use datafusion_common::types::{logical_string, NativeType};
 use datafusion_common::utils::take_function_args;
 use datafusion_common::{internal_err, Result};
 use datafusion_expr::{
-    Coercion, ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature,
-    TypeSignatureClass, Volatility,
+    Coercion, ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl,
+    Signature, TypeSignatureClass, Volatility,
 };
 use datafusion_functions::utils::make_scalar_function;
 use sha1::{Digest, Sha1};
@@ -82,7 +82,17 @@ impl ScalarUDFImpl for SparkSha1 {
     }
 
     fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
-        Ok(DataType::Utf8)
+        internal_err!("return_field_from_args should be used instead")
+    }
+
+    fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+        let nullable = args.arg_fields.iter().any(|f| f.is_nullable())
+            || args
+                .scalar_arguments
+                .iter()
+                .any(|scalar| scalar.is_some_and(|s| s.is_null()));
+
+        Ok(Arc::new(Field::new(self.name(), DataType::Utf8, nullable)))
     }
 
     fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
@@ -133,3 +143,43 @@ fn spark_sha1(args: &[ArrayRef]) -> Result<ArrayRef> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use datafusion_common::ScalarValue;
+
+    #[test]
+    fn test_sha1_nullability() -> Result<()> {
+        let func = SparkSha1::new();
+
+        // Non-nullable input keeps output non-nullable
+        let non_nullable: FieldRef = Arc::new(Field::new("col", DataType::Binary, false));
+        let out = func.return_field_from_args(ReturnFieldArgs {
+            arg_fields: &[Arc::clone(&non_nullable)],
+            scalar_arguments: &[None],
+        })?;
+        assert!(!out.is_nullable());
+        assert_eq!(out.data_type(), &DataType::Utf8);
+
+        // Nullable input makes output nullable
+        let nullable: FieldRef = Arc::new(Field::new("col", DataType::Binary, true));
+        let out = func.return_field_from_args(ReturnFieldArgs {
+            arg_fields: &[Arc::clone(&nullable)],
+            scalar_arguments: &[None],
+        })?;
+        assert!(out.is_nullable());
+        assert_eq!(out.data_type(), &DataType::Utf8);
+
+        // Null scalar argument also makes output nullable
+        let null_scalar = ScalarValue::Binary(None);
+        let out = func.return_field_from_args(ReturnFieldArgs {
+            arg_fields: &[Arc::clone(&non_nullable)],
+            scalar_arguments: &[Some(&null_scalar)],
+        })?;
+        assert!(out.is_nullable());
+        assert_eq!(out.data_type(), &DataType::Utf8);
+
+        Ok(())
+    }
+}