fix : correct nullability propagation for spark.bitwise_not (#19224)

## Which issue does this PR close?

- Closes #19150.

## Rationale for this change

The Spark `bitwise_not` UDF always appeared as **nullable** in logical
plans, even when its input column was **non-nullable**.

This happened because the UDF implemented only `return_type()`, which
returns a `DataType` but **does not propagate nullability**.
DataFusion requires UDFs to implement `return_field_from_args()` when
nullability depends on input fields.

As a result:

- `bitwise_not(non_nullable_col)` incorrectly produced a **nullable**
output.
- Downstream query planning and schema inference became inconsistent.
- This differed from both **Spark semantics** and **Arrow kernel
behavior**, where nullability is preserved.

This PR corrects the nullability inference.

## What changes are included in this PR?

- Implemented `return_field_from_args()` for the Spark `bitwise_not`
UDF.
  - Output type = input type  
  - Output nullability = input nullability  
- Updated `return_type()` to return an error, per DataFusion API
guidelines when overriding nullability.
- Added unit tests verifying:
  - Non-nullable input → non-nullable output  
  - Nullable input → nullable output  
  - Behavior across multiple integer types (`Int32`, `Int64`)
- Code comments and minor cleanup.

## Are these changes tested?

Yes.

This PR includes new unit tests that validate:

- correct nullability propagation
- correct output types  
- consistent behavior across supported integer types

## Are there any user-facing changes?

Yes, but they are **behavior-correcting**, not breaking:

- The `spark.bitwise_not` UDF now correctly reports nullability in
schemas and logical plans.
- No API changes.
- No behavioral change for actual runtime values — Arrow kernels already
preserved null bitmaps; only planner metadata was incorrect.

This is not considered a breaking change.

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: shifluxxc

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

@@ -15,12 +15,14 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use arrow::array::*;
 use arrow::compute::kernels::bitwise;
-use arrow::datatypes::{Int16Type, Int32Type, Int64Type, Int8Type};
-use arrow::{array::*, datatypes::DataType};
-use datafusion_common::{plan_err, Result};
+use arrow::datatypes::{
+    DataType, Field, FieldRef, Int16Type, Int32Type, Int64Type, Int8Type,
+};
+use datafusion_common::{internal_err, plan_err, Result};
 use datafusion_expr::{ColumnarValue, TypeSignature, Volatility};
-use datafusion_expr::{ScalarFunctionArgs, ScalarUDFImpl, Signature};
+use datafusion_expr::{ReturnFieldArgs, ScalarFunctionArgs, ScalarUDFImpl, Signature};
 use datafusion_functions::utils::make_scalar_function;
 use std::{any::Any, sync::Arc};
 
@@ -64,8 +66,32 @@ impl ScalarUDFImpl for SparkBitwiseNot {
         &self.signature
     }
 
-    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
-        Ok(arg_types[0].clone())
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        internal_err!(
+            "SparkBitwiseNot: return_type() is not used; return_field_from_args() is implemented"
+        )
+    }
+
+    fn return_field_from_args(&self, args: ReturnFieldArgs) -> Result<FieldRef> {
+        if args.arg_fields.len() != 1 {
+            return plan_err!("bitwise_not expects exactly 1 argument");
+        }
+
+        let input_field = &args.arg_fields[0];
+
+        let out_dt = input_field.data_type().clone();
+        let mut out_nullable = input_field.is_nullable();
+
+        let scalar_null_present = args
+            .scalar_arguments
+            .iter()
+            .any(|opt_s| opt_s.is_some_and(|sv| sv.is_null()));
+
+        if scalar_null_present {
+            out_nullable = true;
+        }
+
+        Ok(Arc::new(Field::new(self.name(), out_dt, out_nullable)))
     }
 
     fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
@@ -107,3 +133,95 @@ pub fn spark_bitwise_not(args: &[ArrayRef]) -> Result<ArrayRef> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow::datatypes::{DataType, Field};
+    use std::sync::Arc;
+
+    use datafusion_expr::ReturnFieldArgs;
+
+    #[test]
+    fn test_bitwise_not_nullability() {
+        let bitwise_not = SparkBitwiseNot::new();
+
+        // --- non-nullable Int32 input ---
+        let non_nullable_i32 = Arc::new(Field::new("c", DataType::Int32, false));
+        let out_non_null = bitwise_not
+            .return_field_from_args(ReturnFieldArgs {
+                arg_fields: &[Arc::clone(&non_nullable_i32)],
+                // single-argument function -> one scalar_argument slot (None)
+                scalar_arguments: &[None],
+            })
+            .unwrap();
+
+        // result should be non-nullable and the same DataType as input
+        assert!(!out_non_null.is_nullable());
+        assert_eq!(out_non_null.data_type(), &DataType::Int32);
+
+        // --- nullable Int32 input ---
+        let nullable_i32 = Arc::new(Field::new("c", DataType::Int32, true));
+        let out_nullable = bitwise_not
+            .return_field_from_args(ReturnFieldArgs {
+                arg_fields: &[Arc::clone(&nullable_i32)],
+                scalar_arguments: &[None],
+            })
+            .unwrap();
+
+        // result should be nullable and the same DataType as input
+        assert!(out_nullable.is_nullable());
+        assert_eq!(out_nullable.data_type(), &DataType::Int32);
+
+        // --- also test another integer type (Int64) for completeness ---
+        let non_nullable_i64 = Arc::new(Field::new("c", DataType::Int64, false));
+        let out_i64 = bitwise_not
+            .return_field_from_args(ReturnFieldArgs {
+                arg_fields: &[Arc::clone(&non_nullable_i64)],
+                scalar_arguments: &[None],
+            })
+            .unwrap();
+
+        assert!(!out_i64.is_nullable());
+        assert_eq!(out_i64.data_type(), &DataType::Int64);
+
+        let nullable_i64 = Arc::new(Field::new("c", DataType::Int64, true));
+        let out_i64_null = bitwise_not
+            .return_field_from_args(ReturnFieldArgs {
+                arg_fields: &[Arc::clone(&nullable_i64)],
+                scalar_arguments: &[None],
+            })
+            .unwrap();
+
+        assert!(out_i64_null.is_nullable());
+        assert_eq!(out_i64_null.data_type(), &DataType::Int64);
+    }
+
+    #[test]
+    fn test_bitwise_not_nullability_with_null_scalar() -> Result<()> {
+        use arrow::datatypes::{DataType, Field};
+        use datafusion_common::ScalarValue;
+        use std::sync::Arc;
+
+        let func = SparkBitwiseNot::new();
+
+        let non_nullable: FieldRef = Arc::new(Field::new("col", DataType::Int32, 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::Int32);
+
+        let null_scalar = ScalarValue::Int32(None);
+        let out_with_null_scalar = func.return_field_from_args(ReturnFieldArgs {
+            arg_fields: &[Arc::clone(&non_nullable)],
+            scalar_arguments: &[Some(&null_scalar)],
+        })?;
+        assert!(out_with_null_scalar.is_nullable());
+        assert_eq!(out_with_null_scalar.data_type(), &DataType::Int32);
+
+        Ok(())
+    }
+}