Repository: narwhals-dev/narwhals
Pull request #1345 (2024-11-10 11:31:24)

chroe: refactor `ArrowDataFrame.with_columns`
## What type of PR is this? (check all applicable)

- [x] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Checklist

- [x] Code follows style guide (ruff)
- [x] Tests added 
- [x] Documented the changes

## If you have comments or can explain your changes, please do so below.

While profiling for plotly, py-spy indicates that we spend a lot of time in `validate_dataframe_comparand` for pyarrow case. This is called only in `with_columns` methods.

This PR proposed two changes:

- Creates constant array via `np.full` in `validate_dataframe_comparand` instead of `[const] * length`
- Changes the logic in `with_columns` to use pyarrow native methods to insert a column value. I expect this to be faster than the current approach of concatenating the already existing columns with the new ones - caveat is if the number of new columns is order(s) of magnitude greater than the existing ones. In _majority_ of scenarios I would expect this to not be the case, but this is the reason I am opening this as **RFC**


MarcoGorelli: nice - there may be some issue on py38, but the rest looks good!

should we wait until altair / marimo's cis are fixed to merge?
FBruzzesi: Fixed old versions, what's the deal with TPCH taking so long now 🙈?
MarcoGorelli: 🤔 looks like yesterday it went from 2 minutes to 15 minutes

Patch:
diff --git a/docs/api-reference/dataframe.md b/docs/api-reference/dataframe.md
index 447acbc1..00ff2122 100644
--- a/docs/api-reference/dataframe.md
+++ b/docs/api-reference/dataframe.md
@@ -26,6 +26,7 @@
         - lazy
         - null_count
         - pipe
+        - pivot
         - rename
         - row
         - rows
diff --git a/docs/api-reference/expr.md b/docs/api-reference/expr.md
index bf68689a..e99ffd4f 100644
--- a/docs/api-reference/expr.md
+++ b/docs/api-reference/expr.md
@@ -27,6 +27,7 @@
         - is_null
         - is_unique
         - len
+        - map_batches
         - max
         - mean
         - median
diff --git a/docs/installation.md b/docs/installation.md
index dc82521c..6da80b12 100644
--- a/docs/installation.md
+++ b/docs/installation.md
@@ -29,7 +29,7 @@ To verify the installation, start the Python REPL and execute:
 ```python
 >>> import narwhals
 >>> narwhals.__version__
-'1.13.3'
+'1.13.4'
 ```
 If you see the version number, then the installation was successful!
 
diff --git a/narwhals/__init__.py b/narwhals/__init__.py
index 974c2505..a16e1a9e 100644
--- a/narwhals/__init__.py
+++ b/narwhals/__init__.py
@@ -68,7 +68,7 @@ from narwhals.utils import maybe_get_index
 from narwhals.utils import maybe_reset_index
 from narwhals.utils import maybe_set_index
 
-__version__ = "1.13.3"
+__version__ = "1.13.4"
 
 __all__ = [
     "dependencies",
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index a357bdeb..999d4905 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -305,35 +305,30 @@ class ArrowDataFrame:
         *exprs: IntoArrowExpr,
         **named_exprs: IntoArrowExpr,
     ) -> Self:
+        native_frame = self._native_frame
         new_columns = evaluate_into_exprs(self, *exprs, **named_exprs)
-        new_column_name_to_new_column_map = {s.name: s for s in new_columns}
-        to_concat = []
-        output_names = []
-        # Make sure to preserve column order
+
         length = len(self)
-        for name in self.columns:
-            if name in new_column_name_to_new_column_map:
-                to_concat.append(
-                    validate_dataframe_comparand(
-                        length=length,
-                        other=new_column_name_to_new_column_map.pop(name),
-                        backend_version=self._backend_version,
-                    )
-                )
-            else:
-                to_concat.append(self._native_frame[name])
-            output_names.append(name)
-        for s in new_column_name_to_new_column_map:
-            to_concat.append(
-                validate_dataframe_comparand(
-                    length=length,
-                    other=new_column_name_to_new_column_map[s],
-                    backend_version=self._backend_version,
+        columns = self.columns
+
+        for col_value in new_columns:
+            col_name = col_value.name
+
+            column = validate_dataframe_comparand(
+                length=length,
+                other=col_value,
+                backend_version=self._backend_version,
+            )
+
+            native_frame = (
+                native_frame.set_column(
+                    columns.index(col_name), field_=col_name, column=column
                 )
+                if col_name in columns
+                else native_frame.append_column(field_=col_name, column=column)
             )
-            output_names.append(s)
-        df = self._native_frame.__class__.from_arrays(to_concat, names=output_names)
-        return self._from_native_frame(df)
+
+        return self._from_native_frame(native_frame)
 
     def group_by(self, *keys: str, drop_null_keys: bool) -> ArrowGroupBy:
         from narwhals._arrow.group_by import ArrowGroupBy
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index 802a900b..91c8bfcc 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -8,6 +8,8 @@ from typing import Sequence
 
 from narwhals._expression_parsing import reuse_series_implementation
 from narwhals._expression_parsing import reuse_series_namespace_implementation
+from narwhals.dependencies import get_numpy
+from narwhals.dependencies import is_numpy_array
 from narwhals.utils import Implementation
 
 if TYPE_CHECKING:
@@ -306,8 +308,15 @@ class ArrowExpr:
             seed=seed,
         )
 
-    def fill_null(self: Self, value: Any) -> Self:
-        return reuse_series_implementation(self, "fill_null", value=value)
+    def fill_null(
+        self: Self,
+        value: Any | None = None,
+        strategy: Literal["forward", "backward"] | None = None,
+        limit: int | None = None,
+    ) -> Self:
+        return reuse_series_implementation(
+            self, "fill_null", value=value, strategy=strategy, limit=limit
+        )
 
     def is_duplicated(self: Self) -> Self:
         return reuse_series_implementation(self, "is_duplicated")
@@ -383,6 +392,44 @@ class ArrowExpr:
     def mode(self: Self) -> Self:
         return reuse_series_implementation(self, "mode")
 
+    def map_batches(
+        self: Self,
+        function: Callable[[Any], Any],
+        return_dtype: DType | None = None,
+    ) -> Self:
+        def func(df: ArrowDataFrame) -> list[ArrowSeries]:
+            input_series_list = self._call(df)
+            output_names = [input_series.name for input_series in input_series_list]
+            result = [function(series) for series in input_series_list]
+
+            if is_numpy_array(result[0]):
+                result = [
+                    df.__narwhals_namespace__()
+                    ._create_compliant_series(array)
+                    .alias(output_name)
+                    for array, output_name in zip(result, output_names)
+                ]
+            elif (np := get_numpy()) is not None and np.isscalar(result[0]):
+                result = [
+                    df.__narwhals_namespace__()
+                    ._create_compliant_series([array])
+                    .alias(output_name)
+                    for array, output_name in zip(result, output_names)
+                ]
+            if return_dtype is not None:
+                result = [series.cast(return_dtype) for series in result]
+            return result
+
+        return self.__class__(
+            func,
+            depth=self._depth + 1,
+            function_name=self._function_name + "->map_batches",
+            root_names=self._root_names,
+            output_names=self._output_names,
+            backend_version=self._backend_version,
+            dtypes=self._dtypes,
+        )
+
     @property
     def dt(self: Self) -> ArrowExprDateTimeNamespace:
         return ArrowExprDateTimeNamespace(self)
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index 120242fa..8a39ff09 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -62,14 +62,16 @@ class ArrowNamespace:
             dtypes=self._dtypes,
         )
 
-    def _create_series_from_scalar(self, value: Any, series: ArrowSeries) -> ArrowSeries:
+    def _create_series_from_scalar(
+        self, value: Any, *, reference_series: ArrowSeries
+    ) -> ArrowSeries:
         from narwhals._arrow.series import ArrowSeries
 
         if self._backend_version < (13,) and hasattr(value, "as_py"):
             value = value.as_py()
         return ArrowSeries._from_iterable(
             [value],
-            name=series.name,
+            name=reference_series.name,
             backend_version=self._backend_version,
             dtypes=self._dtypes,
         )
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index bc22f346..d0c2bf98 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -580,14 +580,57 @@ class ArrowSeries:
 
         return self._from_native_series(pc.take(ser, mask))
 
-    def fill_null(self: Self, value: Any) -> Self:
+    def fill_null(
+        self: Self,
+        value: Any | None = None,
+        strategy: Literal["forward", "backward"] | None = None,
+        limit: int | None = None,
+    ) -> Self:
+        import numpy as np  # ignore-banned-import
         import pyarrow as pa  # ignore-banned-import()
         import pyarrow.compute as pc  # ignore-banned-import()
 
+        def fill_aux(
+            arr: pa.Array,
+            limit: int,
+            direction: Literal["forward", "backward"] | None = None,
+        ) -> pa.Array:
+            # this algorithm first finds the indices of the valid values to fill all the null value positions
+            # then it calculates the distance of each new index and the original index
+            # if the distance is equal to or less than the limit and the original value is null, it is replaced
+            valid_mask = pc.is_valid(arr)
+            indices = pa.array(np.arange(len(arr)), type=pa.int64())
+            if direction == "forward":
+                valid_index = np.maximum.accumulate(np.where(valid_mask, indices, -1))
+                distance = indices - valid_index
+            else:
+                valid_index = np.minimum.accumulate(
+                    np.where(valid_mask[::-1], indices[::-1], len(arr))
+                )[::-1]
+                distance = valid_index - indices
+            return pc.if_else(
+                pc.and_(
+                    pc.is_null(arr),
+                    pc.less_equal(distance, pa.scalar(limit)),
+                ),
+                arr.take(valid_index),
+                arr,
+            )
+
         ser = self._native_series
         dtype = ser.type
 
-        return self._from_native_series(pc.fill_null(ser, pa.scalar(value, dtype)))
+        if value is not None:
+            res_ser = self._from_native_series(pc.fill_null(ser, pa.scalar(value, dtype)))
+        elif limit is None:
+            fill_func = (
+                pc.fill_null_forward if strategy == "forward" else pc.fill_null_backward
+            )
+            res_ser = self._from_native_series(fill_func(ser))
+        else:
+            res_ser = self._from_native_series(fill_aux(ser, limit, strategy))
+
+        return res_ser
 
     def to_frame(self: Self) -> ArrowDataFrame:
         import pyarrow as pa  # ignore-banned-import()
@@ -709,17 +752,32 @@ class ArrowSeries:
         from narwhals._arrow.dataframe import ArrowDataFrame
 
         series = self._native_series
-        da = series.dictionary_encode().combine_chunks()
+        name = self._name
+        da = series.dictionary_encode(null_encoding="encode").combine_chunks()
 
-        columns = np.zeros((len(da.dictionary), len(da)), np.uint8)
+        columns = np.zeros((len(da.dictionary), len(da)), np.int8)
         columns[da.indices, np.arange(len(da))] = 1
-        names = [f"{self._name}{separator}{v}" for v in da.dictionary]
+        null_col_pa, null_col_pl = f"{name}{separator}None", f"{name}{separator}null"
+        cols = [
+            {null_col_pa: null_col_pl}.get(
+                f"{name}{separator}{v}", f"{name}{separator}{v}"
+            )
+            for v in da.dictionary
+        ]
 
+        output_order = (
+            [
+                null_col_pl,
+                *sorted([c for c in cols if c != null_col_pl])[int(drop_first) :],
+            ]
+            if null_col_pl in cols
+            else sorted(cols)[int(drop_first) :]
+        )
         return ArrowDataFrame(
-            pa.Table.from_arrays(columns, names=names),
+            pa.Table.from_arrays(columns, names=cols),
             backend_version=self._backend_version,
             dtypes=self._dtypes,
-        ).select(*sorted(names)[int(drop_first) :])
+        ).select(*output_order)
 
     def quantile(
         self: Self,
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index 8d54a3ca..9d8a8272 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -174,20 +174,24 @@ def validate_dataframe_comparand(
     If the comparison isn't supported, return `NotImplemented` so that the
     "right-hand-side" operation (e.g. `__radd__`) can be tried.
     """
-    from narwhals._arrow.dataframe import ArrowDataFrame
     from narwhals._arrow.series import ArrowSeries
 
-    if isinstance(other, ArrowDataFrame):
-        return NotImplemented
     if isinstance(other, ArrowSeries):
         if len(other) == 1:
+            import numpy as np  # ignore-banned-import
             import pyarrow as pa  # ignore-banned-import
 
-            value = other.item()
+            value = other._native_series[0]
             if backend_version < (13,) and hasattr(value, "as_py"):  # pragma: no cover
                 value = value.as_py()
-            return pa.chunked_array([[value] * length])
+            return pa.array(np.full(shape=length, fill_value=value))
         return other._native_series
+
+    from narwhals._arrow.dataframe import ArrowDataFrame  # pragma: no cover
+
+    if isinstance(other, ArrowDataFrame):  # pragma: no cover
+        return NotImplemented
+
     msg = "Please report a bug"  # pragma: no cover
     raise AssertionError(msg)
 
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index cf5c8eae..d7121a04 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -9,6 +9,7 @@ from typing import Sequence
 from narwhals._dask.utils import add_row_index
 from narwhals._dask.utils import parse_exprs_and_named_exprs
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
+from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals.utils import Implementation
 from narwhals.utils import flatten
 from narwhals.utils import generate_temporary_column_name
@@ -116,7 +117,14 @@ class DaskLazyFrame:
 
         if exprs and all(isinstance(x, str) for x in exprs) and not named_exprs:
             # This is a simple slice => fastpath!
-            return self._from_native_frame(self._native_frame.loc[:, exprs])
+            return self._from_native_frame(
+                select_columns_by_name(
+                    self._native_frame,
+                    list(exprs),  # type: ignore[arg-type]
+                    self._backend_version,
+                    self._implementation,
+                )
+            )
 
         new_series = parse_exprs_and_named_exprs(self, *exprs, **named_exprs)
 
@@ -136,7 +144,12 @@ class DaskLazyFrame:
             )
             return self._from_native_frame(df)
 
-        df = self._native_frame.assign(**new_series).loc[:, list(new_series.keys())]
+        df = select_columns_by_name(
+            self._native_frame.assign(**new_series),
+            list(new_series.keys()),
+            self._backend_version,
+            self._implementation,
+        )
         return self._from_native_frame(df)
 
     def drop_nulls(self: Self, subset: str | list[str] | None) -> Self:
@@ -150,7 +163,7 @@ class DaskLazyFrame:
     def schema(self) -> dict[str, DType]:
         return {
             col: native_to_narwhals_dtype(
-                self._native_frame.loc[:, col], self._dtypes, self._implementation
+                self._native_frame[col], self._dtypes, self._implementation
             )
             for col in self._native_frame.columns
         }
@@ -196,7 +209,7 @@ class DaskLazyFrame:
             subset = subset or self.columns
             token = generate_temporary_column_name(n_bytes=8, columns=subset)
             ser = native_frame.groupby(subset).size().rename(token)
-            ser = ser.loc[ser == 1]
+            ser = ser[ser == 1]
             unique = ser.reset_index().drop(columns=token)
             result = native_frame.merge(unique, on=subset, how="inner")
         else:
@@ -257,8 +270,16 @@ class DaskLazyFrame:
                 n_bytes=8, columns=[*self.columns, *other.columns]
             )
 
+            if right_on is None:  # pragma: no cover
+                msg = "`right_on` cannot be `None` in anti-join"
+                raise TypeError(msg)
             other_native = (
-                other._native_frame.loc[:, right_on]
+                select_columns_by_name(
+                    other._native_frame,
+                    right_on,
+                    self._backend_version,
+                    self._implementation,
+                )
                 .rename(  # rename to avoid creating extra columns in join
                     columns=dict(zip(right_on, left_on))  # type: ignore[arg-type]
                 )
@@ -272,12 +293,20 @@ class DaskLazyFrame:
                 right_on=left_on,
             )
             return self._from_native_frame(
-                df.loc[df[indicator_token] == "left_only"].drop(columns=[indicator_token])
+                df[df[indicator_token] == "left_only"].drop(columns=[indicator_token])
             )
 
         if how == "semi":
+            if right_on is None:  # pragma: no cover
+                msg = "`right_on` cannot be `None` in semi-join"
+                raise TypeError(msg)
             other_native = (
-                other._native_frame.loc[:, right_on]
+                select_columns_by_name(
+                    other._native_frame,
+                    right_on,
+                    self._backend_version,
+                    self._implementation,
+                )
                 .rename(  # rename to avoid creating extra columns in join
                     columns=dict(zip(right_on, left_on))  # type: ignore[arg-type]
                 )
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 169adc43..79766d2a 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -67,7 +67,7 @@ class DaskExpr:
         dtypes: DTypes,
     ) -> Self:
         def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
-            return [df._native_frame.loc[:, column_name] for column_name in column_names]
+            return [df._native_frame[column_name] for column_name in column_names]
 
         return cls(
             func,
@@ -376,6 +376,14 @@ class DaskExpr:
             returns_scalar=False,
         )
 
+    def map_batches(
+        self: Self,
+        function: Callable[[Any], Any],
+        return_dtype: DType | None = None,
+    ) -> NoReturn:
+        msg = "`Expr.map_batches` is not implemented for Dask yet"
+        raise NotImplementedError(msg)
+
     def mean(self) -> Self:
         return self._from_call(
             lambda _input: _input.mean(),
@@ -525,11 +533,34 @@ class DaskExpr:
             returns_scalar=True,
         )
 
-    def fill_null(self, value: Any) -> DaskExpr:
+    def fill_null(
+        self: Self,
+        value: Any | None = None,
+        strategy: Literal["forward", "backward"] | None = None,
+        limit: int | None = None,
+    ) -> DaskExpr:
+        def func(
+            _input: dask_expr.Series,
+            value: Any | None,
+            strategy: str | None,
+            limit: int | None,
+        ) -> dask_expr.Series:
+            if value is not None:
+                res_ser = _input.fillna(value)
+            else:
+                res_ser = (
+                    _input.ffill(limit=limit)
+                    if strategy == "forward"
+                    else _input.bfill(limit=limit)
+                )
+            return res_ser
+
         return self._from_call(
-            lambda _input, _val: _input.fillna(_val),
+            func,
             "fillna",
             value,
+            strategy,
+            limit,
             returns_scalar=False,
         )
 
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index 04365e51..2f1b401f 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -41,7 +41,7 @@ class DaskNamespace:
 
     def all(self) -> DaskExpr:
         def func(df: DaskLazyFrame) -> list[dask_expr.Series]:
-            return [df._native_frame.loc[:, column_name] for column_name in df.columns]
+            return [df._native_frame[column_name] for column_name in df.columns]
 
         return DaskExpr(
             func,
@@ -76,9 +76,9 @@ class DaskNamespace:
 
         return DaskExpr(
             lambda df: [
-                df._native_frame.assign(literal=value)
-                .loc[:, "literal"]
-                .pipe(convert_if_dtype, dtype)
+                df._native_frame.assign(literal=value)["literal"].pipe(
+                    convert_if_dtype, dtype
+                )
             ],
             depth=0,
             function_name="lit",
@@ -126,7 +126,7 @@ class DaskNamespace:
                         npartitions=df._native_frame.npartitions,
                     )
                 ]
-            return [df._native_frame.loc[:, df.columns[0]].size.to_series().rename("len")]
+            return [df._native_frame[df.columns[0]].size.to_series().rename("len")]
 
         # coverage bug? this is definitely hit
         return DaskExpr(  # pragma: no cover
@@ -311,7 +311,9 @@ class DaskNamespace:
         msg = "`_create_compliant_series` for DaskNamespace exists only for compatibility"
         raise NotImplementedError(msg)
 
-    def _create_series_from_scalar(self, *_: Any) -> NoReturn:
+    def _create_series_from_scalar(
+        self, value: Any, *, reference_series: DaskExpr
+    ) -> NoReturn:
         msg = (
             "`_create_series_from_scalar` for DaskNamespace exists only for compatibility"
         )
diff --git a/narwhals/_dask/utils.py b/narwhals/_dask/utils.py
index cf8f9a3f..913beb19 100644
--- a/narwhals/_dask/utils.py
+++ b/narwhals/_dask/utils.py
@@ -43,7 +43,7 @@ def parse_exprs_and_named_exprs(
         if hasattr(expr, "__narwhals_expr__"):
             _results = expr._call(df)
         elif isinstance(expr, str):
-            _results = [df._native_frame.loc[:, expr]]
+            _results = [df._native_frame[expr]]
         else:  # pragma: no cover
             msg = f"Expected expression or column name, got: {expr}"
             raise TypeError(msg)
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index 89d020b0..abd34d69 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -227,12 +227,12 @@ def reuse_series_implementation(
 
         out: list[CompliantSeries] = [
             plx._create_series_from_scalar(
-                getattr(column, attr)(*_args, **_kwargs),
-                column,  # type: ignore[arg-type]
+                getattr(series, attr)(*_args, **_kwargs),
+                reference_series=series,  # type: ignore[arg-type]
             )
             if returns_scalar
-            else getattr(column, attr)(*_args, **_kwargs)
-            for column in expr._call(df)  # type: ignore[arg-type]
+            else getattr(series, attr)(*_args, **_kwargs)
+            for series in expr._call(df)  # type: ignore[arg-type]
         ]
         if expr._output_names is not None and (
             [s.name for s in out] != expr._output_names
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index 8c30187b..25a5c236 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -12,9 +12,10 @@ from narwhals._expression_parsing import evaluate_into_exprs
 from narwhals._pandas_like.expr import PandasLikeExpr
 from narwhals._pandas_like.utils import broadcast_series
 from narwhals._pandas_like.utils import convert_str_slice_to_int_slice
-from narwhals._pandas_like.utils import create_native_series
+from narwhals._pandas_like.utils import create_compliant_series
 from narwhals._pandas_like.utils import horizontal_concat
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
+from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals._pandas_like.utils import validate_dataframe_comparand
 from narwhals.dependencies import is_numpy_array
 from narwhals.utils import Implementation
@@ -232,7 +233,14 @@ class PandasLikeDataFrame:
 
         elif is_sequence_but_not_str(item) or (is_numpy_array(item) and item.ndim == 1):
             if all(isinstance(x, str) for x in item) and len(item) > 0:
-                return self._from_native_frame(self._native_frame.loc[:, item])
+                return self._from_native_frame(
+                    select_columns_by_name(
+                        self._native_frame,
+                        item,
+                        self._backend_version,
+                        self._implementation,
+                    )
+                )
             return self._from_native_frame(self._native_frame.iloc[item])
 
         elif isinstance(item, slice):
@@ -328,7 +336,15 @@ class PandasLikeDataFrame:
     ) -> Self:
         if exprs and all(isinstance(x, str) for x in exprs) and not named_exprs:
             # This is a simple slice => fastpath!
-            return self._from_native_frame(self._native_frame.loc[:, list(exprs)])
+            column_names = list(exprs)
+            return self._from_native_frame(
+                select_columns_by_name(
+                    self._native_frame,
+                    column_names,  # type: ignore[arg-type]
+                    self._backend_version,
+                    self._implementation,
+                )
+            )
         new_series = evaluate_into_exprs(self, *exprs, **named_exprs)
         if not new_series:
             # return empty dataframe, like Polars does
@@ -349,7 +365,7 @@ class PandasLikeDataFrame:
         return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
 
     def with_row_index(self, name: str) -> Self:
-        row_index = create_native_series(
+        row_index = create_compliant_series(
             range(len(self._native_frame)),
             index=self._native_frame.index,
             implementation=self._implementation,
@@ -545,9 +561,17 @@ class PandasLikeDataFrame:
                 indicator_token = generate_temporary_column_name(
                     n_bytes=8, columns=[*self.columns, *other.columns]
                 )
+                if right_on is None:  # pragma: no cover
+                    msg = "`right_on` cannot be `None` in anti-join"
+                    raise TypeError(msg)
 
                 other_native = (
-                    other._native_frame.loc[:, right_on]
+                    select_columns_by_name(
+                        other._native_frame,
+                        right_on,
+                        self._backend_version,
+                        self._implementation,
+                    )
                     .rename(  # rename to avoid creating extra columns in join
                         columns=dict(zip(right_on, left_on)),  # type: ignore[arg-type]
                         copy=False,
@@ -567,8 +591,16 @@ class PandasLikeDataFrame:
                 )
 
         if how == "semi":
+            if right_on is None:  # pragma: no cover
+                msg = "`right_on` cannot be `None` in semi-join"
+                raise TypeError(msg)
             other_native = (
-                other._native_frame.loc[:, right_on]
+                select_columns_by_name(
+                    other._native_frame,
+                    right_on,
+                    self._backend_version,
+                    self._implementation,
+                )
                 .rename(  # rename to avoid creating extra columns in join
                     columns=dict(zip(right_on, left_on)),  # type: ignore[arg-type]
                     copy=False,
@@ -677,7 +709,6 @@ class PandasLikeDataFrame:
         from narwhals._pandas_like.series import PandasLikeSeries
 
         if as_series:
-            # TODO(Unassigned): should this return narwhals series?
             return {
                 col: PandasLikeSeries(
                     self._native_frame[col],
@@ -796,6 +827,90 @@ class PandasLikeDataFrame:
     def gather_every(self: Self, n: int, offset: int = 0) -> Self:
         return self._from_native_frame(self._native_frame.iloc[offset::n])
 
+    def pivot(
+        self: Self,
+        on: str | list[str],
+        *,
+        index: str | list[str] | None,
+        values: str | list[str] | None,
+        aggregate_function: Any | None,
+        maintain_order: bool,
+        sort_columns: bool,
+        separator: str = "_",
+    ) -> Self:
+        if self._implementation is Implementation.PANDAS and (
+            self._backend_version < (1, 1)
+        ):  # pragma: no cover
+            msg = "pivot is only supported for pandas>=1.1"
+            raise NotImplementedError(msg)
+        if self._implementation is Implementation.MODIN:
+            msg = "pivot is not supported for Modin backend due to https://github.com/modin-project/modin/issues/7409."
+            raise NotImplementedError(msg)
+        from itertools import product
+
+        frame = self._native_frame
+
+        if isinstance(on, str):
+            on = [on]
+        if isinstance(index, str):
+            index = [index]
+
+        if values is None:
+            values_ = [c for c in self.columns if c not in {*on, *index}]  # type: ignore[misc]
+        elif isinstance(values, str):  # pragma: no cover
+            values_ = [values]
+        else:
+            values_ = values
+
+        if aggregate_function is None:
+            result = frame.pivot(columns=on, index=index, values=values_)
+
+        elif aggregate_function == "len":
+            result = (
+                frame.groupby([*on, *index])  # type: ignore[misc]
+                .agg({v: "size" for v in values_})
+                .reset_index()
+                .pivot(columns=on, index=index, values=values_)
+            )
+        else:
+            result = frame.pivot_table(
+                values=values_,
+                index=index,
+                columns=on,
+                aggfunc=aggregate_function,
+                margins=False,
+                observed=True,
+            )
+
+        # Put columns in the right order
+        if sort_columns:
+            uniques = {
+                col: sorted(self._native_frame[col].unique().tolist()) for col in on
+            }
+        else:
+            uniques = {col: self._native_frame[col].unique().tolist() for col in on}
+        all_lists = [values_, *list(uniques.values())]
+        ordered_cols = list(product(*all_lists))
+        result = result.loc[:, ordered_cols]
+        columns = result.columns.tolist()
+
+        n_on = len(on)
+        if n_on == 1:
+            new_columns = [
+                separator.join(col).strip() if len(values_) > 1 else col[-1]
+                for col in columns
+            ]
+        else:
+            new_columns = [
+                separator.join([col[0], '{"' + '","'.join(col[-n_on:]) + '"}'])
+                if len(values_) > 1
+                else '{"' + '","'.join(col[-n_on:]) + '"}'
+                for col in columns
+            ]
+        result.columns = new_columns
+        result.columns.names = [""]  # type: ignore[attr-defined]
+        return self._from_native_frame(result.reset_index())
+
     def to_arrow(self: Self) -> Any:
         if self._implementation is Implementation.CUDF:
             return self._native_frame.to_arrow(preserve_index=False)
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index 9ebbc315..408fc0ba 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -9,6 +9,8 @@ from typing import Sequence
 from narwhals._expression_parsing import reuse_series_implementation
 from narwhals._expression_parsing import reuse_series_namespace_implementation
 from narwhals._pandas_like.series import PandasLikeSeries
+from narwhals.dependencies import get_numpy
+from narwhals.dependencies import is_numpy_array
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -259,8 +261,15 @@ class PandasLikeExpr:
     def is_null(self) -> Self:
         return reuse_series_implementation(self, "is_null")
 
-    def fill_null(self, value: Any) -> Self:
-        return reuse_series_implementation(self, "fill_null", value=value)
+    def fill_null(
+        self,
+        value: Any | None = None,
+        strategy: Literal["forward", "backward"] | None = None,
+        limit: int | None = None,
+    ) -> Self:
+        return reuse_series_implementation(
+            self, "fill_null", value=value, strategy=strategy, limit=limit
+        )
 
     def is_in(self, other: Any) -> Self:
         return reuse_series_implementation(self, "is_in", other=other)
@@ -399,6 +408,39 @@ class PandasLikeExpr:
     def mode(self: Self) -> Self:
         return reuse_series_implementation(self, "mode")
 
+    def map_batches(
+        self: Self,
+        function: Callable[[Any], Any],
+        return_dtype: DType | None = None,
+    ) -> Self:
+        def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
+            input_series_list = self._call(df)
+            output_names = [input_series.name for input_series in input_series_list]
+            result = [function(series) for series in input_series_list]
+            if is_numpy_array(result[0]) or (
+                (np := get_numpy()) is not None and np.isscalar(result[0])
+            ):
+                result = [
+                    df.__narwhals_namespace__()
+                    ._create_compliant_series(array)
+                    .alias(output_name)
+                    for array, output_name in zip(result, output_names)
+                ]
+            if return_dtype is not None:
+                result = [series.cast(return_dtype) for series in result]
+            return result
+
+        return self.__class__(
+            func,
+            depth=self._depth + 1,
+            function_name=self._function_name + "->map_batches",
+            root_names=self._root_names,
+            output_names=self._output_names,
+            implementation=self._implementation,
+            backend_version=self._backend_version,
+            dtypes=self._dtypes,
+        )
+
     @property
     def str(self: Self) -> PandasLikeExprStringNamespace:
         return PandasLikeExprStringNamespace(self)
diff --git a/narwhals/_pandas_like/group_by.py b/narwhals/_pandas_like/group_by.py
index c628ecbd..b21392d8 100644
--- a/narwhals/_pandas_like/group_by.py
+++ b/narwhals/_pandas_like/group_by.py
@@ -11,6 +11,7 @@ from typing import Iterator
 from narwhals._expression_parsing import is_simple_aggregation
 from narwhals._expression_parsing import parse_into_exprs
 from narwhals._pandas_like.utils import native_series_from_iterable
+from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals.utils import Implementation
 from narwhals.utils import remove_prefix
 from narwhals.utils import tupleify
@@ -38,7 +39,15 @@ class PandasLikeGroupBy:
         ):  # pragma: no cover
             if (
                 not drop_null_keys
-                and self._df._native_frame.loc[:, self._keys].isna().any().any()
+                and select_columns_by_name(
+                    self._df._native_frame,
+                    self._keys,
+                    self._df._backend_version,
+                    self._df._implementation,
+                )
+                .isna()
+                .any()
+                .any()
             ):
                 msg = "Grouping by null values is not supported in pandas < 1.0.0"
                 raise NotImplementedError(msg)
@@ -227,7 +236,11 @@ def agg_pandas(  # noqa: PLR0915
             result_aggs = native_namespace.DataFrame(
                 list(grouped.groups.keys()), columns=keys
             )
-        return from_dataframe(result_aggs.loc[:, output_names])
+        return from_dataframe(
+            select_columns_by_name(
+                result_aggs, output_names, backend_version, implementation
+            )
+        )
 
     if dataframe_is_empty:
         # Don't even attempt this, it's way too inconsistent across pandas versions.
@@ -275,4 +288,8 @@ def agg_pandas(  # noqa: PLR0915
     # This may need updating, depending on https://github.com/pandas-dev/pandas/pull/51466/files
     result_complex.reset_index(inplace=True)  # noqa: PD002
 
-    return from_dataframe(result_complex.loc[:, output_names])
+    return from_dataframe(
+        select_columns_by_name(
+            result_complex, output_names, backend_version, implementation
+        )
+    )
diff --git a/narwhals/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index c3ca204a..45ba1d85 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -15,7 +15,7 @@ from narwhals._pandas_like.dataframe import PandasLikeDataFrame
 from narwhals._pandas_like.expr import PandasLikeExpr
 from narwhals._pandas_like.selectors import PandasSelectorNamespace
 from narwhals._pandas_like.series import PandasLikeSeries
-from narwhals._pandas_like.utils import create_native_series
+from narwhals._pandas_like.utils import create_compliant_series
 from narwhals._pandas_like.utils import horizontal_concat
 from narwhals._pandas_like.utils import vertical_concat
 
@@ -67,12 +67,12 @@ class PandasLikeNamespace:
         )
 
     def _create_series_from_scalar(
-        self, value: Any, series: PandasLikeSeries
+        self, value: Any, *, reference_series: PandasLikeSeries
     ) -> PandasLikeSeries:
         return PandasLikeSeries._from_iterable(
             [value],
-            name=series._native_series.name,
-            index=series._native_series.index[0:1],
+            name=reference_series._native_series.name,
+            index=reference_series._native_series.index[0:1],
             implementation=self._implementation,
             backend_version=self._backend_version,
             dtypes=self._dtypes,
@@ -91,7 +91,7 @@ class PandasLikeNamespace:
         )
 
     def _create_compliant_series(self, value: Any) -> PandasLikeSeries:
-        return create_native_series(
+        return create_compliant_series(
             value,
             implementation=self._implementation,
             backend_version=self._backend_version,
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index eb3c22c7..1bee0079 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -14,6 +14,7 @@ from narwhals._pandas_like.utils import int_dtype_mapper
 from narwhals._pandas_like.utils import narwhals_to_native_dtype
 from narwhals._pandas_like.utils import native_series_from_iterable
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
+from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals._pandas_like.utils import set_axis
 from narwhals._pandas_like.utils import to_datetime
 from narwhals._pandas_like.utils import validate_column_comparand
@@ -451,9 +452,23 @@ class PandasLikeSeries:
         ser = self._native_series
         return self._from_native_series(ser.isna())
 
-    def fill_null(self, value: Any) -> PandasLikeSeries:
+    def fill_null(
+        self,
+        value: Any | None = None,
+        strategy: Literal["forward", "backward"] | None = None,
+        limit: int | None = None,
+    ) -> PandasLikeSeries:
         ser = self._native_series
-        return self._from_native_series(ser.fillna(value))
+        if value is not None:
+            res_ser = self._from_native_series(ser.fillna(value=value))
+        else:
+            res_ser = self._from_native_series(
+                ser.ffill(limit=limit)
+                if strategy == "forward"
+                else ser.bfill(limit=limit)
+            )
+
+        return res_ser
 
     def drop_nulls(self) -> PandasLikeSeries:
         ser = self._native_series
@@ -693,13 +708,28 @@ class PandasLikeSeries:
         plx = self.__native_namespace__()
         series = self._native_series
         name = str(self._name) if self._name else ""
+
+        null_col_pl = f"{name}{separator}null"
+
+        has_nulls = series.isna().any()
+        result = plx.get_dummies(
+            series,
+            prefix=name,
+            prefix_sep=separator,
+            drop_first=drop_first,
+            # Adds a null column at the end, depending on whether or not there are any.
+            dummy_na=has_nulls,
+            dtype="int8",
+        )
+        if has_nulls:
+            *cols, null_col_pd = list(result.columns)
+            output_order = [null_col_pd, *cols]
+            result = select_columns_by_name(
+                result, output_order, self._backend_version, self._implementation
+            ).rename(columns={null_col_pd: null_col_pl}, copy=False)
+
         return PandasLikeDataFrame(
-            plx.get_dummies(
-                series,
-                prefix=name,
-                prefix_sep=separator,
-                drop_first=drop_first,
-            ).astype(int),
+            result,
             implementation=self._implementation,
             backend_version=self._backend_version,
             dtypes=self._dtypes,
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index 87ed2c94..d68f41be 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 from typing import Iterable
 from typing import Literal
+from typing import Sequence
 from typing import TypeVar
 
 from narwhals._arrow.utils import (
@@ -142,7 +143,7 @@ def validate_dataframe_comparand(index: Any, other: Any) -> Any:
     raise AssertionError(msg)
 
 
-def create_native_series(
+def create_compliant_series(
     iterable: Any,
     index: Any = None,
     *,
@@ -635,3 +636,20 @@ def calculate_timestamp_date(s: pd.Series, time_unit: str) -> pd.Series:
     else:
         result = s * 1_000
     return result
+
+
+def select_columns_by_name(
+    df: T,
+    column_names: Sequence[str],
+    backend_version: tuple[int, ...],
+    implementation: Implementation,
+) -> T:
+    """Select columns by name. Prefer this over `df.loc[:, column_names]` as it's
+    generally more performant."""
+    if (df.columns.dtype.kind == "b") or (  # type: ignore[attr-defined]
+        implementation is Implementation.PANDAS and backend_version < (1, 5)
+    ):
+        # See https://github.com/narwhals-dev/narwhals/issues/1349#issuecomment-2470118122
+        # for why we need this
+        return df.loc[:, column_names]  # type: ignore[no-any-return, attr-defined]
+    return df[column_names]  # type: ignore[no-any-return, index]
diff --git a/narwhals/_polars/dataframe.py b/narwhals/_polars/dataframe.py
index b5640493..fffe5092 100644
--- a/narwhals/_polars/dataframe.py
+++ b/narwhals/_polars/dataframe.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Literal
 from typing import Sequence
 
 from narwhals._polars.namespace import PolarsNamespace
@@ -16,14 +17,16 @@ if TYPE_CHECKING:
     from types import ModuleType
 
     import numpy as np
+    import polars as pl
     from typing_extensions import Self
 
+    from narwhals.dtypes import DType
     from narwhals.typing import DTypes
 
 
 class PolarsDataFrame:
     def __init__(
-        self, df: Any, *, backend_version: tuple[int, ...], dtypes: DTypes
+        self, df: pl.DataFrame, *, backend_version: tuple[int, ...], dtypes: DTypes
     ) -> None:
         self._native_frame = df
         self._backend_version = backend_version
@@ -91,19 +94,21 @@ class PolarsDataFrame:
             return self._native_frame.__array__(dtype)
         return self._native_frame.__array__(dtype)
 
-    def collect_schema(self) -> dict[str, Any]:
+    def collect_schema(self) -> dict[str, DType]:
         if self._backend_version < (1,):
-            schema = self._native_frame.schema
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.schema.items()
+            }
         else:
-            schema = dict(self._native_frame.collect_schema())
-        return {
-            name: native_to_narwhals_dtype(dtype, self._dtypes)
-            for name, dtype in schema.items()
-        }
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.collect_schema().items()
+            }
 
     @property
     def shape(self) -> tuple[int, int]:
-        return self._native_frame.shape  # type: ignore[no-any-return]
+        return self._native_frame.shape
 
     def __getitem__(self, item: Any) -> Any:
         if self._backend_version > (0, 20, 30):
@@ -179,7 +184,7 @@ class PolarsDataFrame:
 
     @property
     def columns(self) -> list[str]:
-        return self._native_frame.columns  # type: ignore[no-any-return]
+        return self._native_frame.columns
 
     def lazy(self) -> PolarsLazyFrame:
         return PolarsLazyFrame(
@@ -243,10 +248,38 @@ class PolarsDataFrame:
             )
         )
 
+    def pivot(
+        self: Self,
+        on: str | list[str],
+        *,
+        index: str | list[str] | None = None,
+        values: str | list[str] | None = None,
+        aggregate_function: Literal[
+            "min", "max", "first", "last", "sum", "mean", "median", "len"
+        ]
+        | None = None,
+        maintain_order: bool = True,
+        sort_columns: bool = False,
+        separator: str = "_",
+    ) -> Self:
+        if self._backend_version < (1, 0, 0):  # pragma: no cover
+            msg = "`pivot` is only supported for Polars>=1.0.0"
+            raise NotImplementedError(msg)
+        result = self._native_frame.pivot(
+            on,
+            index=index,
+            values=values,
+            aggregate_function=aggregate_function,
+            maintain_order=maintain_order,
+            sort_columns=sort_columns,
+            separator=separator,
+        )
+        return self._from_native_object(result)  # type: ignore[no-any-return]
+
 
 class PolarsLazyFrame:
     def __init__(
-        self, df: Any, *, backend_version: tuple[int, ...], dtypes: DTypes
+        self, df: pl.LazyFrame, *, backend_version: tuple[int, ...], dtypes: DTypes
     ) -> None:
         self._native_frame = df
         self._backend_version = backend_version
@@ -285,7 +318,7 @@ class PolarsLazyFrame:
 
     @property
     def columns(self) -> list[str]:
-        return self._native_frame.columns  # type: ignore[no-any-return]
+        return self._native_frame.columns
 
     @property
     def schema(self) -> dict[str, Any]:
@@ -295,15 +328,17 @@ class PolarsLazyFrame:
             for name, dtype in schema.items()
         }
 
-    def collect_schema(self) -> dict[str, Any]:
+    def collect_schema(self) -> dict[str, DType]:
         if self._backend_version < (1,):
-            schema = self._native_frame.schema
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.schema.items()
+            }
         else:
-            schema = dict(self._native_frame.collect_schema())
-        return {
-            name: native_to_narwhals_dtype(dtype, self._dtypes)
-            for name, dtype in schema.items()
-        }
+            return {
+                name: native_to_narwhals_dtype(dtype, self._dtypes)
+                for name, dtype in self._native_frame.collect_schema().items()
+            }
 
     def collect(self) -> PolarsDataFrame:
         return PolarsDataFrame(
diff --git a/narwhals/_polars/expr.py b/narwhals/_polars/expr.py
index 46b415dd..cb2e41f2 100644
--- a/narwhals/_polars/expr.py
+++ b/narwhals/_polars/expr.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Callable
 from typing import Sequence
 
 from narwhals._polars.utils import extract_args_kwargs
@@ -44,21 +45,34 @@ class PolarsExpr:
 
     def cast(self, dtype: DType) -> Self:
         expr = self._native_expr
-        dtype = narwhals_to_native_dtype(dtype, self._dtypes)
-        return self._from_native_expr(expr.cast(dtype))
+        dtype_pl = narwhals_to_native_dtype(dtype, self._dtypes)
+        return self._from_native_expr(expr.cast(dtype_pl))
+
+    def map_batches(
+        self,
+        function: Callable[[Any], Self],
+        return_dtype: DType | None = None,
+    ) -> Self:
+        if return_dtype is not None:
+            return_dtype_pl = narwhals_to_native_dtype(return_dtype, self._dtypes)
+            return self._from_native_expr(
+                self._native_expr.map_batches(function, return_dtype_pl)
+            )
+        else:
+            return self._from_native_expr(self._native_expr.map_batches(function))
 
     def replace_strict(
         self, old: Sequence[Any], new: Sequence[Any], *, return_dtype: DType | None
     ) -> Self:
         expr = self._native_expr
-        return_dtype = (
+        return_dtype_pl = (
             narwhals_to_native_dtype(return_dtype, self._dtypes) if return_dtype else None
         )
         if self._backend_version < (1,):
             msg = f"`replace_strict` is only available in Polars>=1.0, found version {self._backend_version}"
             raise NotImplementedError(msg)
         return self._from_native_expr(
-            expr.replace_strict(old, new, return_dtype=return_dtype)
+            expr.replace_strict(old, new, return_dtype=return_dtype_pl)
         )
 
     def __eq__(self, other: object) -> Self:  # type: ignore[override]
diff --git a/narwhals/_polars/series.py b/narwhals/_polars/series.py
index 3bc684d2..4feff11c 100644
--- a/narwhals/_polars/series.py
+++ b/narwhals/_polars/series.py
@@ -13,6 +13,7 @@ if TYPE_CHECKING:
     from types import ModuleType
 
     import numpy as np
+    import polars as pl
     from typing_extensions import Self
 
     from narwhals._polars.dataframe import PolarsDataFrame
@@ -27,7 +28,7 @@ class PolarsSeries:
     def __init__(
         self, series: Any, *, backend_version: tuple[int, ...], dtypes: DTypes
     ) -> None:
-        self._native_series = series
+        self._native_series: pl.Series = series
         self._backend_version = backend_version
         self._implementation = Implementation.POLARS
         self._dtypes = dtypes
@@ -85,7 +86,7 @@ class PolarsSeries:
 
     @property
     def name(self) -> str:
-        return self._native_series.name  # type: ignore[no-any-return]
+        return self._native_series.name
 
     @property
     def dtype(self: Self) -> DType:
@@ -102,8 +103,8 @@ class PolarsSeries:
 
     def cast(self, dtype: DType) -> Self:
         ser = self._native_series
-        dtype = narwhals_to_native_dtype(dtype, self._dtypes)
-        return self._from_native_series(ser.cast(dtype))
+        dtype_pl = narwhals_to_native_dtype(dtype, self._dtypes)
+        return self._from_native_series(ser.cast(dtype_pl))
 
     def replace_strict(
         self, old: Sequence[Any], new: Sequence[Any], *, return_dtype: DType | None
@@ -203,16 +204,23 @@ class PolarsSeries:
     def to_dummies(
         self: Self, *, separator: str = "_", drop_first: bool = False
     ) -> PolarsDataFrame:
+        import polars as pl  # ignore-banned-import
+
         from narwhals._polars.dataframe import PolarsDataFrame
 
         if self._backend_version < (0, 20, 15):
+            has_nulls = self._native_series.is_null().any()
             result = self._native_series.to_dummies(separator=separator)
-            result = result.select(result.columns[int(drop_first) :])
+            output_columns = result.columns
+            if drop_first:
+                _ = output_columns.pop(int(has_nulls))
+
+            result = result.select(output_columns)
         else:
             result = self._native_series.to_dummies(
                 separator=separator, drop_first=drop_first
             )
-
+        result = result.with_columns(pl.all().cast(pl.Int8))
         return PolarsDataFrame(
             result, backend_version=self._backend_version, dtypes=self._dtypes
         )
diff --git a/narwhals/_polars/utils.py b/narwhals/_polars/utils.py
index 295c03bc..79528af1 100644
--- a/narwhals/_polars/utils.py
+++ b/narwhals/_polars/utils.py
@@ -5,10 +5,11 @@ from typing import Any
 from typing import Literal
 
 if TYPE_CHECKING:
+    import polars as pl
+
     from narwhals.dtypes import DType
     from narwhals.typing import DTypes
 
-from narwhals.dependencies import get_polars
 from narwhals.utils import parse_version
 
 
@@ -33,7 +34,7 @@ def extract_args_kwargs(args: Any, kwargs: Any) -> tuple[list[Any], dict[str, An
     return args, kwargs
 
 
-def native_to_narwhals_dtype(dtype: Any, dtypes: DTypes) -> DType:
+def native_to_narwhals_dtype(dtype: pl.DataType, dtypes: DTypes) -> DType:
     import polars as pl  # ignore-banned-import()
 
     if dtype == pl.Float64:
@@ -79,25 +80,26 @@ def native_to_narwhals_dtype(dtype: Any, dtypes: DTypes) -> DType:
         return dtypes.Struct(
             [
                 dtypes.Field(field_name, native_to_narwhals_dtype(field_type, dtypes))
-                for field_name, field_type in dtype
+                for field_name, field_type in dtype  # type: ignore[attr-defined]
             ]
         )
     if dtype == pl.List:
-        return dtypes.List(native_to_narwhals_dtype(dtype.inner, dtypes))
+        return dtypes.List(native_to_narwhals_dtype(dtype.inner, dtypes))  # type: ignore[attr-defined]
     if dtype == pl.Array:
         if parse_version(pl.__version__) < (0, 20, 30):  # pragma: no cover
             return dtypes.Array(
-                native_to_narwhals_dtype(dtype.inner, dtypes), dtype.width
+                native_to_narwhals_dtype(dtype.inner, dtypes),  # type: ignore[attr-defined]
+                dtype.width,  # type: ignore[attr-defined]
             )
         else:
-            return dtypes.Array(native_to_narwhals_dtype(dtype.inner, dtypes), dtype.size)
+            return dtypes.Array(native_to_narwhals_dtype(dtype.inner, dtypes), dtype.size)  # type: ignore[attr-defined]
     return dtypes.Unknown()
 
 
-def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
-    if (pl := get_polars()) is not None and isinstance(
-        dtype, (pl.DataType, pl.DataType.__class__)
-    ):
+def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> pl.DataType:
+    import polars as pl  # ignore-banned-import()
+
+    if isinstance(dtype, (pl.DataType, pl.DataType.__class__)):  # type: ignore[arg-type]
         msg = (
             f"Expected Narwhals object, got: {type(dtype)}.\n\n"
             "Perhaps you:\n"
@@ -106,8 +108,6 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
         )
         raise TypeError(msg)
 
-    import polars as pl  # ignore-banned-import()
-
     if dtype == dtypes.Float64:
         return pl.Float64()
     if dtype == dtypes.Float32:
@@ -142,7 +142,7 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
     if dtype == dtypes.Date:
         return pl.Date()
     if dtype == dtypes.Datetime or isinstance(dtype, dtypes.Datetime):
-        dt_time_unit = getattr(dtype, "time_unit", "us")
+        dt_time_unit: Literal["ms", "us", "ns"] = getattr(dtype, "time_unit", "us")
         dt_time_zone = getattr(dtype, "time_zone", None)
         return pl.Datetime(dt_time_unit, dt_time_zone)
     if dtype == dtypes.Duration or isinstance(dtype, dtypes.Duration):
@@ -151,13 +151,13 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], dtypes: DTypes) -> Any:
 
     if dtype == dtypes.List:  # pragma: no cover
         msg = "Converting to List dtype is not supported yet"
-        return NotImplementedError(msg)
+        raise NotImplementedError(msg)
     if dtype == dtypes.Struct:  # pragma: no cover
         msg = "Converting to Struct dtype is not supported yet"
-        return NotImplementedError(msg)
+        raise NotImplementedError(msg)
     if dtype == dtypes.Array:  # pragma: no cover
         msg = "Converting to Array dtype is not supported yet"
-        return NotImplementedError(msg)
+        raise NotImplementedError(msg)
     return pl.Unknown()  # pragma: no cover
 
 
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index bb163b28..34482072 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -2567,6 +2567,91 @@ class DataFrame(BaseFrame[DataFrameT]):
         """
         return super().gather_every(n=n, offset=offset)
 
+    def pivot(
+        self: Self,
+        on: str | list[str],
+        *,
+        index: str | list[str] | None = None,
+        values: str | list[str] | None = None,
+        aggregate_function: Literal[
+            "min", "max", "first", "last", "sum", "mean", "median", "len"
+        ]
+        | None = None,
+        maintain_order: bool = True,
+        sort_columns: bool = False,
+        separator: str = "_",
+    ) -> Self:
+        r"""
+        Create a spreadsheet-style pivot table as a DataFrame.
+
+        Arguments:
+            on: Name of the column(s) whose values will be used as the header of the
+                output DataFrame.
+            index: One or multiple keys to group by. If None, all remaining columns not
+                specified on `on` and `values` will be used. At least one of `index` and
+                `values` must be specified.
+            values: One or multiple keys to group by. If None, all remaining columns not
+                specified on `on` and `index` will be used. At least one of `index` and
+                `values` must be specified.
+            aggregate_function: Choose from:
+                - None: no aggregation takes place, will raise error if multiple values
+                    are in group.
+                - A predefined aggregate function string, one of
+                    {'min', 'max', 'first', 'last', 'sum', 'mean', 'median', 'len'}
+            maintain_order: Sort the grouped keys so that the output order is predictable.
+            sort_columns: Sort the transposed columns by name. Default is by order of
+                discovery.
+            separator: Used as separator/delimiter in generated column names in case of
+                multiple `values` columns.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> data = {
+            ...     "ix": [1, 1, 2, 2, 1, 2],
+            ...     "col": ["a", "a", "a", "a", "b", "b"],
+            ...     "foo": [0, 1, 2, 2, 7, 1],
+            ...     "bar": [0, 2, 0, 0, 9, 4],
+            ... }
+            >>> df_pd = pd.DataFrame(data)
+            >>> df_pl = pl.DataFrame(data)
+
+            Let's define a dataframe-agnostic function:
+
+            >>> @nw.narwhalify
+            ... def func(df):
+            ...     return df.pivot("col", index="ix", aggregate_function="sum")
+
+            We can then pass either pandas or Polars to `func`:
+
+            >>> func(df_pd)
+               ix  foo_a  foo_b  bar_a  bar_b
+            0   1      1      7      2      9
+            1   2      4      1      0      4
+            >>> func(df_pl)
+            shape: (2, 5)
+            ┌─────┬───────┬───────┬───────┬───────┐
+            │ ix  ┆ foo_a ┆ foo_b ┆ bar_a ┆ bar_b │
+            │ --- ┆ ---   ┆ ---   ┆ ---   ┆ ---   │
+            │ i64 ┆ i64   ┆ i64   ┆ i64   ┆ i64   │
+            ╞═════╪═══════╪═══════╪═══════╪═══════╡
+            │ 1   ┆ 1     ┆ 7     ┆ 2     ┆ 9     │
+            │ 2   ┆ 4     ┆ 1     ┆ 0     ┆ 4     │
+            └─────┴───────┴───────┴───────┴───────┘
+        """
+        return self._from_compliant_dataframe(
+            self._compliant_frame.pivot(
+                on=on,
+                index=index,
+                values=values,
+                aggregate_function=aggregate_function,
+                maintain_order=maintain_order,
+                sort_columns=sort_columns,
+                separator=separator,
+            )
+        )
+
     def to_arrow(self: Self) -> pa.Table:
         r"""
         Convert to arrow table.
diff --git a/narwhals/expr.py b/narwhals/expr.py
index 9349dd7d..7a280c02 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -539,6 +539,75 @@ class Expr:
         """
         return self.__class__(lambda plx: self._call(plx).std(ddof=ddof))
 
+    def map_batches(
+        self,
+        function: Callable[[Any], Self],
+        return_dtype: DType | None = None,
+    ) -> Self:
+        """
+        Apply a custom python function to a whole Series or sequence of Series.
+
+        The output of this custom function is presumed to be either a Series,
+        or a NumPy array (in which case it will be automatically converted into
+        a Series).
+
+        Arguments:
+            return_dtype: Dtype of the output Series.
+                          If not set, the dtype will be inferred based on the first non-null value
+                          that is returned by the function.
+
+        Examples:
+            >>> import polars as pl
+            >>> import pandas as pd
+            >>> import pyarrow as pa
+            >>> import narwhals as nw
+            >>> data = {"a": [1, 2, 3], "b": [4, 5, 6]}
+            >>> df_pd = pd.DataFrame(data)
+            >>> df_pl = pl.DataFrame(data)
+            >>> df_pa = pa.table(data)
+
+            Let's define a dataframe-agnostic function:
+
+            >>> @nw.narwhalify
+            ... def func(df):
+            ...     return df.select(
+            ...         nw.col("a", "b").map_batches(
+            ...             lambda s: s.to_numpy() + 1, return_dtype=nw.Float64
+            ...         )
+            ...     )
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> func(df_pd)
+                 a    b
+            0  2.0  5.0
+            1  3.0  6.0
+            2  4.0  7.0
+            >>> func(df_pl)
+            shape: (3, 2)
+            ┌─────┬─────┐
+            │ a   ┆ b   │
+            │ --- ┆ --- │
+            │ f64 ┆ f64 │
+            ╞═════╪═════╡
+            │ 2.0 ┆ 5.0 │
+            │ 3.0 ┆ 6.0 │
+            │ 4.0 ┆ 7.0 │
+            └─────┴─────┘
+            >>> func(df_pa)
+            pyarrow.Table
+            a: double
+            b: double
+            ----
+            a: [[2,3,4]]
+            b: [[5,6,7]]
+        """
+        return self.__class__(
+            lambda plx: self._call(plx).map_batches(
+                function=function, return_dtype=return_dtype
+            )
+        )
+
     def sum(self) -> Expr:
         """
         Return the sum value.
@@ -1472,10 +1541,22 @@ class Expr:
         """
         return self.__class__(lambda plx: self._call(plx).arg_true())
 
-    def fill_null(self, value: Any) -> Self:
+    def fill_null(
+        self,
+        value: Any | None = None,
+        strategy: Literal["forward", "backward"] | None = None,
+        limit: int | None = None,
+    ) -> Self:
         """
         Fill null values with given value.
 
+        Arguments:
+            value: Value used to fill null values.
+
+            strategy: Strategy used to fill null values.
+
+            limit: Number of consecutive null values to fill when using the 'forward' or 'backward' strategy.
+
         Notes:
             pandas and Polars handle null values differently. Polars distinguishes
             between NaN and Null, whereas pandas doesn't.
@@ -1486,13 +1567,22 @@ class Expr:
             >>> import pyarrow as pa
             >>> import narwhals as nw
             >>> df_pd = pd.DataFrame(
-            ...     {"a": [2, 4, None, 3, 5], "b": [2.0, 4.0, float("nan"), 3.0, 5.0]}
+            ...     {
+            ...         "a": [2, 4, None, None, 3, 5],
+            ...         "b": [2.0, 4.0, float("nan"), float("nan"), 3.0, 5.0],
+            ...     }
             ... )
             >>> df_pl = pl.DataFrame(
-            ...     {"a": [2, 4, None, 3, 5], "b": [2.0, 4.0, float("nan"), 3.0, 5.0]}
+            ...     {
+            ...         "a": [2, 4, None, None, 3, 5],
+            ...         "b": [2.0, 4.0, float("nan"), float("nan"), 3.0, 5.0],
+            ...     }
             ... )
             >>> df_pa = pa.table(
-            ...     {"a": [2, 4, None, 3, 5], "b": [2.0, 4.0, float("nan"), 3.0, 5.0]}
+            ...     {
+            ...         "a": [2, 4, None, None, 3, 5],
+            ...         "b": [2.0, 4.0, float("nan"), float("nan"), 3.0, 5.0],
+            ...     }
             ... )
 
             Let's define a dataframe-agnostic function:
@@ -1508,11 +1598,12 @@ class Expr:
             0  2.0  2.0
             1  4.0  4.0
             2  0.0  0.0
-            3  3.0  3.0
-            4  5.0  5.0
+            3  0.0  0.0
+            4  3.0  3.0
+            5  5.0  5.0
 
             >>> func(df_pl)  # nan != null for polars
-            shape: (5, 2)
+            shape: (6, 2)
             ┌─────┬─────┐
             │ a   ┆ b   │
             │ --- ┆ --- │
@@ -1521,6 +1612,7 @@ class Expr:
             │ 2   ┆ 2.0 │
             │ 4   ┆ 4.0 │
             │ 0   ┆ NaN │
+            │ 0   ┆ NaN │
             │ 3   ┆ 3.0 │
             │ 5   ┆ 5.0 │
             └─────┴─────┘
@@ -1530,10 +1622,69 @@ class Expr:
             a: int64
             b: double
             ----
-            a: [[2,4,0,3,5]]
-            b: [[2,4,nan,3,5]]
+            a: [[2,4,0,0,3,5]]
+            b: [[2,4,nan,nan,3,5]]
+
+            Using a strategy:
+
+            >>> @nw.narwhalify
+            ... def func_strategies(df):
+            ...     return df.with_columns(
+            ...         nw.col("a", "b")
+            ...         .fill_null(strategy="forward", limit=1)
+            ...         .name.suffix("_filled")
+            ...     )
+
+            >>> func_strategies(df_pd)
+                 a    b  a_filled  b_filled
+            0  2.0  2.0       2.0       2.0
+            1  4.0  4.0       4.0       4.0
+            2  NaN  NaN       4.0       4.0
+            3  NaN  NaN       NaN       NaN
+            4  3.0  3.0       3.0       3.0
+            5  5.0  5.0       5.0       5.0
+
+            >>> func_strategies(df_pl)  # nan != null for polars
+            shape: (6, 4)
+            ┌──────┬─────┬──────────┬──────────┐
+            │ a    ┆ b   ┆ a_filled ┆ b_filled │
+            │ ---  ┆ --- ┆ ---      ┆ ---      │
+            │ i64  ┆ f64 ┆ i64      ┆ f64      │
+            ╞══════╪═════╪══════════╪══════════╡
+            │ 2    ┆ 2.0 ┆ 2        ┆ 2.0      │
+            │ 4    ┆ 4.0 ┆ 4        ┆ 4.0      │
+            │ null ┆ NaN ┆ 4        ┆ NaN      │
+            │ null ┆ NaN ┆ null     ┆ NaN      │
+            │ 3    ┆ 3.0 ┆ 3        ┆ 3.0      │
+            │ 5    ┆ 5.0 ┆ 5        ┆ 5.0      │
+            └──────┴─────┴──────────┴──────────┘
+
+            >>> func_strategies(df_pa)  # nan != null for pyarrow
+            pyarrow.Table
+            a: int64
+            b: double
+            a_filled: int64
+            b_filled: double
+            ----
+            a: [[2,4,null,null,3,5]]
+            b: [[2,4,nan,nan,3,5]]
+            a_filled: [[2,4,4,null,3,5]]
+            b_filled: [[2,4,nan,nan,3,5]]
         """
-        return self.__class__(lambda plx: self._call(plx).fill_null(value))
+        if value is not None and strategy is not None:
+            msg = "cannot specify both `value` and `strategy`"
+            raise ValueError(msg)
+        if value is None and strategy is None:
+            msg = "must specify either a fill `value` or `strategy`"
+            raise ValueError(msg)
+        if strategy is not None and strategy not in {"forward", "backward"}:
+            msg = f"strategy not supported: {strategy}"
+            raise ValueError(msg)
+        return self.__class__(
+            lambda plx: self._call(plx).fill_null(
+                value=value, strategy=strategy, limit=limit
+            )
+        )
 
     # --- partial reduction ---
     def drop_nulls(self) -> Self:
@@ -5203,8 +5354,8 @@ def lit(value: Any, dtype: DType | None = None) -> Expr:
 
         >>> func(df_pd)
            a  literal
-        0  1  3
-        1  2  3
+        0  1        3
+        1  2        3
         >>> func(df_pl)
         shape: (2, 2)
         ┌─────┬─────────┐
diff --git a/narwhals/functions.py b/narwhals/functions.py
index 3fd2025a..9d7d0553 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -164,7 +164,7 @@ def new_series(
     native_namespace: ModuleType,
 ) -> Series:
     """
-    Instantiate Narwhals Series from raw data.
+    Instantiate Narwhals Series from iterable (e.g. list or array).
 
     Arguments:
         name: Name of resulting Series.
@@ -188,7 +188,12 @@ def new_series(
         ... def func(df):
         ...     values = [4, 1, 2]
         ...     native_namespace = nw.get_native_namespace(df)
-        ...     return nw.new_series("c", values, nw.Int32, native_namespace=native_namespace)
+        ...     return nw.new_series(
+        ...         name="c",
+        ...         values=values,
+        ...         dtype=nw.Int32,
+        ...         native_namespace=native_namespace,
+        ...     )
 
         Let's see what happens when passing pandas / Polars input:
 
@@ -233,9 +238,11 @@ def _new_series_impl(
                 narwhals_to_native_dtype as polars_narwhals_to_native_dtype,
             )
 
-            dtype = polars_narwhals_to_native_dtype(dtype, dtypes=dtypes)
+            dtype_pl = polars_narwhals_to_native_dtype(dtype, dtypes=dtypes)
+        else:
+            dtype_pl = None
 
-        native_series = native_namespace.Series(name=name, values=values, dtype=dtype)
+        native_series = native_namespace.Series(name=name, values=values, dtype=dtype_pl)
     elif implementation in {
         Implementation.PANDAS,
         Implementation.MODIN,
@@ -376,12 +383,14 @@ def _from_dict_impl(
                 narwhals_to_native_dtype as polars_narwhals_to_native_dtype,
             )
 
-            schema = {
+            schema_pl = {
                 name: polars_narwhals_to_native_dtype(dtype, dtypes=dtypes)
                 for name, dtype in schema.items()
             }
+        else:
+            schema_pl = None
 
-        native_frame = native_namespace.from_dict(data, schema=schema)
+        native_frame = native_namespace.from_dict(data, schema=schema_pl)
     elif implementation in {
         Implementation.PANDAS,
         Implementation.MODIN,
diff --git a/narwhals/series.py b/narwhals/series.py
index 8b3ad25f..0b5dfcc1 100644
--- a/narwhals/series.py
+++ b/narwhals/series.py
@@ -1565,13 +1565,22 @@ class Series:
         """
         return self._from_compliant_series(self._compliant_series.is_null())
 
-    def fill_null(self, value: Any) -> Self:
+    def fill_null(
+        self,
+        value: Any | None = None,
+        strategy: Literal["forward", "backward"] | None = None,
+        limit: int | None = None,
+    ) -> Self:
         """
         Fill null values using the specified value.
 
         Arguments:
             value: Value used to fill null values.
 
+            strategy: Strategy used to fill null values.
+
+            limit: Number of consecutive null values to fill when using the 'forward' or 'backward' strategy.
+
         Notes:
             pandas and Polars handle null values differently. Polars distinguishes
             between NaN and Null, whereas pandas doesn't.
@@ -1605,8 +1614,40 @@ class Series:
                2
                5
             ]
+
+            Using a strategy:
+
+            >>> @nw.narwhalify
+            ... def func_strategies(s):
+            ...     return s.fill_null(strategy="forward", limit=1)
+
+            >>> func_strategies(s_pd)
+            0    1.0
+            1    2.0
+            2    2.0
+            dtype: float64
+
+            >>> func_strategies(s_pl)  # doctest: +NORMALIZE_WHITESPACE
+            shape: (3,)
+            Series: '' [i64]
+            [
+               1
+               2
+               2
+            ]
         """
-        return self._from_compliant_series(self._compliant_series.fill_null(value))
+        if value is not None and strategy is not None:
+            msg = "cannot specify both `value` and `strategy`"
+            raise ValueError(msg)
+        if value is None and strategy is None:
+            msg = "must specify either a fill `value` or `strategy`"
+            raise ValueError(msg)
+        if strategy is not None and strategy not in {"forward", "backward"}:
+            msg = f"strategy not supported: {strategy}"
+            raise ValueError(msg)
+        return self._from_compliant_series(
+            self._compliant_series.fill_null(value=value, strategy=strategy, limit=limit)
+        )
 
     def is_between(
         self, lower_bound: Any, upper_bound: Any, closed: str = "both"
@@ -2509,7 +2550,7 @@ class Series:
             ┌─────┬─────┬─────┐
             │ a_1 ┆ a_2 ┆ a_3 │
             │ --- ┆ --- ┆ --- │
-            │ u8  ┆ u8  ┆ u8  │
+            │ i8  ┆ i8  ┆ i8  │
             ╞═════╪═════╪═════╡
             │ 1   ┆ 0   ┆ 0   │
             │ 0   ┆ 1   ┆ 0   │
@@ -2520,7 +2561,7 @@ class Series:
             ┌─────┬─────┐
             │ a_2 ┆ a_3 │
             │ --- ┆ --- │
-            │ u8  ┆ u8  │
+            │ i8  ┆ i8  │
             ╞═════╪═════╡
             │ 0   ┆ 0   │
             │ 1   ┆ 0   │
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index 85e226e9..738f19ec 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -1453,8 +1453,8 @@ def lit(value: Any, dtype: DType | None = None) -> Expr:
 
         >>> func(df_pd)
            a  literal
-        0  1  3
-        1  2  3
+        0  1        3
+        1  2        3
         >>> func(df_pl)
         shape: (2, 2)
         ┌─────┬─────────┐
@@ -2363,7 +2363,7 @@ def new_series(
     native_namespace: ModuleType,
 ) -> Series:
     """
-    Instantiate Narwhals Series from raw data.
+    Instantiate Narwhals Series from iterable (e.g. list or array).
 
     Arguments:
         name: Name of resulting Series.
@@ -2387,7 +2387,12 @@ def new_series(
         ... def func(df):
         ...     values = [4, 1, 2]
         ...     native_namespace = nw.get_native_namespace(df)
-        ...     return nw.new_series("c", values, nw.Int32, native_namespace=native_namespace)
+        ...     return nw.new_series(
+        ...         name="c",
+        ...         values=values,
+        ...         dtype=nw.Int32,
+        ...         native_namespace=native_namespace,
+        ...     )
 
         Let's see what happens when passing pandas / Polars input:
 
diff --git a/narwhals/utils.py b/narwhals/utils.py
index e538bee1..e1a7b812 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -274,15 +274,37 @@ def maybe_get_index(obj: T) -> Any | None:
     return None
 
 
-def maybe_set_index(df: T, column_names: str | list[str]) -> T:
+def maybe_set_index(
+    obj: T,
+    column_names: str | list[str] | None = None,
+    *,
+    index: Series | list[Series] | None = None,
+) -> T:
     """
-    Set columns `columns` to be the index of `df`, if `df` is pandas-like.
+    Set the index of a DataFrame or a Series, if it's pandas-like.
+
+    Arguments:
+        obj: object for which maybe set the index (can be either a Narwhals `DataFrame`
+            or `Series`).
+        column_names: name or list of names of the columns to set as index.
+            For dataframes, only one of `column_names` and `index` can be specified but
+            not both. If `column_names` is passed and `df` is a Series, then a
+            `ValueError` is raised.
+        index: series or list of series to set as index.
+
+    Raises:
+        ValueError: If one of the following condition happens:
+
+            - none of `column_names` and `index` are provided
+            - both `column_names` and `index` are provided
+            - `column_names` is provided and `df` is a Series
 
     Notes:
-        This is only really intended for backwards-compatibility purposes,
-        for example if your library already aligns indices for users.
+        This is only really intended for backwards-compatibility purposes, for example if
+        your library already aligns indices for users.
         If you're designing a new library, we highly encourage you to not
         rely on the Index.
+
         For non-pandas-like inputs, this is a no-op.
 
     Examples:
@@ -297,15 +319,49 @@ def maybe_set_index(df: T, column_names: str | list[str]) -> T:
         4  1
         5  2
     """
-    df_any = cast(Any, df)
-    native_frame = to_native(df_any)
-    if is_pandas_like_dataframe(native_frame):
+
+    df_any = cast(Any, obj)
+    native_obj = to_native(df_any)
+
+    if column_names is not None and index is not None:
+        msg = "Only one of `column_names` or `index` should be provided"
+        raise ValueError(msg)
+
+    if not column_names and not index:
+        msg = "Either `column_names` or `index` should be provided"
+        raise ValueError(msg)
+
+    if index is not None:
+        keys = (
+            [to_native(idx, pass_through=True) for idx in index]
+            if _is_iterable(index)
+            else to_native(index, pass_through=True)
+        )
+    else:
+        keys = column_names
+
+    if is_pandas_like_dataframe(native_obj):
         return df_any._from_compliant_dataframe(  # type: ignore[no-any-return]
-            df_any._compliant_frame._from_native_frame(
-                native_frame.set_index(column_names)
-            )
+            df_any._compliant_frame._from_native_frame(native_obj.set_index(keys))
+        )
+    elif is_pandas_like_series(native_obj):
+        if column_names:
+            msg = "Cannot set index using column names on a Series"
+            raise ValueError(msg)
+
+        if (
+            df_any._compliant_series._implementation is Implementation.PANDAS
+            and df_any._compliant_series._backend_version < (1,)
+        ):  # pragma: no cover
+            native_obj = native_obj.set_axis(keys, inplace=False)
+        else:
+            native_obj = native_obj.set_axis(keys)
+
+        return df_any._from_compliant_series(  # type: ignore[no-any-return]
+            df_any._compliant_series._from_native_series(native_obj)
         )
-    return df_any  # type: ignore[no-any-return]
+    else:
+        return df_any  # type: ignore[no-any-return]
 
 
 def maybe_reset_index(obj: T) -> T:
@@ -482,12 +538,11 @@ def is_ordered_categorical(series: Series) -> bool:
 
 
 def generate_unique_token(n_bytes: int, columns: list[str]) -> str:  # pragma: no cover
-    warn(
+    msg = (
         "Use `generate_temporary_column_name` instead. `generate_unique_token` is "
-        "deprecated and it will be removed in future versions",
-        DeprecationWarning,
-        stacklevel=2,
+        "deprecated and it will be removed in future versions"
     )
+    issue_deprecation_warning(msg, _version="1.13.0")
     return generate_temporary_column_name(n_bytes=n_bytes, columns=columns)
 
 
diff --git a/pyproject.toml b/pyproject.toml
index fdc7aecf..9f20bcf7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "narwhals"
-version = "1.13.3"
+version = "1.13.4"
 authors = [
   { name="Marco Gorelli", email="33491632+MarcoGorelli@users.noreply.github.com" },
 ]
@@ -65,6 +65,7 @@ lint.ignore = [
   "ISC001",
   "NPY002",
   "PD901",  # This is a auxiliary library so dataframe variables have no concrete business meaning
+  "PD010",
   "PLR0911",
   "PLR0912",
   "PLR0913",
@@ -116,7 +117,7 @@ filterwarnings = [
   'ignore:.*You are using pyarrow version',
   'ignore:.*but when imported by',
   'ignore:Distributing .*This may take some time',
-  'ignore:.*The default coalesce behavior'
+  'ignore:.*The default coalesce behavior',
 ]
 xfail_strict = true
 markers = ["slow: marks tests as slow (deselect with '-m \"not slow\"')"]
diff --git a/tpch/execute/q1.py b/tpch/execute/q1.py
index d0ebce58..6bfdc1a1 100644
--- a/tpch/execute/q1.py
+++ b/tpch/execute/q1.py
@@ -8,4 +8,3 @@ from . import lineitem
 print(q1.query(IO_FUNCS["pandas[pyarrow]"](lineitem)))
 print(q1.query(IO_FUNCS["polars[lazy]"](lineitem)).collect())
 print(q1.query(IO_FUNCS["pyarrow"](lineitem)))
-print(q1.query(IO_FUNCS["dask"](lineitem)).compute())
diff --git a/tpch/execute/q10.py b/tpch/execute/q10.py
index 1f610932..fefcd0e8 100644
--- a/tpch/execute/q10.py
+++ b/tpch/execute/q10.py
@@ -19,7 +19,3 @@ print(q10.query(fn(customer), fn(nation), fn(lineitem), fn(orders)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q10.query(fn(customer), fn(nation), fn(lineitem), fn(orders)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q10.query(fn(customer), fn(nation), fn(lineitem), fn(orders)).compute())
diff --git a/tpch/execute/q11.py b/tpch/execute/q11.py
index 0dd8a243..ad0146d3 100644
--- a/tpch/execute/q11.py
+++ b/tpch/execute/q11.py
@@ -18,7 +18,3 @@ print(q11.query(fn(nation), fn(partsupp), fn(supplier)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q11.query(fn(nation), fn(partsupp), fn(supplier)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q11.query(fn(nation), fn(partsupp), fn(supplier)).compute())
diff --git a/tpch/execute/q12.py b/tpch/execute/q12.py
index f684e22a..42d6d245 100644
--- a/tpch/execute/q12.py
+++ b/tpch/execute/q12.py
@@ -17,7 +17,3 @@ print(q12.query(fn(line_item), fn(orders)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q12.query(fn(line_item), fn(orders)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q12.query(fn(line_item), fn(orders)).compute())
diff --git a/tpch/execute/q13.py b/tpch/execute/q13.py
index 7b03a2f2..b7162d97 100644
--- a/tpch/execute/q13.py
+++ b/tpch/execute/q13.py
@@ -17,7 +17,3 @@ print(q13.query(fn(customer), fn(orders)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q13.query(fn(customer), fn(orders)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q13.query(fn(customer), fn(orders)).compute())
diff --git a/tpch/execute/q14.py b/tpch/execute/q14.py
index a8233013..44313010 100644
--- a/tpch/execute/q14.py
+++ b/tpch/execute/q14.py
@@ -17,7 +17,3 @@ print(q14.query(fn(line_item), fn(part)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q14.query(fn(line_item), fn(part)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q14.query(fn(line_item), fn(part)).compute())
diff --git a/tpch/execute/q15.py b/tpch/execute/q15.py
index 40b4432b..1fca0b07 100644
--- a/tpch/execute/q15.py
+++ b/tpch/execute/q15.py
@@ -17,7 +17,3 @@ print(q15.query(fn(lineitem), fn(supplier)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q15.query(fn(lineitem), fn(supplier)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q15.query(fn(lineitem), fn(supplier)).compute())
diff --git a/tpch/execute/q16.py b/tpch/execute/q16.py
index ef30f935..4d65a1b5 100644
--- a/tpch/execute/q16.py
+++ b/tpch/execute/q16.py
@@ -18,7 +18,3 @@ print(q16.query(fn(part), fn(partsupp), fn(supplier)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q16.query(fn(part), fn(partsupp), fn(supplier)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q16.query(fn(part), fn(partsupp), fn(supplier)).compute())
diff --git a/tpch/execute/q17.py b/tpch/execute/q17.py
index 0b7ca4a6..bec1abdf 100644
--- a/tpch/execute/q17.py
+++ b/tpch/execute/q17.py
@@ -17,7 +17,3 @@ print(q17.query(fn(lineitem), fn(part)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q17.query(fn(lineitem), fn(part)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q17.query(fn(lineitem), fn(part)).compute())
diff --git a/tpch/execute/q18.py b/tpch/execute/q18.py
index a096deb2..727780a7 100644
--- a/tpch/execute/q18.py
+++ b/tpch/execute/q18.py
@@ -18,7 +18,3 @@ print(q18.query(fn(customer), fn(lineitem), fn(orders)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q18.query(fn(customer), fn(lineitem), fn(orders)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q18.query(fn(customer), fn(lineitem), fn(orders)).compute())
diff --git a/tpch/execute/q19.py b/tpch/execute/q19.py
index 23095a89..3dd966ea 100644
--- a/tpch/execute/q19.py
+++ b/tpch/execute/q19.py
@@ -14,6 +14,3 @@ print(q19.query(fn(lineitem), fn(part)).collect())
 
 fn = IO_FUNCS["pyarrow"]
 print(q19.query(fn(lineitem), fn(part)))
-
-fn = IO_FUNCS["dask"]
-print(q19.query(fn(lineitem), fn(part)).compute())
diff --git a/tpch/execute/q2.py b/tpch/execute/q2.py
index 0e2d0701..2aff3c0a 100644
--- a/tpch/execute/q2.py
+++ b/tpch/execute/q2.py
@@ -42,14 +42,3 @@ print(
         fn(partsupp),
     )
 )
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(
-    q2.query(
-        fn(region),
-        fn(nation),
-        fn(supplier),
-        fn(part),
-        fn(partsupp),
-    ).compute()
-)
diff --git a/tpch/execute/q20.py b/tpch/execute/q20.py
index c4ffa43b..b61d2cdd 100644
--- a/tpch/execute/q20.py
+++ b/tpch/execute/q20.py
@@ -17,6 +17,3 @@ print(q20.query(fn(part), fn(partsupp), fn(nation), fn(lineitem), fn(supplier)).
 
 fn = IO_FUNCS["pyarrow"]
 print(q20.query(fn(part), fn(partsupp), fn(nation), fn(lineitem), fn(supplier)))
-
-fn = IO_FUNCS["dask"]
-print(q20.query(fn(part), fn(partsupp), fn(nation), fn(lineitem), fn(supplier)).compute())
diff --git a/tpch/execute/q21.py b/tpch/execute/q21.py
index d6fb272a..3fb41232 100644
--- a/tpch/execute/q21.py
+++ b/tpch/execute/q21.py
@@ -16,6 +16,3 @@ print(q21.query(fn(lineitem), fn(nation), fn(orders), fn(supplier)).collect())
 
 fn = IO_FUNCS["pyarrow"]
 print(q21.query(fn(lineitem), fn(nation), fn(orders), fn(supplier)))
-
-fn = IO_FUNCS["dask"]
-print(q21.query(fn(lineitem), fn(nation), fn(orders), fn(supplier)).compute())
diff --git a/tpch/execute/q22.py b/tpch/execute/q22.py
index f71fc422..96744ea5 100644
--- a/tpch/execute/q22.py
+++ b/tpch/execute/q22.py
@@ -17,7 +17,3 @@ print(q22.query(fn(customer), fn(orders)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q22.query(fn(customer), fn(orders)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q22.query(fn(customer), fn(orders)).compute())
diff --git a/tpch/execute/q3.py b/tpch/execute/q3.py
index bbcc51d5..17b7cf56 100644
--- a/tpch/execute/q3.py
+++ b/tpch/execute/q3.py
@@ -18,7 +18,3 @@ print(q3.query(fn(customer), fn(lineitem), fn(orders)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q3.query(fn(customer), fn(lineitem), fn(orders)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q3.query(fn(customer), fn(lineitem), fn(orders)).compute())
diff --git a/tpch/execute/q4.py b/tpch/execute/q4.py
index bcfd3a15..a1bfe95f 100644
--- a/tpch/execute/q4.py
+++ b/tpch/execute/q4.py
@@ -17,7 +17,3 @@ print(q4.query(fn(line_item), fn(orders)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q4.query(fn(line_item), fn(orders)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q4.query(fn(line_item), fn(orders)).compute())
diff --git a/tpch/execute/q5.py b/tpch/execute/q5.py
index 66524c5a..2bdb0554 100644
--- a/tpch/execute/q5.py
+++ b/tpch/execute/q5.py
@@ -33,11 +33,3 @@ print(
         fn(region), fn(nation), fn(customer), fn(line_item), fn(orders), fn(supplier)
     )
 )
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(
-    q5.query(
-        fn(region), fn(nation), fn(customer), fn(line_item), fn(orders), fn(supplier)
-    ).compute()
-)
diff --git a/tpch/execute/q6.py b/tpch/execute/q6.py
index 1d650b79..8a87bd70 100644
--- a/tpch/execute/q6.py
+++ b/tpch/execute/q6.py
@@ -16,7 +16,3 @@ print(q6.query(fn(lineitem)).collect())
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q6.query(fn(lineitem)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(q6.query(fn(lineitem)).compute())
diff --git a/tpch/execute/q7.py b/tpch/execute/q7.py
index 069fb258..ea15568f 100644
--- a/tpch/execute/q7.py
+++ b/tpch/execute/q7.py
@@ -22,9 +22,3 @@ print(
 tool = "pyarrow"
 fn = IO_FUNCS[tool]
 print(q7.query(fn(nation), fn(customer), fn(lineitem), fn(orders), fn(supplier)))
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(
-    q7.query(fn(nation), fn(customer), fn(lineitem), fn(orders), fn(supplier)).compute()
-)
diff --git a/tpch/execute/q8.py b/tpch/execute/q8.py
index 8c3aa5de..9e4f706a 100644
--- a/tpch/execute/q8.py
+++ b/tpch/execute/q8.py
@@ -53,17 +53,3 @@ print(
         fn(region),
     )
 )
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(
-    q8.query(
-        fn(part),
-        fn(supplier),
-        fn(lineitem),
-        fn(orders),
-        fn(customer),
-        fn(nation),
-        fn(region),
-    ).compute()
-)
diff --git a/tpch/execute/q9.py b/tpch/execute/q9.py
index 4c8e6874..6f191d76 100644
--- a/tpch/execute/q9.py
+++ b/tpch/execute/q9.py
@@ -29,11 +29,3 @@ fn = IO_FUNCS[tool]
 print(
     q9.query(fn(part), fn(partsupp), fn(nation), fn(lineitem), fn(orders), fn(supplier))
 )
-
-tool = "dask"
-fn = IO_FUNCS[tool]
-print(
-    q9.query(
-        fn(part), fn(partsupp), fn(nation), fn(lineitem), fn(orders), fn(supplier)
-    ).compute()
-)
diff --git a/utils/check_api_reference.py b/utils/check_api_reference.py
index b7d8595a..199b60ed 100644
--- a/utils/check_api_reference.py
+++ b/utils/check_api_reference.py
@@ -12,7 +12,7 @@ from narwhals.utils import remove_suffix
 ret = 0
 
 NAMESPACES = {"dt", "str", "cat", "name"}
-EXPR_ONLY_METHODS = {"over"}
+EXPR_ONLY_METHODS = {"over", "map_batches"}
 SERIES_ONLY_METHODS = {
     "dtype",
     "is_empty",
diff --git a/utils/generate_backend_completeness.py b/utils/generate_backend_completeness.py
index d7d05daa..2aafc567 100644
--- a/utils/generate_backend_completeness.py
+++ b/utils/generate_backend_completeness.py
@@ -68,7 +68,7 @@ def render_table_and_write_to_output(
     results: list[pl.DataFrame], title: str, output_filename: str
 ) -> None:
     results = (
-        pl.concat(results)  # noqa: PD010
+        pl.concat(results)
         .with_columns(supported=pl.lit(":white_check_mark:"))
         .pivot(on="Backend", values="supported", index=["Class", "Method"])
         .filter(pl.col("narwhals").is_not_null())


Test patch:
diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
index 7847939b..17fff6b3 100644
--- a/.github/workflows/pytest.yml
+++ b/.github/workflows/pytest.yml
@@ -86,9 +86,9 @@ jobs:
         run: uv pip install ibis-framework[duckdb]>=6.0.0 --system
         # Ibis puts upper bounds on dependencies, and requires Python3.10+,
         # which messes with other dependencies on lower Python versions
-        if: matrix.python-version == '3.12'
+        if: matrix.python-version == '3.11'
       - name: Run pytest
         run: pytest tests --cov=narwhals --cov=tests --cov-fail-under=100 --runslow
       - name: Run doctests
-        if: matrix.python-version == '3.12'
+        if: matrix.python-version == '3.13'
         run: pytest narwhals --doctest-modules
diff --git a/tests/expr_and_series/fill_null_test.py b/tests/expr_and_series/fill_null_test.py
index 47117269..32e4b9cd 100644
--- a/tests/expr_and_series/fill_null_test.py
+++ b/tests/expr_and_series/fill_null_test.py
@@ -1,21 +1,26 @@
 from __future__ import annotations
 
+import warnings
+from contextlib import nullcontext as does_not_raise
+from typing import Any
+
+import pytest
+
 import narwhals.stable.v1 as nw
 from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
-data = {
-    "a": [0.0, None, 2, 3, 4],
-    "b": [1.0, None, None, 5, 3],
-    "c": [5.0, None, 3, 2, 1],
-}
-
 
 def test_fill_null(constructor: Constructor) -> None:
+    data = {
+        "a": [0.0, None, 2, 3, 4],
+        "b": [1.0, None, None, 5, 3],
+        "c": [5.0, None, 3, 2, 1],
+    }
     df = nw.from_native(constructor(data))
 
-    result = df.with_columns(nw.col("a", "b", "c").fill_null(99))
+    result = df.with_columns(nw.col("a", "b", "c").fill_null(value=99))
     expected = {
         "a": [0.0, 99, 2, 3, 4],
         "b": [1.0, 99, 99, 5, 3],
@@ -24,17 +29,297 @@ def test_fill_null(constructor: Constructor) -> None:
     assert_equal_data(result, expected)
 
 
+def test_fill_null_exceptions(constructor: Constructor) -> None:
+    data = {
+        "a": [0.0, None, 2, 3, 4],
+    }
+    df = nw.from_native(constructor(data))
+
+    with pytest.raises(ValueError, match="cannot specify both `value` and `strategy`"):
+        df.with_columns(nw.col("a").fill_null(value=99, strategy="forward"))
+
+    with pytest.raises(
+        ValueError, match="must specify either a fill `value` or `strategy`"
+    ):
+        df.with_columns(nw.col("a").fill_null())
+
+    with pytest.raises(ValueError, match="strategy not supported:"):
+        df.with_columns(nw.col("a").fill_null(strategy="invalid"))  # type: ignore  # noqa: PGH003
+
+
+def test_fill_null_strategies_with_limit_as_none(constructor: Constructor) -> None:
+    data_limits = {
+        "a": [1, None, None, None, 5, 6, None, None, None, 10],
+        "b": ["a", None, None, None, "b", "c", None, None, None, "d"],
+    }
+    df = nw.from_native(constructor(data_limits))
+
+    expected_forward = {
+        "a": [1, 1, 1, 1, 5, 6, 6, 6, 6, 10],
+        "b": ["a", "a", "a", "a", "b", "c", "c", "c", "c", "d"],
+    }
+    if (
+        "pandas_pyarrow_constructor" in str(constructor)
+        or "modin" in str(constructor)
+        or "dask" in str(constructor)
+    ):
+        with warnings.catch_warnings():
+            # case for modin and dask
+            warnings.filterwarnings(
+                "ignore", message="The 'downcast' keyword in fillna is deprecated"
+            )
+            # case for pandas_pyarrow_constructor
+            warnings.filterwarnings(
+                "ignore", message="Falling back on a non-pyarrow code path which"
+            )
+            result_forward = df.with_columns(
+                nw.col("a", "b").fill_null(strategy="forward", limit=None)
+            )
+            assert_equal_data(result_forward, expected_forward)
+    else:
+        result_forward = df.with_columns(
+            nw.col("a", "b").fill_null(strategy="forward", limit=None)
+        )
+
+        assert_equal_data(result_forward, expected_forward)
+
+    expected_backward = {
+        "a": [1, 5, 5, 5, 5, 6, 10, 10, 10, 10],
+        "b": ["a", "b", "b", "b", "b", "c", "d", "d", "d", "d"],
+    }
+    if (
+        "pandas_pyarrow_constructor" in str(constructor)
+        or "modin" in str(constructor)
+        or "dask" in str(constructor)
+    ):
+        with warnings.catch_warnings():
+            # case for modin and dask
+            warnings.filterwarnings(
+                "ignore", message="The 'downcast' keyword in fillna is deprecated"
+            )
+
+            # case for pandas_pyarrow_constructor
+            warnings.filterwarnings(
+                "ignore", message="Falling back on a non-pyarrow code path which"
+            )
+            result_backward = df.with_columns(
+                nw.col("a", "b").fill_null(strategy="backward", limit=None)
+            )
+            assert_equal_data(result_backward, expected_backward)
+    else:
+        result_backward = df.with_columns(
+            nw.col("a", "b").fill_null(strategy="backward", limit=None)
+        )
+        assert_equal_data(result_backward, expected_backward)
+
+
+def test_fill_null_limits(constructor: Constructor) -> None:
+    context: Any = (
+        pytest.raises(NotImplementedError, match="The limit keyword is not supported")
+        if "cudf" in str(constructor)
+        else warnings.catch_warnings()
+        if "modin" in str(constructor)
+        else does_not_raise()
+    )
+    data_limits = {
+        "a": [1, None, None, None, 5, 6, None, None, None, 10],
+        "b": ["a", None, None, None, "b", "c", None, None, None, "d"],
+    }
+    df = nw.from_native(constructor(data_limits))
+    with context:
+        if "modin" in str(constructor):
+            warnings.filterwarnings(
+                "ignore", message="The 'downcast' keyword in fillna is deprecated"
+            )
+
+        result_forward = df.with_columns(
+            nw.col("a", "b").fill_null(strategy="forward", limit=2)
+        )
+        expected_forward = {
+            "a": [1, 1, 1, float("nan"), 5, 6, 6, 6, float("nan"), 10],
+            "b": ["a", "a", "a", None, "b", "c", "c", "c", None, "d"],
+        }
+        assert_equal_data(result_forward, expected_forward)
+
+        result_backward = df.with_columns(
+            nw.col("a", "b").fill_null(strategy="backward", limit=2)
+        )
+
+        expected_backward = {
+            "a": [1, float("nan"), 5, 5, 5, 6, float("nan"), 10, 10, 10],
+            "b": ["a", None, "b", "b", "b", "c", None, "d", "d", "d"],
+        }
+        assert_equal_data(result_backward, expected_backward)
+
+
 def test_fill_null_series(constructor_eager: ConstructorEager) -> None:
-    df = nw.from_native(constructor_eager(data), eager_only=True)
+    data_series_float = {
+        "a": [0.0, 1, None, 2, None, 3],
+    }
+    df_float = nw.from_native(constructor_eager(data_series_float), eager_only=True)
 
-    expected = {
-        "a": [0.0, 99, 2, 3, 4],
-        "b": [1.0, 99, 99, 5, 3],
-        "c": [5.0, 99, 3, 2, 1],
+    expected_float = {
+        "a_zero_digit": [0.0, 1, 0, 2, 0, 3],
     }
-    result = df.with_columns(
-        a=df["a"].fill_null(99),
-        b=df["b"].fill_null(99),
-        c=df["c"].fill_null(99),
+    result_float = df_float.select(
+        a_zero_digit=df_float["a"].fill_null(value=0),
     )
-    assert_equal_data(result, expected)
+
+    assert_equal_data(result_float, expected_float)
+
+    data_series_str = {
+        "a": ["a", None, "c", None, "e"],
+    }
+    df_str = nw.from_native(constructor_eager(data_series_str), eager_only=True)
+
+    expected_str = {
+        "a_z_str": ["a", "z", "c", "z", "e"],
+    }
+
+    result_str = df_str.select(
+        a_z_str=df_str["a"].fill_null(value="z"),
+    )
+
+    assert_equal_data(result_str, expected_str)
+
+
+def test_fill_null_series_limits(constructor_eager: ConstructorEager) -> None:
+    context: Any = (
+        pytest.raises(NotImplementedError, match="The limit keyword is not supported")
+        if "cudf" in str(constructor_eager)
+        else warnings.catch_warnings()
+        if "modin" in str(constructor_eager)
+        else does_not_raise()
+    )
+    data_series_float = {
+        "a": [0.0, 1, None, None, 2, None, None, 3],
+        "b": ["", "a", None, None, "c", None, None, "e"],
+    }
+    df = nw.from_native(constructor_eager(data_series_float), eager_only=True)
+
+    with context:
+        if "modin" in str(constructor_eager):
+            warnings.filterwarnings(
+                "ignore", message="The 'downcast' keyword in fillna is deprecated"
+            )
+        expected_forward = {
+            "a_forward": [0.0, 1, 1, float("nan"), 2, 2, float("nan"), 3],
+            "b_forward": ["", "a", "a", None, "c", "c", None, "e"],
+        }
+        result_forward = df.select(
+            a_forward=df["a"].fill_null(strategy="forward", limit=1),
+            b_forward=df["b"].fill_null(strategy="forward", limit=1),
+        )
+
+        assert_equal_data(result_forward, expected_forward)
+
+        expected_backward = {
+            "a_backward": [0.0, 1, float("nan"), 2, 2, float("nan"), 3, 3],
+            "b_backward": ["", "a", None, "c", "c", None, "e", "e"],
+        }
+
+        result_backward = df.select(
+            a_backward=df["a"].fill_null(strategy="backward", limit=1),
+            b_backward=df["b"].fill_null(strategy="backward", limit=1),
+        )
+
+        assert_equal_data(result_backward, expected_backward)
+
+
+def test_fill_null_series_limit_as_none(constructor_eager: ConstructorEager) -> None:
+    data_series = {
+        "a": [1, None, None, None, 5, 6, None, None, None, 10],
+    }
+    df = nw.from_native(constructor_eager(data_series), eager_only=True)
+
+    expected_forward = {
+        "a_forward": [1, 1, 1, 1, 5, 6, 6, 6, 6, 10],
+        "a_backward": [1, 5, 5, 5, 5, 6, 10, 10, 10, 10],
+    }
+    if (
+        "pandas_pyarrow_constructor" in str(constructor_eager)
+        or "modin" in str(constructor_eager)
+        or "dask" in str(constructor_eager)
+    ):
+        with warnings.catch_warnings():
+            # case for modin and dask
+            warnings.filterwarnings(
+                "ignore", message="The 'downcast' keyword in fillna is deprecated"
+            )
+            # case for pandas_pyarrow_constructor
+            warnings.filterwarnings(
+                "ignore", message="Falling back on a non-pyarrow code path which"
+            )
+            result_forward = df.select(
+                a_forward=df["a"].fill_null(strategy="forward", limit=None),
+                a_backward=df["a"].fill_null(strategy="backward", limit=None),
+            )
+            assert_equal_data(result_forward, expected_forward)
+    else:
+        result_forward = df.select(
+            a_forward=df["a"].fill_null(strategy="forward", limit=None),
+            a_backward=df["a"].fill_null(strategy="backward", limit=None),
+        )
+
+        assert_equal_data(result_forward, expected_forward)
+
+    data_series_str = {
+        "a": ["a", None, None, None, "b", "c", None, None, None, "d"],
+    }
+
+    df_str = nw.from_native(constructor_eager(data_series_str), eager_only=True)
+
+    expected_forward_str = {
+        "a_forward": ["a", "a", "a", "a", "b", "c", "c", "c", "c", "d"],
+        "a_backward": ["a", "b", "b", "b", "b", "c", "d", "d", "d", "d"],
+    }
+
+    if (
+        "pandas_pyarrow_constructor" in str(constructor_eager)
+        or "modin" in str(constructor_eager)
+        or "dask" in str(constructor_eager)
+    ):
+        with warnings.catch_warnings():
+            # case for modin and dask
+            warnings.filterwarnings(
+                "ignore", message="The 'downcast' keyword in fillna is deprecated"
+            )
+            # case for pandas_pyarrow_constructor
+            warnings.filterwarnings(
+                "ignore", message="Falling back on a non-pyarrow code path which"
+            )
+            result_forward_str = df_str.select(
+                a_forward=df_str["a"].fill_null(strategy="forward", limit=None),
+                a_backward=df_str["a"].fill_null(strategy="backward", limit=None),
+            )
+            assert_equal_data(result_forward_str, expected_forward_str)
+    else:
+        result_forward_str = df_str.select(
+            a_forward=df_str["a"].fill_null(strategy="forward", limit=None),
+            a_backward=df_str["a"].fill_null(strategy="backward", limit=None),
+        )
+        assert_equal_data(result_forward_str, expected_forward_str)
+
+
+def test_fill_null_series_exceptions(constructor_eager: ConstructorEager) -> None:
+    data_series_float = {
+        "a": [0.0, 1, None, 2, None, 3],
+    }
+    df_float = nw.from_native(constructor_eager(data_series_float), eager_only=True)
+
+    with pytest.raises(ValueError, match="cannot specify both `value` and `strategy`"):
+        df_float.select(
+            a_zero_digit=df_float["a"].fill_null(value=0, strategy="forward"),
+        )
+
+    with pytest.raises(
+        ValueError, match="must specify either a fill `value` or `strategy`"
+    ):
+        df_float.select(
+            a_zero_digit=df_float["a"].fill_null(),
+        )
+
+    with pytest.raises(ValueError, match="strategy not supported:"):
+        df_float.select(
+            a_zero_digit=df_float["a"].fill_null(strategy="invalid"),  # type: ignore  # noqa: PGH003
+        )
diff --git a/tests/expr_and_series/map_batches_test.py b/tests/expr_and_series/map_batches_test.py
new file mode 100644
index 00000000..f2720e20
--- /dev/null
+++ b/tests/expr_and_series/map_batches_test.py
@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+import narwhals.stable.v1 as nw
+from tests.utils import POLARS_VERSION
+from tests.utils import Constructor
+from tests.utils import assert_equal_data
+
+data = {"a": [1, 2, 3], "b": [4, 5, 6], "z": [7.0, 8.0, 9.0]}
+
+
+def test_map_batches_expr(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    if "dask" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor(data))
+    expected = df.select(nw.col("a", "b").map_batches(lambda s: s + 1))
+    assert_equal_data(expected, {"a": [2, 3, 4], "b": [5, 6, 7]})
+
+
+def test_map_batches_expr_numpy(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    if "dask" in str(constructor) or (
+        "polars" in str(constructor) and POLARS_VERSION <= (0, 20, 3)
+    ):
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor(data))
+    expected = df.select(
+        nw.col("a")
+        .map_batches(lambda s: s.to_numpy() + 1, return_dtype=nw.Float64())
+        .sum()
+    )
+    assert_equal_data(expected, {"a": [9.0]})
+
+    expected = df.select(nw.all().map_batches(lambda s: s.to_numpy().argmax()))
+    assert_equal_data(expected, {"a": [2], "b": [2], "z": [2]})
+
+
+def test_map_batches_expr_names(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    if "dask" in str(constructor) or (
+        "polars" in str(constructor) and POLARS_VERSION <= (0, 20, 3)
+    ):
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor(data))
+    expected = nw.from_native(df.select(nw.all().map_batches(lambda x: x.to_numpy())))
+    assert_equal_data(expected, {"a": [1, 2, 3], "b": [4, 5, 6], "z": [7.0, 8.0, 9.0]})
+
+
+def test_map_batches_raise() -> None:
+    pytest.importorskip("dask")
+    pytest.importorskip("dask_expr", exc_type=ImportError)
+    import dask.dataframe as dd
+
+    df = nw.from_native(dd.from_pandas(pd.DataFrame({"a": [1, 2, 3]})))
+    with pytest.raises(
+        NotImplementedError, match="`Expr.map_batches` is not implemented for Dask yet"
+    ):
+        df.select(nw.col("a").map_batches(lambda x: x.to_numpy()))
diff --git a/tests/frame/pivot_test.py b/tests/frame/pivot_test.py
new file mode 100644
index 00000000..76217361
--- /dev/null
+++ b/tests/frame/pivot_test.py
@@ -0,0 +1,240 @@
+from __future__ import annotations
+
+from contextlib import nullcontext as does_not_raise
+from typing import Any
+
+import polars as pl
+import pytest
+
+import narwhals.stable.v1 as nw
+from tests.utils import PANDAS_VERSION
+from tests.utils import POLARS_VERSION
+from tests.utils import assert_equal_data
+
+data = {
+    "ix": [1, 2, 1, 1, 2, 2],
+    "iy": [1, 2, 2, 1, 2, 1],
+    "col": ["b", "b", "a", "a", "a", "a"],
+    "col_b": ["x", "y", "x", "y", "x", "y"],
+    "foo": [7, 1, 0, 1, 2, 2],
+    "bar": [9, 4, 0, 2, 0, 0],
+}
+
+data_no_dups = {
+    "ix": [1, 1, 2, 2],
+    "col": ["a", "b", "a", "b"],
+    "foo": [1, 2, 3, 4],
+    "bar": ["x", "y", "z", "w"],
+}
+
+
+@pytest.mark.parametrize(
+    ("agg_func", "expected"),
+    [
+        (
+            "min",
+            {
+                "ix": [1, 2],
+                "foo_a": [0, 2],
+                "foo_b": [7, 1],
+                "bar_a": [0, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "max",
+            {
+                "ix": [1, 2],
+                "foo_a": [1, 2],
+                "foo_b": [7, 1],
+                "bar_a": [2, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "first",
+            {
+                "ix": [1, 2],
+                "foo_a": [0, 2],
+                "foo_b": [7, 1],
+                "bar_a": [0, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "last",
+            {
+                "ix": [1, 2],
+                "foo_a": [1, 2],
+                "foo_b": [7, 1],
+                "bar_a": [2, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "sum",
+            {
+                "ix": [1, 2],
+                "foo_a": [1, 4],
+                "foo_b": [7, 1],
+                "bar_a": [2, 0],
+                "bar_b": [9, 4],
+            },
+        ),
+        (
+            "mean",
+            {
+                "ix": [1, 2],
+                "foo_a": [0.5, 2.0],
+                "foo_b": [7.0, 1.0],
+                "bar_a": [1.0, 0.0],
+                "bar_b": [9.0, 4.0],
+            },
+        ),
+        (
+            "median",
+            {
+                "ix": [1, 2],
+                "foo_a": [0.5, 2.0],
+                "foo_b": [7.0, 1.0],
+                "bar_a": [1.0, 0.0],
+                "bar_b": [9.0, 4.0],
+            },
+        ),
+        (
+            "len",
+            {
+                "ix": [1, 2],
+                "foo_a": [2, 2],
+                "foo_b": [1, 1],
+                "bar_a": [2, 2],
+                "bar_b": [1, 1],
+            },
+        ),
+    ],
+)
+@pytest.mark.parametrize(("on", "index"), [("col", "ix"), (["col"], ["ix"])])
+def test_pivot(
+    constructor_eager: Any,
+    agg_func: str,
+    expected: dict[str, list[Any]],
+    on: str | list[str],
+    index: str | list[str],
+    request: pytest.FixtureRequest,
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.pivot(
+        on=on,
+        index=index,
+        values=["foo", "bar"],
+        aggregate_function=agg_func,  # type: ignore[arg-type]
+    )
+
+    assert_equal_data(result, expected)
+
+
+@pytest.mark.parametrize(
+    ("data_", "context"),
+    [
+        (data_no_dups, does_not_raise()),
+        (data, pytest.raises((ValueError, pl.exceptions.ComputeError))),
+    ],
+)
+def test_pivot_no_agg(
+    request: Any, constructor_eager: Any, data_: Any, context: Any
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data_), eager_only=True)
+    with context:
+        df.pivot("col", index="ix", aggregate_function=None)
+
+
+@pytest.mark.parametrize(
+    ("sort_columns", "expected"),
+    [
+        (True, ["ix", "foo_a", "foo_b", "bar_a", "bar_b"]),
+        (False, ["ix", "foo_b", "foo_a", "bar_b", "bar_a"]),
+    ],
+)
+def test_pivot_sort_columns(
+    request: Any, constructor_eager: Any, sort_columns: Any, expected: list[str]
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.pivot(
+        on="col",
+        index="ix",
+        values=["foo", "bar"],
+        aggregate_function="sum",
+        sort_columns=sort_columns,
+    )
+    assert result.columns == expected
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "expected"),
+    [
+        ({"on": ["col"], "values": ["foo"]}, ["ix", "b", "a"]),
+        (
+            {"on": ["col"], "values": ["foo", "bar"]},
+            ["ix", "foo_b", "foo_a", "bar_b", "bar_a"],
+        ),
+        (
+            {"on": ["col", "col_b"], "values": ["foo"]},
+            ["ix", '{"b","x"}', '{"b","y"}', '{"a","x"}', '{"a","y"}'],
+        ),
+        (
+            {"on": ["col", "col_b"], "values": ["foo", "bar"]},
+            [
+                "ix",
+                'foo_{"b","x"}',
+                'foo_{"b","y"}',
+                'foo_{"a","x"}',
+                'foo_{"a","y"}',
+                'bar_{"b","x"}',
+                'bar_{"b","y"}',
+                'bar_{"a","x"}',
+                'bar_{"a","y"}',
+            ],
+        ),
+    ],
+)
+def test_pivot_names_out(
+    request: Any, constructor_eager: Any, kwargs: Any, expected: list[str]
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "modin")):
+        request.applymarker(pytest.mark.xfail)
+    if ("polars" in str(constructor_eager) and POLARS_VERSION < (1, 0)) or (
+        "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1)
+    ):
+        # not implemented
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+
+    result = (
+        df.pivot(aggregate_function="min", index="ix", **kwargs).collect_schema().names()
+    )
+    assert result == expected
diff --git a/tests/frame/select_test.py b/tests/frame/select_test.py
index 83b8e1f5..52c2a71f 100644
--- a/tests/frame/select_test.py
+++ b/tests/frame/select_test.py
@@ -4,6 +4,7 @@ import pandas as pd
 import pytest
 
 import narwhals.stable.v1 as nw
+from tests.utils import PANDAS_VERSION
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
@@ -32,3 +33,14 @@ def test_non_string_select_invalid() -> None:
     df = nw.from_native(pd.DataFrame({0: [1, 2], "b": [3, 4]}))
     with pytest.raises(TypeError, match="\n\nHint: if you were trying to select"):
         nw.to_native(df.select(0))  # type: ignore[arg-type]
+
+
+def test_select_boolean_cols(request: pytest.FixtureRequest) -> None:
+    if PANDAS_VERSION < (1, 1):
+        # bug in old pandas
+        request.applymarker(pytest.mark.xfail)
+    df = nw.from_native(pd.DataFrame({True: [1, 2], False: [3, 4]}), eager_only=True)
+    result = df.group_by(True).agg(nw.col(False).max())  # type: ignore[arg-type]# noqa: FBT003
+    assert_equal_data(result.to_dict(as_series=False), {True: [1, 2]})  # type: ignore[dict-item]
+    result = df.select(nw.col([False, True]))  # type: ignore[list-item]
+    assert_equal_data(result.to_dict(as_series=False), {True: [1, 2], False: [3, 4]})  # type: ignore[dict-item]
diff --git a/tests/series_only/to_dummy_test.py b/tests/series_only/to_dummy_test.py
index 10d6e971..9ebbee57 100644
--- a/tests/series_only/to_dummy_test.py
+++ b/tests/series_only/to_dummy_test.py
@@ -6,14 +6,15 @@ import narwhals.stable.v1 as nw
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
-data = [1, 2, 3]
+data = ["x", "y", "z"]
+data_na = ["x", "y", None]
 
 
 @pytest.mark.parametrize("sep", ["_", "-"])
 def test_to_dummies(constructor_eager: ConstructorEager, sep: str) -> None:
     s = nw.from_native(constructor_eager({"a": data}), eager_only=True)["a"].alias("a")
     result = s.to_dummies(separator=sep)
-    expected = {f"a{sep}1": [1, 0, 0], f"a{sep}2": [0, 1, 0], f"a{sep}3": [0, 0, 1]}
+    expected = {f"a{sep}x": [1, 0, 0], f"a{sep}y": [0, 1, 0], f"a{sep}z": [0, 0, 1]}
 
     assert_equal_data(result, expected)
 
@@ -22,6 +23,27 @@ def test_to_dummies(constructor_eager: ConstructorEager, sep: str) -> None:
 def test_to_dummies_drop_first(constructor_eager: ConstructorEager, sep: str) -> None:
     s = nw.from_native(constructor_eager({"a": data}), eager_only=True)["a"].alias("a")
     result = s.to_dummies(drop_first=True, separator=sep)
-    expected = {f"a{sep}2": [0, 1, 0], f"a{sep}3": [0, 0, 1]}
+    expected = {f"a{sep}y": [0, 1, 0], f"a{sep}z": [0, 0, 1]}
 
     assert_equal_data(result, expected)
+
+
+@pytest.mark.parametrize("sep", ["_", "-"])
+def test_to_dummies_with_nulls(constructor_eager: ConstructorEager, sep: str) -> None:
+    if "pandas_nullable_constructor" not in str(constructor_eager):
+        pytest.skip()
+    s = nw.from_native(constructor_eager({"a": data_na}), eager_only=True)["a"].alias("a")
+    result = s.to_dummies(separator=sep)
+    expected = {f"a{sep}null": [0, 0, 1], f"a{sep}x": [1, 0, 0], f"a{sep}y": [0, 1, 0]}
+
+    assert_equal_data(result, expected)
+
+
+@pytest.mark.parametrize("sep", ["_", "-"])
+def test_to_dummies_drop_first_na(constructor_eager: ConstructorEager, sep: str) -> None:
+    s = nw.from_native(constructor_eager({"a": data_na}), eager_only=True)["a"].alias("a")
+    result = s.to_dummies(drop_first=True, separator=sep)
+    expected = {f"a{sep}null": [0, 0, 1], f"a{sep}y": [0, 1, 0]}
+
+    assert_equal_data(result, expected)
+    assert result.schema == {f"a{sep}null": nw.Int8, f"a{sep}y": nw.Int8}
diff --git a/tests/series_only/to_list_test.py b/tests/series_only/to_list_test.py
index 84b4fad4..f241d792 100644
--- a/tests/series_only/to_list_test.py
+++ b/tests/series_only/to_list_test.py
@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import pytest
-
 import narwhals.stable.v1 as nw
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
@@ -9,10 +7,6 @@ from tests.utils import assert_equal_data
 data = [1, 2, 3]
 
 
-def test_to_list(
-    constructor_eager: ConstructorEager, request: pytest.FixtureRequest
-) -> None:
-    if "cudf" in str(constructor_eager):  # pragma: no cover
-        request.applymarker(pytest.mark.xfail)
+def test_to_list(constructor_eager: ConstructorEager) -> None:
     s = nw.from_native(constructor_eager({"a": data}), eager_only=True)["a"]
     assert_equal_data({"a": s.to_list()}, {"a": [1, 2, 3]})
diff --git a/tests/utils_test.py b/tests/utils_test.py
index fb668b4d..dc2415c8 100644
--- a/tests/utils_test.py
+++ b/tests/utils_test.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import string
+from typing import TYPE_CHECKING
 
 import hypothesis.strategies as st
 import pandas as pd
@@ -15,6 +16,9 @@ import narwhals.stable.v1 as nw
 from tests.utils import PANDAS_VERSION
 from tests.utils import get_module_version_as_tuple
 
+if TYPE_CHECKING:
+    from narwhals.series import Series
+
 
 def test_maybe_align_index_pandas() -> None:
     df = nw.from_native(pd.DataFrame({"a": [1, 2, 3]}, index=[1, 2, 0]))
@@ -58,21 +62,109 @@ def test_maybe_align_index_polars() -> None:
         nw.maybe_align_index(df, s[1:])
 
 
-def test_maybe_set_index_pandas() -> None:
-    df = nw.from_native(pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}, index=[1, 2, 0]))
-    result = nw.maybe_set_index(df, "b")
-    expected = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}, index=[1, 2, 0]).set_index(
-        "b"
-    )
+@pytest.mark.parametrize(
+    "column_names",
+    ["b", ["a", "b"]],
+)
+def test_maybe_set_index_pandas_column_names(
+    column_names: str | list[str] | None,
+) -> None:
+    df = nw.from_native(pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}))
+    result = nw.maybe_set_index(df, column_names)
+    expected = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}).set_index(column_names)
     assert_frame_equal(nw.to_native(result), expected)
 
 
-def test_maybe_set_index_polars() -> None:
+@pytest.mark.parametrize(
+    "column_names",
+    [
+        "b",
+        ["a", "b"],
+    ],
+)
+def test_maybe_set_index_polars_column_names(
+    column_names: str | list[str] | None,
+) -> None:
+    df = nw.from_native(pl.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}))
+    result = nw.maybe_set_index(df, column_names)
+    assert result is df
+
+
+@pytest.mark.parametrize(
+    "native_df_or_series",
+    [pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}), pd.Series([0, 1, 2])],
+)
+@pytest.mark.parametrize(
+    ("narwhals_index", "pandas_index"),
+    [
+        (nw.from_native(pd.Series([1, 2, 0]), series_only=True), pd.Series([1, 2, 0])),
+        (
+            [
+                nw.from_native(pd.Series([0, 1, 2]), series_only=True),
+                nw.from_native(pd.Series([1, 2, 0]), series_only=True),
+            ],
+            [
+                pd.Series([0, 1, 2]),
+                pd.Series([1, 2, 0]),
+            ],
+        ),
+    ],
+)
+def test_maybe_set_index_pandas_direct_index(
+    narwhals_index: Series | list[Series] | None,
+    pandas_index: pd.Series | list[pd.Series] | None,
+    native_df_or_series: pd.DataFrame | pd.Series,
+) -> None:
+    df = nw.from_native(native_df_or_series, allow_series=True)
+    result = nw.maybe_set_index(df, index=narwhals_index)
+    if isinstance(native_df_or_series, pd.Series):
+        native_df_or_series.index = pandas_index
+        assert_series_equal(nw.to_native(result), native_df_or_series)
+    else:
+        expected = native_df_or_series.set_index(pandas_index)
+        assert_frame_equal(nw.to_native(result), expected)
+
+
+@pytest.mark.parametrize(
+    "index",
+    [
+        nw.from_native(pd.Series([1, 2, 0]), series_only=True),
+        [
+            nw.from_native(pd.Series([0, 1, 2]), series_only=True),
+            nw.from_native(pd.Series([1, 2, 0]), series_only=True),
+        ],
+    ],
+)
+def test_maybe_set_index_polars_direct_index(
+    index: Series | list[Series] | None,
+) -> None:
     df = nw.from_native(pl.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}))
-    result = nw.maybe_set_index(df, "b")
+    result = nw.maybe_set_index(df, index=index)
     assert result is df
 
 
+def test_maybe_set_index_pandas_series_column_names() -> None:
+    df = nw.from_native(pd.Series([0, 1, 2]), allow_series=True)
+    with pytest.raises(
+        ValueError, match="Cannot set index using column names on a Series"
+    ):
+        nw.maybe_set_index(df, column_names=["a"])
+
+
+def test_maybe_set_index_pandas_either_index_or_column_names() -> None:
+    df = nw.from_native(pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}))
+    column_names = ["a", "b"]
+    index = nw.from_native(pd.Series([0, 1, 2]), series_only=True)
+    with pytest.raises(
+        ValueError, match="Only one of `column_names` or `index` should be provided"
+    ):
+        nw.maybe_set_index(df, column_names=column_names, index=index)
+    with pytest.raises(
+        ValueError, match="Either `column_names` or `index` should be provided"
+    ):
+        nw.maybe_set_index(df)
+
+
 def test_maybe_get_index_pandas() -> None:
     pandas_df = pd.DataFrame({"a": [1, 2, 3]}, index=[1, 2, 0])
     result = nw.maybe_get_index(nw.from_native(pandas_df))