fix: match Spark ordering in native array extrema

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

@@ -91,19 +91,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, use Spark's JVM codegen dispatcher inside the Comet pipeline by default. If the dispatcher is disabled, these cases fall back to Spark unless incompatible native execution is explicitly enabled ([#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 use the same JVM codegen dispatch and dispatcher-disabled fallback as `array_max` ([#4496](https://github.com/apache/datafusion-comet/issues/4496)).
 
 ## array_position
 

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

@@ -28,6 +28,13 @@ to Spark in some cases, especially when the data contains both positive and nega
 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.
+
 ## Ordering: NaN and signed zero (`-0.0` vs `+0.0`)
 
 Spark's `ORDER BY`, `RANK`, `DENSE_RANK`, and window frame comparisons route through

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 | Native for literal or column delimiter and null replacement; other cases and non-UTF8_BINARY collations use the JVM codegen dispatcher ([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` | ✅ | Hybrid | Native Spark-compatible floating-point and nested ordering; non-default string collations use the JVM codegen dispatcher ([details](compatibility/expressions/array.md)) |
+| `array_min` | ✅ | Hybrid | Native Spark-compatible floating-point and nested ordering; non-default string collations use the JVM codegen dispatcher ([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

@@ -259,6 +259,10 @@ harness = false
 name = "arrays_overlap"
 harness = false
 
+[[bench]]
+name = "array_extrema"
+harness = false
+
 [[bench]]
 name = "checked_arithmetic"
 harness = false

native/spark-expr/benches/array_extrema.rs

@@ -0,0 +1,137 @@
+// 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.
+
+//! Compare ordinary-data extrema with DataFusion; special-value semantics belong in tests.
+
+use arrow::array::{ArrayRef, Float32Array, Float64Array, ListArray};
+use arrow::buffer::{NullBuffer, OffsetBuffer};
+use arrow::datatypes::Field;
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+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 list(values: ArrayRef, len: usize, nullable: bool) -> ArrayRef {
+    let rows = values.len() / len;
+    Arc::new(ListArray::new(
+        Arc::new(Field::new_list_field(values.data_type().clone(), true)),
+        OffsetBuffer::from_lengths(std::iter::repeat_n(len, rows)),
+        values,
+        nullable.then(|| NullBuffer::from_iter((0..rows).map(|i| i % 10 != 0))),
+    ))
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("array_extrema");
+    group.sample_size(20);
+    group.warm_up_time(Duration::from_millis(250));
+    group.measurement_time(Duration::from_secs(1));
+    for len in [8, 1024] {
+        for nullable in [false, true] {
+            let values = (0..64 * len)
+                .map(|i| (!nullable || i % 10 != 0).then_some(((i * 17) % 1000 + 1) as f64));
+            let inputs = [
+                (
+                    "float32",
+                    list(
+                        Arc::new(Float32Array::from_iter(
+                            values.clone().map(|v| v.map(|v| v as f32)),
+                        )),
+                        len,
+                        nullable,
+                    ),
+                ),
+                (
+                    "float64",
+                    list(Arc::new(Float64Array::from_iter(values)), len, nullable),
+                ),
+                (
+                    "nested",
+                    // Null list elements are skipped by both engines. Keep inner floats non-null
+                    // so the fixture does not depend on their different nested null ordering.
+                    list(
+                        list(
+                            Arc::new(Float64Array::from_iter_values(
+                                (0..64 * len * 4).map(|i| ((i * 17) % 1000 + 1) as f64),
+                            )),
+                            4,
+                            nullable,
+                        ),
+                        len,
+                        nullable,
+                    ),
+                ),
+            ];
+            for (kind, input) in inputs {
+                for is_min in [true, false] {
+                    let comet = ScalarUDF::from(SparkArrayExtrema::new(is_min));
+                    let datafusion = if is_min {
+                        array_min_udf()
+                    } else {
+                        array_max_udf()
+                    };
+                    let args = 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",
+                            comet.return_type(&[input.data_type().clone()]).unwrap(),
+                            true,
+                        )),
+                        config_options: Arc::new(ConfigOptions::default()),
+                    };
+                    let evaluate = |udf: &ScalarUDF| {
+                        udf.invoke_with_args(args.clone())
+                            .unwrap()
+                            .into_array(input.len())
+                            .unwrap()
+                    };
+                    assert_eq!(evaluate(&comet).to_data(), evaluate(&datafusion).to_data());
+                    let op = if is_min { "min" } else { "max" };
+                    for (engine, udf) in [("comet", &comet), ("datafusion", datafusion.as_ref())] {
+                        group.bench_function(
+                            BenchmarkId::new(
+                                format!("{kind}_{op}_{engine}"),
+                                format!("len={len}_null={nullable}"),
+                            ),
+                            |b| {
+                                b.iter(|| {
+                                    black_box(
+                                        udf.invoke_with_args(black_box(args.clone())).unwrap(),
+                                    )
+                                })
+                            },
+                        );
+                    }
+                }
+            }
+        }
+    }
+    group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);