Repository: narwhals-dev/narwhals

Issue:
[Enh]: Support for anti-join
### We would like to learn about your use case. For example, if this feature is needed  to adopt Narwhals in an open source project, could you please enter the link to it below?


_No response_

### Please describe the purpose of the new feature or describe the problem to solve.

Hands down one of my favorite feature 😂
The fact that polars has native support is really convenient

### Suggest a solution if possible.

It should look something like the following:

```py
(df
.merge(other, how='outer', indicator = True)
.loc[lambda t: t["_merge"]=="left_only"]
.drop("_merge", axis=1)
)
```

We can decide to name the indicator column with another non-default value

### If you have tried alternatives, please describe them below.

_No response_

### Additional information that may help us understand your needs.

_No response_

Discussion:
FBruzzesi: It seems I can replicate the error when running with pytest, but can't on ipython  🤔 I am getting somewhat confused

```py
import modin.pandas as mpd
import narwhals as nw

df = mpd.DataFrame({"a": [1, 3, 2], "b": [4, 4, 6], "z": [7.0, 8, 9]})
df_ = nw.from_native(df)
other = df_.filter(nw.col("b") < 6)

result = df_.join(other, how="anti", left_on=["a", "b"], right_on=["a", "b"])
nw.to_native(result)
```

![image](https://github.com/narwhals-dev/narwhals/assets/42817048/33c4174f-1439-4e3f-8143-be970434a7aa)

MarcoGorelli: I think in ci we have `modin[dask]`, looks like you have `modin[ray]`?
FBruzzesi: I tried both backends, ending up with the same result:

- failing tests
- non failing ipython

Patch:
diff --git a/README.md b/README.md
index 1ee3056e..c0a28537 100644
--- a/README.md
+++ b/README.md
@@ -13,7 +13,8 @@ Extremely lightweight and extensible compatibility layer between Polars, pandas,
 
 - [Read the documentation](https://narwhals-dev.github.io/narwhals/)
 - [Chat with us on Discord!](https://discord.gg/V3PqtB4VA4)
-- [Join our community call](https://calendar.google.com/calendar/embed?src=27ff6dc5f598c1d94c1f6e627a1aaae680e2fac88f848bda1f2c7946ae74d5ab%40group.calendar.google.com&ctz=Europe%2FLondon).
+- [Join our community call](https://calendar.google.com/calendar/embed?src=27ff6dc5f598c1d94c1f6e627a1aaae680e2fac88f848bda1f2c7946ae74d5ab%40group.calendar.google.com&ctz=Europe%2FLondon)
+- [Read the contributing guide](https://github.com/narwhals-dev/narwhals/blob/main/CONTRIBUTING.md)
 
 Seamlessly support all, without depending on any!
 
@@ -26,14 +27,18 @@ Seamlessly support all, without depending on any!
 - ✅ **100% branch coverage**, tested against pandas and Polars nightly builds
 - ✅ **Negligible overhead**, see [overhead](https://narwhals-dev.github.io/narwhals/overhead/)
 
-## Used by
+## Used by / integrates with
 
 Join the party!
 
+- [Hamilton](https://github.com/DAGWorks-Inc/hamilton/tree/main/examples/narwhals)
 - [scikit-lego](https://github.com/koaning/scikit-lego)
 - [scikit-playtime](https://github.com/koaning/scikit-playtime)
 - [timebasedcv](https://github.com/FBruzzesi/timebasedcv)
 
+Feel free to add your project to the list if it's missing, and/or
+[chat with us on Discord](https://discord.gg/V3PqtB4VA4) if you'd like any support.
+
 ## Installation
 
 - pip (recommended, as it's the most up-to-date)
diff --git a/docs/installation.md b/docs/installation.md
index fae46721..4a98fe7b 100644
--- a/docs/installation.md
+++ b/docs/installation.md
@@ -11,6 +11,6 @@ Then, if you start the Python REPL and see the following:
 ```python
 >>> import narwhals
 >>> narwhals.__version__
-'0.9.23'
+'0.9.24'
 ```
 then installation worked correctly!
diff --git a/narwhals/__init__.py b/narwhals/__init__.py
index 406d6ee7..5d197924 100644
--- a/narwhals/__init__.py
+++ b/narwhals/__init__.py
@@ -42,7 +42,7 @@ from narwhals.utils import maybe_align_index
 from narwhals.utils import maybe_convert_dtypes
 from narwhals.utils import maybe_set_index
 
-__version__ = "0.9.23"
+__version__ = "0.9.24"
 
 __all__ = [
     "selectors",
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index 9cf65e72..c3f2c0fb 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -2,10 +2,12 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Sequence
 from typing import overload
 
 from narwhals._arrow.utils import translate_dtype
 from narwhals._pandas_like.utils import evaluate_into_exprs
+from narwhals.dependencies import get_numpy
 from narwhals.dependencies import get_pyarrow
 
 if TYPE_CHECKING:
@@ -65,17 +67,22 @@ class ArrowDataFrame:
             return ArrowSeries(self._dataframe[item], name=item)
 
         elif isinstance(item, slice):
-            from narwhals._arrow.dataframe import ArrowDataFrame
-
             if item.step is not None and item.step != 1:
                 msg = "Slicing with step is not supported on PyArrow tables"
                 raise NotImplementedError(msg)
             start = item.start or 0
             stop = item.stop or len(self._dataframe)
-            return ArrowDataFrame(
+            return self._from_dataframe(
                 self._dataframe.slice(item.start, stop - start),
             )
 
+        elif isinstance(item, Sequence) or (
+            (np := get_numpy()) is not None
+            and isinstance(item, np.ndarray)
+            and item.ndim == 1
+        ):
+            return self._from_dataframe(self._dataframe.take(item))
+
         else:  # pragma: no cover
             msg = f"Expected str or slice, got: {type(item)}"
             raise TypeError(msg)
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index 80d0740d..77516813 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -3,7 +3,6 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 from typing import Callable
 
-from narwhals._arrow.series import ArrowSeries
 from narwhals._pandas_like.utils import reuse_series_implementation
 from narwhals._pandas_like.utils import reuse_series_namespace_implementation
 
@@ -12,6 +11,8 @@ if TYPE_CHECKING:
 
     from narwhals._arrow.dataframe import ArrowDataFrame
     from narwhals._arrow.namespace import ArrowNamespace
+    from narwhals._arrow.series import ArrowSeries
+    from narwhals.dtypes import DType
 
 
 class ArrowExpr:
@@ -43,6 +44,8 @@ class ArrowExpr:
 
     @classmethod
     def from_column_names(cls: type[Self], *column_names: str) -> Self:
+        from narwhals._arrow.series import ArrowSeries
+
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             return [
                 ArrowSeries(
@@ -65,9 +68,18 @@ class ArrowExpr:
 
         return ArrowNamespace()
 
+    def cast(self, dtype: DType) -> Self:
+        return reuse_series_implementation(self, "cast", dtype)  # type: ignore[type-var]
+
     def cum_sum(self) -> Self:
         return reuse_series_implementation(self, "cum_sum")  # type: ignore[type-var]
 
+    def any(self) -> Self:
+        return reuse_series_implementation(self, "any", returns_scalar=True)  # type: ignore[type-var]
+
+    def all(self) -> Self:
+        return reuse_series_implementation(self, "all", returns_scalar=True)  # type: ignore[type-var]
+
     @property
     def dt(self) -> ArrowExprDateTimeNamespace:
         return ArrowExprDateTimeNamespace(self)
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index bc7182b7..266fbc68 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
+from typing import Any
 from typing import Iterable
 
 from narwhals import dtypes
@@ -66,6 +67,14 @@ class ArrowNamespace:
             output_names=None,
         )
 
+    def _create_series_from_scalar(self, value: Any, series: ArrowSeries) -> ArrowSeries:
+        from narwhals._arrow.series import ArrowSeries
+
+        return ArrowSeries.from_iterable(
+            [value],
+            name=series.name,
+        )
+
     # --- not in spec ---
     def __init__(self) -> None: ...
 
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index 0044e3da..dd0791f4 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -2,8 +2,13 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Iterable
 
+from narwhals._arrow.namespace import ArrowNamespace
+from narwhals._arrow.utils import reverse_translate_dtype
 from narwhals._arrow.utils import translate_dtype
+from narwhals._pandas_like.utils import native_series_from_iterable
+from narwhals.dependencies import get_pyarrow
 from narwhals.dependencies import get_pyarrow_compute
 
 if TYPE_CHECKING:
@@ -29,9 +34,21 @@ class ArrowSeries:
             name=self._name,
         )
 
+    @classmethod
+    def from_iterable(cls: type[Self], data: Iterable[Any], name: str) -> Self:
+        return cls(
+            native_series_from_iterable(
+                data, name=name, index=None, implementation="arrow"
+            ),
+            name=name,
+        )
+
     def __len__(self) -> int:
         return len(self._series)
 
+    def __narwhals_namespace__(self) -> ArrowNamespace:
+        return ArrowNamespace()
+
     @property
     def name(self) -> str:
         return self._name
@@ -62,6 +79,23 @@ class ArrowSeries:
         pc = get_pyarrow_compute()
         return self._from_series(pc.cumulative_sum(self._series))
 
+    def any(self) -> bool:
+        pc = get_pyarrow_compute()
+        return pc.any(self._series)  # type: ignore[no-any-return]
+
+    def all(self) -> bool:
+        pc = get_pyarrow_compute()
+        return pc.all(self._series)  # type: ignore[no-any-return]
+
+    def is_empty(self) -> bool:
+        return len(self) == 0
+
+    def cast(self, dtype: DType) -> Self:
+        pc = get_pyarrow_compute()
+        ser = self._series
+        dtype = reverse_translate_dtype(dtype)
+        return self._from_series(pc.cast(ser, dtype))
+
     @property
     def shape(self) -> tuple[int]:
         return (len(self._series),)
@@ -70,6 +104,10 @@ class ArrowSeries:
     def dt(self) -> ArrowSeriesDateTimeNamespace:
         return ArrowSeriesDateTimeNamespace(self)
 
+    @property
+    def cat(self) -> ArrowSeriesCatNamespace:
+        return ArrowSeriesCatNamespace(self)
+
 
 class ArrowSeriesDateTimeNamespace:
     def __init__(self, series: ArrowSeries) -> None:
@@ -82,3 +120,17 @@ class ArrowSeriesDateTimeNamespace:
         # https://arrow.apache.org/docs/python/generated/pyarrow.compute.strftime.html
         format = format.replace("%S.%f", "%S").replace("%S%.f", "%S")
         return self._series._from_series(pc.strftime(self._series._series, format))
+
+
+class ArrowSeriesCatNamespace:
+    def __init__(self, series: ArrowSeries) -> None:
+        self._series = series
+
+    def get_categories(self) -> ArrowSeries:
+        pa = get_pyarrow()
+        ca = self._series._series
+        # todo: this looks potentially expensive - is there no better way?
+        out = pa.chunked_array(
+            [pa.concat_arrays([x.dictionary for x in ca.chunks]).unique()]
+        )
+        return self._series._from_series(out)
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index 9b129a32..fe60e2fd 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -1,7 +1,10 @@
+from __future__ import annotations
+
 from typing import Any
 
 from narwhals import dtypes
 from narwhals.dependencies import get_pyarrow
+from narwhals.utils import isinstance_or_issubclass
 
 
 def translate_dtype(dtype: Any) -> dtypes.DType:
@@ -45,3 +48,48 @@ def translate_dtype(dtype: Any) -> dtypes.DType:
     if pa.types.is_dictionary(dtype):
         return dtypes.Categorical()
     raise AssertionError
+
+
+def reverse_translate_dtype(dtype: dtypes.DType | type[dtypes.DType]) -> Any:
+    from narwhals import dtypes
+
+    pa = get_pyarrow()
+
+    if isinstance_or_issubclass(dtype, dtypes.Float64):
+        return pa.float64()
+    if isinstance_or_issubclass(dtype, dtypes.Float32):
+        return pa.float32()
+    if isinstance_or_issubclass(dtype, dtypes.Int64):
+        return pa.int64()
+    if isinstance_or_issubclass(dtype, dtypes.Int32):
+        return pa.int32()
+    if isinstance_or_issubclass(dtype, dtypes.Int16):
+        return pa.int16()
+    if isinstance_or_issubclass(dtype, dtypes.Int8):
+        return pa.int8()
+    if isinstance_or_issubclass(dtype, dtypes.UInt64):
+        return pa.uint64()
+    if isinstance_or_issubclass(dtype, dtypes.UInt32):
+        return pa.uint32()
+    if isinstance_or_issubclass(dtype, dtypes.UInt16):
+        return pa.uint16()
+    if isinstance_or_issubclass(dtype, dtypes.UInt8):
+        return pa.uint8()
+    if isinstance_or_issubclass(dtype, dtypes.String):
+        return pa.string()
+    if isinstance_or_issubclass(dtype, dtypes.Boolean):
+        return pa.bool_()
+    if isinstance_or_issubclass(dtype, dtypes.Categorical):
+        # todo: what should the key be? let's keep it consistent
+        # with Polars for now
+        return pa.dictionary(pa.uint32(), pa.string())
+    if isinstance_or_issubclass(dtype, dtypes.Datetime):
+        # Use Polars' default
+        return pa.timestamp("us")
+    if isinstance_or_issubclass(dtype, dtypes.Duration):
+        # Use Polars' default
+        return pa.duration("us")
+    if isinstance_or_issubclass(dtype, dtypes.Date):
+        return pa.date32()
+    msg = f"Unknown dtype: {dtype}"  # pragma: no cover
+    raise AssertionError(msg)
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index c58f08ce..e4886195 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -6,24 +6,25 @@ from typing import Any
 from typing import Iterable
 from typing import Iterator
 from typing import Literal
+from typing import Sequence
 from typing import overload
 
 from narwhals._pandas_like.expr import PandasExpr
 from narwhals._pandas_like.utils import create_native_series
 from narwhals._pandas_like.utils import evaluate_into_exprs
+from narwhals._pandas_like.utils import generate_unique_token
 from narwhals._pandas_like.utils import horizontal_concat
 from narwhals._pandas_like.utils import translate_dtype
 from narwhals._pandas_like.utils import validate_dataframe_comparand
 from narwhals._pandas_like.utils import validate_indices
 from narwhals.dependencies import get_cudf
 from narwhals.dependencies import get_modin
+from narwhals.dependencies import get_numpy
 from narwhals.dependencies import get_pandas
 from narwhals.utils import flatten
 from narwhals.utils import parse_version
 
 if TYPE_CHECKING:
-    from collections.abc import Sequence
-
     from typing_extensions import Self
 
     from narwhals._pandas_like.group_by import PandasGroupBy
@@ -101,12 +102,18 @@ class PandasDataFrame:
                 implementation=self._implementation,
             )
 
-        elif isinstance(item, slice):
+        elif isinstance(item, (slice, Sequence)):
             from narwhals._pandas_like.dataframe import PandasDataFrame
 
             return PandasDataFrame(
                 self._dataframe.iloc[item], implementation=self._implementation
             )
+        elif (
+            (np := get_numpy()) is not None
+            and isinstance(item, np.ndarray)
+            and item.ndim == 1
+        ):
+            return self._from_dataframe(self._dataframe.iloc[item])
 
         else:  # pragma: no cover
             msg = f"Expected str or slice, got: {type(item)}"
@@ -280,7 +287,7 @@ class PandasDataFrame:
         self,
         other: Self,
         *,
-        how: Literal["left", "inner", "outer", "cross"] = "inner",
+        how: Literal["left", "inner", "outer", "cross", "anti"] = "inner",
         left_on: str | list[str] | None = None,
         right_on: str | list[str] | None = None,
     ) -> Self:
@@ -295,27 +302,9 @@ class PandasDataFrame:
                 and (pd := get_pandas()) is not None
                 and parse_version(pd.__version__) < parse_version("1.4.0")
             ):
-
-                def generate_unique_token(
-                    n_bytes: int, columns: list[str]
-                ) -> str:  # pragma: no cover
-                    import secrets
-
-                    counter = 0
-                    while True:
-                        token = secrets.token_hex(n_bytes)
-                        if token not in columns:
-                            return token
-
-                        counter += 1
-                        if counter > 100:  # pragma: no cover
-                            msg = (
-                                "Internal Error: Narwhals was not able to generate a column name to perform cross "
-                                "join operation"
-                            )
-                            raise AssertionError(msg)
-
-                key_token = generate_unique_token(8, self.columns)
+                key_token = generate_unique_token(
+                    n_bytes=8, columns=[*self.columns, *other.columns]
+                )
 
                 return self._from_dataframe(
                     self._dataframe.assign(**{key_token: 0}).merge(
@@ -335,6 +324,31 @@ class PandasDataFrame:
                     ),
                 )
 
+        if how == "anti":
+            indicator_token = generate_unique_token(
+                n_bytes=8, columns=[*self.columns, *other.columns]
+            )
+
+            other = (
+                other._dataframe.loc[:, right_on]
+                .rename(  # rename to avoid creating extra columns in join
+                    columns=dict(zip(right_on, left_on))  # type: ignore[arg-type]
+                )
+                .drop_duplicates(ignore_index=True)
+            )
+            return self._from_dataframe(
+                self._dataframe.merge(
+                    other,
+                    how="outer",
+                    indicator=indicator_token,
+                    left_on=left_on,
+                    right_on=left_on,
+                )
+                .loc[lambda t: t[indicator_token] == "left_only"]
+                .drop(columns=[indicator_token])
+                .reset_index(drop=True)
+            )
+
         return self._from_dataframe(
             self._dataframe.merge(
                 other._dataframe,
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index 1ff880e5..3f00142b 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import secrets
 from copy import copy
 from typing import TYPE_CHECKING
 from typing import Any
@@ -10,6 +11,7 @@ from narwhals.dependencies import get_cudf
 from narwhals.dependencies import get_modin
 from narwhals.dependencies import get_numpy
 from narwhals.dependencies import get_pandas
+from narwhals.dependencies import get_pyarrow
 from narwhals.utils import flatten
 from narwhals.utils import isinstance_or_issubclass
 from narwhals.utils import parse_version
@@ -365,6 +367,9 @@ def native_series_from_iterable(
         mpd = get_modin()
 
         return mpd.Series(data, name=name, index=index)
+    if implementation == "arrow":
+        pa = get_pyarrow()
+        return pa.chunked_array([data])
     msg = f"Unknown implementation: {implementation}"  # pragma: no cover
     raise TypeError(msg)  # pragma: no cover
 
@@ -608,3 +613,31 @@ def int_dtype_mapper(dtype: Any) -> str:
     if str(dtype).lower() != str(dtype):  # pragma: no cover
         return "Int64"
     return "int64"
+
+
+def generate_unique_token(n_bytes: int, columns: list[str]) -> str:  # pragma: no cover
+    """Generates a unique token of specified n_bytes that is not present in the given list of columns.
+
+    Arguments:
+        n_bytes : The number of bytes to generate for the token.
+        columns : The list of columns to check for uniqueness.
+
+    Returns:
+        A unique token that is not present in the given list of columns.
+
+    Raises:
+        AssertionError: If a unique token cannot be generated after 100 attempts.
+    """
+    counter = 0
+    while True:
+        token = secrets.token_hex(n_bytes)
+        if token not in columns:
+            return token
+
+        counter += 1
+        if counter > 100:
+            msg = (
+                "Internal Error: Narwhals was not able to generate a column name to perform cross "
+                "join operation"
+            )
+            raise AssertionError(msg)
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index 9e770774..b69688d8 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -13,6 +13,7 @@ from narwhals._arrow.dataframe import ArrowDataFrame
 from narwhals._pandas_like.dataframe import PandasDataFrame
 from narwhals.dependencies import get_cudf
 from narwhals.dependencies import get_modin
+from narwhals.dependencies import get_numpy
 from narwhals.dependencies import get_pandas
 from narwhals.dependencies import get_polars
 from narwhals.dependencies import get_pyarrow
@@ -174,11 +175,11 @@ class BaseFrame:
         self,
         other: Self,
         *,
-        how: Literal["inner", "cross"] = "inner",
+        how: Literal["inner", "cross", "anti"] = "inner",
         left_on: str | list[str] | None = None,
         right_on: str | list[str] | None = None,
     ) -> Self:
-        _supported_joins = {"inner", "cross"}
+        _supported_joins = {"inner", "cross", "anti"}
         if how not in _supported_joins:
             msg = f"Only the following join stragies are supported: {_supported_joins}"
             raise NotImplementedError(msg)
@@ -388,19 +389,26 @@ class DataFrame(BaseFrame):
         """
         return self._dataframe.shape  # type: ignore[no-any-return]
 
+    @overload
+    def __getitem__(self, item: Sequence[int]) -> Series: ...
+
     @overload
     def __getitem__(self, item: str) -> Series: ...
 
     @overload
     def __getitem__(self, item: slice) -> DataFrame: ...
 
-    def __getitem__(self, item: str | slice) -> Series | DataFrame:
+    def __getitem__(self, item: str | slice | Sequence[int]) -> Series | DataFrame:
         if isinstance(item, str):
             from narwhals.series import Series
 
             return Series(self._dataframe[item])
 
-        elif isinstance(item, (range, slice)):
+        elif isinstance(item, (Sequence, slice)) or (
+            (np := get_numpy()) is not None
+            and isinstance(item, np.ndarray)
+            and item.ndim == 1
+        ):
             return self._from_dataframe(self._dataframe[item])
 
         else:
@@ -1447,7 +1455,7 @@ class DataFrame(BaseFrame):
         self,
         other: Self,
         *,
-        how: Literal["inner", "cross"] = "inner",
+        how: Literal["inner", "cross", "anti"] = "inner",
         left_on: str | list[str] | None = None,
         right_on: str | list[str] | None = None,
     ) -> Self:
@@ -1461,6 +1469,7 @@ class DataFrame(BaseFrame):
 
                   * *inner*: Returns rows that have matching values in both tables
                   * *cross*: Returns the Cartesian product of rows from both tables
+                  * *anti*: Filter rows that do not have a match in the right table.
 
             left_on: Name(s) of the left join column(s).
 
@@ -2858,7 +2867,7 @@ class LazyFrame(BaseFrame):
         self,
         other: Self,
         *,
-        how: Literal["inner", "cross"] = "inner",
+        how: Literal["inner", "cross", "anti"] = "inner",
         left_on: str | list[str] | None = None,
         right_on: str | list[str] | None = None,
     ) -> Self:
@@ -2872,6 +2881,7 @@ class LazyFrame(BaseFrame):
 
                   * *inner*: Returns rows that have matching values in both tables
                   * *cross*: Returns the Cartesian product of rows from both tables
+                  * *anti*: Filter rows that do not have a match in the right table.
 
             left_on: Join column of the left DataFrame.
 
diff --git a/pyproject.toml b/pyproject.toml
index cd2bdb6a..dcec0e81 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "narwhals"
-version = "0.9.23"
+version = "0.9.24"
 authors = [
   { name="Marco Gorelli", email="33491632+MarcoGorelli@users.noreply.github.com" },
 ]
diff --git a/utils/check_backend_completeness.py b/utils/check_backend_completeness.py
index a5352dbf..dfb8853c 100644
--- a/utils/check_backend_completeness.py
+++ b/utils/check_backend_completeness.py
@@ -37,10 +37,6 @@ MISSING = [
     "DataFrame.with_columns",
     "DataFrame.with_row_index",
     "DataFrame.write_parquet",
-    "Series.all",
-    "Series.any",
-    "Series.cast",
-    "Series.cat",
     "Series.diff",
     "Series.drop_nulls",
     "Series.fill_null",
@@ -49,7 +45,6 @@ MISSING = [
     "Series.head",
     "Series.is_between",
     "Series.is_duplicated",
-    "Series.is_empty",
     "Series.is_first_distinct",
     "Series.is_in",
     "Series.is_last_distinct",
@@ -130,6 +125,18 @@ if __name__ == "__main__":
     missing.extend([x for x in pd_methods if x not in pa_methods and x not in MISSING])
 
     if missing:
-        print(sorted(missing))
+        print(
+            "The following have not been implemented for the Arrow backend: ",
+            sorted(missing),
+        )
         sys.exit(1)
+
+    no_longer_missing = [x for x in MISSING if x in pa_methods and x in pd_methods]
+    if no_longer_missing:
+        print(
+            "Please remove the following from MISSING in utils/check_backend_completeness.py: ",
+            sorted(no_longer_missing),
+        )
+        sys.exit(1)
+
     sys.exit(0)


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 5361dbd2..9a2fa2f3 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,4 +1,3 @@
-import os
 from typing import Any
 from typing import Callable
 
@@ -58,7 +57,7 @@ if parse_version(pd.__version__) >= parse_version("1.5.0"):
 else:  # pragma: no cover
     params = [pandas_constructor]
 params.append(polars_constructor)
-if os.environ.get("CI") and get_modin() is not None:  # pragma: no cover
+if get_modin() is not None:  # pragma: no cover
     params.append(modin_constructor)
 
 
@@ -104,7 +103,7 @@ if parse_version(pd.__version__) >= parse_version("1.5.0"):
 else:  # pragma: no cover
     params_series = [pandas_series_constructor]
 params_series.append(polars_series_constructor)
-if os.environ.get("CI") and get_modin() is not None:  # pragma: no cover
+if get_modin() is not None:  # pragma: no cover
     params_series.append(modin_series_constructor)
 
 
diff --git a/tests/expr/any_all_test.py b/tests/expr/any_all_test.py
index 05aef972..c5bb88cb 100644
--- a/tests/expr/any_all_test.py
+++ b/tests/expr/any_all_test.py
@@ -1,12 +1,20 @@
 from typing import Any
 
+import pyarrow as pa
+import pytest
+
 import narwhals as nw
+from narwhals.utils import parse_version
 from tests.utils import compare_dicts
 
 
-def test_any_all(constructor: Any) -> None:
+def test_any_all(constructor_with_pyarrow: Any, request: Any) -> None:
+    if "table" in str(constructor_with_pyarrow) and parse_version(
+        pa.__version__
+    ) < parse_version("12.0.0"):  # pragma: no cover
+        request.applymarker(pytest.mark.xfail)
     df = nw.from_native(
-        constructor(
+        constructor_with_pyarrow(
             {
                 "a": [True, False, True],
                 "b": [True, True, True],
@@ -14,9 +22,9 @@ def test_any_all(constructor: Any) -> None:
             }
         )
     )
-    result = nw.to_native(df.select(nw.all().all()))
+    result = df.select(nw.all().all())
     expected = {"a": [False], "b": [True], "c": [False]}
     compare_dicts(result, expected)
-    result = nw.to_native(df.select(nw.all().any()))
+    result = df.select(nw.all().any())
     expected = {"a": [True], "b": [True], "c": [False]}
     compare_dicts(result, expected)
diff --git a/tests/expr/cast_test.py b/tests/expr/cast_test.py
new file mode 100644
index 00000000..7e647296
--- /dev/null
+++ b/tests/expr/cast_test.py
@@ -0,0 +1,108 @@
+from typing import Any
+
+import pyarrow as pa
+import pytest
+
+import narwhals as nw
+from narwhals.utils import parse_version
+
+
+def test_cast(constructor_with_pyarrow: Any, request: Any) -> None:
+    if "table" in str(constructor_with_pyarrow) and parse_version(
+        pa.__version__
+    ) <= parse_version("12.0.0"):  # pragma: no cover
+        request.applymarker(pytest.mark.xfail)
+    data = {
+        "a": [1],
+        "b": [1],
+        "c": [1],
+        "d": [1],
+        "e": [1],
+        "f": [1],
+        "g": [1],
+        "h": [1],
+        "i": [1],
+        "j": [1],
+        "k": ["1"],
+        "l": [1],
+        "m": [True],
+        "n": [True],
+        "o": ["a"],
+        "p": [1],
+    }
+    schema = {
+        "a": nw.Int64,
+        "b": nw.Int32,
+        "c": nw.Int16,
+        "d": nw.Int8,
+        "e": nw.UInt64,
+        "f": nw.UInt32,
+        "g": nw.UInt16,
+        "h": nw.UInt8,
+        "i": nw.Float64,
+        "j": nw.Float32,
+        "k": nw.String,
+        "l": nw.Datetime,
+        "m": nw.Boolean,
+        "n": nw.Boolean,
+        "o": nw.Categorical,
+        "p": nw.Int64,
+    }
+    df = nw.from_native(constructor_with_pyarrow(data), eager_only=True).select(
+        nw.col(key).cast(value) for key, value in schema.items()
+    )
+    result = df.select(
+        nw.col("a").cast(nw.Int32),
+        nw.col("b").cast(nw.Int16),
+        nw.col("c").cast(nw.Int8),
+        nw.col("d").cast(nw.Int64),
+        nw.col("e").cast(nw.UInt32),
+        nw.col("f").cast(nw.UInt16),
+        nw.col("g").cast(nw.UInt8),
+        nw.col("h").cast(nw.UInt64),
+        nw.col("i").cast(nw.Float32),
+        nw.col("j").cast(nw.Float64),
+        nw.col("k").cast(nw.String),
+        nw.col("l").cast(nw.Datetime),
+        nw.col("m").cast(nw.Int8),
+        nw.col("n").cast(nw.Int8),
+        nw.col("o").cast(nw.String),
+        nw.col("p").cast(nw.Duration),
+    )
+    expected = {
+        "a": nw.Int32,
+        "b": nw.Int16,
+        "c": nw.Int8,
+        "d": nw.Int64,
+        "e": nw.UInt32,
+        "f": nw.UInt16,
+        "g": nw.UInt8,
+        "h": nw.UInt64,
+        "i": nw.Float32,
+        "j": nw.Float64,
+        "k": nw.String,
+        "l": nw.Datetime,
+        "m": nw.Int8,
+        "n": nw.Int8,
+        "o": nw.String,
+        "p": nw.Duration,
+    }
+    assert result.schema == expected
+    result = df.select(
+        df["a"].cast(nw.Int32),
+        df["b"].cast(nw.Int16),
+        df["c"].cast(nw.Int8),
+        df["d"].cast(nw.Int64),
+        df["e"].cast(nw.UInt32),
+        df["f"].cast(nw.UInt16),
+        df["g"].cast(nw.UInt8),
+        df["h"].cast(nw.UInt64),
+        df["i"].cast(nw.Float32),
+        df["j"].cast(nw.Float64),
+        df["k"].cast(nw.String),
+        df["l"].cast(nw.Datetime),
+        df["m"].cast(nw.Int8),
+        df["n"].cast(nw.Int8),
+        df["o"].cast(nw.String),
+        df["p"].cast(nw.Duration),
+    )
diff --git a/tests/expr/cat/get_categories_test.py b/tests/expr/cat/get_categories_test.py
index fabce934..66088173 100644
--- a/tests/expr/cat/get_categories_test.py
+++ b/tests/expr/cat/get_categories_test.py
@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from typing import Any
 
+import pyarrow as pa
+
 import narwhals as nw
 from tests.utils import compare_dicts
 
@@ -16,3 +18,15 @@ def test_get_categories(constructor: Any) -> None:
     compare_dicts(result, expected)
     result = df.select(df["a"].cat.get_categories())
     compare_dicts(result, expected)
+
+
+def test_get_categories_pyarrow() -> None:
+    # temporary test until we have `cast` in pyarrow - later, fuse
+    # this with test above
+    table = pa.table(
+        {"a": pa.array(["a", "b", None, "d"], pa.dictionary(pa.int64(), pa.utf8()))}
+    )
+    df = nw.from_native(table, eager_only=True)
+    result = df["a"].cat.get_categories().to_list()
+    expected = ["a", "b", "d"]
+    assert result == expected
diff --git a/tests/frame/slice_test.py b/tests/frame/slice_test.py
index 5c47f287..83ed31fe 100644
--- a/tests/frame/slice_test.py
+++ b/tests/frame/slice_test.py
@@ -1,5 +1,7 @@
 from typing import Any
 
+import numpy as np
+import pandas as pd
 import polars as pl
 import pyarrow as pa
 import pytest
@@ -47,3 +49,23 @@ def test_slice_lazy_fails() -> None:
 def test_slice_int_fails(constructor_with_pyarrow: Any) -> None:
     with pytest.raises(TypeError, match="Expected str or slice, got: <class 'int'>"):
         _ = nw.from_native(constructor_with_pyarrow(data))[1]  # type: ignore[call-overload,index]
+
+
+def test_gather(constructor_with_pyarrow: Any) -> None:
+    df = nw.from_native(constructor_with_pyarrow(data), eager_only=True)
+    result = df[[0, 3, 1]]
+    expected = {
+        "a": [1.0, 4.0, 2.0],
+        "b": [11, 14, 12],
+    }
+    compare_dicts(result, expected)
+    result = df[np.array([0, 3, 1])]
+    compare_dicts(result, expected)
+
+
+def test_gather_pandas_index() -> None:
+    # check that we're slicing positionally, and not on the pandas index
+    df = pd.DataFrame({"a": [4, 1, 2], "b": [1, 4, 2]}, index=[2, 1, 3])
+    result = nw.from_native(df, eager_only=True)[[1, 2]]
+    expected = {"a": [1, 2], "b": [4, 2]}
+    compare_dicts(result, expected)
diff --git a/tests/frame/test_common.py b/tests/frame/test_common.py
index 1f09edc0..8533d9fc 100644
--- a/tests/frame/test_common.py
+++ b/tests/frame/test_common.py
@@ -325,30 +325,12 @@ def test_join(df_raw: Any) -> None:
     compare_dicts(result_native, expected)
 
 
-@pytest.mark.parametrize(
-    ("df_raw", "expected"),
-    [
-        (
-            df_polars,
-            {"a": [1, 1, 1, 3, 3, 3, 2, 2, 2], "a_right": [1, 3, 2, 1, 3, 2, 1, 3, 2]},
-        ),
-        (
-            df_lazy,
-            {"a": [1, 1, 1, 3, 3, 3, 2, 2, 2], "a_right": [1, 3, 2, 1, 3, 2, 1, 3, 2]},
-        ),
-        (
-            df_pandas,
-            {"a": [1, 1, 1, 3, 3, 3, 2, 2, 2], "a_right": [1, 3, 2, 1, 3, 2, 1, 3, 2]},
-        ),
-        (
-            df_mpd,
-            {"a": [1, 1, 1, 3, 3, 3, 2, 2, 2], "a_right": [1, 3, 2, 1, 3, 2, 1, 3, 2]},
-        ),
-    ],
-)
-def test_cross_join(df_raw: Any, expected: dict[str, list[Any]]) -> None:
+@pytest.mark.parametrize("df_raw", [df_polars, df_lazy, df_pandas, df_mpd])
+def test_cross_join(df_raw: Any) -> None:
     df = nw.from_native(df_raw).select("a")
     result = df.join(df, how="cross")  # type: ignore[arg-type]
+
+    expected = {"a": [1, 1, 1, 3, 3, 3, 2, 2, 2], "a_right": [1, 3, 2, 1, 3, 2, 1, 3, 2]}
     compare_dicts(result, expected)
 
     with pytest.raises(ValueError, match="Can not pass left_on, right_on for cross join"):
@@ -364,6 +346,32 @@ def test_cross_join_non_pandas() -> None:
     compare_dicts(result, expected)
 
 
+@pytest.mark.parametrize(
+    "df_raw",
+    [
+        df_polars,
+        df_lazy,
+        df_pandas,
+        # df_mpd, (todo: understand the difference between ipython/jupyter and pytest runs)
+    ],
+)
+@pytest.mark.parametrize(
+    ("join_key", "filter_expr", "expected"),
+    [
+        (["a", "b"], (nw.col("b") < 5), {"a": [2], "b": [6], "z": [9]}),
+        (["b"], (nw.col("b") < 5), {"a": [2], "b": [6], "z": [9]}),
+        (["b"], (nw.col("b") > 5), {"a": [1, 3], "b": [4, 4], "z": [7.0, 8.0]}),
+    ],
+)
+def test_anti_join(
+    df_raw: Any, join_key: list[str], filter_expr: nw.Expr, expected: dict[str, list[Any]]
+) -> None:
+    df = nw.from_native(df_raw)
+    other = df.filter(filter_expr)
+    result = df.join(other, how="anti", left_on=join_key, right_on=join_key)  # type: ignore[arg-type]
+    compare_dicts(result, expected)
+
+
 @pytest.mark.parametrize(
     "df_raw", [df_pandas, df_lazy, df_pandas_nullable, df_pandas_pyarrow]
 )
diff --git a/tests/series/cast_test.py b/tests/series/cast_test.py
index 0026ecee..3362291f 100644
--- a/tests/series/cast_test.py
+++ b/tests/series/cast_test.py
@@ -3,6 +3,7 @@ from datetime import datetime
 
 import pandas as pd
 import polars as pl
+import pyarrow as pa
 import pytest
 from polars.testing import assert_frame_equal
 
@@ -43,6 +44,24 @@ def test_cast_date_datetime_polars() -> None:
     assert df.schema == {"a": nw.Date}
 
 
+def test_cast_date_datetime_pyarrow() -> None:
+    # polars: date to datetime
+    dfpa = pa.table({"a": [date(2020, 1, 1), date(2020, 1, 2)]})
+    df = nw.from_native(dfpa)
+    df = df.select(nw.col("a").cast(nw.Datetime))
+    result = nw.to_native(df)
+    expected = pa.table({"a": [datetime(2020, 1, 1), datetime(2020, 1, 2)]})
+    assert result == expected
+
+    # pyarrow: datetime to date
+    dfpa = pa.table({"a": [datetime(2020, 1, 1), datetime(2020, 1, 2)]})
+    df = nw.from_native(dfpa)
+    df = df.select(nw.col("a").cast(nw.Date))
+    result = nw.to_native(df)
+    expected = pa.table({"a": [date(2020, 1, 1), date(2020, 1, 2)]})
+    assert result == expected
+
+
 @pytest.mark.skipif(
     parse_version(pd.__version__) < parse_version("2.0.0"),
     reason="pyarrow dtype not available",
diff --git a/tests/series/is_empty_test.py b/tests/series/is_empty_test.py
new file mode 100644
index 00000000..791dc732
--- /dev/null
+++ b/tests/series/is_empty_test.py
@@ -0,0 +1,11 @@
+from typing import Any
+
+import narwhals as nw
+
+
+def test_is_empty(constructor_series_with_pyarrow: Any) -> None:
+    series = nw.from_native(constructor_series_with_pyarrow([1, 2, 3]), series_only=True)
+    assert not series.is_empty()
+    assert not series[:1].is_empty()
+    assert len(series[:1]) == 1
+    assert series[:0].is_empty()
diff --git a/tests/series/test_common.py b/tests/series/test_common.py
index 38fb798f..d6ca809c 100644
--- a/tests/series/test_common.py
+++ b/tests/series/test_common.py
@@ -148,185 +148,6 @@ def test_convert(df_raw: Any) -> None:
     assert_series_equal(result, pd.Series([1, 3, 2], name="a"))
 
 
-def test_cast() -> None:
-    df_raw = pl.DataFrame(
-        {
-            "a": [1],
-            "b": [1],
-            "c": [1],
-            "d": [1],
-            "e": [1],
-            "f": [1],
-            "g": [1],
-            "h": [1],
-            "i": [1],
-            "j": [1],
-            "k": ["1"],
-            "l": [1],
-            "m": [True],
-            "n": [True],
-            "o": ["a"],
-            "p": [1],
-        },
-        schema={
-            "a": pl.Int64,
-            "b": pl.Int32,
-            "c": pl.Int16,
-            "d": pl.Int8,
-            "e": pl.UInt64,
-            "f": pl.UInt32,
-            "g": pl.UInt16,
-            "h": pl.UInt8,
-            "i": pl.Float64,
-            "j": pl.Float32,
-            "k": pl.String,
-            "l": pl.Datetime,
-            "m": pl.Boolean,
-            "n": pl.Boolean,
-            "o": pl.Categorical,
-            "p": pl.Int64,
-        },
-    )
-    df = nw.from_native(df_raw, eager_only=True).select(
-        nw.col("a").cast(nw.Int32),
-        nw.col("b").cast(nw.Int16),
-        nw.col("c").cast(nw.Int8),
-        nw.col("d").cast(nw.Int64),
-        nw.col("e").cast(nw.UInt32),
-        nw.col("f").cast(nw.UInt16),
-        nw.col("g").cast(nw.UInt8),
-        nw.col("h").cast(nw.UInt64),
-        nw.col("i").cast(nw.Float32),
-        nw.col("j").cast(nw.Float64),
-        nw.col("k").cast(nw.String),
-        nw.col("l").cast(nw.Datetime),
-        nw.col("m").cast(nw.Int8),
-        nw.col("n").cast(nw.Int8),
-        nw.col("o").cast(nw.String),
-        nw.col("p").cast(nw.Duration),
-    )
-    result = df.schema
-    expected = {
-        "a": nw.Int32,
-        "b": nw.Int16,
-        "c": nw.Int8,
-        "d": nw.Int64,
-        "e": nw.UInt32,
-        "f": nw.UInt16,
-        "g": nw.UInt8,
-        "h": nw.UInt64,
-        "i": nw.Float32,
-        "j": nw.Float64,
-        "k": nw.String,
-        "l": nw.Datetime,
-        "m": nw.Int8,
-        "n": nw.Int8,
-        "o": nw.String,
-        "p": nw.Duration,
-    }
-    assert result == expected
-    result_pd = nw.from_native(df.to_pandas(), eager_only=True).schema
-    assert result_pd == expected
-    result = df.select(
-        df["a"].cast(nw.Int32),
-        df["b"].cast(nw.Int16),
-        df["c"].cast(nw.Int8),
-        df["d"].cast(nw.Int64),
-        df["e"].cast(nw.UInt32),
-        df["f"].cast(nw.UInt16),
-        df["g"].cast(nw.UInt8),
-        df["h"].cast(nw.UInt64),
-        df["i"].cast(nw.Float32),
-        df["j"].cast(nw.Float64),
-        df["k"].cast(nw.String),
-        df["l"].cast(nw.Datetime),
-        df["m"].cast(nw.Int8),
-        df["n"].cast(nw.Boolean),
-        df["o"].cast(nw.Categorical),
-        df["p"].cast(nw.Duration),
-    ).schema
-    expected = {
-        "a": nw.Int32,
-        "b": nw.Int16,
-        "c": nw.Int8,
-        "d": nw.Int64,
-        "e": nw.UInt32,
-        "f": nw.UInt16,
-        "g": nw.UInt8,
-        "h": nw.UInt64,
-        "i": nw.Float32,
-        "j": nw.Float64,
-        "k": nw.String,
-        "l": nw.Datetime,
-        "m": nw.Int8,
-        "n": nw.Boolean,
-        "o": nw.Categorical,
-        "p": nw.Duration,
-    }
-    df = nw.from_native(df.to_pandas())  # type: ignore[assignment]
-    result_pd = df.select(
-        df["a"].cast(nw.Int32),
-        df["b"].cast(nw.Int16),
-        df["c"].cast(nw.Int8),
-        df["d"].cast(nw.Int64),
-        df["e"].cast(nw.UInt32),
-        df["f"].cast(nw.UInt16),
-        df["g"].cast(nw.UInt8),
-        df["h"].cast(nw.UInt64),
-        df["i"].cast(nw.Float32),
-        df["j"].cast(nw.Float64),
-        df["k"].cast(nw.String),
-        df["l"].cast(nw.Datetime),
-        df["m"].cast(nw.Int8),
-        df["n"].cast(nw.Boolean),
-        df["o"].cast(nw.Categorical),
-        df["p"].cast(nw.Duration),
-    ).schema
-    assert result == expected
-    df = nw.from_native(df.to_pandas().convert_dtypes())  # type: ignore[assignment]
-    result_pd = df.select(
-        df["a"].cast(nw.Int32),
-        df["b"].cast(nw.Int16),
-        df["c"].cast(nw.Int8),
-        df["d"].cast(nw.Int64),
-        df["e"].cast(nw.UInt32),
-        df["f"].cast(nw.UInt16),
-        df["g"].cast(nw.UInt8),
-        df["h"].cast(nw.UInt64),
-        df["i"].cast(nw.Float32),
-        df["j"].cast(nw.Float64),
-        df["k"].cast(nw.String),
-        df["l"].cast(nw.Datetime),
-        df["m"].cast(nw.Int8),
-        df["n"].cast(nw.Boolean),
-        df["o"].cast(nw.Categorical),
-        df["p"].cast(nw.Duration),
-    ).schema
-    assert result == expected
-    if parse_version(pd.__version__) < parse_version("2.0.0"):  # pragma: no cover
-        return
-    df = nw.from_native(df.to_pandas().convert_dtypes(dtype_backend="pyarrow"))  # type: ignore[assignment]
-    result_pd = df.select(
-        df["a"].cast(nw.Int32),
-        df["b"].cast(nw.Int16),
-        df["c"].cast(nw.Int8),
-        df["d"].cast(nw.Int64),
-        df["e"].cast(nw.UInt32),
-        df["f"].cast(nw.UInt16),
-        df["g"].cast(nw.UInt8),
-        df["h"].cast(nw.UInt64),
-        df["i"].cast(nw.Float32),
-        df["j"].cast(nw.Float64),
-        df["k"].cast(nw.String),
-        df["l"].cast(nw.Datetime),
-        df["m"].cast(nw.Int8),
-        df["n"].cast(nw.Boolean),
-        df["o"].cast(nw.Categorical),
-        df["p"].cast(nw.Duration),
-    ).schema
-    assert result == expected
-
-
 def test_to_numpy() -> None:
     s = pd.Series([1, 2, None], dtype="Int64")
     nw_series = nw.from_native(s, series_only=True)
@@ -343,14 +164,6 @@ def test_is_duplicated(df_raw: Any) -> None:
     assert (result.to_numpy() == expected).all()
 
 
-@pytest.mark.parametrize("df_raw", [df_pandas, df_polars])
-@pytest.mark.parametrize(("threshold", "expected"), [(0, False), (10, True)])
-def test_is_empty(df_raw: Any, threshold: Any, expected: Any) -> None:
-    series = nw.from_native(df_raw["b"], series_only=True)
-    result = series.filter(series > threshold).is_empty()
-    assert result == expected
-
-
 @pytest.mark.parametrize("df_raw", [df_pandas, df_polars])
 def test_is_unique(df_raw: Any) -> None:
     series = nw.from_native(df_raw["b"], series_only=True)
diff --git a/tests/utils.py b/tests/utils.py
index 83bf016a..202c332e 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import math
-import os
 import sys
 import warnings
 from typing import TYPE_CHECKING
@@ -43,17 +42,14 @@ def compare_dicts(result: Any, expected: dict[str, Any]) -> None:
 
 def maybe_get_modin_df(df_pandas: pd.DataFrame) -> Any:
     """Convert a pandas DataFrame to a Modin DataFrame if Modin is available."""
-    if os.environ.get("CI", None):  # pragma: no cover
-        try:
-            import modin.pandas as mpd
-        except ImportError:
-            return df_pandas.copy()
-        else:
-            with warnings.catch_warnings():
-                warnings.filterwarnings("ignore", category=UserWarning)
-                return mpd.DataFrame(df_pandas.to_dict(orient="list"))
-    else:  # pragma: no cover
-        return df_pandas
+    try:
+        import modin.pandas as mpd
+    except ImportError:  # pragma: no cover
+        return df_pandas.copy()
+    else:
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore", category=UserWarning)
+            return mpd.DataFrame(df_pandas.to_dict(orient="list"))
 
 
 def is_windows() -> bool: