fix: make array null argument handling follow SQL semantics (#22508)

## Which issue does this PR close?

  - Closes #22507.

  ## Rationale for this change

Some array functions were not handling null index or count arguments
correctly.

A null `size` or `max` value was sometimes treated like `0`, which gave
wrong results like `[]` or the original array. `array_element` also
depended on Arrow buffer values in null slots.

  This change makes these functions follow normal SQL null rules.

  ## What changes are included in this PR?

  - Make `array_resize` return `NULL` when `size` is `NULL`
  - Make `array_replace_n` return `NULL` when `max` is `NULL`
  - Make `array_remove_n` return `NULL` when `max` is `NULL`
  - Make `array_element` check for null indexes explicitly
- Clean up `array_repeat` so null counts stay explicit in offset
building
- Update `array_remove_n` field nullability so planner metadata matches
runtime behavior
  - Add regression tests for these cases
  - Update SQL logic test outputs for the changed null behavior

  ## Are these changes tested?

  Yes.

I added regression tests for the changed Rust paths and updated the SQL
logic tests.


  ## Are there any user-facing changes?

These functions now return `NULL` for null index or count arguments
instead of returning `[]`, an unchanged array, or relying on accidental
behavior.

Author: kumarUjjawal

datafusion/functions-nested/src/array_any_match.rs

@@ -171,12 +171,12 @@ impl HigherOrderUDF for ArrayAnyMatch {
         &self,
         args: HigherOrderReturnFieldArgs,
     ) -> Result<Arc<Field>> {
-        let [ValueOrLambda::Value(list), _] =
+        let [ValueOrLambda::Value(list), ValueOrLambda::Lambda(lambda)] =
             take_function_args(self.name(), args.arg_fields)?
         else {
             return plan_err!("{} expects a value as first argument", self.name());
         };
-        let nullable = list.is_nullable();
+        let nullable = list.is_nullable() || lambda.is_nullable();
         Ok(Arc::new(Field::new("", DataType::Boolean, nullable)))
     }
 
@@ -272,14 +272,14 @@ mod tests {
     };
     use datafusion_common::{DFSchema, Result};
     use datafusion_expr::{
-        Expr, col,
+        Expr, HigherOrderReturnFieldArgs, HigherOrderUDF, ValueOrLambda, col,
         execution_props::ExecutionProps,
         expr::{HigherOrderFunction, LambdaVariable},
         lambda, lit,
     };
     use datafusion_physical_expr::create_physical_expr;
 
-    use crate::array_any_match::array_any_match_higher_order_function;
+    use crate::array_any_match::{ArrayAnyMatch, array_any_match_higher_order_function};
 
     fn run_any_match(
         list: impl arrow::array::Array + Clone + 'static,
@@ -413,6 +413,44 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_any_match_return_field_nullability() -> Result<()> {
+        for list_nullable in [true, false] {
+            for lambda_nullable in [true, false] {
+                let list = Arc::new(Field::new(
+                    "list",
+                    DataType::new_list(DataType::Int32, true),
+                    list_nullable,
+                ));
+                let lambda =
+                    Arc::new(Field::new("predicate", DataType::Boolean, lambda_nullable));
+                let arg_fields = [
+                    ValueOrLambda::Value(Arc::clone(&list)),
+                    ValueOrLambda::Lambda(Arc::clone(&lambda)),
+                ];
+                let scalar_arguments = [None, None];
+
+                let result = ArrayAnyMatch::new().return_field_from_args(
+                    HigherOrderReturnFieldArgs {
+                        arg_fields: &arg_fields,
+                        scalar_arguments: &scalar_arguments,
+                    },
+                )?;
+
+                assert_eq!(
+                    result,
+                    Arc::new(Field::new(
+                        "",
+                        DataType::Boolean,
+                        list_nullable || lambda_nullable,
+                    ))
+                );
+            }
+        }
+
+        Ok(())
+    }
+
     // Predicate must not be evaluated on elements belonging to null rows.
     // The 10 in the null row would satisfy x > 5, but the row result must be None.
     #[test]

datafusion/functions-nested/src/extract.rs

@@ -256,8 +256,8 @@ where
         let end = offset_window[1];
         let len = end - start;
 
-        // array is null
-        if array.is_null(row_index) {
+        // array or index is null
+        if array.is_null(row_index) || indexes.is_null(row_index) {
             mutable.extend_nulls(1);
             continue;
         }
@@ -1107,7 +1107,7 @@ mod tests {
     };
     use arrow::array::{ListArray, RecordBatch};
     use arrow::buffer::{NullBuffer, OffsetBuffer, ScalarBuffer};
-    use arrow::datatypes::{DataType, Field};
+    use arrow::datatypes::{DataType, Field, Int32Type};
     use datafusion_common::{Column, DFSchema, Result, assert_batches_eq};
     use datafusion_expr::expr::ScalarFunction;
     use datafusion_expr::{Expr, ExprSchemable};
@@ -1198,6 +1198,26 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_array_element_null_index_with_non_zero_buffer_returns_null() -> Result<()> {
+        let list_array = ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
+            Some(vec![Some(1), Some(2), Some(3)]),
+            Some(vec![Some(4)]),
+            Some(vec![Some(5)]),
+        ]);
+        let indexes = Int64Array::new(
+            ScalarBuffer::from(vec![1, 1, 1]),
+            Some(NullBuffer::from(vec![true, false, true])),
+        );
+
+        let result = general_array_element(&list_array, &indexes)?;
+        let expected = Int32Array::from(vec![Some(1), None, Some(5)]);
+
+        assert_eq!(result.as_primitive::<Int32Type>(), &expected);
+
+        Ok(())
+    }
+
     #[test]
     fn test_array_any_null_handling() -> Result<()> {
         let values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]));