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

// 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.

//! Spark-compatible `arrays_overlap` with correct null handling.
//!
//! DataFusion's `array_has_any` uses `RowConverter` for element comparison, which
//! treats NULL == NULL as true (grouping semantics). For outer-level null elements,
//! Spark's `arrays_overlap` uses three-valued logic: NULL elements are skipped but
//! cause the result to be null if no definite overlap is found. For comparing
//! non-null elements (including nested types), Spark uses structural equality via
//! `ordering.equiv` where NULL == NULL is true.
//!
//! This implementation returns:
//!   - true  if any non-null element appears in both arrays
//!   - null  if no definite overlap but either array contains null elements
//!   - false if no overlap and neither array contains null elements

use arrow::array::{
    make_comparator, Array, ArrayRef, AsArray, BooleanArray, GenericListArray, GenericStringArray,
    OffsetSizeTrait, PrimitiveArray, Scalar,
};
use arrow::buffer::NullBuffer;
use arrow::compute::kernels::cmp::eq;
use arrow::compute::SortOptions;
use arrow::datatypes::{
    ArrowPrimitiveType, DataType, Date32Type, Date64Type, Decimal128Type, FieldRef, Float32Type,
    Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, TimeUnit, TimestampMicrosecondType,
    TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType, UInt16Type, UInt32Type,
    UInt64Type, UInt8Type,
};
use datafusion::common::{exec_err, utils::take_function_args, HashSet, Result, ScalarValue};
use datafusion::logical_expr::{
    ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, TypeSignature, Volatility,
};
use std::cmp::Ordering;
use std::hash::Hash;
use std::ops::Range;
use std::sync::Arc;

use crate::float_semantics::{canonicalize_nan, has_float_leaf, normalize_nested_floats};

#[derive(Debug, PartialEq, Eq, Hash)]
pub struct SparkArraysOverlap {
    signature: Signature,
}

impl Default for SparkArraysOverlap {
    fn default() -> Self {
        Self::new()
    }
}

impl SparkArraysOverlap {
    pub fn new() -> Self {
        Self {
            signature: Signature::new(TypeSignature::Any(2), Volatility::Immutable),
        }
    }
}

impl ScalarUDFImpl for SparkArraysOverlap {
    fn name(&self) -> &str {
        "spark_arrays_overlap"
    }

    fn signature(&self) -> &Signature {
        &self.signature
    }

    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
        Ok(DataType::Boolean)
    }

    fn return_field_from_args(
        &self,
        _args: datafusion::logical_expr::ReturnFieldArgs,
    ) -> Result<FieldRef> {
        Ok(Arc::new(arrow::datatypes::Field::new(
            self.name(),
            DataType::Boolean,
            true,
        )))
    }

    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
        let [left, right] = take_function_args(self.name(), &args.args)?;

        // Return null if either input is a null scalar
        if let ColumnarValue::Scalar(s) = &left {
            if s.is_null() {
                return Ok(ColumnarValue::Scalar(ScalarValue::Boolean(None)));
            }
        }
        if let ColumnarValue::Scalar(s) = &right {
            if s.is_null() {
                return Ok(ColumnarValue::Scalar(ScalarValue::Boolean(None)));
            }
        }

        match (left, right) {
            (ColumnarValue::Array(left_arr), ColumnarValue::Array(right_arr)) => {
                let result = match (left_arr.data_type(), right_arr.data_type()) {
                    (DataType::List(_), DataType::List(_)) => arrays_overlap_list::<i32>(
                        left_arr.as_any().downcast_ref().unwrap(),
                        right_arr.as_any().downcast_ref().unwrap(),
                    )?,
                    (DataType::LargeList(_), DataType::LargeList(_)) => arrays_overlap_list::<i64>(
                        left_arr.as_any().downcast_ref().unwrap(),
                        right_arr.as_any().downcast_ref().unwrap(),
                    )?,
                    (l, r) => {
                        return exec_err!(
                            "spark_arrays_overlap does not support types '{l}' and '{r}'"
                        )
                    }
                };
                Ok(ColumnarValue::Array(result))
            }
            (left, right) => {
                // Handle scalar inputs by converting to arrays
                let left_arr = left.to_array(1)?;
                let right_arr = right.to_array(1)?;
                let result = match (left_arr.data_type(), right_arr.data_type()) {
                    (DataType::List(_), DataType::List(_)) => arrays_overlap_list::<i32>(
                        left_arr.as_any().downcast_ref().unwrap(),
                        right_arr.as_any().downcast_ref().unwrap(),
                    )?,
                    (DataType::LargeList(_), DataType::LargeList(_)) => arrays_overlap_list::<i64>(
                        left_arr.as_any().downcast_ref().unwrap(),
                        right_arr.as_any().downcast_ref().unwrap(),
                    )?,
                    (l, r) => {
                        return exec_err!(
                            "spark_arrays_overlap does not support types '{l}' and '{r}'"
                        )
                    }
                };
                let scalar = ScalarValue::try_from_array(&result, 0)?;
                Ok(ColumnarValue::Scalar(scalar))
            }
        }
    }
}

/// Spark-compatible arrays_overlap with SQL three-valued null logic.
///
/// For each row, compares elements of two list arrays and returns:
/// - null if either array is null
/// - true if any non-null element appears in both arrays
/// - null if no definite overlap but either array contains null elements
/// - false otherwise
fn arrays_overlap_list<OffsetSize: OffsetSizeTrait>(
    left: &GenericListArray<OffsetSize>,
    right: &GenericListArray<OffsetSize>,
) -> Result<ArrayRef> {
    let left_values = left.values();
    let right_values = right.values();

    if left_values.data_type() != right_values.data_type() {
        return arrays_overlap_list_generic(left, right);
    }

    // Fast paths for flat element types: probe the flat value buffers directly instead of
    // slicing each row and running an Arrow compare kernel once per probe element.
    macro_rules! flat_fast_path {
        ($l:expr, $r:expr) => {
            return Ok(overlap_rows(left, right, flat_row_overlap($l, $r)))
        };
    }
    macro_rules! primitive_fast_path {
        ($t:ty) => {
            flat_fast_path!(
                left_values.as_primitive::<$t>(),
                right_values.as_primitive::<$t>()
            )
        };
    }

    match left_values.data_type() {
        DataType::Boolean => flat_fast_path!(left_values.as_boolean(), right_values.as_boolean()),
        DataType::Int8 => primitive_fast_path!(Int8Type),
        DataType::Int16 => primitive_fast_path!(Int16Type),
        DataType::Int32 => primitive_fast_path!(Int32Type),
        DataType::Int64 => primitive_fast_path!(Int64Type),
        DataType::UInt8 => primitive_fast_path!(UInt8Type),
        DataType::UInt16 => primitive_fast_path!(UInt16Type),
        DataType::UInt32 => primitive_fast_path!(UInt32Type),
        DataType::UInt64 => primitive_fast_path!(UInt64Type),
        DataType::Float32 => primitive_fast_path!(Float32Type),
        DataType::Float64 => primitive_fast_path!(Float64Type),
        DataType::Date32 => primitive_fast_path!(Date32Type),
        DataType::Date64 => primitive_fast_path!(Date64Type),
        DataType::Decimal128(_, _) => primitive_fast_path!(Decimal128Type),
        DataType::Timestamp(TimeUnit::Second, _) => primitive_fast_path!(TimestampSecondType),
        DataType::Timestamp(TimeUnit::Millisecond, _) => {
            primitive_fast_path!(TimestampMillisecondType)
        }
        DataType::Timestamp(TimeUnit::Microsecond, _) => {
            primitive_fast_path!(TimestampMicrosecondType)
        }
        DataType::Timestamp(TimeUnit::Nanosecond, _) => {
            primitive_fast_path!(TimestampNanosecondType)
        }
        DataType::Utf8 => flat_fast_path!(
            left_values.as_string::<i32>(),
            right_values.as_string::<i32>()
        ),
        DataType::LargeUtf8 => flat_fast_path!(
            left_values.as_string::<i64>(),
            right_values.as_string::<i64>()
        ),
        _ => arrays_overlap_list_generic(left, right),
    }
}

/// Drives the row loop for the flat fast paths. `row_overlap` reports whether the two element
/// ranges share a non-null value; null bookkeeping is identical to the generic path: when there
/// is no definite overlap, the row is null if either side holds a null element.
fn overlap_rows<OffsetSize: OffsetSizeTrait>(
    left: &GenericListArray<OffsetSize>,
    right: &GenericListArray<OffsetSize>,
    mut row_overlap: impl FnMut(Range<usize>, Range<usize>) -> bool,
) -> ArrayRef {
    let len = left.len();
    let left_offsets = left.offsets();
    let right_offsets = right.offsets();
    let left_element_nulls = left.values().nulls();
    let right_element_nulls = right.values().nulls();

    let mut builder = BooleanArray::builder(len);

    for i in 0..len {
        if left.is_null(i) || right.is_null(i) {
            builder.append_null();
            continue;
        }

        let left_range = left_offsets[i].as_usize()..left_offsets[i + 1].as_usize();
        let right_range = right_offsets[i].as_usize()..right_offsets[i + 1].as_usize();

        if left_range.is_empty() || right_range.is_empty() {
            builder.append_value(false);
        } else if row_overlap(left_range.clone(), right_range.clone()) {
            builder.append_value(true);
        } else if range_has_null(left_element_nulls, left_range)
            || range_has_null(right_element_nulls, right_range)
        {
            builder.append_null();
        } else {
            builder.append_value(false);
        }
    }

    Arc::new(builder.finish())
}

/// True if the validity bitmap marks any element in `range` as null. Slicing counts the bitmap a
/// word at a time rather than testing each element.
fn range_has_null(nulls: Option<&NullBuffer>, range: Range<usize>) -> bool {
    nulls.is_some_and(|n| n.null_count() > 0 && n.slice(range.start, range.len()).null_count() > 0)
}

/// Projects a native value onto a Spark-compatible hashable key. Floating-point keys canonicalize
/// every NaN representation while preserving the distinct bit patterns of positive and negative
/// zero.
trait OverlapKey: Copy {
    type Key: Hash + Eq + Copy;

    fn overlap_key(self) -> Self::Key;
}

macro_rules! identity_overlap_key {
    ($($t:ty),*) => {
        $(impl OverlapKey for $t {
            type Key = $t;

            fn overlap_key(self) -> $t {
                self
            }
        })*
    };
}
identity_overlap_key!(i8, i16, i32, i64, i128, u8, u16, u32, u64);

impl OverlapKey for f32 {
    type Key = u32;

    fn overlap_key(self) -> u32 {
        canonicalize_nan(self).to_bits()
    }
}

impl OverlapKey for f64 {
    type Key = u64;

    fn overlap_key(self) -> u64 {
        canonicalize_nan(self).to_bits()
    }
}

/// A flat element array whose value at an index reduces to a hashable key, or `None` when the
/// element is null and so can never take part in an overlap.
trait KeyedValues<'a> {
    type Key: Hash + Eq + Copy;

    fn key_at(&self, i: usize) -> Option<Self::Key>;
}

impl<'a, T: ArrowPrimitiveType> KeyedValues<'a> for &'a PrimitiveArray<T>
where
    T::Native: OverlapKey,
{
    type Key = <T::Native as OverlapKey>::Key;

    fn key_at(&self, i: usize) -> Option<Self::Key> {
        (!self.is_null(i)).then(|| self.value(i).overlap_key())
    }
}

impl<'a> KeyedValues<'a> for &'a BooleanArray {
    type Key = bool;

    fn key_at(&self, i: usize) -> Option<bool> {
        (!self.is_null(i)).then(|| self.value(i))
    }
}

impl<'a, S: OffsetSizeTrait> KeyedValues<'a> for &'a GenericStringArray<S> {
    type Key = &'a str;

    fn key_at(&self, i: usize) -> Option<&'a str> {
        let values: &'a GenericStringArray<S> = self;
        (!values.is_null(i)).then(|| values.value(i))
    }
}

/// Above this many pairwise comparisons a hash probe beats the nested scan.
const NESTED_SCAN_BUDGET: usize = 256;

/// Row overlap for flat element types. The scratch set is reused across rows.
fn flat_row_overlap<'a, V>(left: V, right: V) -> impl FnMut(Range<usize>, Range<usize>) -> bool + 'a
where
    V: KeyedValues<'a> + Copy + 'a,
{
    let mut seen: HashSet<V::Key> = HashSet::new();

    move |left_range, right_range| {
        // Probe with the smaller side so the inner loop, and the hash table, stay small.
        let (probe, probe_range, search, search_range) = if left_range.len() <= right_range.len() {
            (left, left_range, right, right_range)
        } else {
            (right, right_range, left, left_range)
        };

        if probe_range.len() * search_range.len() <= NESTED_SCAN_BUDGET {
            for pi in probe_range {
                let Some(key) = probe.key_at(pi) else {
                    continue;
                };
                for si in search_range.clone() {
                    if search.key_at(si) == Some(key) {
                        return true;
                    }
                }
            }
            return false;
        }

        seen.clear();
        seen.reserve(probe_range.len());
        seen.extend(probe_range.filter_map(|pi| probe.key_at(pi)));
        search_range
            .into_iter()
            .any(|si| search.key_at(si).is_some_and(|key| seen.contains(&key)))
    }
}

fn normalize_list_element_floats<OffsetSize: OffsetSizeTrait>(
    list: &GenericListArray<OffsetSize>,
) -> GenericListArray<OffsetSize> {
    let field = match list.data_type() {
        DataType::List(f) | DataType::LargeList(f) => Arc::clone(f),
        _ => unreachable!("GenericListArray always has List or LargeList data type"),
    };
    let normalized_values = normalize_nested_floats(list.values());
    GenericListArray::new(
        field,
        list.offsets().clone(),
        normalized_values,
        list.nulls().cloned(),
    )
}

/// Fallback for nested and otherwise unhandled element types.
///
/// note: Spark's flat arrays_overlap (HashSet<Double>) treats -0.0 and 0.0 as different,
/// only the nested path here treats them as equal. this normalization can't move into the
/// flat fast path in arrays_overlap_list without breaking that difference.
fn arrays_overlap_list_generic<OffsetSize: OffsetSizeTrait>(
    left: &GenericListArray<OffsetSize>,
    right: &GenericListArray<OffsetSize>,
) -> Result<ArrayRef> {
    let left_owned =
        has_float_leaf(left.values().data_type()).then(|| normalize_list_element_floats(left));
    let left: &GenericListArray<OffsetSize> = left_owned.as_ref().unwrap_or(left);
    let right_owned =
        has_float_leaf(right.values().data_type()).then(|| normalize_list_element_floats(right));
    let right: &GenericListArray<OffsetSize> = right_owned.as_ref().unwrap_or(right);

    let len = left.len();
    let mut builder = BooleanArray::builder(len);

    for i in 0..len {
        if left.is_null(i) || right.is_null(i) {
            builder.append_null();
            continue;
        }

        let left_values = left.value(i);
        let right_values = right.value(i);

        if left_values.is_empty() || right_values.is_empty() {
            builder.append_value(false);
            continue;
        }

        // DataFusion's make_array(NULL) produces a List<Null> with NullArray values.
        if left_values.data_type() == &DataType::Null || right_values.data_type() == &DataType::Null
        {
            builder.append_null();
            continue;
        }

        let mut found_overlap = false;
        let mut has_null = false;

        // Put smaller array on the probe side: fewer find_in_array calls means
        // fewer kernel dispatches and allocations in the flat vectorized path.
        let (probe, search) = if left_values.len() <= right_values.len() {
            (&left_values, &right_values)
        } else {
            (&right_values, &left_values)
        };

        let comparator = if needs_comparator(probe.data_type()) {
            Some(make_comparator(
                probe.as_ref(),
                search.as_ref(),
                SortOptions::default(),
            )?)
        } else {
            None
        };

        for pi in 0..probe.len() {
            if probe.is_null(pi) {
                has_null = true;
                continue;
            }
            let (found, null_eq) = if let Some(comparator) = &comparator {
                find_in_array_nested(pi, search, comparator.as_ref())
            } else {
                find_in_array_flat(probe, pi, search)?
            };
            if null_eq {
                has_null = true;
            }
            if found {
                found_overlap = true;
                break;
            }
        }

        if found_overlap {
            builder.append_value(true);
        } else if has_null {
            builder.append_null();
        } else {
            builder.append_value(false);
        }
    }

    Ok(Arc::new(builder.finish()))
}

/// Vectorized search using Arrow's `eq` kernel. One SIMD call per probe element.
fn find_in_array_flat(probe: &ArrayRef, pi: usize, search: &ArrayRef) -> Result<(bool, bool)> {
    let scalar = Scalar::new(probe.slice(pi, 1));
    let eq_result = eq(search, &scalar)
        .map_err(|e| datafusion::error::DataFusionError::ArrowError(Box::new(e), None))?;
    Ok((eq_result.true_count() > 0, eq_result.null_count() > 0))
}

/// Element-by-element search using Arrow's nested comparator.
fn find_in_array_nested(
    pi: usize,
    search: &ArrayRef,
    comparator: &dyn Fn(usize, usize) -> Ordering,
) -> (bool, bool) {
    let mut has_null = false;
    for si in 0..search.len() {
        if search.is_null(si) {
            has_null = true;
            continue;
        }
        if comparator(pi, si) == Ordering::Equal {
            return (true, has_null);
        }
    }
    (false, has_null)
}

fn needs_comparator(dt: &DataType) -> bool {
    matches!(
        dt,
        DataType::List(_)
            | DataType::LargeList(_)
            | DataType::FixedSizeList(_, _)
            | DataType::Struct(_)
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use arrow::array::{
        Float64Array, Float64Builder, Int32Array, Int32Builder, ListArray, ListBuilder,
        StructArray, StructBuilder,
    };
    use arrow::buffer::{NullBuffer, OffsetBuffer};
    use arrow::datatypes::{Field, Fields};

    fn make_list_array(
        values: &Int32Array,
        offsets: &[i32],
        nulls: Option<NullBuffer>,
    ) -> ListArray {
        ListArray::new(
            Arc::new(Field::new("item", DataType::Int32, true)),
            OffsetBuffer::new(offsets.to_vec().into()),
            Arc::new(values.clone()),
            nulls,
        )
    }

    #[test]
    fn test_basic_overlap() -> Result<()> {
        // [1, 2, 3] vs [3, 4, 5] => true
        let left = make_list_array(&Int32Array::from(vec![1, 2, 3]), &[0, 3], None);
        let right = make_list_array(&Int32Array::from(vec![3, 4, 5]), &[0, 3], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.value(0));
        assert!(result.is_valid(0));
        Ok(())
    }

    #[test]
    fn test_no_overlap() -> Result<()> {
        // [1, 2] vs [3, 4] => false
        let left = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);
        let right = make_list_array(&Int32Array::from(vec![3, 4]), &[0, 2], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(!result.value(0));
        assert!(result.is_valid(0));
        Ok(())
    }

    #[test]
    fn test_flat_float32_nan_payloads_and_signed_zero() -> Result<()> {
        let positive_nan = f32::from_bits(0x7fc0_0001);
        let negative_nan = f32::from_bits(0xffc0_0002);
        let signaling_nan = f32::from_bits(0x7f80_0001);

        let hash_nan_left = (1..=16)
            .map(|value| Some(value as f32))
            .chain([Some(positive_nan)])
            .collect::<Vec<_>>();
        let hash_nan_right = (17..=32)
            .map(|value| Some(value as f32))
            .chain([Some(negative_nan)])
            .collect::<Vec<_>>();
        let hash_zero_left = (1..=16)
            .map(|value| Some(value as f32))
            .chain([Some(0.0)])
            .collect::<Vec<_>>();
        let hash_zero_right = (17..=32)
            .map(|value| Some(value as f32))
            .chain([Some(-0.0)])
            .collect::<Vec<_>>();

        let left = ListArray::from_iter_primitive::<Float32Type, _, _>([
            Some(vec![Some(positive_nan)]),
            Some(vec![Some(negative_nan)]),
            Some(vec![Some(signaling_nan)]),
            Some(vec![Some(0.0)]),
            Some(vec![Some(-0.0)]),
            Some(vec![Some(positive_nan), None]),
            Some(vec![Some(0.0), None]),
            Some(hash_nan_left),
            Some(hash_zero_left),
        ]);
        let right = ListArray::from_iter_primitive::<Float32Type, _, _>([
            Some(vec![Some(f32::NAN)]),
            Some(vec![Some(positive_nan)]),
            Some(vec![Some(negative_nan)]),
            Some(vec![Some(-0.0)]),
            Some(vec![Some(0.0)]),
            Some(vec![Some(negative_nan)]),
            Some(vec![Some(-0.0)]),
            Some(hash_nan_right),
            Some(hash_zero_right),
        ]);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        let expected = BooleanArray::from(vec![
            Some(true),
            Some(true),
            Some(true),
            Some(false),
            Some(false),
            Some(true),
            None,
            Some(true),
            Some(false),
        ]);
        assert_eq!(result, &expected);
        Ok(())
    }

    #[test]
    fn test_flat_float64_nan_payloads_and_signed_zero() -> Result<()> {
        let positive_nan = f64::from_bits(0x7ff8_0000_0000_0001);
        let negative_nan = f64::from_bits(0xfff8_0000_0000_0002);
        let signaling_nan = f64::from_bits(0x7ff0_0000_0000_0001);

        let hash_nan_left = (1..=16)
            .map(|value| Some(value as f64))
            .chain([Some(positive_nan)])
            .collect::<Vec<_>>();
        let hash_nan_right = (17..=32)
            .map(|value| Some(value as f64))
            .chain([Some(negative_nan)])
            .collect::<Vec<_>>();
        let hash_zero_left = (1..=16)
            .map(|value| Some(value as f64))
            .chain([Some(0.0)])
            .collect::<Vec<_>>();
        let hash_zero_right = (17..=32)
            .map(|value| Some(value as f64))
            .chain([Some(-0.0)])
            .collect::<Vec<_>>();

        let left = ListArray::from_iter_primitive::<Float64Type, _, _>([
            Some(vec![Some(positive_nan)]),
            Some(vec![Some(negative_nan)]),
            Some(vec![Some(signaling_nan)]),
            Some(vec![Some(0.0)]),
            Some(vec![Some(-0.0)]),
            Some(vec![Some(positive_nan), None]),
            Some(vec![Some(0.0), None]),
            Some(hash_nan_left),
            Some(hash_zero_left),
        ]);
        let right = ListArray::from_iter_primitive::<Float64Type, _, _>([
            Some(vec![Some(f64::NAN)]),
            Some(vec![Some(positive_nan)]),
            Some(vec![Some(negative_nan)]),
            Some(vec![Some(-0.0)]),
            Some(vec![Some(0.0)]),
            Some(vec![Some(negative_nan)]),
            Some(vec![Some(-0.0)]),
            Some(hash_nan_right),
            Some(hash_zero_right),
        ]);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        let expected = BooleanArray::from(vec![
            Some(true),
            Some(true),
            Some(true),
            Some(false),
            Some(false),
            Some(true),
            None,
            Some(true),
            Some(false),
        ]);
        assert_eq!(result, &expected);
        Ok(())
    }

    #[test]
    fn test_null_only_overlap() -> Result<()> {
        // [1, NULL] vs [NULL, 2] => null (no definite overlap, but nulls present)
        let left = make_list_array(&Int32Array::from(vec![Some(1), None]), &[0, 2], None);
        let right = make_list_array(&Int32Array::from(vec![None, Some(2)]), &[0, 2], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.is_null(0));
        Ok(())
    }

    #[test]
    fn test_null_with_overlap() -> Result<()> {
        // [1, NULL] vs [1, 2] => true (definite overlap on 1)
        let left = make_list_array(&Int32Array::from(vec![Some(1), None]), &[0, 2], None);
        let right = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.value(0));
        assert!(result.is_valid(0));
        Ok(())
    }

    #[test]
    fn test_empty_array() -> Result<()> {
        // [1, NULL, 3] vs [] => false
        let left = make_list_array(
            &Int32Array::from(vec![Some(1), None, Some(3)]),
            &[0, 3],
            None,
        );
        let right = make_list_array(&Int32Array::from(Vec::<i32>::new()), &[0, 0], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(!result.value(0));
        assert!(result.is_valid(0));
        Ok(())
    }

    #[test]
    fn test_null_array() -> Result<()> {
        // NULL vs [1, 2] => null
        let left = make_list_array(
            &Int32Array::from(Vec::<i32>::new()),
            &[0, 0],
            Some(NullBuffer::from(vec![false])),
        );
        let right = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.is_null(0));
        Ok(())
    }

    #[test]
    fn test_both_null_elements() -> Result<()> {
        // [NULL] vs [NULL] => null
        let left = make_list_array(&Int32Array::from(vec![None::<i32>]), &[0, 1], None);
        let right = make_list_array(&Int32Array::from(vec![None::<i32>]), &[0, 1], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.is_null(0));
        Ok(())
    }

    #[test]
    fn test_both_null_elements_via_null_array() -> Result<()> {
        // Simulate what DataFusion's make_array(NULL) produces: List<Null> with NullArray values
        use arrow::array::NullArray;

        let null_values = Arc::new(NullArray::new(1)) as ArrayRef;
        let null_field = Arc::new(Field::new("item", DataType::Null, true));
        let left = ListArray::new(
            Arc::clone(&null_field),
            OffsetBuffer::new(vec![0, 1].into()),
            Arc::clone(&null_values),
            None,
        );
        let right = ListArray::new(
            null_field,
            OffsetBuffer::new(vec![0, 1].into()),
            null_values,
            None,
        );

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(
            result.is_null(0),
            "Expected null for [NULL] vs [NULL] (NullArray representation), got {:?}",
            result
        );
        Ok(())
    }

    #[test]
    fn test_one_null_element_no_overlap() -> Result<()> {
        // [3, NULL] vs [1, 2] => null
        let left = make_list_array(&Int32Array::from(vec![Some(3), None]), &[0, 2], None);
        let right = make_list_array(&Int32Array::from(vec![1, 2]), &[0, 2], None);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.is_null(0));
        Ok(())
    }

    /// Build a single-row ListArray of nested lists: List<List<Int32>>
    fn make_nested_list(elements: Vec<Option<Vec<Option<i32>>>>) -> ListArray {
        let inner_builder = ListBuilder::new(Int32Builder::new());
        let mut outer_builder = ListBuilder::new(inner_builder);

        for elem in &elements {
            match elem {
                Some(inner) => {
                    let inner_list_builder = outer_builder.values();
                    for val in inner {
                        match val {
                            Some(v) => inner_list_builder.values().append_value(*v),
                            None => inner_list_builder.values().append_null(),
                        }
                    }
                    inner_list_builder.append(true);
                }
                None => {
                    outer_builder.values().append(false);
                }
            }
        }
        outer_builder.append(true);
        outer_builder.finish()
    }

    fn make_nested_float_list(elements: &[&[f64]]) -> ListArray {
        let mut builder = ListBuilder::new(ListBuilder::new(Float64Builder::new()));
        for element in elements {
            for value in *element {
                builder.values().values().append_value(*value);
            }
            builder.values().append(true);
        }
        builder.append(true);
        builder.finish()
    }

    #[test]
    fn test_nested_float_spark_equality() -> Result<()> {
        // 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)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.value(0));

        let left = make_nested_float_list(&[&[0.0]]);
        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));
        Ok(())
    }

    #[test]
    fn test_nested_float_signed_nan_spark_equality() -> Result<()> {
        // [[-NaN]] vs [[NaN]] => true
        let left = make_nested_float_list(&[&[-f64::NAN]]);
        let right = make_nested_float_list(&[&[f64::NAN]]);
        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.value(0));
        Ok(())
    }

    #[test]
    fn test_nested_array_basic_overlap() -> Result<()> {
        // [[1,2], [3,4]] vs [[3,4], [5,6]] => true
        let left = make_nested_list(vec![
            Some(vec![Some(1), Some(2)]),
            Some(vec![Some(3), Some(4)]),
        ]);
        let right = make_nested_list(vec![
            Some(vec![Some(3), Some(4)]),
            Some(vec![Some(5), Some(6)]),
        ]);

        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_nested_array_no_overlap() -> Result<()> {
        // [[1,2]] vs [[3,4]] => false
        let left = make_nested_list(vec![Some(vec![Some(1), Some(2)])]);
        let right = make_nested_list(vec![Some(vec![Some(3), Some(4)])]);

        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_nested_array_inner_nulls_match() -> Result<()> {
        // [[1,NULL]] vs [[1,NULL]] => true (structural equality: NULL == NULL)
        let left = make_nested_list(vec![Some(vec![Some(1), None])]);
        let right = make_nested_list(vec![Some(vec![Some(1), None])]);

        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_nested_array_inner_nulls_no_match() -> Result<()> {
        // [[1,NULL]] vs [[1,2], [3,4]] => false (structural: [1,NULL] != [1,2], [1,NULL] != [3,4])
        let left = make_nested_list(vec![Some(vec![Some(1), None])]);
        let right = make_nested_list(vec![
            Some(vec![Some(1), Some(2)]),
            Some(vec![Some(3), Some(4)]),
        ]);

        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_nested_array_all_null_elements_match() -> Result<()> {
        // [[NULL]] vs [[NULL]] => true (structural equality: [NULL] == [NULL])
        let left = make_nested_list(vec![Some(vec![None])]);
        let right = make_nested_list(vec![Some(vec![None])]);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(
            result.is_valid(0),
            "Expected true for [[NULL]] vs [[NULL]], got null"
        );
        assert!(
            result.value(0),
            "Expected true for [[NULL]] vs [[NULL]], got false"
        );
        Ok(())
    }

    #[test]
    fn test_nested_array_definite_match_despite_inner_nulls() -> Result<()> {
        // [[1,2], [1,NULL]] vs [[1,2]] => true (definite match on [1,2])
        let left = make_nested_list(vec![
            Some(vec![Some(1), Some(2)]),
            Some(vec![Some(1), None]),
        ]);
        let right = make_nested_list(vec![Some(vec![Some(1), Some(2)])]);

        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(())
    }

    /// Build a single-row ListArray of structs: List<Struct<a: Int32, b: Int32>>
    fn make_struct_list(elements: Vec<Option<(Option<i32>, Option<i32>)>>) -> ListArray {
        let fields = vec![
            Arc::new(Field::new("a", DataType::Int32, true)),
            Arc::new(Field::new("b", DataType::Int32, true)),
        ];
        let struct_builder = StructBuilder::new(
            fields.clone(),
            vec![Box::new(Int32Builder::new()), Box::new(Int32Builder::new())],
        );
        let mut list_builder = ListBuilder::new(struct_builder);

        for elem in &elements {
            let sb = list_builder.values();
            match elem {
                Some((a, b)) => {
                    sb.field_builder::<Int32Builder>(0)
                        .unwrap()
                        .append_option(*a);
                    sb.field_builder::<Int32Builder>(1)
                        .unwrap()
                        .append_option(*b);
                    sb.append(true);
                }
                None => {
                    sb.field_builder::<Int32Builder>(0).unwrap().append_null();
                    sb.field_builder::<Int32Builder>(1).unwrap().append_null();
                    sb.append(false);
                }
            }
        }
        list_builder.append(true);
        list_builder.finish()
    }

    #[test]
    fn test_struct_basic_overlap() -> Result<()> {
        // [{1,2}, {3,4}] vs [{3,4}, {5,6}] => true
        let left = make_struct_list(vec![Some((Some(1), Some(2))), Some((Some(3), Some(4)))]);
        let right = make_struct_list(vec![Some((Some(3), Some(4))), Some((Some(5), Some(6)))]);

        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_no_overlap() -> Result<()> {
        // [{1,2}] vs [{3,4}] => false
        let left = make_struct_list(vec![Some((Some(1), Some(2)))]);
        let right = make_struct_list(vec![Some((Some(3), Some(4)))]);

        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_with_null_field_match() -> Result<()> {
        // [{1,NULL}] vs [{1,NULL}] => true (structural equality: NULL == NULL)
        let left = make_struct_list(vec![Some((Some(1), None))]);
        let right = make_struct_list(vec![Some((Some(1), None))]);

        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_definite_match_with_null_field() -> Result<()> {
        // [{1,2}, {1,NULL}] vs [{1,2}] => true (definite match on {1,2})
        let left = make_struct_list(vec![Some((Some(1), Some(2))), Some((Some(1), None))]);
        let right = make_struct_list(vec![Some((Some(1), Some(2)))]);

        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(())
    }

    /// 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()
    }

    /// Regression test for an empty struct alongside a float field. Iceberg exposes
    /// `_partition` as `struct<>` on an unpartitioned table, so
    /// `arrays_overlap(array(named_struct('x', x, 'p', _partition)), ...)` reaches the
    /// nested path with a zero column struct child. Rebuilding that child used to panic.
    #[test]
    fn test_struct_with_empty_struct_field_overlap() -> Result<()> {
        fn make_list(values: Vec<f64>) -> ListArray {
            let len = values.len();
            let x: ArrayRef = Arc::new(Float64Array::from(values));
            let partition: ArrayRef = Arc::new(StructArray::new_empty_fields(len, None));
            let fields: Fields = vec![
                Arc::new(Field::new("x", DataType::Float64, true)),
                Arc::new(Field::new("p", partition.data_type().clone(), true)),
            ]
            .into();
            let element: ArrayRef =
                Arc::new(StructArray::new(fields.clone(), vec![x, partition], None));
            ListArray::new(
                Arc::new(Field::new("item", DataType::Struct(fields), true)),
                OffsetBuffer::new((0..=len as i32).collect::<Vec<_>>().into()),
                element,
                None,
            )
        }

        // One row per element, mirroring `SELECT ... FROM t` over values 1.0 and 2.0.
        let left = make_list(vec![1.0, 2.0]);
        let right = make_list(vec![1.0, 2.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(1));
        Ok(())
    }

    #[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)
        let left = make_struct_list(vec![None]);
        let right = make_struct_list(vec![Some((Some(1), Some(2)))]);

        let result = arrays_overlap_list::<i32>(&left, &right)?;
        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
        assert!(result.is_null(0));
        Ok(())
    }
}