fix(datafusion): treat signed zeros as equal in array_has

[jonasdedden]

Problem

#6236 makes filters treat -0.0 and 0.0 as one value when one side is a zero literal, but array_has still compares by encoding (second part of #9316):

ds.to_table(filter="array_has(l, 0.0)")   # misses lists holding only -0.0
ds.to_table(filter="array_has(l, -0.0)")  # misses lists holding only +0.0

array_has uses Arrow's total-order eq kernel (compare_with_eq), so the two encodings compare unequal. This affects Float32/Float64 (and Float16 via the same helper), nested fields (s.x), computed haystacks (array_append(l, 1.0)), computed needles (-1.0 * 0.0), and integer needles against float lists (which coerce to float). Aliases list_has / array_contains / list_contains resolve to the same kernel.

Chosen rule

Extend the existing signed_zero planner rewrite, the same way 0.0 IN (a, b) is already handled:

array_has(l, 0.0)  ->  array_has(l, -0.0) OR array_has(l, 0.0)

• Applies to any haystack expression (column, nested field, computed list), not just bare columns.
• Needle must be a floating-point zero literal post-coercion (Float16/32/64, either encoding). Integer, NULL, NaN, and non-literal needles are left alone.
• Canonical form is negative-first, so both encodings map to the same disjunction and a second optimize_expr deduplicates rather than growing (same fixed-point pattern as the existing IN-list dedup).
• normalize_zero_comparisons folds array_has operands first, so computed zeros (-1.0 * 0.0, 1.0 - 1.0) present as literals before the rewrite; otherwise the simplifier would fold the whole probe by total order first.
• NULL semantics preserved: array_has(NULL, 0.0) rewrites to NULL OR NULL = NULL; a null element never matches, so OR of two falses stays false.

Alternatives considered

• Engine-level equality fix (change array_has/eq kernel to IEEE equality): would also fix column needles (array_has(l, m[0])), but that is the column-to-column case owned separately, and it diverges from fix: make float filters treat -0.0 and 0.0 as the same value #6236's planner-rewrite approach which deliberately preserves bit-pattern index keying. array_has never uses a scalar index anyway (BTREE/BITMAP reject list columns; plans stay as plain filters), so the OR keeps the same plan shape.
• Widening the haystack list: impossible when the haystack is a column; the needle-side OR is the only uniform spelling.
• array_has_any / array_has_all: left out (see limitations).

Compatibility

• No API change. Filters that already matched both encodings (non-zero needles, integer lists) are untouched: only float-zero needles rewrite.
• Integer semantics preserved: array_has(int_list, 0) never presents as a floating zero post-coercion, so it stays a single probe. array_has(float_list, 0) coerces 0 to float and correctly becomes the OR.
• Plan shape: array_has stays a plain filter (full_filter/refine_filter); no new index path. Verified list columns cannot take BTREE/BITMAP indices.

Test results

• cargo test -p lance-datafusion --lib: 189 passed (76 signed_zero, including 7 new array_has tests).
• cargo clippy -p lance-datafusion --lib: clean. cargo fmt --check -p lance-datafusion: clean.
• Raw SQL on pylance 13.0.0b4 (before fix): array_has(l, 0.0) -> [1, 5], array_has(l, -0.0) -> [0, 5]; the rewrite output array_has(l, -0.0) OR array_has(l, 0.0) -> [0, 1, 5] (correct) for Float64/Float32/nested, NOT (...) correctly excludes the null-list row, non-zero needles unchanged. This validates the emitted form on the current engine; the new unit tests validate the planner emits exactly that form (Float64/32 via coercion, nested/computed/aliases, folded needles, idempotent re-optimization).

Remaining limitations

• Related to bug: -0.0 and 0.0 still compare unequal between columns and in array_has #9316 (array_has part only; does not close it — column-to-column -0.0 = 0.0 is separate).
• Column needles (array_has(l, m)) stay sign-sensitive; owned by the column-column fix.
• array_has_any / array_has_all with zero-containing needle lists are not rewritten. ANY could widen its needle list; ALL needs an OR-per-zero expansion and is not a simple widening. Left as follow-up.

[lance-gatekeeper]

❌ Gate recommendation: request changes.

The signed-zero correction is sound for stable haystacks, but the current OR lowering violates DataFusion’s volatile-expression contract by evaluating the haystack twice and can silently drop rows. Preserve one haystack evaluation; a single array_has_any(haystack, [-0.0, 0.0]) probe verified the intended semantics, or the equality kernel can normalize signed zeros directly.

[lance-gatekeeper]

Cloning the haystack into both OR arms changes the semantics of supported volatile expressions: DataFusion permits a Volatility::Volatile function to return a different value on each evaluation. I ran a volatile list UDF that returns [+0.0] on its first call and [-0.0] on its second. Optimizing array_has(volatile_zeros(), 0.0) produced the two calls shown below and returned false for all 10,000 rows even though every UDF result contained a zero. Preserve a single haystack evaluation—for example, lower this to one array_has_any(haystack, [-0.0, 0.0]) probe (the same experiment then made one call and returned all 10,000 rows), or normalize equality in the kernel.

Reproducer run on this head

let schema = Arc::new(Schema::empty());
let planner = Planner::new(schema.clone());
let calls = Arc::new(AtomicUsize::new(0));
let calls_in_udf = Arc::clone(&calls);
let list_type = DataType::List(Arc::new(Field::new("item", DataType::Float64, true)));
let volatile_zeros = create_udf(
    "volatile_zeros",
    vec![],
    list_type,
    Volatility::Volatile,
    Arc::new(move |_| {
        let value = if calls_in_udf.fetch_add(1, Ordering::SeqCst) % 2 == 0 {
            0.0
        } else {
            -0.0
        };
        let list = ListArray::from_iter_primitive::<Float64Type, _, _>(
            [Some(vec![Some(value)])],
        );
        Ok(ColumnarValue::Scalar(ScalarValue::List(Arc::new(list))))
    }),
);
let optimized = planner.optimize_expr(array_has(volatile_zeros.call(vec![]), lit(0.0)))?;
let batch = RecordBatch::try_new_with_options(
    schema,
    vec![],
    &RecordBatchOptions::new().with_row_count(Some(10_000)),
)?;
let values = planner
    .create_physical_expr(&optimized)?
    .evaluate(&batch)?
    .into_array(10_000)?;
let values = values.as_any().downcast_ref::<BooleanArray>().unwrap();
assert_eq!(calls.load(Ordering::SeqCst), 1); // observed: 2
assert_eq!(values.true_count(), 10_000); // observed: 0

Observed optimized expression:

array_has(volatile_zeros(), Float64(-0)) OR array_has(volatile_zeros(), Float64(0))
calls=2 true=0 false=10000

[jonasdedden]

Closing in favor of #9323.