Repository: narwhals-dev/narwhals
Pull request #1717 (2025-01-04 09:49:21)

feat: enable to cast to `Struct` dtype
<!--
# Thanks for contributing a pull request! 
## Please make sure you see our contribution guidelines: https://github.com/narwhals-dev/narwhals/blob/main/CONTRIBUTING.md
-->

## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [x] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #\<issue number\>
- Closes #\<issue number\>

## Checklist

- [x] Code follows style guide (ruff)
- [x] Tests added
- [ ] Documented the changes

## If you have comments or can explain your changes, please do so below



Patch:
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index c2898f20..c0efa50f 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -485,9 +485,12 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
         import pyarrow as pa
 
         df = self._native_frame
+        cols = self.columns
 
         row_indices = pa.array(range(df.num_rows))
-        return self._from_native_frame(df.append_column(name, row_indices))
+        return self._from_native_frame(
+            df.append_column(name, row_indices).select([name, *cols])
+        )
 
     def filter(self: Self, *predicates: IntoArrowExpr, **constraints: Any) -> Self:
         if (
diff --git a/narwhals/_arrow/group_by.py b/narwhals/_arrow/group_by.py
index 66008eea..ffb16578 100644
--- a/narwhals/_arrow/group_by.py
+++ b/narwhals/_arrow/group_by.py
@@ -73,6 +73,7 @@ class ArrowGroupBy:
             exprs,
             self._keys,
             self._df._from_native_frame,
+            backend_version=self._df._backend_version,
         )
 
     def __iter__(self: Self) -> Iterator[tuple[Any, ArrowDataFrame]]:
@@ -114,6 +115,7 @@ def agg_arrow(
     exprs: Sequence[CompliantExpr[ArrowSeries]],
     keys: list[str],
     from_dataframe: Callable[[Any], ArrowDataFrame],
+    backend_version: tuple[int, ...],
 ) -> ArrowDataFrame:
     import pyarrow.compute as pc
 
@@ -207,6 +209,10 @@ def agg_arrow(
         expected_old_names_indices[item].pop(0) for item in result_simple.column_names
     ]
     new_column_names = [new_column_names[i] for i in index_map]
-
     result_simple = result_simple.rename_columns(new_column_names)
+    if backend_version < (12, 0, 0):
+        columns = result_simple.column_names
+        result_simple = result_simple.select(
+            [*keys, *[col for col in columns if col not in keys]]
+        )
     return from_dataframe(result_simple)
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index ecc5b084..cf7760d4 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -869,13 +869,20 @@ class ArrowSeries(CompliantSeries):
     def gather_every(self: Self, n: int, offset: int = 0) -> Self:
         return self._from_native_series(self._native_series[offset::n])
 
-    def clip(self: Self, lower_bound: Any | None, upper_bound: Any | None) -> Self:
-        import pyarrow as pa
+    def clip(
+        self: Self, lower_bound: Self | Any | None, upper_bound: Self | Any | None
+    ) -> Self:
         import pyarrow.compute as pc
 
         arr = self._native_series
-        arr = pc.max_element_wise(arr, pa.scalar(lower_bound, type=arr.type))
-        arr = pc.min_element_wise(arr, pa.scalar(upper_bound, type=arr.type))
+        _, lower_bound = broadcast_and_extract_native(
+            self, lower_bound, self._backend_version
+        )
+        _, upper_bound = broadcast_and_extract_native(
+            self, upper_bound, self._backend_version
+        )
+        arr = pc.max_element_wise(arr, lower_bound)
+        arr = pc.min_element_wise(arr, upper_bound)
 
         return self._from_native_series(arr)
 
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index 23907465..4dbc17a9 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -129,9 +129,19 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], version: Version) -> pa
                 version=version,
             )
         )
-    if isinstance_or_issubclass(dtype, dtypes.Struct):  # pragma: no cover
-        msg = "Converting to Struct dtype is not supported yet"
-        return NotImplementedError(msg)
+    if isinstance_or_issubclass(dtype, dtypes.Struct):
+        return pa.struct(
+            [
+                (
+                    field.name,
+                    narwhals_to_native_dtype(
+                        field.dtype,
+                        version=version,
+                    ),
+                )
+                for field in dtype.fields  # type: ignore[union-attr]
+            ]
+        )
     if isinstance_or_issubclass(dtype, dtypes.Array):  # pragma: no cover
         msg = "Converting to Array dtype is not supported yet"
         return NotImplementedError(msg)
@@ -153,6 +163,11 @@ def broadcast_and_extract_native(
     from narwhals._arrow.dataframe import ArrowDataFrame
     from narwhals._arrow.series import ArrowSeries
 
+    if rhs is None:
+        import pyarrow as pa
+
+        return lhs._native_series, pa.scalar(None, type=lhs._native_series.type)
+
     # If `rhs` is the output of an expression evaluation, then it is
     # a list of Series. So, we verify that that list is of length-1,
     # and take the first (and only) element.
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index 36d85bdb..65bf5eeb 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -186,7 +186,11 @@ class DaskLazyFrame(CompliantLazyFrame):
     def with_row_index(self: Self, name: str) -> Self:
         # Implementation is based on the following StackOverflow reply:
         # https://stackoverflow.com/questions/60831518/in-dask-how-does-one-add-a-range-of-integersauto-increment-to-a-new-column/60852409#60852409
-        return self._from_native_frame(add_row_index(self._native_frame, name))
+        return self._from_native_frame(
+            add_row_index(
+                self._native_frame, name, self._backend_version, self._implementation
+            )
+        )
 
     def rename(self: Self, mapping: dict[str, str]) -> Self:
         return self._from_native_frame(self._native_frame.rename(columns=mapping))
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 9b4b5f6a..e4107c04 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -443,26 +443,20 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         )
 
     def std(self, ddof: int) -> Self:
-        expr = self._from_call(
+        return self._from_call(
             lambda _input, ddof: _input.std(ddof=ddof),
             "std",
             ddof=ddof,
             returns_scalar=True,
         )
-        if ddof != 1:
-            expr._depth += 1
-        return expr
 
     def var(self, ddof: int) -> Self:
-        expr = self._from_call(
+        return self._from_call(
             lambda _input, ddof: _input.var(ddof=ddof),
             "var",
             ddof=ddof,
             returns_scalar=True,
         )
-        if ddof != 1:
-            expr._depth += 1
-        return expr
 
     def skew(self: Self) -> Self:
         return self._from_call(
@@ -536,8 +530,8 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
 
     def is_between(
         self,
-        lower_bound: Any,
-        upper_bound: Any,
+        lower_bound: Self | Any,
+        upper_bound: Self | Any,
         closed: str = "both",
     ) -> Self:
         if closed == "none":
@@ -673,8 +667,8 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
 
     def clip(
         self: Self,
-        lower_bound: Any | None = None,
-        upper_bound: Any | None = None,
+        lower_bound: Self | Any | None,
+        upper_bound: Self | Any | None,
     ) -> Self:
         return self._from_call(
             lambda _input, lower_bound, upper_bound: _input.clip(
@@ -755,7 +749,12 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         def func(_input: dask_expr.Series) -> dask_expr.Series:
             _name = _input.name
             col_token = generate_temporary_column_name(n_bytes=8, columns=[_name])
-            _input = add_row_index(_input.to_frame(), col_token)
+            _input = add_row_index(
+                _input.to_frame(),
+                col_token,
+                backend_version=self._backend_version,
+                implementation=self._implementation,
+            )
             first_distinct_index = _input.groupby(_name).agg({col_token: "min"})[
                 col_token
             ]
@@ -772,7 +771,12 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
         def func(_input: dask_expr.Series) -> dask_expr.Series:
             _name = _input.name
             col_token = generate_temporary_column_name(n_bytes=8, columns=[_name])
-            _input = add_row_index(_input.to_frame(), col_token)
+            _input = add_row_index(
+                _input.to_frame(),
+                col_token,
+                backend_version=self._backend_version,
+                implementation=self._implementation,
+            )
             last_distinct_index = _input.groupby(_name).agg({col_token: "max"})[col_token]
 
             return _input[col_token].isin(last_distinct_index)
diff --git a/narwhals/_dask/group_by.py b/narwhals/_dask/group_by.py
index af269abd..7bda88ee 100644
--- a/narwhals/_dask/group_by.py
+++ b/narwhals/_dask/group_by.py
@@ -36,14 +36,38 @@ def n_unique() -> dd.Aggregation:
     )
 
 
+def var(
+    ddof: int = 1,
+) -> Callable[
+    [pd.core.groupby.generic.SeriesGroupBy], pd.core.groupby.generic.SeriesGroupBy
+]:
+    from functools import partial
+
+    import dask_expr as dx
+
+    return partial(dx._groupby.GroupBy.var, ddof=ddof)
+
+
+def std(
+    ddof: int = 1,
+) -> Callable[
+    [pd.core.groupby.generic.SeriesGroupBy], pd.core.groupby.generic.SeriesGroupBy
+]:
+    from functools import partial
+
+    import dask_expr as dx
+
+    return partial(dx._groupby.GroupBy.std, ddof=ddof)
+
+
 POLARS_TO_DASK_AGGREGATIONS = {
     "sum": "sum",
     "mean": "mean",
     "median": "median",
     "max": "max",
     "min": "min",
-    "std": "std",
-    "var": "var",
+    "std": std,
+    "var": var,
     "len": "size",
     "n_unique": n_unique,
     "count": "count",
@@ -137,8 +161,12 @@ def agg_dask(
                 function_name = POLARS_TO_DASK_AGGREGATIONS.get(
                     expr._function_name, expr._function_name
                 )
-                for output_name in expr._output_names:
-                    simple_aggregations[output_name] = (keys[0], function_name)
+                simple_aggregations.update(
+                    {
+                        output_name: (keys[0], function_name)
+                        for output_name in expr._output_names
+                    }
+                )
                 continue
 
             # e.g. agg(nw.mean('a')) # noqa: ERA001
@@ -149,13 +177,26 @@ def agg_dask(
                 raise AssertionError(msg)
 
             function_name = remove_prefix(expr._function_name, "col->")
-            function_name = POLARS_TO_DASK_AGGREGATIONS.get(function_name, function_name)
+            kwargs = (
+                {"ddof": expr._kwargs.get("ddof", 1)}
+                if function_name in {"std", "var"}
+                else {}
+            )
 
+            agg_function = POLARS_TO_DASK_AGGREGATIONS.get(function_name, function_name)
             # deal with n_unique case in a "lazy" mode to not depend on dask globally
-            function_name = function_name() if callable(function_name) else function_name
-
-            for root_name, output_name in zip(expr._root_names, expr._output_names):
-                simple_aggregations[output_name] = (root_name, function_name)
+            agg_function = (
+                agg_function(**kwargs) if callable(agg_function) else agg_function
+            )
+
+            simple_aggregations.update(
+                {
+                    output_name: (root_name, agg_function)
+                    for root_name, output_name in zip(
+                        expr._root_names, expr._output_names
+                    )
+                }
+            )
         result_simple = grouped.agg(**simple_aggregations)
         return from_dataframe(result_simple.reset_index())
 
diff --git a/narwhals/_dask/utils.py b/narwhals/_dask/utils.py
index 71e17b3f..cb123249 100644
--- a/narwhals/_dask/utils.py
+++ b/narwhals/_dask/utils.py
@@ -3,9 +3,11 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 from typing import Any
 
+from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals.dependencies import get_pandas
 from narwhals.dependencies import get_pyarrow
 from narwhals.exceptions import InvalidIntoExprError
+from narwhals.utils import Implementation
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
 from narwhals.utils import parse_version
@@ -63,9 +65,20 @@ def parse_exprs_and_named_exprs(
     return results
 
 
-def add_row_index(frame: dd.DataFrame, name: str) -> dd.DataFrame:
+def add_row_index(
+    frame: dd.DataFrame,
+    name: str,
+    backend_version: tuple[int, ...],
+    implementation: Implementation,
+) -> dd.DataFrame:
+    original_cols = frame.columns
     frame = frame.assign(**{name: 1})
-    return frame.assign(**{name: frame[name].cumsum(method="blelloch") - 1})
+    return select_columns_by_name(
+        frame.assign(**{name: frame[name].cumsum(method="blelloch") - 1}),
+        [name, *original_cols],
+        backend_version,
+        implementation,
+    )
 
 
 def validate_comparand(lhs: dask_expr.Series, rhs: dask_expr.Series) -> None:
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index 04086664..cf8972de 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -994,10 +994,13 @@ class PandasLikeSeries(CompliantSeries):
         return self._from_native_series(self._native_series.iloc[offset::n])
 
     def clip(
-        self: Self, lower_bound: Any | None = None, upper_bound: Any | None = None
+        self: Self, lower_bound: Self | Any | None, upper_bound: Self | Any | None
     ) -> Self:
+        _, lower_bound = broadcast_align_and_extract_native(self, lower_bound)
+        _, upper_bound = broadcast_align_and_extract_native(self, upper_bound)
+        kwargs = {"axis": 0} if self._implementation is Implementation.MODIN else {}
         return self._from_native_series(
-            self._native_series.clip(lower_bound, upper_bound)
+            self._native_series.clip(lower_bound, upper_bound, **kwargs)
         )
 
     def to_arrow(self: Self) -> Any:
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index 1a888ed7..655e6077 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+import warnings
 from functools import lru_cache
 from typing import TYPE_CHECKING
 from typing import Any
@@ -211,7 +212,16 @@ def horizontal_concat(
 
     Should be in namespace.
     """
-    if implementation in PANDAS_LIKE_IMPLEMENTATION:
+    if implementation is Implementation.CUDF:
+        with warnings.catch_warnings():
+            warnings.filterwarnings(
+                "ignore",
+                message="The behavior of array concatenation with empty entries is deprecated",
+                category=FutureWarning,
+            )
+            return implementation.to_native_namespace().concat(dfs, axis=1)
+
+    if implementation.is_pandas_like():
         extra_kwargs = (
             {"copy": False}
             if implementation is Implementation.PANDAS and backend_version < (3,)
@@ -626,7 +636,6 @@ def narwhals_to_native_dtype(  # noqa: PLR0915
             if dtype_backend == "pyarrow-nullable"
             else f"timedelta64[{du_time_unit}]"
         )
-
     if isinstance_or_issubclass(dtype, dtypes.Date):
         if dtype_backend == "pyarrow-nullable":
             return "date32[pyarrow]"
@@ -658,9 +667,35 @@ def narwhals_to_native_dtype(  # noqa: PLR0915
                 f"{implementation} and version {version}."
             )
             return NotImplementedError(msg)
-    if isinstance_or_issubclass(dtype, dtypes.Struct):  # pragma: no cover
-        msg = "Converting to Struct dtype is not supported yet"
-        return NotImplementedError(msg)
+    if isinstance_or_issubclass(dtype, dtypes.Struct):
+        if implementation is Implementation.PANDAS and backend_version >= (2, 2):
+            try:
+                import pandas as pd
+                import pyarrow as pa  # ignore-banned-import
+            except ImportError as exc:  # pragma: no cover
+                msg = f"Unable to convert to {dtype} to to the following exception: {exc.msg}"
+                raise ImportError(msg) from exc
+
+            return pd.ArrowDtype(
+                pa.struct(
+                    [
+                        (
+                            field.name,
+                            arrow_narwhals_to_native_dtype(
+                                field.dtype,
+                                version=version,
+                            ),
+                        )
+                        for field in dtype.fields  # type: ignore[union-attr]
+                    ]
+                )
+            )
+        else:  # pragma: no cover
+            msg = (
+                "Converting to Struct dtype is not supported for implementation "
+                f"{implementation} and version {version}."
+            )
+            return NotImplementedError(msg)
     if isinstance_or_issubclass(dtype, dtypes.Array):  # pragma: no cover
         msg = "Converting to Array dtype is not supported yet"
         return NotImplementedError(msg)
diff --git a/narwhals/_polars/utils.py b/narwhals/_polars/utils.py
index 0e445a4f..e85132f8 100644
--- a/narwhals/_polars/utils.py
+++ b/narwhals/_polars/utils.py
@@ -194,12 +194,18 @@ def narwhals_to_native_dtype(dtype: DType | type[DType], version: Version) -> pl
     if dtype == dtypes.Duration or isinstance(dtype, dtypes.Duration):
         du_time_unit: Literal["us", "ns", "ms"] = getattr(dtype, "time_unit", "us")
         return pl.Duration(time_unit=du_time_unit)
-
     if dtype == dtypes.List:
         return pl.List(narwhals_to_native_dtype(dtype.inner, version))  # type: ignore[union-attr]
-    if dtype == dtypes.Struct:  # pragma: no cover
-        msg = "Converting to Struct dtype is not supported yet"
-        raise NotImplementedError(msg)
+    if dtype == dtypes.Struct:
+        return pl.Struct(
+            fields=[
+                pl.Field(
+                    name=field.name,
+                    dtype=narwhals_to_native_dtype(field.dtype, version),
+                )
+                for field in dtype.fields  # type: ignore[union-attr]
+            ]
+        )
     if dtype == dtypes.Array:  # pragma: no cover
         msg = "Converting to Array dtype is not supported yet"
         raise NotImplementedError(msg)
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index e57f9e1b..00fac84d 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -245,4 +245,21 @@ class SparkLikeLazyFrame:
         other = other_native.select(
             [F.col(old).alias(new) for old, new in rename_mapping.items()]
         )
-        return self._from_native_frame(self_native.join(other=other, on=left_on, how=how))
+
+        # If how in {"semi", "anti"}, then resulting columns are same as left columns
+        # Otherwise, we add the right columns with the new mapping, while keeping the
+        # original order of right_columns.
+        col_order = left_columns
+
+        if how in {"inner", "left", "cross"}:
+            col_order.extend(
+                [
+                    rename_mapping[colname]
+                    for colname in right_columns
+                    if colname not in (right_on or [])
+                ]
+            )
+
+        return self._from_native_frame(
+            self_native.join(other=other, on=left_on, how=how).select(col_order)
+        )
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index 0e401d46..7a5d5b4d 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -1297,13 +1297,13 @@ class DataFrame(BaseFrame[DataFrameT]):
             └───────┴─────┴─────┘
             >>> agnostic_with_row_index(df_pa)
             pyarrow.Table
+            index: int64
             a: int64
             b: int64
-            index: int64
             ----
+            index: [[0,1,2]]
             a: [[1,2,3]]
             b: [[4,5,6]]
-            index: [[0,1,2]]
         """
         return super().with_row_index(name)
 
@@ -3586,14 +3586,23 @@ class LazyFrame(BaseFrame[FrameT]):
         Examples:
             >>> import narwhals as nw
             >>> import polars as pl
-            >>> lf_native = pl.LazyFrame({"a": [1, 2, 3]})
-            >>> lf = nw.from_native(lf_native)
+            >>> import dask.dataframe as dd
+            >>> lf_pl = pl.LazyFrame({"a": [1, 2, 3]})
+            >>> lf_dask = dd.from_dict({"a": [1, 2, 3]}, npartitions=2)
+
+            >>> lf = nw.from_native(lf_pl)
             >>> lf.implementation
             <Implementation.POLARS: 6>
             >>> lf.implementation.is_pandas()
             False
             >>> lf.implementation.is_polars()
             True
+
+            >>> lf = nw.from_native(lf_dask)
+            >>> lf.implementation
+            <Implementation.DASK: 7>
+            >>> lf.implementation.is_dask()
+            True
         """
         return self._compliant_frame._implementation  # type: ignore[no-any-return]
 
@@ -3610,13 +3619,15 @@ class LazyFrame(BaseFrame[FrameT]):
         Examples:
             >>> import narwhals as nw
             >>> import polars as pl
-            >>> lf_pl = pl.LazyFrame(
-            ...     {
-            ...         "a": ["a", "b", "a", "b", "b", "c"],
-            ...         "b": [1, 2, 3, 4, 5, 6],
-            ...         "c": [6, 5, 4, 3, 2, 1],
-            ...     }
-            ... )
+            >>> import dask.dataframe as dd
+            >>> data = {
+            ...     "a": ["a", "b", "a", "b", "b", "c"],
+            ...     "b": [1, 2, 3, 4, 5, 6],
+            ...     "c": [6, 5, 4, 3, 2, 1],
+            ... }
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
+
             >>> lf = nw.from_native(lf_pl)
             >>> lf  # doctest:+ELLIPSIS
             ┌─────────────────────────────┐
@@ -3636,6 +3647,27 @@ class LazyFrame(BaseFrame[FrameT]):
             │ b   ┆ 11  ┆ 10  │
             │ c   ┆ 6   ┆ 1   │
             └─────┴─────┴─────┘
+
+            >>> lf = nw.from_native(lf_dask)
+            >>> lf
+            ┌───────────────────────────────────┐
+            |        Narwhals LazyFrame         |
+            |-----------------------------------|
+            |Dask DataFrame Structure:          |
+            |                    a      b      c|
+            |npartitions=2                      |
+            |0              string  int64  int64|
+            |3                 ...    ...    ...|
+            |5                 ...    ...    ...|
+            |Dask Name: frompandas, 1 expression|
+            |Expr=df                            |
+            └───────────────────────────────────┘
+            >>> df = lf.group_by("a").agg(nw.col("b", "c").sum()).collect()
+            >>> df.to_native()
+               a   b   c
+            0  a   4  10
+            1  b  11  10
+            2  c   6   1
         """
         return self._dataframe(
             self._compliant_frame.collect(),
@@ -3649,23 +3681,16 @@ class LazyFrame(BaseFrame[FrameT]):
             Object of class that user started with.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
-            >>> import pyarrow as pa
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>>
             >>> data = {"foo": [1, 2, 3], "bar": [6.0, 7.0, 8.0], "ham": ["a", "b", "c"]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
-            >>> df_pa = pa.table(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
-            Calling `to_native` on a Narwhals DataFrame returns the native object:
+            Calling `to_native` on a Narwhals LazyFrame returns the native object:
 
-            >>> nw.from_native(df_pd).lazy().to_native()
-               foo  bar ham
-            0    1  6.0   a
-            1    2  7.0   b
-            2    3  8.0   c
             >>> nw.from_native(lf_pl).to_native().collect()
             shape: (3, 3)
             ┌─────┬─────┬─────┐
@@ -3677,6 +3702,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   ┆ 7.0 ┆ b   │
             │ 3   ┆ 8.0 ┆ c   │
             └─────┴─────┴─────┘
+            >>> nw.from_native(lf_dask).to_native().compute()
+               foo  bar ham
+            0    1  6.0   a
+            1    2  7.0   b
+            2    3  8.0   c
         """
         return to_native(narwhals_object=self, pass_through=False)
 
@@ -3694,28 +3724,23 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import polars as pl
-            >>> import pandas as pd
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {"a": [1, 2, 3], "ba": [4, 5, 6]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function:
 
             >>> def agnostic_pipe(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.pipe(lambda _df: _df.select("a")).to_native()
+            ...     return df.pipe(lambda _df: _df.select("a")).collect().to_native()
 
-            We can then pass either pandas or Polars:
+            We can then pass any supported library such as Polars or Dask to `agnostic_pipe`:
 
-            >>> agnostic_pipe(df_pd)
-               a
-            0  1
-            1  2
-            2  3
-            >>> agnostic_pipe(lf_pl).collect()
+            >>> agnostic_pipe(lf_pl)
             shape: (3, 1)
             ┌─────┐
             │ a   │
@@ -3726,6 +3751,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   │
             │ 3   │
             └─────┘
+            >>> agnostic_pipe(lf_dask)
+               a
+            0  1
+            1  2
+            2  3
         """
         return super().pipe(function, *args, **kwargs)
 
@@ -3746,26 +3776,23 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import polars as pl
-            >>> import pandas as pd
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {"a": [1.0, 2.0, None], "ba": [1.0, None, 2.0]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function:
 
             >>> def agnostic_drop_nulls(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.drop_nulls().to_native()
+            ...     return df.drop_nulls().collect().to_native()
 
-            We can then pass any supported library such as Pandas or Polars to `agnostic_drop_nulls`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_drop_nulls`:
 
-            >>> agnostic_drop_nulls(df_pd)
-                 a   ba
-            0  1.0  1.0
-            >>> agnostic_drop_nulls(lf_pl).collect()
+            >>> agnostic_drop_nulls(lf_pl)
             shape: (1, 2)
             ┌─────┬─────┐
             │ a   ┆ ba  │
@@ -3774,6 +3801,9 @@ class LazyFrame(BaseFrame[FrameT]):
             ╞═════╪═════╡
             │ 1.0 ┆ 1.0 │
             └─────┴─────┘
+            >>> agnostic_drop_nulls(lf_dask)
+                 a   ba
+            0  1.0  1.0
         """
         return super().drop_nulls(subset=subset)
 
@@ -3788,28 +3818,23 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import polars as pl
-            >>> import pandas as pd
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {"a": [1, 2, 3], "b": [4, 5, 6]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function:
 
             >>> def agnostic_with_row_index(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.with_row_index().to_native()
+            ...     return df.with_row_index().collect().to_native()
 
-            We can then pass either pandas or Polars:
+            We can then pass any supported library such as Polars or Dask to `agnostic_with_row_index`:
 
-            >>> agnostic_with_row_index(df_pd)
-               index  a  b
-            0      0  1  4
-            1      1  2  5
-            2      2  3  6
-            >>> agnostic_with_row_index(lf_pl).collect()
+            >>> agnostic_with_row_index(lf_pl)
             shape: (3, 3)
             ┌───────┬─────┬─────┐
             │ index ┆ a   ┆ b   │
@@ -3820,6 +3845,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 1     ┆ 2   ┆ 5   │
             │ 2     ┆ 3   ┆ 6   │
             └───────┴─────┴─────┘
+            >>> agnostic_with_row_index(lf_dask)
+               index  a  b
+            0      0  1  4
+            1      1  2  5
+            2      2  3  6
         """
         return super().with_row_index(name)
 
@@ -3832,17 +3862,23 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
-            >>> lf_pl = pl.LazyFrame(
-            ...     {
-            ...         "foo": [1, 2, 3],
-            ...         "bar": [6.0, 7.0, 8.0],
-            ...         "ham": ["a", "b", "c"],
-            ...     }
-            ... )
+            >>> data = {
+            ...     "foo": [1, 2, 3],
+            ...     "bar": [6.0, 7.0, 8.0],
+            ...     "ham": ["a", "b", "c"],
+            ... }
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
+
             >>> lf = nw.from_native(lf_pl)
             >>> lf.schema  # doctest: +SKIP
-            Schema({'foo': Int64, 'bar': Float64, 'ham', String})
+            Schema({'foo': Int64, 'bar': Float64, 'ham': String})
+
+            >>> lf = nw.from_native(lf_dask)
+            >>> lf.schema  # doctest: +SKIP
+            Schema({'foo': Int64, 'bar': Float64, 'ham': String})
         """
         return super().schema
 
@@ -3854,17 +3890,23 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
-            >>> lf_pl = pl.LazyFrame(
-            ...     {
-            ...         "foo": [1, 2, 3],
-            ...         "bar": [6.0, 7.0, 8.0],
-            ...         "ham": ["a", "b", "c"],
-            ...     }
-            ... )
+            >>> data = {
+            ...     "foo": [1, 2, 3],
+            ...     "bar": [6.0, 7.0, 8.0],
+            ...     "ham": ["a", "b", "c"],
+            ... }
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
+
             >>> lf = nw.from_native(lf_pl)
             >>> lf.collect_schema()
             Schema({'foo': Int64, 'bar': Float64, 'ham': String})
+
+            >>> lf = nw.from_native(lf_dask)
+            >>> lf.collect_schema()
+            Schema({'foo': Int64, 'bar': Float64, 'ham': String})
         """
         return super().collect_schema()
 
@@ -3876,14 +3918,14 @@ class LazyFrame(BaseFrame[FrameT]):
             The column names stored in a list.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrame
             >>>
-            >>> df = {"foo": [1, 2, 3], "bar": [6.0, 7.0, 8.0], "ham": ["a", "b", "c"]}
-            >>> df_pd = pd.DataFrame(df)
-            >>> lf_pl = pl.LazyFrame(df)
+            >>> data = {"foo": [1, 2, 3], "bar": [6.0, 7.0, 8.0], "ham": ["a", "b", "c"]}
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             We define a library agnostic function:
 
@@ -3891,12 +3933,12 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     df = nw.from_native(df_native)
             ...     return df.columns
 
-            We can then pass either pandas or Polars to `agnostic_columns`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_columns`:
 
-            >>> agnostic_columns(df_pd)
-            ['foo', 'bar', 'ham']
             >>> agnostic_columns(lf_pl)  # doctest: +SKIP
             ['foo', 'bar', 'ham']
+            >>> agnostic_columns(lf_dask)
+            ['foo', 'bar', 'ham']
         """
         return super().columns
 
@@ -3923,48 +3965,31 @@ class LazyFrame(BaseFrame[FrameT]):
             existing data.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
-            >>> df = {
+            >>> data = {
             ...     "a": [1, 2, 3, 4],
             ...     "b": [0.5, 4, 10, 13],
             ...     "c": [True, True, False, True],
             ... }
-            >>> df_pd = pd.DataFrame(df)
-            >>> df_pl = pl.DataFrame(df)
-            >>> lf_pl = pl.LazyFrame(df)
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we pass an expression
             to add it as a new column:
 
             >>> def agnostic_with_columns(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.with_columns((nw.col("a") * 2).alias("2a")).to_native()
+            ...     return (
+            ...         df.with_columns((nw.col("a") * 2).alias("2a")).collect().to_native()
+            ...     )
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_with_columns`:
 
-            >>> agnostic_with_columns(df_pd)
-               a     b      c  2a
-            0  1   0.5   True   2
-            1  2   4.0   True   4
-            2  3  10.0  False   6
-            3  4  13.0   True   8
-            >>> agnostic_with_columns(df_pl)
-            shape: (4, 4)
-            ┌─────┬──────┬───────┬─────┐
-            │ a   ┆ b    ┆ c     ┆ 2a  │
-            │ --- ┆ ---  ┆ ---   ┆ --- │
-            │ i64 ┆ f64  ┆ bool  ┆ i64 │
-            ╞═════╪══════╪═══════╪═════╡
-            │ 1   ┆ 0.5  ┆ true  ┆ 2   │
-            │ 2   ┆ 4.0  ┆ true  ┆ 4   │
-            │ 3   ┆ 10.0 ┆ false ┆ 6   │
-            │ 4   ┆ 13.0 ┆ true  ┆ 8   │
-            └─────┴──────┴───────┴─────┘
-            >>> agnostic_with_columns(lf_pl).collect()
+            >>> agnostic_with_columns(lf_pl)
             shape: (4, 4)
             ┌─────┬──────┬───────┬─────┐
             │ a   ┆ b    ┆ c     ┆ 2a  │
@@ -3976,6 +4001,12 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 3   ┆ 10.0 ┆ false ┆ 6   │
             │ 4   ┆ 13.0 ┆ true  ┆ 8   │
             └─────┴──────┴───────┴─────┘
+            >>> agnostic_with_columns(lf_dask)
+               a     b      c  2a
+            0  1   0.5   True   2
+            1  2   4.0   True   4
+            2  3  10.0  False   6
+            3  4  13.0   True   8
         """
         return super().with_columns(*exprs, **named_exprs)
 
@@ -4002,46 +4033,29 @@ class LazyFrame(BaseFrame[FrameT]):
             `0` use `df.select(nw.col(0))`, not `df.select(0)`.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
-            >>> df = {
+            >>> data = {
             ...     "foo": [1, 2, 3],
             ...     "bar": [6, 7, 8],
             ...     "ham": ["a", "b", "c"],
             ... }
-            >>> df_pd = pd.DataFrame(df)
-            >>> df_pl = pl.DataFrame(df)
-            >>> lf_pl = pl.LazyFrame(df)
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we pass the name of a
             column to select that column.
 
             >>> def agnostic_select(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.select("foo").to_native()
+            ...     return df.select("foo").collect().to_native()
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_select`:
 
-            >>> agnostic_select(df_pd)
-               foo
-            0    1
-            1    2
-            2    3
-            >>> agnostic_select(df_pl)
-            shape: (3, 1)
-            ┌─────┐
-            │ foo │
-            │ --- │
-            │ i64 │
-            ╞═════╡
-            │ 1   │
-            │ 2   │
-            │ 3   │
-            └─────┘
-            >>> agnostic_select(lf_pl).collect()
+            >>> agnostic_select(lf_pl)
             shape: (3, 1)
             ┌─────┐
             │ foo │
@@ -4052,30 +4066,19 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   │
             │ 3   │
             └─────┘
+            >>> agnostic_select(lf_dask)
+               foo
+            0    1
+            1    2
+            2    3
 
             Multiple columns can be selected by passing a list of column names.
 
             >>> def agnostic_select(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.select(["foo", "bar"]).to_native()
-            >>>
-            >>> agnostic_select(df_pd)
-               foo  bar
-            0    1    6
-            1    2    7
-            2    3    8
-            >>> agnostic_select(df_pl)
-            shape: (3, 2)
-            ┌─────┬─────┐
-            │ foo ┆ bar │
-            │ --- ┆ --- │
-            │ i64 ┆ i64 │
-            ╞═════╪═════╡
-            │ 1   ┆ 6   │
-            │ 2   ┆ 7   │
-            │ 3   ┆ 8   │
-            └─────┴─────┘
-            >>> agnostic_select(lf_pl).collect()
+            ...     return df.select(["foo", "bar"]).collect().to_native()
+
+            >>> agnostic_select(lf_pl)
             shape: (3, 2)
             ┌─────┬─────┐
             │ foo ┆ bar │
@@ -4086,31 +4089,20 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   ┆ 7   │
             │ 3   ┆ 8   │
             └─────┴─────┘
+            >>> agnostic_select(lf_dask)
+               foo  bar
+            0    1    6
+            1    2    7
+            2    3    8
 
             Multiple columns can also be selected using positional arguments instead of a
             list. Expressions are also accepted.
 
             >>> def agnostic_select(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.select(nw.col("foo"), nw.col("bar") + 1).to_native()
-            >>>
-            >>> agnostic_select(df_pd)
-               foo  bar
-            0    1    7
-            1    2    8
-            2    3    9
-            >>> agnostic_select(df_pl)
-            shape: (3, 2)
-            ┌─────┬─────┐
-            │ foo ┆ bar │
-            │ --- ┆ --- │
-            │ i64 ┆ i64 │
-            ╞═════╪═════╡
-            │ 1   ┆ 7   │
-            │ 2   ┆ 8   │
-            │ 3   ┆ 9   │
-            └─────┴─────┘
-            >>> agnostic_select(lf_pl).collect()
+            ...     return df.select(nw.col("foo"), nw.col("bar") + 1).collect().to_native()
+
+            >>> agnostic_select(lf_pl)
             shape: (3, 2)
             ┌─────┬─────┐
             │ foo ┆ bar │
@@ -4121,30 +4113,19 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   ┆ 8   │
             │ 3   ┆ 9   │
             └─────┴─────┘
+            >>> agnostic_select(lf_dask)
+               foo  bar
+            0    1    7
+            1    2    8
+            2    3    9
 
             Use keyword arguments to easily name your expression inputs.
 
             >>> def agnostic_select(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.select(threshold=nw.col("foo") * 2).to_native()
-            >>>
-            >>> agnostic_select(df_pd)
-               threshold
-            0          2
-            1          4
-            2          6
-            >>> agnostic_select(df_pl)
-            shape: (3, 1)
-            ┌───────────┐
-            │ threshold │
-            │ ---       │
-            │ i64       │
-            ╞═══════════╡
-            │ 2         │
-            │ 4         │
-            │ 6         │
-            └───────────┘
-            >>> agnostic_select(lf_pl).collect()
+            ...     return df.select(threshold=nw.col("foo") * 2).collect().to_native()
+
+            >>> agnostic_select(lf_pl)
             shape: (3, 1)
             ┌───────────┐
             │ threshold │
@@ -4155,6 +4136,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 4         │
             │ 6         │
             └───────────┘
+            >>> agnostic_select(lf_dask)
+               threshold
+            0          2
+            1          4
+            2          6
         """
         return super().select(*exprs, **named_exprs)
 
@@ -4170,29 +4156,24 @@ class LazyFrame(BaseFrame[FrameT]):
             The LazyFrame with the specified columns renamed.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {"foo": [1, 2, 3], "bar": [6, 7, 8], "ham": ["a", "b", "c"]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             We define a library agnostic function:
 
             >>> def agnostic_rename(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.rename({"foo": "apple"}).to_native()
+            ...     return df.rename({"foo": "apple"}).collect().to_native()
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_rename`:
 
-            >>> agnostic_rename(df_pd)
-               apple  bar ham
-            0      1    6   a
-            1      2    7   b
-            2      3    8   c
-            >>> agnostic_rename(lf_pl).collect()
+            >>> agnostic_rename(lf_pl)
             shape: (3, 3)
             ┌───────┬─────┬─────┐
             │ apple ┆ bar ┆ ham │
@@ -4203,6 +4184,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2     ┆ 7   ┆ b   │
             │ 3     ┆ 8   ┆ c   │
             └───────┴─────┴─────┘
+            >>> agnostic_rename(lf_dask)
+               apple  bar ham
+            0      1    6   a
+            1      2    7   b
+            2      3    8   c
         """
         return super().rename(mapping)
 
@@ -4217,43 +4203,26 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {
             ...     "a": [1, 2, 3, 4, 5, 6],
             ...     "b": [7, 8, 9, 10, 11, 12],
             ... }
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function that gets the first 3 rows.
 
             >>> def agnostic_head(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.head(3).to_native()
+            ...     return df.head(3).collect().to_native()
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_head`:
 
-            >>> agnostic_head(df_pd)
-               a  b
-            0  1  7
-            1  2  8
-            2  3  9
-            >>> agnostic_head(df_pl)
-            shape: (3, 2)
-            ┌─────┬─────┐
-            │ a   ┆ b   │
-            │ --- ┆ --- │
-            │ i64 ┆ i64 │
-            ╞═════╪═════╡
-            │ 1   ┆ 7   │
-            │ 2   ┆ 8   │
-            │ 3   ┆ 9   │
-            └─────┴─────┘
-            >>> agnostic_head(lf_pl).collect()
+            >>> agnostic_head(lf_pl)
             shape: (3, 2)
             ┌─────┬─────┐
             │ a   ┆ b   │
@@ -4264,6 +4233,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   ┆ 8   │
             │ 3   ┆ 9   │
             └─────┴─────┘
+            >>> agnostic_head(lf_dask)
+               a  b
+            0  1  7
+            1  2  8
+            2  3  9
         """
         return super().head(n)
 
@@ -4276,45 +4250,32 @@ class LazyFrame(BaseFrame[FrameT]):
         Returns:
             A subset of the LazyFrame of shape (n, n_columns).
 
+        Notes:
+            `LazyFrame.tail` is not supported for the Dask backend with multiple
+            partitions.
+
         Examples:
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {
             ...     "a": [1, 2, 3, 4, 5, 6],
             ...     "b": [7, 8, 9, 10, 11, 12],
             ... }
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=1)
 
             Let's define a dataframe-agnostic function that gets the last 3 rows.
 
             >>> def agnostic_tail(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.tail(3).to_native()
+            ...     return df.tail(3).collect().to_native()
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_tail`:
 
-            >>> agnostic_tail(df_pd)
-               a   b
-            3  4  10
-            4  5  11
-            5  6  12
-            >>> agnostic_tail(df_pl)
-            shape: (3, 2)
-            ┌─────┬─────┐
-            │ a   ┆ b   │
-            │ --- ┆ --- │
-            │ i64 ┆ i64 │
-            ╞═════╪═════╡
-            │ 4   ┆ 10  │
-            │ 5   ┆ 11  │
-            │ 6   ┆ 12  │
-            └─────┴─────┘
-            >>> agnostic_tail(lf_pl).collect()
+            >>> agnostic_tail(lf_pl)
             shape: (3, 2)
             ┌─────┬─────┐
             │ a   ┆ b   │
@@ -4325,6 +4286,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 5   ┆ 11  │
             │ 6   ┆ 12  │
             └─────┴─────┘
+            >>> agnostic_tail(lf_dask)
+               a   b
+            3  4  10
+            4  5  11
+            5  6  12
         """
         return super().tail(n)
 
@@ -4345,29 +4311,24 @@ class LazyFrame(BaseFrame[FrameT]):
             Please consider upgrading to a newer version or pass to eager mode.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {"foo": [1, 2, 3], "bar": [6.0, 7.0, 8.0], "ham": ["a", "b", "c"]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             We define a library agnostic function:
 
             >>> def agnostic_drop(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.drop("ham").to_native()
+            ...     return df.drop("ham").collect().to_native()
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_drop`:
 
-            >>> agnostic_drop(df_pd)
-               foo  bar
-            0    1  6.0
-            1    2  7.0
-            2    3  8.0
-            >>> agnostic_drop(lf_pl).collect()
+            >>> agnostic_drop(lf_pl)
             shape: (3, 2)
             ┌─────┬─────┐
             │ foo ┆ bar │
@@ -4378,19 +4339,19 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   ┆ 7.0 │
             │ 3   ┆ 8.0 │
             └─────┴─────┘
+            >>> agnostic_drop(lf_dask)
+               foo  bar
+            0    1  6.0
+            1    2  7.0
+            2    3  8.0
 
             Use positional arguments to drop multiple columns.
 
             >>> def agnostic_drop(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.drop("foo", "ham").to_native()
+            ...     return df.drop("foo", "ham").collect().to_native()
 
-            >>> agnostic_drop(df_pd)
-               bar
-            0  6.0
-            1  7.0
-            2  8.0
-            >>> agnostic_drop(lf_pl).collect()
+            >>> agnostic_drop(lf_pl)
             shape: (3, 1)
             ┌─────┐
             │ bar │
@@ -4401,6 +4362,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 7.0 │
             │ 8.0 │
             └─────┘
+            >>> agnostic_drop(lf_dask)
+               bar
+            0  6.0
+            1  7.0
+            2  8.0
         """
         return super().drop(*flatten(columns), strict=strict)
 
@@ -4425,11 +4391,11 @@ class LazyFrame(BaseFrame[FrameT]):
             maintain_order: Has no effect and is kept around only for backwards-compatibility.
 
         Returns:
-            LazyFrame: LazyFrame with unique rows.
+            The LazyFrame with unique rows.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
@@ -4438,21 +4404,18 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     "bar": ["a", "a", "a", "a"],
             ...     "ham": ["b", "b", "b", "b"],
             ... }
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             We define a library agnostic function:
 
             >>> def agnostic_unique(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.unique(["bar", "ham"]).to_native()
+            ...     return df.unique(["bar", "ham"]).collect().to_native()
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_unique`:
 
-            >>> agnostic_unique(df_pd)
-               foo bar ham
-            0    1   a   b
-            >>> agnostic_unique(lf_pl).collect()
+            >>> agnostic_unique(lf_pl)
             shape: (1, 3)
             ┌─────┬─────┬─────┐
             │ foo ┆ bar ┆ ham │
@@ -4461,6 +4424,9 @@ class LazyFrame(BaseFrame[FrameT]):
             ╞═════╪═════╪═════╡
             │ 1   ┆ a   ┆ b   │
             └─────┴─────┴─────┘
+            >>> agnostic_unique(lf_dask)
+               foo bar ham
+            0    1   a   b
         """
         if keep not in {"any", "none"}:
             msg = (
@@ -4501,8 +4467,8 @@ class LazyFrame(BaseFrame[FrameT]):
             The filtered LazyFrame.
 
         Examples:
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
@@ -4511,34 +4477,19 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     "bar": [6, 7, 8],
             ...     "ham": ["a", "b", "c"],
             ... }
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we filter on
             one condition.
 
             >>> def agnostic_filter(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.filter(nw.col("foo") > 1).to_native()
+            ...     return df.filter(nw.col("foo") > 1).collect().to_native()
 
-            We can then pass either pandas or Polars to `agnostic_filter`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_filter`:
 
-            >>> agnostic_filter(df_pd)
-               foo  bar ham
-            1    2    7   b
-            2    3    8   c
-            >>> agnostic_filter(df_pl)
-            shape: (2, 3)
-            ┌─────┬─────┬─────┐
-            │ foo ┆ bar ┆ ham │
-            │ --- ┆ --- ┆ --- │
-            │ i64 ┆ i64 ┆ str │
-            ╞═════╪═════╪═════╡
-            │ 2   ┆ 7   ┆ b   │
-            │ 3   ┆ 8   ┆ c   │
-            └─────┴─────┴─────┘
-            >>> agnostic_filter(lf_pl).collect()
+            >>> agnostic_filter(lf_pl)
             shape: (2, 3)
             ┌─────┬─────┬─────┐
             │ foo ┆ bar ┆ ham │
@@ -4548,26 +4499,22 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   ┆ 7   ┆ b   │
             │ 3   ┆ 8   ┆ c   │
             └─────┴─────┴─────┘
+            >>> agnostic_filter(lf_dask)
+               foo  bar ham
+            1    2    7   b
+            2    3    8   c
 
             Filter on multiple conditions:
 
             >>> def agnostic_filter(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.filter((nw.col("foo") < 3) & (nw.col("ham") == "a")).to_native()
-            >>>
-            >>> agnostic_filter(df_pd)
-               foo  bar ham
-            0    1    6   a
-            >>> agnostic_filter(df_pl)
-            shape: (1, 3)
-            ┌─────┬─────┬─────┐
-            │ foo ┆ bar ┆ ham │
-            │ --- ┆ --- ┆ --- │
-            │ i64 ┆ i64 ┆ str │
-            ╞═════╪═════╪═════╡
-            │ 1   ┆ 6   ┆ a   │
-            └─────┴─────┴─────┘
-            >>> agnostic_filter(lf_pl).collect()
+            ...     return (
+            ...         df.filter((nw.col("foo") < 3) & (nw.col("ham") == "a"))
+            ...         .collect()
+            ...         .to_native()
+            ...     )
+
+            >>> agnostic_filter(lf_pl)
             shape: (1, 3)
             ┌─────┬─────┬─────┐
             │ foo ┆ bar ┆ ham │
@@ -4576,29 +4523,24 @@ class LazyFrame(BaseFrame[FrameT]):
             ╞═════╪═════╪═════╡
             │ 1   ┆ 6   ┆ a   │
             └─────┴─────┴─────┘
+            >>> agnostic_filter(lf_dask)
+               foo  bar ham
+            0    1    6   a
 
             Provide multiple filters using `*args` syntax:
 
             >>> def agnostic_filter(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.filter(
-            ...         nw.col("foo") == 1,
-            ...         nw.col("ham") == "a",
-            ...     ).to_native()
-            >>>
-            >>> agnostic_filter(df_pd)
-               foo  bar ham
-            0    1    6   a
-            >>> agnostic_filter(df_pl)
-            shape: (1, 3)
-            ┌─────┬─────┬─────┐
-            │ foo ┆ bar ┆ ham │
-            │ --- ┆ --- ┆ --- │
-            │ i64 ┆ i64 ┆ str │
-            ╞═════╪═════╪═════╡
-            │ 1   ┆ 6   ┆ a   │
-            └─────┴─────┴─────┘
-            >>> agnostic_filter(lf_pl).collect()
+            ...     return (
+            ...         df.filter(
+            ...             nw.col("foo") == 1,
+            ...             nw.col("ham") == "a",
+            ...         )
+            ...         .collect()
+            ...         .to_native()
+            ...     )
+
+            >>> agnostic_filter(lf_pl)
             shape: (1, 3)
             ┌─────┬─────┬─────┐
             │ foo ┆ bar ┆ ham │
@@ -4607,30 +4549,21 @@ class LazyFrame(BaseFrame[FrameT]):
             ╞═════╪═════╪═════╡
             │ 1   ┆ 6   ┆ a   │
             └─────┴─────┴─────┘
+            >>> agnostic_filter(lf_dask)
+               foo  bar ham
+            0    1    6   a
 
             Filter on an OR condition:
 
             >>> def agnostic_filter(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.filter(
-            ...         (nw.col("foo") == 1) | (nw.col("ham") == "c")
-            ...     ).to_native()
-            >>>
-            >>> agnostic_filter(df_pd)
-               foo  bar ham
-            0    1    6   a
-            2    3    8   c
-            >>> agnostic_filter(df_pl)
-            shape: (2, 3)
-            ┌─────┬─────┬─────┐
-            │ foo ┆ bar ┆ ham │
-            │ --- ┆ --- ┆ --- │
-            │ i64 ┆ i64 ┆ str │
-            ╞═════╪═════╪═════╡
-            │ 1   ┆ 6   ┆ a   │
-            │ 3   ┆ 8   ┆ c   │
-            └─────┴─────┴─────┘
-            >>> agnostic_filter(lf_pl).collect()
+            ...     return (
+            ...         df.filter((nw.col("foo") == 1) | (nw.col("ham") == "c"))
+            ...         .collect()
+            ...         .to_native()
+            ...     )
+
+            >>> agnostic_filter(lf_pl)
             shape: (2, 3)
             ┌─────┬─────┬─────┐
             │ foo ┆ bar ┆ ham │
@@ -4640,26 +4573,18 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 1   ┆ 6   ┆ a   │
             │ 3   ┆ 8   ┆ c   │
             └─────┴─────┴─────┘
+            >>> agnostic_filter(lf_dask)
+               foo  bar ham
+            0    1    6   a
+            2    3    8   c
 
             Provide multiple filters using `**kwargs` syntax:
 
             >>> def agnostic_filter(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.filter(foo=2, ham="b").to_native()
-            >>>
-            >>> agnostic_filter(df_pd)
-               foo  bar ham
-            1    2    7   b
-            >>> agnostic_filter(df_pl)
-            shape: (1, 3)
-            ┌─────┬─────┬─────┐
-            │ foo ┆ bar ┆ ham │
-            │ --- ┆ --- ┆ --- │
-            │ i64 ┆ i64 ┆ str │
-            ╞═════╪═════╪═════╡
-            │ 2   ┆ 7   ┆ b   │
-            └─────┴─────┴─────┘
-            >>> agnostic_filter(lf_pl).collect()
+            ...     return df.filter(foo=2, ham="b").collect().to_native()
+
+            >>> agnostic_filter(lf_pl)
             shape: (1, 3)
             ┌─────┬─────┬─────┐
             │ foo ┆ bar ┆ ham │
@@ -4668,6 +4593,9 @@ class LazyFrame(BaseFrame[FrameT]):
             ╞═════╪═════╪═════╡
             │ 2   ┆ 7   ┆ b   │
             └─────┴─────┴─────┘
+            >>> agnostic_filter(lf_dask)
+               foo  bar ham
+            1    2    7   b
         """
         return super().filter(*predicates, **constraints)
 
@@ -4684,52 +4612,41 @@ class LazyFrame(BaseFrame[FrameT]):
                 included in the result.
 
         Returns:
-            LazyGroupBy: Object which can be used to perform aggregations.
+            Object which can be used to perform aggregations.
 
         Examples:
             Group by one column and call `agg` to compute the grouped sum of
             another column.
 
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> import narwhals as nw
             >>> from narwhals.typing import IntoFrameT
             >>>
-            >>> df = {
+            >>> data = {
             ...     "a": ["a", "b", "a", "b", "c"],
             ...     "b": [1, 2, 1, 3, 3],
             ...     "c": [5, 4, 3, 2, 1],
             ... }
-            >>> df_pd = pd.DataFrame(df)
-            >>> df_pl = pl.DataFrame(df)
-            >>> lf_pl = pl.LazyFrame(df)
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we group by one column
             and call `agg` to compute the grouped sum of another column.
 
             >>> def agnostic_group_by_agg(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.group_by("a").agg(nw.col("b").sum()).sort("a").to_native()
+            ...     return (
+            ...         df.group_by("a")
+            ...         .agg(nw.col("b").sum())
+            ...         .sort("a")
+            ...         .collect()
+            ...         .to_native()
+            ...     )
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_group_by_agg`:
 
-            >>> agnostic_group_by_agg(df_pd)
-               a  b
-            0  a  2
-            1  b  5
-            2  c  3
-            >>> agnostic_group_by_agg(df_pl)
-            shape: (3, 2)
-            ┌─────┬─────┐
-            │ a   ┆ b   │
-            │ --- ┆ --- │
-            │ str ┆ i64 │
-            ╞═════╪═════╡
-            │ a   ┆ 2   │
-            │ b   ┆ 5   │
-            │ c   ┆ 3   │
-            └─────┴─────┘
-            >>> agnostic_group_by_agg(lf_pl).collect()
+            >>> agnostic_group_by_agg(lf_pl)
             shape: (3, 2)
             ┌─────┬─────┐
             │ a   ┆ b   │
@@ -4740,34 +4657,25 @@ class LazyFrame(BaseFrame[FrameT]):
             │ b   ┆ 5   │
             │ c   ┆ 3   │
             └─────┴─────┘
+            >>> agnostic_group_by_agg(lf_dask)
+               a  b
+            0  a  2
+            1  b  5
+            2  c  3
 
             Group by multiple columns by passing a list of column names.
 
             >>> def agnostic_group_by_agg(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
             ...     return (
-            ...         df.group_by(["a", "b"]).agg(nw.max("c")).sort(["a", "b"]).to_native()
+            ...         df.group_by(["a", "b"])
+            ...         .agg(nw.max("c"))
+            ...         .sort(["a", "b"])
+            ...         .collect()
+            ...         .to_native()
             ...     )
-            >>>
-            >>> agnostic_group_by_agg(df_pd)
-               a  b  c
-            0  a  1  5
-            1  b  2  4
-            2  b  3  2
-            3  c  3  1
-            >>> agnostic_group_by_agg(df_pl)
-            shape: (4, 3)
-            ┌─────┬─────┬─────┐
-            │ a   ┆ b   ┆ c   │
-            │ --- ┆ --- ┆ --- │
-            │ str ┆ i64 ┆ i64 │
-            ╞═════╪═════╪═════╡
-            │ a   ┆ 1   ┆ 5   │
-            │ b   ┆ 2   ┆ 4   │
-            │ b   ┆ 3   ┆ 2   │
-            │ c   ┆ 3   ┆ 1   │
-            └─────┴─────┴─────┘
-            >>> agnostic_group_by_agg(lf_pl).collect()
+
+            >>> agnostic_group_by_agg(lf_pl)
             shape: (4, 3)
             ┌─────┬─────┬─────┐
             │ a   ┆ b   ┆ c   │
@@ -4779,6 +4687,12 @@ class LazyFrame(BaseFrame[FrameT]):
             │ b   ┆ 3   ┆ 2   │
             │ c   ┆ 3   ┆ 1   │
             └─────┴─────┴─────┘
+            >>> agnostic_group_by_agg(lf_dask)
+               a  b  c
+            0  a  1  5
+            1  b  2  4
+            2  b  3  2
+            3  c  3  1
         """
         from narwhals.expr import Expr
         from narwhals.group_by import LazyGroupBy
@@ -4820,8 +4734,8 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {
@@ -4829,24 +4743,19 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     "b": [6.0, 5.0, 4.0],
             ...     "c": ["a", "c", "b"],
             ... }
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we sort by multiple
             columns in different orders
 
             >>> def agnostic_sort(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.sort("c", "a", descending=[False, True]).to_native()
+            ...     return df.sort("c", "a", descending=[False, True]).collect().to_native()
 
-            We can then pass either pandas or Polars to `func`:
+            We can then pass any supported library such as Polars or Dask to `agnostic_sort`:
 
-            >>> agnostic_sort(df_pd)
-                 a    b  c
-            0  1.0  6.0  a
-            2  NaN  4.0  b
-            1  2.0  5.0  c
-            >>> agnostic_sort(lf_pl).collect()
+            >>> agnostic_sort(lf_pl)
             shape: (3, 3)
             ┌──────┬─────┬─────┐
             │ a    ┆ b   ┆ c   │
@@ -4857,6 +4766,11 @@ class LazyFrame(BaseFrame[FrameT]):
             │ null ┆ 4.0 ┆ b   │
             │ 2    ┆ 5.0 ┆ c   │
             └──────┴─────┴─────┘
+            >>> agnostic_sort(lf_dask)
+                 a    b  c
+            0  1.0  6.0  a
+            2  NaN  4.0  b
+            1  2.0  5.0  c
         """
         return super().sort(by, *more_by, descending=descending, nulls_last=nulls_last)
 
@@ -4892,8 +4806,8 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {
@@ -4906,11 +4820,10 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     "ham": ["a", "b", "d"],
             ... }
 
-            >>> df_pd = pd.DataFrame(data)
-            >>> other_pd = pd.DataFrame(data_other)
-
             >>> lf_pl = pl.LazyFrame(data)
             >>> other_pl = pl.LazyFrame(data_other)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
+            >>> other_dask = dd.from_dict(data_other, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we join over "ham" column:
 
@@ -4920,16 +4833,16 @@ class LazyFrame(BaseFrame[FrameT]):
             ... ) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
             ...     other = nw.from_native(other_native)
-            ...     return df.join(other, left_on="ham", right_on="ham").to_native()
-
-            We can now pass either pandas or Polars to the function:
+            ...     return (
+            ...         df.join(other, left_on="ham", right_on="ham")
+            ...         .sort("ham")
+            ...         .collect()
+            ...         .to_native()
+            ...     )
 
-            >>> agnostic_join_on_ham(df_pd, other_pd)
-               foo  bar ham apple
-            0    1  6.0   a     x
-            1    2  7.0   b     y
+            We can then pass any supported library such as Polars or Dask to `agnostic_join_on_ham`:
 
-            >>> agnostic_join_on_ham(lf_pl, other_pl).collect()
+            >>> agnostic_join_on_ham(lf_pl, other_pl)
             shape: (2, 4)
             ┌─────┬─────┬─────┬───────┐
             │ foo ┆ bar ┆ ham ┆ apple │
@@ -4939,6 +4852,10 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 1   ┆ 6.0 ┆ a   ┆ x     │
             │ 2   ┆ 7.0 ┆ b   ┆ y     │
             └─────┴─────┴─────┴───────┘
+            >>> agnostic_join_on_ham(lf_dask, other_dask)
+               foo  bar ham apple
+            0    1  6.0   a     x
+            0    2  7.0   b     y
         """
         return super().join(
             other, how=how, left_on=left_on, right_on=right_on, on=on, suffix=suffix
@@ -4989,8 +4906,8 @@ class LazyFrame(BaseFrame[FrameT]):
         Examples:
             >>> from datetime import datetime
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from typing import Literal
             >>> from narwhals.typing import IntoFrameT
             >>>
@@ -5012,10 +4929,10 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     ],
             ...     "population": [82.19, 82.66, 83.12],
             ... }
-            >>> gdp_pd = pd.DataFrame(data_gdp)
-            >>> population_pd = pd.DataFrame(data_population)
-            >>> gdp_pl = pl.LazyFrame(data_gdp).sort("datetime")
-            >>> population_pl = pl.LazyFrame(data_population).sort("datetime")
+            >>> gdp_pl = pl.LazyFrame(data_gdp)
+            >>> population_pl = pl.LazyFrame(data_population)
+            >>> gdp_dask = dd.from_dict(data_gdp, npartitions=2)
+            >>> population_dask = dd.from_dict(data_population, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we join over "datetime" column:
 
@@ -5026,19 +4943,16 @@ class LazyFrame(BaseFrame[FrameT]):
             ... ) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
             ...     other = nw.from_native(other_native)
-            ...     return df.join_asof(other, on="datetime", strategy=strategy).to_native()
-
-            We can now pass either pandas or Polars to the function:
+            ...     return (
+            ...         df.sort("datetime")
+            ...         .join_asof(other, on="datetime", strategy=strategy)
+            ...         .collect()
+            ...         .to_native()
+            ...     )
 
-            >>> agnostic_join_asof_datetime(population_pd, gdp_pd, strategy="backward")
-                datetime  population   gdp
-            0 2016-03-01       82.19  4164
-            1 2018-08-01       82.66  4566
-            2 2019-01-01       83.12  4696
+            We can then pass any supported library such as Polars or Dask to `agnostic_join_asof_datetime`:
 
-            >>> agnostic_join_asof_datetime(
-            ...     population_pl, gdp_pl, strategy="backward"
-            ... ).collect()
+            >>> agnostic_join_asof_datetime(population_pl, gdp_pl, strategy="backward")
             shape: (3, 3)
             ┌─────────────────────┬────────────┬──────┐
             │ datetime            ┆ population ┆ gdp  │
@@ -5049,13 +4963,18 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2018-08-01 00:00:00 ┆ 82.66      ┆ 4566 │
             │ 2019-01-01 00:00:00 ┆ 83.12      ┆ 4696 │
             └─────────────────────┴────────────┴──────┘
+            >>> agnostic_join_asof_datetime(population_dask, gdp_dask, strategy="backward")
+                datetime  population   gdp
+            0 2016-03-01       82.19  4164
+            1 2018-08-01       82.66  4566
+            0 2019-01-01       83.12  4696
 
             Here is a real-world times-series example that uses `by` argument.
 
             >>> from datetime import datetime
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data_quotes = {
@@ -5094,10 +5013,10 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     "price": [51.95, 51.95, 720.77, 720.92, 98.0],
             ...     "quantity": [75, 155, 100, 100, 100],
             ... }
-            >>> quotes_pd = pd.DataFrame(data_quotes)
-            >>> trades_pd = pd.DataFrame(data_trades)
-            >>> quotes_pl = pl.LazyFrame(data_quotes).sort("datetime")
-            >>> trades_pl = pl.LazyFrame(data_trades).sort("datetime")
+            >>> quotes_pl = pl.LazyFrame(data_quotes)
+            >>> trades_pl = pl.LazyFrame(data_trades)
+            >>> quotes_dask = dd.from_dict(data_quotes, npartitions=2)
+            >>> trades_dask = dd.from_dict(data_trades, npartitions=2)
 
             Let's define a dataframe-agnostic function in which we join over "datetime" and by "ticker" columns:
 
@@ -5107,19 +5026,16 @@ class LazyFrame(BaseFrame[FrameT]):
             ... ) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
             ...     other = nw.from_native(other_native)
-            ...     return df.join_asof(other, on="datetime", by="ticker").to_native()
-
-            We can now pass either pandas or Polars to the function:
+            ...     return (
+            ...         df.sort("datetime")
+            ...         .join_asof(other, on="datetime", by="ticker")
+            ...         .collect()
+            ...         .to_native()
+            ...     )
 
-            >>> agnostic_join_asof_datetime_by_ticker(trades_pd, quotes_pd)
-                                datetime ticker   price  quantity     bid     ask
-            0 2016-05-25 13:30:00.000023   MSFT   51.95        75   51.95   51.96
-            1 2016-05-25 13:30:00.000038   MSFT   51.95       155   51.97   51.98
-            2 2016-05-25 13:30:00.000048   GOOG  720.77       100  720.50  720.93
-            3 2016-05-25 13:30:00.000048   GOOG  720.92       100  720.50  720.93
-            4 2016-05-25 13:30:00.000048   AAPL   98.00       100     NaN     NaN
+            We can then pass any supported library such as Polars or Dask to `agnostic_join_asof_datetime_by_ticker`:
 
-            >>> agnostic_join_asof_datetime_by_ticker(trades_pl, quotes_pl).collect()
+            >>> agnostic_join_asof_datetime_by_ticker(trades_pl, quotes_pl)
             shape: (5, 6)
             ┌────────────────────────────┬────────┬────────┬──────────┬───────┬────────┐
             │ datetime                   ┆ ticker ┆ price  ┆ quantity ┆ bid   ┆ ask    │
@@ -5132,6 +5048,13 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2016-05-25 13:30:00.000048 ┆ GOOG   ┆ 720.92 ┆ 100      ┆ 720.5 ┆ 720.93 │
             │ 2016-05-25 13:30:00.000048 ┆ AAPL   ┆ 98.0   ┆ 100      ┆ null  ┆ null   │
             └────────────────────────────┴────────┴────────┴──────────┴───────┴────────┘
+            >>> agnostic_join_asof_datetime_by_ticker(trades_dask, quotes_dask)
+                                datetime ticker   price  quantity     bid     ask
+            0 2016-05-25 13:30:00.000023   MSFT   51.95        75   51.95   51.96
+            0 2016-05-25 13:30:00.000038   MSFT   51.95       155   51.97   51.98
+            1 2016-05-25 13:30:00.000048   GOOG  720.92       100  720.50  720.93
+            2 2016-05-25 13:30:00.000048   AAPL   98.00       100     NaN     NaN
+            3 2016-05-25 13:30:00.000048   GOOG  720.77       100  720.50  720.93
         """
         return super().join_asof(
             other,
@@ -5152,26 +5075,21 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {"a": [1, 2], "b": [3, 4]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
 
             Let's define a dataframe-agnostic function in which we copy the DataFrame:
 
             >>> def agnostic_clone(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.clone().to_native()
+            ...     return df.clone().collect().to_native()
 
-            >>> agnostic_clone(df_pd)
-               a  b
-            0  1  3
-            1  2  4
+            We can then pass any supported library such as Polars to `agnostic_clone`:
 
-            >>> agnostic_clone(lf_pl).collect()
+            >>> agnostic_clone(lf_pl)
             shape: (2, 2)
             ┌─────┬─────┐
             │ a   ┆ b   │
@@ -5234,27 +5152,24 @@ class LazyFrame(BaseFrame[FrameT]):
 
         Examples:
             >>> import narwhals as nw
-            >>> import pandas as pd
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {"a": [1, 2, 3, 4], "b": [5, 6, 7, 8]}
-            >>> df_pd = pd.DataFrame(data)
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
-            Let's define a dataframe-agnostic function in which gather every 2 rows,
+            Let's define a dataframe-agnostic function in which we gather every 2 rows,
             starting from a offset of 1:
 
             >>> def agnostic_gather_every(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
-            ...     return df.gather_every(n=2, offset=1).to_native()
+            ...     return df.gather_every(n=2, offset=1).collect().to_native()
 
-            >>> agnostic_gather_every(df_pd)
-               a  b
-            1  2  6
-            3  4  8
+            We can then pass any supported library such as Polars or Dask to `agnostic_gather_every`:
 
-            >>> agnostic_gather_every(lf_pl).collect()
+            >>> agnostic_gather_every(lf_pl)
             shape: (2, 2)
             ┌─────┬─────┐
             │ a   ┆ b   │
@@ -5264,6 +5179,10 @@ class LazyFrame(BaseFrame[FrameT]):
             │ 2   ┆ 6   │
             │ 4   ┆ 8   │
             └─────┴─────┘
+            >>> agnostic_gather_every(lf_dask)
+               a  b
+            1  2  6
+            3  4  8
         """
         return super().gather_every(n=n, offset=offset)
 
@@ -5302,6 +5221,7 @@ class LazyFrame(BaseFrame[FrameT]):
         Examples:
             >>> import narwhals as nw
             >>> import polars as pl
+            >>> import dask.dataframe as dd
             >>> from narwhals.typing import IntoFrameT
             >>>
             >>> data = {
@@ -5310,16 +5230,21 @@ class LazyFrame(BaseFrame[FrameT]):
             ...     "c": [2, 4, 6],
             ... }
             >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
 
             We define a library agnostic function:
 
             >>> def agnostic_unpivot(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
             ...     return (
-            ...         df.unpivot(on=["b", "c"], index="a").sort(["variable", "a"])
-            ...     ).to_native()
+            ...         (df.unpivot(on=["b", "c"], index="a").sort(["variable", "a"]))
+            ...         .collect()
+            ...         .to_native()
+            ...     )
+
+            We can then pass any supported library such as Polars or Dask to `agnostic_unpivot`:
 
-            >>> agnostic_unpivot(lf_pl).collect()
+            >>> agnostic_unpivot(lf_pl)
             shape: (6, 3)
             ┌─────┬──────────┬───────┐
             │ a   ┆ variable ┆ value │
@@ -5333,6 +5258,14 @@ class LazyFrame(BaseFrame[FrameT]):
             │ y   ┆ c        ┆ 4     │
             │ z   ┆ c        ┆ 6     │
             └─────┴──────────┴───────┘
+            >>> agnostic_unpivot(lf_dask)
+               a variable  value
+            0  x        b      1
+            1  y        b      3
+            0  z        b      5
+            2  x        c      2
+            3  y        c      4
+            1  z        c      6
         """
         return super().unpivot(
             on=on, index=index, variable_name=variable_name, value_name=value_name
@@ -5342,7 +5275,7 @@ class LazyFrame(BaseFrame[FrameT]):
         """Explode the dataframe to long format by exploding the given columns.
 
         Notes:
-            It is possible to explode multiple columns only if these columns must have
+            It is possible to explode multiple columns only if these columns have
             matching element counts.
 
         Arguments:
@@ -5369,13 +5302,13 @@ class LazyFrame(BaseFrame[FrameT]):
             ...         nw.from_native(df_native)
             ...         .with_columns(nw.col("lst1", "lst2").cast(nw.List(nw.Int32())))
             ...         .explode("lst1", "lst2")
+            ...         .collect()
             ...         .to_native()
             ...     )
 
-            We can then pass any supported library such as pandas, Polars (eager),
-            or PyArrow to `agnostic_explode`:
+            We can then pass any supported library such as Polars to `agnostic_explode`:
 
-            >>> agnostic_explode(pl.LazyFrame(data)).collect()
+            >>> agnostic_explode(pl.LazyFrame(data))
             shape: (5, 3)
             ┌─────┬──────┬──────┐
             │ a   ┆ lst1 ┆ lst2 │
diff --git a/narwhals/expr.py b/narwhals/expr.py
index fe5dc929..6e3cacb0 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -3002,8 +3002,8 @@ class Expr:
     # TODO @aivanoved: make type alias for numeric type
     def clip(
         self,
-        lower_bound: Any | None = None,
-        upper_bound: Any | None = None,
+        lower_bound: IntoExpr | Any | None = None,
+        upper_bound: IntoExpr | Any | None = None,
     ) -> Self:
         r"""Clip values in the Series.
 
@@ -3130,7 +3130,10 @@ class Expr:
             s: [[-1,1,-1,3,-1,3]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).clip(lower_bound, upper_bound)
+            lambda plx: self._to_compliant_expr(plx).clip(
+                extract_compliant(plx, lower_bound),
+                extract_compliant(plx, upper_bound),
+            )
         )
 
     def mode(self: Self) -> Self:
diff --git a/narwhals/series.py b/narwhals/series.py
index fb607795..c3e6f181 100644
--- a/narwhals/series.py
+++ b/narwhals/series.py
@@ -1342,7 +1342,7 @@ class Series(Generic[IntoSeriesT]):
         return self._compliant_series.var(ddof=ddof)
 
     def clip(
-        self, lower_bound: Any | None = None, upper_bound: Any | None = None
+        self, lower_bound: Self | Any | None = None, upper_bound: Self | Any | None = None
     ) -> Self:
         r"""Clip values in the Series.
 
@@ -1484,7 +1484,10 @@ class Series(Generic[IntoSeriesT]):
             ]
         """
         return self._from_compliant_series(
-            self._compliant_series.clip(lower_bound=lower_bound, upper_bound=upper_bound)
+            self._compliant_series.clip(
+                lower_bound=self._extract_native(lower_bound),
+                upper_bound=self._extract_native(upper_bound),
+            )
         )
 
     def is_in(self, other: Any) -> Self:
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index 8bf4a4b1..ba511742 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -422,13 +422,15 @@ class LazyFrame(NwLazyFrame[IntoFrameT]):
         Examples:
             >>> import narwhals as nw
             >>> import polars as pl
-            >>> lf_pl = pl.LazyFrame(
-            ...     {
-            ...         "a": ["a", "b", "a", "b", "b", "c"],
-            ...         "b": [1, 2, 3, 4, 5, 6],
-            ...         "c": [6, 5, 4, 3, 2, 1],
-            ...     }
-            ... )
+            >>> import dask.dataframe as dd
+            >>> data = {
+            ...     "a": ["a", "b", "a", "b", "b", "c"],
+            ...     "b": [1, 2, 3, 4, 5, 6],
+            ...     "c": [6, 5, 4, 3, 2, 1],
+            ... }
+            >>> lf_pl = pl.LazyFrame(data)
+            >>> lf_dask = dd.from_dict(data, npartitions=2)
+
             >>> lf = nw.from_native(lf_pl)
             >>> lf  # doctest:+ELLIPSIS
             ┌─────────────────────────────┐
@@ -448,6 +450,27 @@ class LazyFrame(NwLazyFrame[IntoFrameT]):
             │ b   ┆ 11  ┆ 10  │
             │ c   ┆ 6   ┆ 1   │
             └─────┴─────┴─────┘
+
+            >>> lf = nw.from_native(lf_dask)
+            >>> lf
+            ┌───────────────────────────────────┐
+            |        Narwhals LazyFrame         |
+            |-----------------------------------|
+            |Dask DataFrame Structure:          |
+            |                    a      b      c|
+            |npartitions=2                      |
+            |0              string  int64  int64|
+            |3                 ...    ...    ...|
+            |5                 ...    ...    ...|
+            |Dask Name: frompandas, 1 expression|
+            |Expr=df                            |
+            └───────────────────────────────────┘
+            >>> df = lf.group_by("a").agg(nw.col("b", "c").sum()).collect()
+            >>> df.to_native()
+               a   b   c
+            0  a   4  10
+            1  b  11  10
+            2  c   6   1
         """
         return super().collect()  # type: ignore[return-value]
 


Test patch:
diff --git a/tests/expr_and_series/cast_test.py b/tests/expr_and_series/cast_test.py
index 8cdceb9c..e956dd45 100644
--- a/tests/expr_and_series/cast_test.py
+++ b/tests/expr_and_series/cast_test.py
@@ -221,6 +221,28 @@ def test_cast_datetime_tz_aware(
     assert_equal_data(result, expected)
 
 
+def test_cast_struct(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if any(backend in str(constructor) for backend in ("dask", "modin", "cudf")):
+        request.applymarker(pytest.mark.xfail)
+
+    if "pandas" in str(constructor) and PANDAS_VERSION < (2, 2):
+        request.applymarker(pytest.mark.xfail)
+
+    data = {
+        "a": [
+            {"movie": "Cars", "rating": 4.5},
+            {"movie": "Toy Story", "rating": 4.9},
+        ]
+    }
+
+    dtype = nw.Struct([nw.Field("movie", nw.String()), nw.Field("rating", nw.Float64())])
+    result = (
+        nw.from_native(constructor(data)).select(nw.col("a").cast(dtype)).lazy().collect()
+    )
+
+    assert result.schema == {"a": dtype}
+
+
 @pytest.mark.parametrize("dtype", [pl.String, pl.String()])
 def test_raise_if_polars_dtype(constructor: Constructor, dtype: Any) -> None:
     df = nw.from_native(constructor({"a": [1, 2, 3], "b": [4, 5, 6]}))
diff --git a/tests/expr_and_series/clip_test.py b/tests/expr_and_series/clip_test.py
index 86fe7dad..838ca6b0 100644
--- a/tests/expr_and_series/clip_test.py
+++ b/tests/expr_and_series/clip_test.py
@@ -1,12 +1,14 @@
 from __future__ import annotations
 
+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
 
 
-def test_clip(constructor: Constructor) -> None:
+def test_clip_expr(constructor: Constructor) -> None:
     df = nw.from_native(constructor({"a": [1, 2, 3, -4, 5]}))
     result = df.select(
         lower_only=nw.col("a").clip(lower_bound=3),
@@ -21,6 +23,19 @@ def test_clip(constructor: Constructor) -> None:
     assert_equal_data(result, expected)
 
 
+def test_clip_expr_expressified(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    if "modin_pyarrow" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+
+    data = {"a": [1, 2, 3, -4, 5], "lb": [3, 2, 1, 1, 1], "ub": [4, 4, 2, 2, 2]}
+    df = nw.from_native(constructor(data))
+    result = df.select(nw.col("a").clip(nw.col("lb"), nw.col("ub") + 1))
+    expected_dict = {"a": [3, 2, 3, 1, 3]}
+    assert_equal_data(result, expected_dict)
+
+
 def test_clip_series(constructor_eager: ConstructorEager) -> None:
     df = nw.from_native(constructor_eager({"a": [1, 2, 3, -4, 5]}), eager_only=True)
     result = {
@@ -35,3 +50,16 @@ def test_clip_series(constructor_eager: ConstructorEager) -> None:
         "both": [3, 3, 3, 3, 4],
     }
     assert_equal_data(result, expected)
+
+
+def test_clip_series_expressified(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if "modin_pyarrow" in str(constructor_eager):
+        request.applymarker(pytest.mark.xfail)
+
+    data = {"a": [1, 2, 3, -4, 5], "lb": [3, 2, 1, 1, 1], "ub": [4, 4, 2, 2, 2]}
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df["a"].clip(df["lb"], df["ub"] + 1).to_frame()
+    expected_dict = {"a": [3, 2, 3, 1, 3]}
+    assert_equal_data(result, expected_dict)
diff --git a/tests/expr_and_series/double_test.py b/tests/expr_and_series/double_test.py
index 321defad..6f43ad13 100644
--- a/tests/expr_and_series/double_test.py
+++ b/tests/expr_and_series/double_test.py
@@ -18,9 +18,9 @@ def test_double_alias(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.with_columns(nw.col("a").alias("o"), nw.all() * 2)
     expected = {
-        "o": [1, 3, 2],
         "a": [2, 6, 4],
         "b": [8, 8, 12],
         "z": [14.0, 16.0, 18.0],
+        "o": [1, 3, 2],
     }
     assert_equal_data(result, expected)
diff --git a/tests/expr_and_series/is_nan_test.py b/tests/expr_and_series/is_nan_test.py
index d98b39e8..806dc753 100644
--- a/tests/expr_and_series/is_nan_test.py
+++ b/tests/expr_and_series/is_nan_test.py
@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import os
+from contextlib import nullcontext as does_not_raise
 from typing import Any
 
 import pytest
+from polars.exceptions import ComputeError
 
 import narwhals.stable.v1 as nw
 from tests.conftest import dask_lazy_p1_constructor
@@ -48,7 +51,16 @@ def test_nan(constructor: Constructor) -> None:
             "float_na": [True, False, None],
         }
 
-    assert_equal_data(result, expected)
+    context = (
+        pytest.raises(
+            ComputeError, match="NAN is not supported in a Non-floating point type column"
+        )
+        if "polars_lazy" in str(constructor)
+        and os.environ.get("NARWHALS_POLARS_GPU", False)
+        else does_not_raise()
+    )
+    with context:
+        assert_equal_data(result, expected)
 
 
 def test_nan_series(constructor_eager: ConstructorEager) -> None:
diff --git a/tests/frame/double_test.py b/tests/frame/double_test.py
index 87ff66af..a99cef5e 100644
--- a/tests/frame/double_test.py
+++ b/tests/frame/double_test.py
@@ -15,9 +15,9 @@ def test_double(constructor: Constructor) -> None:
 
     result = df.with_columns(nw.col("a").alias("o"), nw.all() * 2)
     expected = {
-        "o": [1, 3, 2],
         "a": [2, 6, 4],
         "b": [8, 8, 12],
         "z": [14.0, 16.0, 18.0],
+        "o": [1, 3, 2],
     }
     assert_equal_data(result, expected)
diff --git a/tests/frame/join_test.py b/tests/frame/join_test.py
index 1abe2b90..faeac5b2 100644
--- a/tests/frame/join_test.py
+++ b/tests/frame/join_test.py
@@ -37,8 +37,8 @@ def test_inner_join_two_keys(constructor: Constructor) -> None:
         "antananarivo": [1, 3, 2],
         "bob": [4, 4, 6],
         "zorro": [7.0, 8, 9],
-        "zorro_right": [7.0, 8, 9],
         "index": [0, 1, 2],
+        "zorro_right": [7.0, 8, 9],
     }
     assert_equal_data(result, expected)
     assert_equal_data(result_on, expected)
@@ -65,10 +65,10 @@ def test_inner_join_single_key(constructor: Constructor) -> None:
     expected = {
         "antananarivo": [1, 3, 2],
         "bob": [4, 4, 6],
-        "bob_right": [4, 4, 6],
         "zorro": [7.0, 8, 9],
-        "zorro_right": [7.0, 8, 9],
         "index": [0, 1, 2],
+        "bob_right": [4, 4, 6],
+        "zorro_right": [7.0, 8, 9],
     }
     assert_equal_data(result, expected)
     assert_equal_data(result_on, expected)
@@ -241,8 +241,8 @@ def test_left_join(constructor: Constructor) -> None:
     expected = {
         "antananarivo": [1, 2, 3],
         "bob": [4, 5, 6],
-        "antananarivo_right": [1, 2, None],
         "index": [0, 1, 2],
+        "antananarivo_right": [1, 2, None],
     }
     result_on_list = df_left.join(
         df_right,  # type: ignore[arg-type]
@@ -300,9 +300,9 @@ def test_left_join_overlapping_column(constructor: Constructor) -> None:
         "antananarivo": [1, 2, 3],
         "bob": [4, 5, 6],
         "d": [1, 4, 2],
+        "index": [0, 1, 2],
         "antananarivo_right": [1, 2, 3],
         "d_right": [1, 4, 2],
-        "index": [0, 1, 2],
     }
     assert_equal_data(result, expected)
     result = df_left.join(
@@ -317,9 +317,9 @@ def test_left_join_overlapping_column(constructor: Constructor) -> None:
         "antananarivo": [1, 2, 3],
         "bob": [4, 5, 6],
         "d": [1, 4, 2],
+        "index": [0, 1, 2],
         "antananarivo_right": [1.0, 3.0, None],
         "c": [4.0, 6.0, None],
-        "index": [0, 1, 2],
     }
     assert_equal_data(result, expected)
 
diff --git a/tests/frame/pivot_test.py b/tests/frame/pivot_test.py
index 0e386029..9bb5380f 100644
--- a/tests/frame/pivot_test.py
+++ b/tests/frame/pivot_test.py
@@ -137,6 +137,7 @@ def test_pivot(
         index=index,
         values=["foo", "bar"],
         aggregate_function=agg_func,  # type: ignore[arg-type]
+        sort_columns=True,
     )
 
     assert_equal_data(result, expected)
diff --git a/tests/frame/with_row_index_test.py b/tests/frame/with_row_index_test.py
index b6ad9c82..e19d3c99 100644
--- a/tests/frame/with_row_index_test.py
+++ b/tests/frame/with_row_index_test.py
@@ -12,5 +12,5 @@ data = {
 
 def test_with_row_index(constructor: Constructor) -> None:
     result = nw.from_native(constructor(data)).with_row_index()
-    expected = {"a": ["foo", "bars"], "ab": ["foo", "bars"], "index": [0, 1]}
+    expected = {"index": [0, 1], "a": ["foo", "bars"], "ab": ["foo", "bars"]}
     assert_equal_data(result, expected)
diff --git a/tests/group_by_test.py b/tests/group_by_test.py
index 3c57ce02..22c3b6f1 100644
--- a/tests/group_by_test.py
+++ b/tests/group_by_test.py
@@ -137,12 +137,7 @@ def test_group_by_depth_1_std_var(
     constructor: Constructor,
     attr: str,
     ddof: int,
-    request: pytest.FixtureRequest,
 ) -> None:
-    if "dask" in str(constructor):
-        # Complex aggregation for dask
-        request.applymarker(pytest.mark.xfail)
-
     data = {"a": [1, 1, 1, 2, 2, 2], "b": [4, 5, 6, 0, 5, 5]}
     _pow = 0.5 if attr == "std" else 1
     expected = {
diff --git a/tests/spark_like_test.py b/tests/spark_like_test.py
index 27777139..b597ecd6 100644
--- a/tests/spark_like_test.py
+++ b/tests/spark_like_test.py
@@ -359,10 +359,10 @@ def test_double_alias(pyspark_constructor: Constructor) -> None:
     df = nw.from_native(pyspark_constructor(data))
     result = df.with_columns(nw.col("a").alias("o"), nw.all() * 2)
     expected = {
-        "o": [1, 3, 2],
         "a": [2, 6, 4],
         "b": [8, 8, 12],
         "z": [14.0, 16.0, 18.0],
+        "o": [1, 3, 2],
     }
     assert_equal_data(result, expected)
 
@@ -643,15 +643,19 @@ def test_inner_join_two_keys(pyspark_constructor: Constructor) -> None:
         right_on=["antananarivo", "bob"],
         how="inner",
     )
-    result_on = df.join(df_right, on=["antananarivo", "bob"], how="inner")  # type: ignore[arg-type]
     result = result.sort("idx").drop("idx_right")
+
+    df = nw.from_native(pyspark_constructor(data))
+    df_right = nw.from_native(pyspark_constructor(data))
+
+    result_on = df.join(df_right, on=["antananarivo", "bob"], how="inner")  # type: ignore[arg-type]
     result_on = result_on.sort("idx").drop("idx_right")
     expected = {
         "antananarivo": [1, 3, 2],
         "bob": [4, 4, 6],
         "zorro": [7.0, 8, 9],
-        "zorro_right": [7.0, 8, 9],
         "idx": [0, 1, 2],
+        "zorro_right": [7.0, 8, 9],
     }
     assert_equal_data(result, expected)
     assert_equal_data(result_on, expected)
@@ -666,22 +670,36 @@ def test_inner_join_single_key(pyspark_constructor: Constructor) -> None:
     }
     df = nw.from_native(pyspark_constructor(data))
     df_right = nw.from_native(pyspark_constructor(data))
-    result = df.join(
-        df_right,  # type: ignore[arg-type]
-        left_on="antananarivo",
-        right_on="antananarivo",
-        how="inner",
-    ).sort("idx")
-    result_on = df.join(df_right, on="antananarivo", how="inner").sort("idx")  # type: ignore[arg-type]
-    result = result.drop("idx_right")
-    result_on = result_on.drop("idx_right")
+    result = (
+        df.join(
+            df_right,  # type: ignore[arg-type]
+            left_on="antananarivo",
+            right_on="antananarivo",
+            how="inner",
+        )
+        .sort("idx")
+        .drop("idx_right")
+    )
+
+    df = nw.from_native(pyspark_constructor(data))
+    df_right = nw.from_native(pyspark_constructor(data))
+    result_on = (
+        df.join(
+            df_right,  # type: ignore[arg-type]
+            on="antananarivo",
+            how="inner",
+        )
+        .sort("idx")
+        .drop("idx_right")
+    )
+
     expected = {
         "antananarivo": [1, 3, 2],
         "bob": [4, 4, 6],
-        "bob_right": [4, 4, 6],
         "zorro": [7.0, 8, 9],
-        "zorro_right": [7.0, 8, 9],
         "idx": [0, 1, 2],
+        "bob_right": [4, 4, 6],
+        "zorro_right": [7.0, 8, 9],
     }
     assert_equal_data(result, expected)
     assert_equal_data(result_on, expected)
@@ -831,9 +849,13 @@ def test_left_join(pyspark_constructor: Constructor) -> None:
     expected = {
         "antananarivo": [1, 2, 3],
         "bob": [4, 5, 6],
-        "antananarivo_right": [1, 2, None],
         "idx": [0, 1, 2],
+        "antananarivo_right": [1, 2, None],
     }
+    assert_equal_data(result, expected)
+
+    df_left = nw.from_native(pyspark_constructor(data_left))
+    df_right = nw.from_native(pyspark_constructor(data_right))
     result_on_list = df_left.join(
         df_right,  # type: ignore[arg-type]
         on=["antananarivo", "idx"],
@@ -846,7 +868,6 @@ def test_left_join(pyspark_constructor: Constructor) -> None:
         "idx": [0, 1, 2],
         "co": [4, 5, 7],
     }
-    assert_equal_data(result, expected)
     assert_equal_data(result_on_list, expected_on_list)
 
 
@@ -892,12 +913,14 @@ def test_left_join_overlapping_column(pyspark_constructor: Constructor) -> None:
         "antananarivo": [1, 2, 3],
         "bob": [4, 5, 6],
         "d": [1, 4, 2],
+        "idx": [0, 1, 2],
         "antananarivo_right": [1, 2, 3],
         "d_right": [1, 4, 2],
-        "idx": [0, 1, 2],
     }
     assert_equal_data(result, expected)
 
+    df_left = nw.from_native(pyspark_constructor(data_left))
+    df_right = nw.from_native(pyspark_constructor(data_right))
     result = (
         df_left.join(
             df_right,  # type: ignore[arg-type]
@@ -912,8 +935,8 @@ def test_left_join_overlapping_column(pyspark_constructor: Constructor) -> None:
         "antananarivo": [1, 2, 3],
         "bob": [4, 5, 6],
         "d": [1, 4, 2],
+        "idx": [0, 1, 2],
         "antananarivo_right": [1.0, 3.0, None],
         "c": [4.0, 6.0, None],
-        "idx": [0, 1, 2],
     }
     assert_equal_data(result, expected)
diff --git a/tests/utils.py b/tests/utils.py
index e7c9c7d8..34f1bfa1 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -82,8 +82,8 @@ def assert_equal_data(result: Any, expected: dict[str, Any]) -> None:
             result = result.collect()
 
     if hasattr(result, "columns"):
-        for key in result.columns:
-            assert key in expected, (key, expected)
+        for idx, (col, key) in enumerate(zip(result.columns, expected.keys())):
+            assert col == key, f"Expected column name {key} at index {idx}, found {col}"
     result = {key: _to_comparable_list(result[key]) for key in expected}
     if is_pyspark and expected:  # pragma: no cover
         sort_key = next(iter(expected.keys()))