fix(concat): correct nullability inference (nullable only if all arguments nullable) (#19189)

Fixes #19171

### Problem

The `concat` scalar function currently reports its result as **always
nullable** at
planning/schema time. However, its **runtime semantics** are:

- `concat` ignores NULL inputs.
- The result becomes NULL **only if all input arguments are NULL**.

This mismatch causes incorrect nullability in inferred schemas and can
affect
optimizer behavior.

### What this PR does

Implements `return_field_from_args` for `ConcatFunc` so that:

1. The return `DataType` is derived using the existing `return_type`
logic.
2. The return field’s nullability is computed as:


(If there are no argument fields, the return is considered nullable
defensively.)

This aligns schema-time nullability with runtime behavior and matches
the
semantics used by other SQL engines.

### Tests

- All existing unit tests for `concat` pass.
- Parquet & CSV tests verified locally after initializing test
submodules.
- No behavior changes to runtime concatenation: only planner-side
metadata improved.

### Notes

If CI finds any compatibility adjustments required across DataFusion
crates,
I will update this PR accordingly.

Author: ujjwaltwri

datafusion/spark/src/function/string/concat.rs

@@ -17,8 +17,10 @@
 
 use arrow::array::Array;
 use arrow::buffer::NullBuffer;
-use arrow::datatypes::DataType;
+use arrow::datatypes::{DataType, Field};
+use datafusion_common::arrow::datatypes::FieldRef;
 use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::ReturnFieldArgs;
 use datafusion_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature,
     Volatility,
@@ -71,10 +73,6 @@ impl ScalarUDFImpl for SparkConcat {
         &self.signature
     }
 
-    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
-        Ok(DataType::Utf8)
-    }
-
     fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
         spark_concat(args)
     }
@@ -83,6 +81,17 @@ impl ScalarUDFImpl for SparkConcat {
         // Accept any string types, including zero arguments
         Ok(arg_types.to_vec())
     }
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        datafusion_common::internal_err!(
+            "return_type should not be called for Spark concat"
+        )
+    }
+    fn return_field_from_args(&self, args: ReturnFieldArgs<'_>) -> Result<FieldRef> {
+        // Spark semantics: concat returns NULL if ANY input is NULL
+        let nullable = args.arg_fields.iter().any(|f| f.is_nullable());
+
+        Ok(Arc::new(Field::new("concat", DataType::Utf8, nullable)))
+    }
 }
 
 /// Represents the null state for Spark concat
@@ -231,8 +240,10 @@ mod tests {
     use super::*;
     use crate::function::utils::test::test_scalar_function;
     use arrow::array::StringArray;
-    use arrow::datatypes::DataType;
+    use arrow::datatypes::{DataType, Field};
     use datafusion_common::Result;
+    use datafusion_expr::ReturnFieldArgs;
+    use std::sync::Arc;
 
     #[test]
     fn test_concat_basic() -> Result<()> {
@@ -266,4 +277,50 @@ mod tests {
         );
         Ok(())
     }
+    #[test]
+    fn test_spark_concat_return_field_non_nullable() -> Result<()> {
+        let func = SparkConcat::new();
+
+        let fields = vec![
+            Arc::new(Field::new("a", DataType::Utf8, false)),
+            Arc::new(Field::new("b", DataType::Utf8, false)),
+        ];
+
+        let args = ReturnFieldArgs {
+            arg_fields: &fields,
+            scalar_arguments: &[],
+        };
+
+        let field = func.return_field_from_args(args)?;
+
+        assert!(
+            !field.is_nullable(),
+            "Expected concat result to be non-nullable when all inputs are non-nullable"
+        );
+
+        Ok(())
+    }
+    #[test]
+    fn test_spark_concat_return_field_nullable() -> Result<()> {
+        let func = SparkConcat::new();
+
+        let fields = vec![
+            Arc::new(Field::new("a", DataType::Utf8, false)),
+            Arc::new(Field::new("b", DataType::Utf8, true)),
+        ];
+
+        let args = ReturnFieldArgs {
+            arg_fields: &fields,
+            scalar_arguments: &[],
+        };
+
+        let field = func.return_field_from_args(args)?;
+
+        assert!(
+            field.is_nullable(),
+            "Expected concat result to be nullable when any input is nullable"
+        );
+
+        Ok(())
+    }
 }