fix: raise InvalidOperationError for mismatched list.contains items on pyarrow-backed backends

src/narwhals/_arrow/utils.py

@@ -9,7 +9,7 @@
 
 from narwhals._compliant import EagerSeriesNamespace
 from narwhals._utils import Implementation, Version, isinstance_or_issubclass
-from narwhals.exceptions import ColumnNotFoundError
+from narwhals.exceptions import ColumnNotFoundError, InvalidOperationError
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator, Mapping, Sequence
@@ -575,10 +575,28 @@ def list_contains(array: ChunkedArrayAny, item: NonNestedLiteral) -> ChunkedArra
     A running count of matches over the flattened values changes within a list iff the
     list holds a match. That keeps this linear, where a group-by per list would sort.
     """
+    list_type = cast("pa.ListType[Any] | pa.LargeListType[Any]", array.type)
+    try:
+        # Probe a single null, so that a mismatched `item` raises even without rows.
+        # Not an empty array: pyarrow skips its timezone check on those.
+        _item_matches(pa.nulls(1, list_type.value_type), item)
+    except (pa.ArrowInvalid, pa.ArrowNotImplementedError, pa.ArrowTypeError) as exc:
+        msg = (
+            f"Unable to compare item of type {type(item)} with list of type {list_type}."
+        )
+        raise InvalidOperationError(msg) from exc
     blocks = [_list_block_contains(block, item) for block in _list_blocks(array)]
     return pa.chunked_array(blocks, pa.bool_())
 
 
+def _item_matches(values: ArrayAny, item: NonNestedLiteral) -> pa.BooleanArray:
+    if item is None:
+        return pc.is_null(values)
+    if isinstance(item, float) and math.isnan(item):
+        return pc.is_nan(values)  # NaN matches NaN, as in Polars.
+    return pc.equal(values, lit(item))
+
+
 def _list_blocks(array: ChunkedArrayAny) -> Iterator[ListArrayAny]:
     """Split `array` into runs of whole lists, of up to about `_LIST_BLOCK_VALUES` values.
 
@@ -615,12 +633,7 @@ def _list_block_contains(block: ListArrayAny, item: NonNestedLiteral) -> pa.Bool
     offsets = pc.subtract(offsets, lit(first, offsets.type))  # type: ignore[arg-type]
     ends, starts = offsets.slice(1), offsets.slice(0, len(block))
 
-    if item is None:
-        matches = pc.is_null(values)
-    elif isinstance(item, float) and math.isnan(item):
-        matches = pc.is_nan(values)  # NaN matches NaN, as in Polars.
-    else:
-        matches = pc.equal(values, lit(item))
+    matches = _item_matches(values, item)
 
     # Counting modulo 2^k stays exact within lists shorter than 2^k, so the narrowest
     # type that fits the longest list is enough.

tests/expr_and_series/list/contains_test.py

@@ -1,22 +1,35 @@
 from __future__ import annotations
 
 import math
-from datetime import date
+from datetime import date, datetime, timezone
 from decimal import Decimal
 from typing import TYPE_CHECKING, Any
 
 import pytest
 
 import narwhals as nw
-from tests.utils import PANDAS_VERSION, POLARS_VERSION, assert_equal_data, pyspark_session
+from narwhals.exceptions import InvalidOperationError
+from tests.utils import (
+    PANDAS_VERSION,
+    POLARS_VERSION,
+    assert_equal_data,
+    maybe_collect,
+    pyspark_session,
+)
 
 if TYPE_CHECKING:
     from narwhals.dtypes import DType
+    from narwhals.typing import NonNestedLiteral
     from tests.utils import Constructor, ConstructorEager
 
 data = {"a": [[2, 2, 3, None, None], None, []]}
 expected = {"a": [True, None, False]}
 
+INT_LIST_DATA = {"a": [[2, 2, 3, None, None], None, [], [None], [1], [0, -1]]}
+
+SQL_BACKENDS = ("duckdb", "ibis", "pyspark", "sqlframe")
+"""Backends which coerce the item to the inner dtype instead of matching polars."""
+
 
 def skip_or_xfail_unsupported(
     request: pytest.FixtureRequest, constructor: Constructor | ConstructorEager
@@ -37,6 +50,137 @@ def test_contains_expr(request: pytest.FixtureRequest, constructor: Constructor)
     assert_equal_data(result, expected)
 
 
+@pytest.mark.parametrize(
+    ("data", "inner", "item", "expected"),
+    [
+        pytest.param(
+            INT_LIST_DATA,
+            nw.Int64(),
+            2.0,
+            [True, None, False, False, False, False],
+            id="float_matches_int",
+        ),
+        pytest.param(
+            INT_LIST_DATA,
+            nw.Int64(),
+            1.5,
+            [False, None, False, False, False, False],
+            id="non_integer",
+        ),
+        pytest.param(
+            INT_LIST_DATA,
+            nw.Int8(),
+            300,
+            [False, None, False, False, False, False],
+            id="overflow",
+        ),
+        pytest.param(
+            {"a": [[1.0, 2.0], [3.0], None, []]},
+            nw.Float64(),
+            1,
+            [True, False, None, False],
+            id="int_matches_float",
+        ),
+    ],
+)
+def test_contains_numeric_coercion_expr(
+    request: pytest.FixtureRequest,
+    constructor: Constructor,
+    data: dict[str, Any],
+    inner: DType,
+    item: float,
+    expected: list[bool | None],
+) -> None:
+    # Mixing numeric kinds is deliberately unspecified, see
+    # https://github.com/narwhals-dev/narwhals/issues/3900. This pins what each backend
+    # does today so a future change is visible, it is not a guarantee.
+    # Polars 2.0 requires an explicit cast rather than lossily coercing to `Float64`.
+    mixes_kinds = isinstance(item, float) != inner.is_float()
+    if "polars" in str(constructor) and POLARS_VERSION >= (2,) and mixes_kinds:
+        request.applymarker(pytest.mark.xfail(reason="polars 2.0 needs a cast"))
+    skip_or_xfail_unsupported(request, constructor)
+    result = nw.from_native(constructor(data)).select(
+        nw.col("a").cast(nw.List(inner)).list.contains(item)
+    )
+    assert_equal_data(result, {"a": expected})
+
+
+@pytest.mark.parametrize(
+    ("data", "inner", "item"),
+    [
+        pytest.param({"a": [[2, 3], [1], None]}, nw.Int64(), True, id="bool_item"),
+        pytest.param({"a": [[2, 3], [1], None]}, nw.Int64(), "2", id="str_item"),
+        pytest.param({"a": [["x"], [], None]}, nw.String(), float("nan"), id="nan_item"),
+        pytest.param(
+            {"a": [[datetime(2020, 1, 1, 1, 2, 3)], [], None]},
+            nw.Datetime("us"),
+            datetime(2020, 1, 1, 1, 2, 3, tzinfo=timezone.utc),
+            id="datetime_time_zone",
+        ),
+    ],
+)
+def test_contains_invalid_item_raises(
+    request: pytest.FixtureRequest,
+    constructor: Constructor,
+    data: dict[str, Any],
+    inner: DType,
+    item: NonNestedLiteral,
+) -> None:
+    if "polars" in str(constructor) and (
+        (isinstance(item, float) and POLARS_VERSION < (1, 28, 0))
+        or (isinstance(item, datetime) and POLARS_VERSION < (2,))
+    ):
+        request.applymarker(pytest.mark.xfail(reason="old polars coerces the item"))
+    if any(backend in str(constructor) for backend in SQL_BACKENDS):
+        request.applymarker(pytest.mark.xfail(reason="mismatched item coerced"))
+    skip_or_xfail_unsupported(request, constructor)
+    df = nw.from_native(constructor(data))
+    with pytest.raises(InvalidOperationError):
+        maybe_collect(df.select(nw.col("a").cast(nw.List(inner)).list.contains(item)))
+
+
+def test_contains_datetime_precision_coerced(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    # Before 2.0, Polars doesn't convert the item to the list's time unit, so it never
+    # finds the match (1.28-1.x raises instead when the list's unit is finer).
+    if "polars" in str(constructor) and POLARS_VERSION < (2,):
+        request.applymarker(pytest.mark.xfail(reason="old polars mismatches units"))
+    if any(backend in str(constructor) for backend in ("pyspark", "sqlframe")):
+        pytest.skip(reason="only microsecond precision is supported")
+    skip_or_xfail_unsupported(request, constructor)
+    df = nw.from_native(constructor({"a": [[datetime(2020, 1, 1, 1, 2, 3)], [], None]}))
+    result = df.select(
+        nw.col("a")
+        .cast(nw.List(nw.Datetime("ms")))
+        .list.contains(datetime(2020, 1, 1, 1, 2, 3))
+    )
+    assert_equal_data(result, {"a": [True, False, None]})
+
+
+def test_contains_invalid_item_raises_without_rows(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    if any(backend in str(constructor) for backend in SQL_BACKENDS):
+        request.applymarker(pytest.mark.xfail(reason="mismatched item coerced"))
+    skip_or_xfail_unsupported(request, constructor)
+    df = nw.from_native(constructor({"a": [[2, 3]]})).filter(nw.col("a").is_null())
+    with pytest.raises(InvalidOperationError):
+        maybe_collect(df.select(nw.col("a").cast(nw.List(nw.Int64())).list.contains("2")))
+
+
+def test_contains_all_null_inner_expr(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    if any(backend in str(constructor) for backend in ("ibis", "pyspark")):
+        pytest.skip(reason="cannot infer the type of an all-null column")
+    skip_or_xfail_unsupported(request, constructor)
+    result = nw.from_native(constructor({"a": [[None, None]]})).select(
+        nw.col("a").cast(nw.List(nw.Int64())).list.contains(1)
+    )
+    assert_equal_data(result, {"a": [False]})
+
+
 def test_contains_no_match_with_null_elements_expr(
     request: pytest.FixtureRequest, constructor: Constructor
 ) -> None: