fix: normalize signed zero in nested float array comparisons

native/spark-expr/src/array_funcs/array_position.rs

@@ -33,6 +33,8 @@ use num::Float;
 use std::cmp::Ordering;
 use std::sync::Arc;
 
+use super::nested_float_normalize::normalize_negative_zero;
+
 /// Spark array_position() function that returns the 1-based position of an element in an array.
 /// Returns 0 if the element is not found (Spark behavior differs from DataFusion which returns null).
 fn spark_array_position(args: &[ColumnarValue]) -> Result<ColumnarValue, DataFusionError> {
@@ -273,7 +275,13 @@ fn position_fallback<O: OffsetSizeTrait>(
     let num_rows = list_array.len();
     let nulls = combined_nulls(list_array.nulls(), element.nulls());
     let mut result = vec![0i64; num_rows];
-    let comparator = make_comparator(values.as_ref(), element.as_ref(), SortOptions::default())?;
+    let values_normalized = normalize_negative_zero(values);
+    let element_normalized = normalize_negative_zero(element);
+    let comparator = make_comparator(
+        values_normalized.as_ref(),
+        element_normalized.as_ref(),
+        SortOptions::default(),
+    )?;
 
     for (row_index, w) in offsets.windows(2).enumerate() {
         if nulls.as_ref().is_some_and(|n| n.is_null(row_index)) {
@@ -301,8 +309,6 @@ mod tests {
 
     #[test]
     fn test_nested_float_and_null_position() -> DataFusionResult<()> {
-        // Arrow and the previous ScalarValue fallback distinguish signed zeros, so the second
-        // row matches at position 2 rather than position 1.
         let values = ListArray::from_iter_primitive::<Float64Type, _, _>([
             Some(vec![Some(1.0)]),
             Some(vec![Some(f64::NAN)]),
@@ -324,7 +330,44 @@ mod tests {
 
         let result = array_position_inner(&[Arc::new(array), Arc::new(element)])?;
         let result = result.as_any().downcast_ref::<Int64Array>().unwrap();
-        assert_eq!(result, &Int64Array::from(vec![2, 2, 1]));
+        assert_eq!(result, &Int64Array::from(vec![2, 1, 1]));
+        Ok(())
+    }
+
+    #[test]
+    fn test_struct_float_field_signed_zero_position() -> DataFusionResult<()> {
+        use arrow::array::{Float64Builder, StructBuilder};
+
+        let fields = vec![Arc::new(Field::new("a", DataType::Float64, true))];
+        let mut values_builder =
+            StructBuilder::new(fields.clone(), vec![Box::new(Float64Builder::new())]);
+        for v in [-0.0, 1.0] {
+            values_builder
+                .field_builder::<Float64Builder>(0)
+                .unwrap()
+                .append_value(v);
+            values_builder.append(true);
+        }
+        let values = Arc::new(values_builder.finish());
+        let array = ListArray::new(
+            Arc::new(Field::new("item", values.data_type().clone(), true)),
+            OffsetBuffer::new(vec![0, 2].into()),
+            values,
+            None,
+        );
+
+        let mut element_builder = StructBuilder::new(fields, vec![Box::new(Float64Builder::new())]);
+        element_builder
+            .field_builder::<Float64Builder>(0)
+            .unwrap()
+            .append_value(0.0);
+        element_builder.append(true);
+        let element = element_builder.finish();
+
+        let result = array_position_inner(&[Arc::new(array), Arc::new(element)])?;
+        let result = result.as_any().downcast_ref::<Int64Array>().unwrap();
+        // {-0.0} is the first element and now matches {0.0}, matching Spark.
+        assert_eq!(result, &Int64Array::from(vec![1]));
         Ok(())
     }
 }

native/spark-expr/src/array_funcs/arrays_overlap.rs

@@ -51,6 +51,8 @@ use std::hash::Hash;
 use std::ops::Range;
 use std::sync::Arc;
 
+use super::nested_float_normalize::normalize_negative_zero;
+
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub struct SparkArraysOverlap {
     signature: Signature,
@@ -429,9 +431,11 @@ fn arrays_overlap_list_generic<OffsetSize: OffsetSizeTrait>(
         };
 
         let comparator = if needs_comparator(probe.data_type()) {
+            let probe_normalized = normalize_negative_zero(probe);
+            let search_normalized = normalize_negative_zero(search);
             Some(make_comparator(
-                probe.as_ref(),
-                search.as_ref(),
+                probe_normalized.as_ref(),
+                search_normalized.as_ref(),
                 SortOptions::default(),
             )?)
         } else {
@@ -706,8 +710,7 @@ mod tests {
 
     #[test]
     fn test_nested_float_total_order() -> Result<()> {
-        // Preserve the existing Arrow total-order behavior: NaN matches itself, while signed
-        // zeros are distinct.
+        // NaN matches itself, and signed zeros are equal, matching Spark.
         let left = make_nested_float_list(&[&[f64::NAN]]);
         let right = make_nested_float_list(&[&[f64::NAN]]);
         let result = arrays_overlap_list::<i32>(&left, &right)?;
@@ -718,7 +721,7 @@ mod tests {
         let right = make_nested_float_list(&[&[-0.0]]);
         let result = arrays_overlap_list::<i32>(&left, &right)?;
         let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
-        assert!(!result.value(0));
+        assert!(result.value(0));
         Ok(())
     }
 
@@ -905,6 +908,37 @@ mod tests {
         Ok(())
     }
 
+    /// Build a single-row ListArray of structs: List<Struct<a: Float64>>
+    fn make_struct_float_list(elements: Vec<Option<f64>>) -> ListArray {
+        let fields = vec![Arc::new(Field::new("a", DataType::Float64, true))];
+        let struct_builder =
+            StructBuilder::new(fields.clone(), vec![Box::new(Float64Builder::new())]);
+        let mut list_builder = ListBuilder::new(struct_builder);
+
+        for elem in &elements {
+            let sb = list_builder.values();
+            sb.field_builder::<Float64Builder>(0)
+                .unwrap()
+                .append_option(*elem);
+            sb.append(true);
+        }
+        list_builder.append(true);
+        list_builder.finish()
+    }
+
+    #[test]
+    fn test_struct_float_field_signed_zero_overlap() -> Result<()> {
+        // [{-0.0}] vs [{0.0}] => true, matching Spark
+        let left = make_struct_float_list(vec![Some(-0.0)]);
+        let right = make_struct_float_list(vec![Some(0.0)]);
+
+        let result = arrays_overlap_list::<i32>(&left, &right)?;
+        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
+        assert!(result.is_valid(0));
+        assert!(result.value(0));
+        Ok(())
+    }
+
     #[test]
     fn test_struct_null_element() -> Result<()> {
         // [NULL] vs [{1,2}] => null (null outer element)

native/spark-expr/src/array_funcs/mod.rs

@@ -23,6 +23,7 @@ mod arrays_zip;
 mod flatten;
 mod get_array_struct_fields;
 mod list_extract;
+mod nested_float_normalize;
 mod size;
 
 pub use array_insert::ArrayInsert;

native/spark-expr/src/array_funcs/nested_float_normalize.rs

@@ -0,0 +1,183 @@
+// 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.
+
+use arrow::array::{
+    Array, ArrayRef, AsArray, FixedSizeListArray, Float32Array, Float64Array, LargeListArray,
+    ListArray, StructArray,
+};
+use arrow::datatypes::DataType;
+use std::sync::Arc;
+
+/// Recursively rebuilds nested arrays with `-0.0` normalized to `0.0` in any
+/// Float32/Float64 leaves, leaving NaN untouched.
+pub(super) fn normalize_negative_zero(array: &ArrayRef) -> ArrayRef {
+    match array.data_type() {
+        DataType::Float32 => {
+            let arr = array.as_primitive::<arrow::datatypes::Float32Type>();
+            let normalized: Float32Array = arr
+                .iter()
+                .map(|v| v.map(|v| if v == 0.0 { 0.0f32 } else { v }))
+                .collect();
+            Arc::new(normalized)
+        }
+        DataType::Float64 => {
+            let arr = array.as_primitive::<arrow::datatypes::Float64Type>();
+            let normalized: Float64Array = arr
+                .iter()
+                .map(|v| v.map(|v| if v == 0.0 { 0.0f64 } else { v }))
+                .collect();
+            Arc::new(normalized)
+        }
+        DataType::List(field) => {
+            let list = array.as_list::<i32>();
+            let normalized_values = normalize_negative_zero(list.values());
+            Arc::new(ListArray::new(
+                Arc::clone(field),
+                list.offsets().clone(),
+                normalized_values,
+                list.nulls().cloned(),
+            ))
+        }
+        DataType::LargeList(field) => {
+            let list = array.as_list::<i64>();
+            let normalized_values = normalize_negative_zero(list.values());
+            Arc::new(LargeListArray::new(
+                Arc::clone(field),
+                list.offsets().clone(),
+                normalized_values,
+                list.nulls().cloned(),
+            ))
+        }
+        DataType::FixedSizeList(field, size) => {
+            let list = array.as_fixed_size_list();
+            let normalized_values = normalize_negative_zero(list.values());
+            Arc::new(FixedSizeListArray::new(
+                Arc::clone(field),
+                *size,
+                normalized_values,
+                list.nulls().cloned(),
+            ))
+        }
+        DataType::Struct(_) => {
+            let s = array.as_struct();
+            let normalized_columns: Vec<ArrayRef> =
+                s.columns().iter().map(normalize_negative_zero).collect();
+            Arc::new(StructArray::new(
+                s.fields().clone(),
+                normalized_columns,
+                s.nulls().cloned(),
+            ))
+        }
+        _ => Arc::clone(array),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow::array::Float64Builder;
+    use arrow::array::ListBuilder;
+    use arrow::datatypes::Field;
+
+    #[test]
+    fn test_normalize_flat_floats() {
+        let arr: ArrayRef = Arc::new(Float64Array::from(vec![
+            Some(-0.0),
+            Some(0.0),
+            Some(f64::NAN),
+            None,
+            Some(1.5),
+        ]));
+        let normalized = normalize_negative_zero(&arr);
+        let normalized = normalized.as_primitive::<arrow::datatypes::Float64Type>();
+
+        assert_eq!(normalized.value(0).to_bits(), 0.0f64.to_bits());
+        assert_eq!(normalized.value(1).to_bits(), 0.0f64.to_bits());
+        assert!(normalized.value(2).is_nan());
+        assert!(normalized.is_null(3));
+        assert_eq!(normalized.value(4), 1.5);
+    }
+
+    #[test]
+    fn test_normalize_nested_list_floats() {
+        let mut builder = ListBuilder::new(Float64Builder::new());
+        builder.values().append_value(-0.0);
+        builder.values().append_value(f64::NAN);
+        builder.append(true);
+        let arr: ArrayRef = Arc::new(builder.finish());
+
+        let normalized = normalize_negative_zero(&arr);
+        let normalized = normalized.as_list::<i32>();
+        let inner = normalized.value(0);
+        let inner = inner.as_primitive::<arrow::datatypes::Float64Type>();
+
+        assert_eq!(inner.value(0).to_bits(), 0.0f64.to_bits());
+        assert!(inner.value(1).is_nan());
+    }
+
+    #[test]
+    fn test_normalize_struct_floats() {
+        let a = Float64Array::from(vec![Some(-0.0), Some(1.0)]);
+        let b = Float64Array::from(vec![Some(f64::NAN), Some(-0.0)]);
+        let fields = vec![
+            Arc::new(Field::new("a", DataType::Float64, true)),
+            Arc::new(Field::new("b", DataType::Float64, true)),
+        ];
+        let arr: ArrayRef = Arc::new(StructArray::new(
+            fields.into(),
+            vec![Arc::new(a), Arc::new(b)],
+            None,
+        ));
+
+        let normalized = normalize_negative_zero(&arr);
+        let normalized = normalized.as_struct();
+        let col_a = normalized
+            .column(0)
+            .as_primitive::<arrow::datatypes::Float64Type>();
+        let col_b = normalized
+            .column(1)
+            .as_primitive::<arrow::datatypes::Float64Type>();
+
+        assert_eq!(col_a.value(0).to_bits(), 0.0f64.to_bits());
+        assert_eq!(col_a.value(1), 1.0);
+        assert!(col_b.value(0).is_nan());
+        assert_eq!(col_b.value(1).to_bits(), 0.0f64.to_bits());
+    }
+
+    #[test]
+    fn test_normalize_fixed_size_list_floats() {
+        let values = Float64Array::from(vec![Some(-0.0), Some(f64::NAN), Some(1.0), Some(-0.0)]);
+        let field = Arc::new(Field::new("item", DataType::Float64, true));
+        let arr: ArrayRef = Arc::new(FixedSizeListArray::new(
+            Arc::clone(&field),
+            2,
+            Arc::new(values),
+            None,
+        ));
+
+        let normalized = normalize_negative_zero(&arr);
+        let normalized = normalized.as_fixed_size_list();
+        let flat = normalized
+            .values()
+            .as_primitive::<arrow::datatypes::Float64Type>();
+
+        assert_eq!(flat.value(0).to_bits(), 0.0f64.to_bits());
+        assert!(flat.value(1).is_nan());
+        assert_eq!(flat.value(2), 1.0);
+        assert_eq!(flat.value(3).to_bits(), 0.0f64.to_bits());
+    }
+}