fix: match Spark ordering in native array extrema

docs/source/contributor-guide/expression-audits/array_funcs.md

@@ -90,19 +90,19 @@
 
 ## array_max
 
-- Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `ArrayMax(child) extends UnaryExpression with ImplicitCastInputTypes`; skips NULL elements; for float/double Spark's `SQLOrderingUtil` treats NaN as greater than any non-NaN. Wired as `CometScalarFunction("array_max")`.
-- Spark 4.0.1 (audited 2026-05-27): `NullIntolerant` -> `nullIntolerant` field refactor.
-- Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
-- Float/double arrays containing NaN match Spark: NaN is treated as greater than any non-NaN value.
+- Spark 3.4.3 (audited 2026-08-22): identical to 3.5.8.
+- Spark 3.5.8 (audited 2026-08-22): `ArrayMax` skips NULL elements and returns NULL for an empty or all-NULL array. `SQLOrderingUtil` treats all NaNs as equal and greater than non-NaN values, and signed zeros as equal. The first equal maximum is retained. Nested arrays and structs compare lexicographically, with NULL fields or elements ordered first.
+- Spark 4.0.1 (audited 2026-08-22): `NullIntolerant` becomes a `nullIntolerant` field. Extrema semantics are unchanged; string ordering can use non-default collations.
+- Spark 4.1.1 (audited 2026-08-22): identical to 4.0.1.
+- Current status: `CometArrayMax` uses the native `SparkArrayExtrema` UDF. Typed float/double scans and recursive array/struct comparisons follow Spark's ordering and preserve the original first equal element, including its zero sign and NaN representation. This path is used in both strict and non-strict floating-point modes without the JVM codegen dispatcher. Other scalar element types retain the existing DataFusion implementation. Non-UTF8_BINARY string collations, including nested fields, are flagged `Incompatible` ([#4496](https://github.com/apache/datafusion-comet/issues/4496)).
 
 ## array_min
 
-- Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): mirror of `ArrayMax` with `evalInternal` returning the minimum. Same NULL-skip and NaN-ordering semantics. Wired as `CometScalarFunction("array_min")`.
-- Spark 4.0.1 (audited 2026-05-27): same trait refactor as `array_max`.
-- Spark 4.1.1 (audited 2026-05-27): identical to 4.0.1.
-- Float/double arrays containing NaN match Spark, mirroring `array_max`.
+- Spark 3.4.3 (audited 2026-08-22): identical to 3.5.8.
+- Spark 3.5.8 (audited 2026-08-22): mirrors `ArrayMax`, retaining the first equal minimum. The NULL, NaN, signed-zero, and nested comparison rules are the same.
+- Spark 4.0.1 (audited 2026-08-22): same trait refactor and collation support as `array_max`, with no change in floating-point extrema semantics.
+- Spark 4.1.1 (audited 2026-08-22): identical to 4.0.1.
+- Current status: `CometArrayMin` shares the native `SparkArrayExtrema` implementation and support boundary with `array_max`. Both floating-point modes use Spark-compatible native ordering, preserving the original first equal minimum. Non-default string collations remain `Incompatible` ([#4496](https://github.com/apache/datafusion-comet/issues/4496)).
 
 ## array_position
 

docs/source/user-guide/latest/compatibility/floating-point.md

@@ -27,3 +27,10 @@ So Comet adds additional normalization expression of NaN and zero for comparison
 to Spark in some cases, especially when the data contains both positive and negative zero. This is likely an edge
 case that is not of concern for many users. If it is a concern, setting `spark.comet.exec.strictFloatingPoint=true`
 will make relevant operations fall back to Spark.
+
+`array_min` and `array_max` use Spark-compatible native comparisons in both strict and non-strict
+floating-point modes. Signed zeros compare equal, and all NaN representations compare equal and
+greater than non-NaN values. The original first equal element is retained: for example,
+`array_min(array(0.0D, -0.0D))` returns `0.0`, while reversing those elements returns `-0.0`.
+The same ordering applies recursively to floating-point fields in arrays and structs. These
+expressions do not require Spark's codegen dispatcher for floating-point compatibility.

docs/source/user-guide/latest/expressions.md

@@ -156,8 +156,8 @@ The tables below list every Spark built-in expression with its current status.
 | `array_insert` | ✅ | Native |  |
 | `array_intersect` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) |
 | `array_join` | ✅ | Hybrid | Routes through the JVM codegen dispatcher by default; the incompatible native path is opt-in via allowIncompatible ([details](compatibility/expressions/array.md)) |
-| `array_max` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) |
-| `array_min` | ✅ | Native | NaN ordering may differ ([details](compatibility/floating-point.md)) |
+| `array_max` | ✅ | Native | Spark-compatible floating-point and nested ordering; non-default string collations fall back ([details](compatibility/expressions/array.md)) |
+| `array_min` | ✅ | Native | Spark-compatible floating-point and nested ordering; non-default string collations fall back ([details](compatibility/expressions/array.md)) |
 | `array_position` | ✅ | Native | Binary/struct/map/null elements fall back |
 | `array_prepend` | ✅ | — |  |
 | `array_remove` | ✅ | Native |  |

native/spark-expr/Cargo.toml

@@ -168,6 +168,10 @@ harness = false
 name = "arrays_overlap"
 harness = false
 
+[[bench]]
+name = "array_extrema"
+harness = false
+
 [[bench]]
 name = "checked_arithmetic"
 harness = false
@@ -222,4 +226,4 @@ harness = false
 
 [[bench]]
 name = "cast_int_to_decimal"
-harness = false
+harness = false

native/spark-expr/benches/array_extrema.rs

@@ -0,0 +1,187 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Compares the Spark-compatible UDF with the DataFusion version pinned by Cargo.lock.
+//! All timed inputs contain ordinary finite, nonzero values. Nested elements contain
+//! no inner nulls: mixed zero signs, NaN payloads, and nested null ordering deliberately
+//! differ from DataFusion and belong in the correctness tests, not parity benchmarks.
+//! The null percentage controls both outer-row and immediate-child validity.
+//!
+//! Run the whole bounded matrix, or filter (for example) by float64 or nested_float64:
+//! cargo bench -p datafusion-comet-spark-expr --bench array_extrema -- float64
+
+use arrow::array::{ArrayRef, Float32Array, Float64Array, Int32Array, ListArray};
+use arrow::buffer::{NullBuffer, OffsetBuffer};
+use arrow::datatypes::Field;
+use criterion::{
+    criterion_group, criterion_main, measurement::WallTime, BenchmarkGroup, BenchmarkId, Criterion,
+    Throughput,
+};
+use datafusion::common::config::ConfigOptions;
+use datafusion::functions_nested::min_max::{array_max_udf, array_min_udf};
+use datafusion::logical_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDF};
+use datafusion_comet_spark_expr::SparkArrayExtrema;
+use std::hint::black_box;
+use std::sync::Arc;
+use std::time::Duration;
+
+fn valid(index: usize, null_percent: usize) -> bool {
+    (index * 17 + 23) % 100 >= null_percent
+}
+
+fn list(values: ArrayRef, rows: usize, len: usize, null_percent: usize) -> ArrayRef {
+    let offsets: Vec<i32> = (0..=rows).map(|row| (row * len) as i32).collect();
+    let nulls = (null_percent != 0).then(|| {
+        NullBuffer::from(
+            (0..rows)
+                .map(|row| valid(row, null_percent))
+                .collect::<Vec<_>>(),
+        )
+    });
+    Arc::new(ListArray::new(
+        Arc::new(Field::new_list_field(values.data_type().clone(), true)),
+        OffsetBuffer::new(offsets.into()),
+        values,
+        nulls,
+    ))
+}
+
+fn finite_value(index: usize) -> i32 {
+    ((index * 104_729 + 51) % 1_000_003 + 1) as i32
+}
+
+fn primitive_input(kind: &str, rows: usize, len: usize, null_percent: usize) -> ArrayRef {
+    let values = (0..rows * len).map(|i| valid(i, null_percent).then(|| finite_value(i)));
+    let values: ArrayRef = match kind {
+        "float32" => Arc::new(Float32Array::from_iter(
+            values.map(|v| v.map(|v| v as f32 / 8.0)),
+        )),
+        "float64" => Arc::new(Float64Array::from_iter(
+            values.map(|v| v.map(|v| f64::from(v) / 8.0)),
+        )),
+        "int32_control" => Arc::new(Int32Array::from_iter(values)),
+        _ => unreachable!(),
+    };
+    list(values, rows, len, null_percent)
+}
+
+fn nested_input(rows: usize, len: usize, null_percent: usize) -> ArrayRef {
+    let children = rows * len;
+    let values: ArrayRef = Arc::new(Float64Array::from_iter_values(
+        (0..children * 4).map(|i| f64::from(finite_value(i)) / 8.0),
+    ));
+    // Null immediate children are skipped by both implementations; there are no
+    // null float values inside a valid child list, so recursive ordering agrees.
+    list(
+        list(values, children, 4, null_percent),
+        rows,
+        len,
+        null_percent,
+    )
+}
+
+fn args(input: &ArrayRef, udf: &ScalarUDF) -> ScalarFunctionArgs {
+    ScalarFunctionArgs {
+        args: vec![ColumnarValue::Array(Arc::clone(input))],
+        arg_fields: vec![Arc::new(Field::new(
+            "input",
+            input.data_type().clone(),
+            true,
+        ))],
+        number_rows: input.len(),
+        return_field: Arc::new(Field::new(
+            "result",
+            udf.return_type(&[input.data_type().clone()]).unwrap(),
+            true,
+        )),
+        config_options: Arc::new(ConfigOptions::default()),
+    }
+}
+
+fn bench_case(
+    group: &mut BenchmarkGroup<'_, WallTime>,
+    input: ArrayRef,
+    len: usize,
+    null_percent: usize,
+) {
+    let case = format!("rows={}_len={len}_null={null_percent}pct", input.len());
+    group.throughput(Throughput::Elements((input.len() * len) as u64));
+    for is_min in [true, false] {
+        let operation = if is_min { "min" } else { "max" };
+        let comet = ScalarUDF::from(SparkArrayExtrema::new(is_min));
+        let datafusion = if is_min {
+            array_min_udf()
+        } else {
+            array_max_udf()
+        };
+        let args = args(&input, &comet);
+        // Validate every fixture before timing. This is a parity/control check for
+        // ordinary data only, not an oracle for Spark's special-value semantics.
+        let comet_result = comet
+            .invoke_with_args(args.clone())
+            .unwrap()
+            .into_array(input.len())
+            .unwrap();
+        let datafusion_result = datafusion
+            .invoke_with_args(args.clone())
+            .unwrap()
+            .into_array(input.len())
+            .unwrap();
+        assert_eq!(
+            comet_result.to_data(),
+            datafusion_result.to_data(),
+            "{operation}/{case}"
+        );
+        for (name, udf) in [("comet", &comet), ("datafusion", datafusion.as_ref())] {
+            group.bench_function(
+                BenchmarkId::new(format!("{operation}_{name}"), &case),
+                |b| b.iter(|| black_box(udf.invoke_with_args(black_box(args.clone())).unwrap())),
+            );
+        }
+    }
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    // 104 cases with one-second measurements and a short warmup: roughly two
+    // minutes on an idle machine. CLI filters can select individual dimensions.
+    for kind in ["float32", "float64", "int32_control", "nested_float64"] {
+        let mut group = c.benchmark_group(format!("array_extrema/{kind}"));
+        group.sample_size(20);
+        group.warm_up_time(Duration::from_millis(250));
+        group.measurement_time(Duration::from_secs(1));
+        let (lengths, null_percentages): (&[usize], &[usize]) = match kind {
+            "float32" | "float64" => (&[8, 32, 1024], &[0, 10, 50]),
+            "int32_control" => (&[8, 1024], &[0, 50]),
+            _ => (&[8, 64], &[0, 50]),
+        };
+        for &len in lengths {
+            let rows = (65_536 / len).min(4096);
+            for &null_percent in null_percentages {
+                let input = if kind == "nested_float64" {
+                    nested_input(rows, len, null_percent)
+                } else {
+                    primitive_input(kind, rows, len, null_percent)
+                };
+                bench_case(&mut group, input, len, null_percent);
+            }
+        }
+        group.finish();
+    }
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);