Repository: narwhals-dev/narwhals
Pull request #1808 (2025-01-14 13:48:25)

feat: add `.over` method for `SparkLikeExpr`
## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [x] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #1714 

## Checklist

- [x] Code follows style guide (ruff)
- [x] Tests added
- [x] Documented the changes

## If you have comments or can explain your changes, please do so below



Patch:
diff --git a/narwhals/_arrow/dataframe.py b/narwhals/_arrow/dataframe.py
index ed738647..e44ba29a 100644
--- a/narwhals/_arrow/dataframe.py
+++ b/narwhals/_arrow/dataframe.py
@@ -312,7 +312,11 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
             col_name = col_value.name
 
             column = validate_dataframe_comparand(
-                length=length, other=col_value, backend_version=self._backend_version
+                length=length,
+                other=col_value,
+                backend_version=self._backend_version,
+                allow_broadcast=True,
+                method_name="with_columns",
             )
 
             native_frame = (
@@ -395,10 +399,9 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
         )
         return self._from_native_frame(self._native_frame.drop(to_drop))
 
-    def drop_nulls(self: Self, subset: str | list[str] | None) -> Self:
+    def drop_nulls(self: Self, subset: list[str] | None) -> Self:
         if subset is None:
             return self._from_native_frame(self._native_frame.drop_null())
-        subset = [subset] if isinstance(subset, str) else subset
         plx = self.__narwhals_namespace__()
         return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
 
@@ -483,7 +486,7 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
             and all(isinstance(x, bool) for x in predicates[0])
             and not constraints
         ):
-            mask = predicates[0]
+            mask_native = predicates[0]
         else:
             plx = self.__narwhals_namespace__()
             expr = plx.all_horizontal(
@@ -492,8 +495,15 @@ class ArrowDataFrame(CompliantDataFrame, CompliantLazyFrame):
                 )
             )
             # `[0]` is safe as all_horizontal's expression only returns a single column
-            mask = expr._call(self)[0]._native_series
-        return self._from_native_frame(self._native_frame.filter(mask))
+            mask = expr._call(self)[0]
+            mask_native = validate_dataframe_comparand(
+                length=len(self),
+                other=mask,
+                backend_version=self._backend_version,
+                allow_broadcast=False,
+                method_name="filter",
+            )
+        return self._from_native_frame(self._native_frame.filter(mask_native))
 
     def null_count(self: Self) -> Self:
         import pyarrow as pa
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index d0bc49b0..8ab13a75 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -15,6 +15,7 @@ from narwhals._arrow.series import ArrowSeries
 from narwhals._expression_parsing import reuse_series_implementation
 from narwhals.dependencies import get_numpy
 from narwhals.dependencies import is_numpy_array
+from narwhals.exceptions import AnonymousExprError
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
@@ -390,12 +391,9 @@ class ArrowExpr(CompliantExpr[ArrowSeries]):
     def over(self: Self, keys: list[str]) -> Self:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             if self._output_names is None:
-                msg = (
-                    "Anonymous expressions are not supported in over.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
-                )
-                raise ValueError(msg)
+                msg = ".over"
+                raise AnonymousExprError.from_expr_name(msg)
+
             tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
             tmp = df.select(*keys).join(
                 tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
diff --git a/narwhals/_arrow/expr_name.py b/narwhals/_arrow/expr_name.py
index 7a2fbede..4c4991bf 100644
--- a/narwhals/_arrow/expr_name.py
+++ b/narwhals/_arrow/expr_name.py
@@ -3,6 +3,8 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 from typing import Callable
 
+from narwhals.exceptions import AnonymousExprError
+
 if TYPE_CHECKING:
     from typing_extensions import Self
 
@@ -17,12 +19,8 @@ class ArrowExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.keep`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.keep"
+            raise AnonymousExprError.from_expr_name(msg)
 
         return self._compliant_expr.__class__(
             lambda df: [
@@ -42,12 +40,8 @@ class ArrowExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.map`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.map"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [function(str(name)) for name in root_names]
 
@@ -68,12 +62,8 @@ class ArrowExprNameNamespace:
     def prefix(self: Self, prefix: str) -> ArrowExpr:
         root_names = self._compliant_expr._root_names
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.prefix`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.prefix"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [prefix + str(name) for name in root_names]
         return self._compliant_expr.__class__(
@@ -93,12 +83,8 @@ class ArrowExprNameNamespace:
     def suffix(self: Self, suffix: str) -> ArrowExpr:
         root_names = self._compliant_expr._root_names
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.suffix`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.suffix"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [str(name) + suffix for name in root_names]
 
@@ -120,12 +106,8 @@ class ArrowExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.to_lowercase`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.to_lowercase"
+            raise AnonymousExprError.from_expr_name(msg)
         output_names = [str(name).lower() for name in root_names]
 
         return self._compliant_expr.__class__(
@@ -146,12 +128,8 @@ class ArrowExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.to_uppercase`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.to_uppercase"
+            raise AnonymousExprError.from_expr_name(msg)
         output_names = [str(name).upper() for name in root_names]
 
         return self._compliant_expr.__class__(
diff --git a/narwhals/_arrow/group_by.py b/narwhals/_arrow/group_by.py
index 11ed914f..aac1decb 100644
--- a/narwhals/_arrow/group_by.py
+++ b/narwhals/_arrow/group_by.py
@@ -9,6 +9,7 @@ from typing import Sequence
 
 from narwhals._expression_parsing import is_simple_aggregation
 from narwhals._expression_parsing import parse_into_exprs
+from narwhals.exceptions import AnonymousExprError
 from narwhals.utils import generate_temporary_column_name
 from narwhals.utils import remove_prefix
 
@@ -61,12 +62,8 @@ class ArrowGroupBy:
         )
         for expr in exprs:
             if expr._output_names is None:
-                msg = (
-                    "Anonymous expressions are not supported in group_by.agg.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
-                )
-                raise ValueError(msg)
+                msg = "group_by.agg"
+                raise AnonymousExprError.from_expr_name(msg)
 
         return agg_arrow(
             self._grouped,
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index b02ad32e..70d372a5 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -435,19 +435,18 @@ class ArrowWhen:
 
         plx = df.__narwhals_namespace__()
         condition = parse_into_expr(self._condition, namespace=plx)(df)[0]
+
         try:
             value_series = parse_into_expr(self._then_value, namespace=plx)(df)[0]
         except TypeError:
-            # `self._otherwise_value` is a scalar and can't be converted to an expression
-            value_series = condition.__class__._from_iterable(
-                pa.repeat(pa.scalar(self._then_value), len(condition)),
-                name="literal",
-                backend_version=self._backend_version,
-                version=self._version,
+            # `self._then_value` is a scalar and can't be converted to an expression
+            value_series = plx._create_series_from_scalar(
+                self._then_value, reference_series=condition
             )
 
-        value_series_native = value_series._native_series
-        condition_native = condition._native_series
+        condition_native, value_series_native = broadcast_series(
+            [condition, value_series]
+        )
 
         if self._otherwise_value is None:
             otherwise_native = pa.repeat(
@@ -472,9 +471,7 @@ class ArrowWhen:
             ]
         else:
             otherwise_series = otherwise_expr(df)[0]
-            condition_native, otherwise_native = broadcast_series(
-                [condition, otherwise_series]
-            )
+            _, otherwise_native = broadcast_series([condition, otherwise_series])
             return [
                 value_series._from_native_series(
                     pc.if_else(condition_native, value_series_native, otherwise_native)
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index ca485265..7f642b94 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -6,6 +6,7 @@ from typing import Any
 from typing import Sequence
 from typing import overload
 
+from narwhals.exceptions import ShapeError
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
 
@@ -212,7 +213,12 @@ def broadcast_and_extract_native(
 
 
 def validate_dataframe_comparand(
-    length: int, other: Any, backend_version: tuple[int, ...]
+    length: int,
+    other: Any,
+    backend_version: tuple[int, ...],
+    *,
+    allow_broadcast: bool,
+    method_name: str,
 ) -> Any:
     """Validate RHS of binary operation.
 
@@ -222,7 +228,14 @@ def validate_dataframe_comparand(
     from narwhals._arrow.series import ArrowSeries
 
     if isinstance(other, ArrowSeries):
-        if len(other) == 1:
+        len_other = len(other)
+        if len_other == 1:
+            if length > 1 and not allow_broadcast:
+                msg = (
+                    f"{method_name}'s length: 1 differs from that of the series: {length}"
+                )
+                raise ShapeError(msg)
+
             import numpy as np  # ignore-banned-import
             import pyarrow as pa
 
@@ -230,6 +243,11 @@ def validate_dataframe_comparand(
             if backend_version < (13,) and hasattr(value, "as_py"):
                 value = value.as_py()
             return pa.array(np.full(shape=length, fill_value=value))
+
+        if length != len_other:
+            msg = f"{method_name}'s length: {len_other} differs from that of the series: {length}"
+            raise ShapeError(msg)
+
         return other._native_series
 
     from narwhals._arrow.dataframe import ArrowDataFrame  # pragma: no cover
diff --git a/narwhals/_dask/dataframe.py b/narwhals/_dask/dataframe.py
index 89e1bc48..92c1f198 100644
--- a/narwhals/_dask/dataframe.py
+++ b/narwhals/_dask/dataframe.py
@@ -96,13 +96,14 @@ class DaskLazyFrame(CompliantLazyFrame):
     def columns(self) -> list[str]:
         return self._native_frame.columns.tolist()  # type: ignore[no-any-return]
 
-    def filter(self, *predicates: DaskExpr, **constraints: Any) -> Self:
+    def filter(self: Self, *predicates: DaskExpr, **constraints: Any) -> Self:
         plx = self.__narwhals_namespace__()
         expr = plx.all_horizontal(
             *chain(predicates, (plx.col(name) == v for name, v in constraints.items()))
         )
         # `[0]` is safe as all_horizontal's expression only returns a single column
         mask = expr._call(self)[0]
+
         return self._from_native_frame(self._native_frame.loc[mask])
 
     def select(
@@ -149,10 +150,9 @@ class DaskLazyFrame(CompliantLazyFrame):
         )
         return self._from_native_frame(df)
 
-    def drop_nulls(self: Self, subset: str | list[str] | None) -> Self:
+    def drop_nulls(self: Self, subset: list[str] | None) -> Self:
         if subset is None:
             return self._from_native_frame(self._native_frame.dropna())
-        subset = [subset] if isinstance(subset, str) else subset
         plx = self.__narwhals_namespace__()
         return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
 
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 951222c3..09a3adf8 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -16,6 +16,7 @@ from narwhals._dask.utils import maybe_evaluate
 from narwhals._dask.utils import narwhals_to_native_dtype
 from narwhals._expression_parsing import infer_new_root_output_names
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
+from narwhals.exceptions import AnonymousExprError
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.exceptions import InvalidOperationError
 from narwhals.typing import CompliantExpr
@@ -359,7 +360,7 @@ class DaskExpr(CompliantExpr["dx.Series"]):
         )
 
     def cum_sum(self: Self, *, reverse: bool) -> Self:
-        if reverse:
+        if reverse:  # pragma: no cover
             msg = "`cum_sum(reverse=True)` is not supported with Dask backend"
             raise NotImplementedError(msg)
 
@@ -370,7 +371,7 @@ class DaskExpr(CompliantExpr["dx.Series"]):
         )
 
     def cum_count(self: Self, *, reverse: bool) -> Self:
-        if reverse:
+        if reverse:  # pragma: no cover
             msg = "`cum_count(reverse=True)` is not supported with Dask backend"
             raise NotImplementedError(msg)
 
@@ -381,7 +382,7 @@ class DaskExpr(CompliantExpr["dx.Series"]):
         )
 
     def cum_min(self: Self, *, reverse: bool) -> Self:
-        if reverse:
+        if reverse:  # pragma: no cover
             msg = "`cum_min(reverse=True)` is not supported with Dask backend"
             raise NotImplementedError(msg)
 
@@ -392,7 +393,7 @@ class DaskExpr(CompliantExpr["dx.Series"]):
         )
 
     def cum_max(self: Self, *, reverse: bool) -> Self:
-        if reverse:
+        if reverse:  # pragma: no cover
             msg = "`cum_max(reverse=True)` is not supported with Dask backend"
             raise NotImplementedError(msg)
 
@@ -403,7 +404,7 @@ class DaskExpr(CompliantExpr["dx.Series"]):
         )
 
     def cum_prod(self: Self, *, reverse: bool) -> Self:
-        if reverse:
+        if reverse:  # pragma: no cover
             msg = "`cum_prod(reverse=True)` is not supported with Dask backend"
             raise NotImplementedError(msg)
 
@@ -690,12 +691,8 @@ class DaskExpr(CompliantExpr["dx.Series"]):
     def over(self: Self, keys: list[str]) -> Self:
         def func(df: DaskLazyFrame) -> list[Any]:
             if self._output_names is None:
-                msg = (
-                    "Anonymous expressions are not supported in over.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
-                )
-                raise ValueError(msg)
+                msg = "over"
+                raise AnonymousExprError.from_expr_name(msg)
 
             if df._native_frame.npartitions == 1:  # pragma: no cover
                 tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
diff --git a/narwhals/_dask/expr_name.py b/narwhals/_dask/expr_name.py
index 3f5a0e4a..bbed6add 100644
--- a/narwhals/_dask/expr_name.py
+++ b/narwhals/_dask/expr_name.py
@@ -3,6 +3,8 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 from typing import Callable
 
+from narwhals.exceptions import AnonymousExprError
+
 if TYPE_CHECKING:
     from typing_extensions import Self
 
@@ -17,12 +19,8 @@ class DaskExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.keep`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.keep"
+            raise AnonymousExprError.from_expr_name(msg)
 
         return self._compliant_expr.__class__(
             lambda df: [
@@ -43,12 +41,8 @@ class DaskExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.map`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.map"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [function(str(name)) for name in root_names]
 
@@ -70,12 +64,8 @@ class DaskExprNameNamespace:
     def prefix(self: Self, prefix: str) -> DaskExpr:
         root_names = self._compliant_expr._root_names
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.prefix`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.prefix"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [prefix + str(name) for name in root_names]
         return self._compliant_expr.__class__(
@@ -96,12 +86,8 @@ class DaskExprNameNamespace:
     def suffix(self: Self, suffix: str) -> DaskExpr:
         root_names = self._compliant_expr._root_names
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.suffix`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.suffix"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [str(name) + suffix for name in root_names]
 
@@ -124,12 +110,8 @@ class DaskExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.to_lowercase`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.to_lowercase"
+            raise AnonymousExprError.from_expr_name(msg)
         output_names = [str(name).lower() for name in root_names]
 
         return self._compliant_expr.__class__(
@@ -151,12 +133,8 @@ class DaskExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.to_uppercase`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.to_uppercase"
+            raise AnonymousExprError.from_expr_name(msg)
         output_names = [str(name).upper() for name in root_names]
 
         return self._compliant_expr.__class__(
diff --git a/narwhals/_dask/group_by.py b/narwhals/_dask/group_by.py
index 60086efa..b92cf471 100644
--- a/narwhals/_dask/group_by.py
+++ b/narwhals/_dask/group_by.py
@@ -8,6 +8,7 @@ from typing import Sequence
 
 from narwhals._expression_parsing import is_simple_aggregation
 from narwhals._expression_parsing import parse_into_exprs
+from narwhals.exceptions import AnonymousExprError
 from narwhals.utils import remove_prefix
 
 if TYPE_CHECKING:
@@ -50,7 +51,7 @@ def var(
 
     try:
         import dask.dataframe.dask_expr as dx
-    except ModuleNotFoundError:
+    except ModuleNotFoundError:  # pragma: no cover
         import dask_expr as dx
 
     return partial(dx._groupby.GroupBy.var, ddof=ddof)
@@ -65,7 +66,7 @@ def std(
 
     try:
         import dask.dataframe.dask_expr as dx
-    except ModuleNotFoundError:
+    except ModuleNotFoundError:  # pragma: no cover
         import dask_expr as dx
 
     return partial(dx._groupby.GroupBy.std, ddof=ddof)
@@ -110,12 +111,8 @@ class DaskLazyGroupBy:
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
             if expr._output_names is None:
-                msg = (
-                    "Anonymous expressions are not supported in group_by.agg.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
-                )
-                raise ValueError(msg)
+                msg = "group_by.agg"
+                raise AnonymousExprError.from_expr_name(msg)
 
             output_names.extend(expr._output_names)
 
diff --git a/narwhals/_dask/namespace.py b/narwhals/_dask/namespace.py
index 23805afd..936b9abb 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -398,13 +398,24 @@ class DaskWhen:
         plx = df.__narwhals_namespace__()
         condition = parse_into_expr(self._condition, namespace=plx)(df)[0]
         condition = cast("dx.Series", condition)
+
         try:
-            value_series = parse_into_expr(self._then_value, namespace=plx)(df)[0]
+            then_expr = parse_into_expr(self._then_value, namespace=plx)
         except TypeError:
-            # `self._otherwise_value` is a scalar and can't be converted to an expression
+            # `self._then_value` is a scalar and can't be converted to an expression
+            value_sequence: Sequence[Any] = [self._then_value]
+            is_scalar = True
+        else:
+            is_scalar = then_expr._returns_scalar  # type: ignore[attr-defined]
+            value_sequence = then_expr(df)[0]
+
+        if is_scalar:
             _df = condition.to_frame("a")
-            _df["tmp"] = self._then_value
+            _df["tmp"] = value_sequence[0]
             value_series = _df["tmp"]
+        else:
+            value_series = value_sequence
+
         value_series = cast("dx.Series", value_series)
         validate_comparand(condition, value_series)
 
diff --git a/narwhals/_dask/utils.py b/narwhals/_dask/utils.py
index cd303d8e..d1438386 100644
--- a/narwhals/_dask/utils.py
+++ b/narwhals/_dask/utils.py
@@ -89,7 +89,7 @@ def add_row_index(
 def validate_comparand(lhs: dx.Series, rhs: dx.Series) -> None:
     try:
         import dask.dataframe.dask_expr as dx
-    except ModuleNotFoundError:
+    except ModuleNotFoundError:  # pragma: no cover
         import dask_expr as dx
 
     if not dx._expr.are_co_aligned(lhs._expr, rhs._expr):  # pragma: no cover
diff --git a/narwhals/_duckdb/dataframe.py b/narwhals/_duckdb/dataframe.py
index 98eca0bd..0bd2c9f0 100644
--- a/narwhals/_duckdb/dataframe.py
+++ b/narwhals/_duckdb/dataframe.py
@@ -147,7 +147,7 @@ class DuckDBLazyFrame:
             result.append(value.alias(col))
         return self._from_native_frame(self._native_frame.select(*result))
 
-    def filter(self, *predicates: DuckDBExpr, **constraints: Any) -> Self:
+    def filter(self: Self, *predicates: DuckDBExpr, **constraints: Any) -> Self:
         plx = self.__narwhals_namespace__()
         expr = plx.all_horizontal(
             *chain(predicates, (plx.col(name) == v for name, v in constraints.items()))
@@ -321,3 +321,12 @@ class DuckDBLazyFrame:
             )
         )
         return self._from_native_frame(result)
+
+    def drop_nulls(self: Self, subset: list[str] | None) -> Self:
+        import duckdb
+
+        rel = self._native_frame
+        subset_ = subset if subset is not None else rel.columns
+        keep_condition = " and ".join(f'"{col}" is not null' for col in subset_)
+        query = f"select * from rel where {keep_condition}"  # noqa: S608
+        return self._from_native_frame(duckdb.sql(query))
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 0cfddbaa..c79acdfa 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -7,6 +7,7 @@ from typing import Literal
 from typing import Sequence
 
 from narwhals._duckdb.expr_dt import DuckDBExprDateTimeNamespace
+from narwhals._duckdb.expr_name import DuckDBExprNameNamespace
 from narwhals._duckdb.expr_str import DuckDBExprStringNamespace
 from narwhals._duckdb.utils import binary_operation_returns_scalar
 from narwhals._duckdb.utils import get_column_name
@@ -571,3 +572,7 @@ class DuckDBExpr(CompliantExpr["duckdb.Expression"]):
     @property
     def dt(self: Self) -> DuckDBExprDateTimeNamespace:
         return DuckDBExprDateTimeNamespace(self)
+
+    @property
+    def name(self: Self) -> DuckDBExprNameNamespace:
+        return DuckDBExprNameNamespace(self)
diff --git a/narwhals/_duckdb/expr_name.py b/narwhals/_duckdb/expr_name.py
new file mode 100644
index 00000000..2ed2b2ea
--- /dev/null
+++ b/narwhals/_duckdb/expr_name.py
@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from typing import Callable
+
+from narwhals.exceptions import AnonymousExprError
+
+if TYPE_CHECKING:
+    from typing_extensions import Self
+
+    from narwhals._duckdb.expr import DuckDBExpr
+
+
+class DuckDBExprNameNamespace:
+    def __init__(self: Self, expr: DuckDBExpr) -> None:
+        self._compliant_expr = expr
+
+    def keep(self: Self) -> DuckDBExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.keep"
+            raise AnonymousExprError.from_expr_name(msg)
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), root_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=root_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs=self._compliant_expr._kwargs,
+        )
+
+    def map(self: Self, function: Callable[[str], str]) -> DuckDBExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.map"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        output_names = [function(str(name)) for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs={**self._compliant_expr._kwargs, "function": function},
+        )
+
+    def prefix(self: Self, prefix: str) -> DuckDBExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.prefix"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        output_names = [prefix + str(name) for name in root_names]
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs={**self._compliant_expr._kwargs, "prefix": prefix},
+        )
+
+    def suffix(self: Self, suffix: str) -> DuckDBExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.suffix"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        output_names = [str(name) + suffix for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs={**self._compliant_expr._kwargs, "suffix": suffix},
+        )
+
+    def to_lowercase(self: Self) -> DuckDBExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.to_lowercase"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        output_names = [str(name).lower() for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs=self._compliant_expr._kwargs,
+        )
+
+    def to_uppercase(self: Self) -> DuckDBExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.to_uppercase"
+            raise AnonymousExprError.from_expr_name(msg)
+        output_names = [str(name).upper() for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs=self._compliant_expr._kwargs,
+        )
diff --git a/narwhals/_duckdb/group_by.py b/narwhals/_duckdb/group_by.py
index 0b312ff0..ce81a157 100644
--- a/narwhals/_duckdb/group_by.py
+++ b/narwhals/_duckdb/group_by.py
@@ -4,6 +4,7 @@ from copy import copy
 from typing import TYPE_CHECKING
 
 from narwhals._expression_parsing import parse_into_exprs
+from narwhals.exceptions import AnonymousExprError
 
 if TYPE_CHECKING:
     from narwhals._duckdb.dataframe import DuckDBLazyFrame
@@ -33,12 +34,8 @@ class DuckDBGroupBy:
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
             if expr._output_names is None:  # pragma: no cover
-                msg = (
-                    "Anonymous expressions are not supported in group_by.agg.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
-                )
-                raise ValueError(msg)
+                msg = "group_by.agg"
+                raise AnonymousExprError.from_expr_name(msg)
 
             output_names.extend(expr._output_names)
 
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index ac7e645f..dc52da00 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -27,6 +27,7 @@ if TYPE_CHECKING:
     from narwhals.typing import CompliantNamespace
     from narwhals.typing import CompliantSeries
     from narwhals.typing import CompliantSeriesT_co
+    from narwhals.typing import IntoExpr
 
     IntoCompliantExpr: TypeAlias = (
         CompliantExpr[CompliantSeriesT_co] | str | CompliantSeriesT_co
@@ -334,3 +335,10 @@ def extract_compliant(
     if isinstance(other, Series):
         return other._compliant_series
     return other
+
+
+def operation_is_order_dependent(*args: IntoExpr | Any) -> bool:
+    # If an arg is an Expr, we look at `_is_order_dependent`. If it isn't,
+    # it means that it was a scalar (e.g. nw.col('a') + 1) or a column name,
+    # neither of which is order-dependent, so we default to `False`.
+    return any(getattr(x, "_is_order_dependent", False) for x in args)
diff --git a/narwhals/_pandas_like/dataframe.py b/narwhals/_pandas_like/dataframe.py
index 47c43a69..05802b6a 100644
--- a/narwhals/_pandas_like/dataframe.py
+++ b/narwhals/_pandas_like/dataframe.py
@@ -371,10 +371,9 @@ class PandasLikeDataFrame:
         )
         return self._from_native_frame(df)
 
-    def drop_nulls(self, subset: str | list[str] | None) -> Self:
+    def drop_nulls(self, subset: list[str] | None) -> Self:
         if subset is None:
             return self._from_native_frame(self._native_frame.dropna(axis=0))
-        subset = [subset] if isinstance(subset, str) else subset
         plx = self.__narwhals_namespace__()
         return self.filter(~plx.any_horizontal(plx.col(*subset).is_null()))
 
@@ -401,7 +400,7 @@ class PandasLikeDataFrame:
     def row(self, row: int) -> tuple[Any, ...]:
         return tuple(x for x in self._native_frame.iloc[row])
 
-    def filter(self, *predicates: IntoPandasLikeExpr, **constraints: Any) -> Self:
+    def filter(self: Self, *predicates: IntoPandasLikeExpr, **constraints: Any) -> Self:
         plx = self.__narwhals_namespace__()
         if (
             len(predicates) == 1
@@ -409,7 +408,7 @@ class PandasLikeDataFrame:
             and all(isinstance(x, bool) for x in predicates[0])
             and not constraints
         ):
-            _mask = predicates[0]
+            mask_native = predicates[0]
         else:
             expr = plx.all_horizontal(
                 *chain(
@@ -418,8 +417,14 @@ class PandasLikeDataFrame:
             )
             # `[0]` is safe as all_horizontal's expression only returns a single column
             mask = expr._call(self)[0]
-            _mask = validate_dataframe_comparand(self._native_frame.index, mask)
-        return self._from_native_frame(self._native_frame.loc[_mask])
+            mask_native = validate_dataframe_comparand(
+                self._native_frame.index,
+                mask,
+                allow_broadcast=False,
+                method_name="filter",
+            )
+
+        return self._from_native_frame(self._native_frame.loc[mask_native])
 
     def with_columns(
         self,
@@ -438,13 +443,21 @@ class PandasLikeDataFrame:
             if name in new_column_name_to_new_column_map:
                 to_concat.append(
                     validate_dataframe_comparand(
-                        index, new_column_name_to_new_column_map.pop(name)
+                        index,
+                        new_column_name_to_new_column_map.pop(name),
+                        allow_broadcast=True,
+                        method_name="with_columns",
                     )
                 )
             else:
                 to_concat.append(self._native_frame[name])
         to_concat.extend(
-            validate_dataframe_comparand(index, new_column_name_to_new_column_map[s])
+            validate_dataframe_comparand(
+                index,
+                new_column_name_to_new_column_map[s],
+                allow_broadcast=True,
+                method_name="with_columns",
+            )
             for s in new_column_name_to_new_column_map
         )
 
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index 3519674d..6eceaaac 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -16,6 +16,7 @@ from narwhals._pandas_like.series import PandasLikeSeries
 from narwhals._pandas_like.utils import rename
 from narwhals.dependencies import get_numpy
 from narwhals.dependencies import is_numpy_array
+from narwhals.exceptions import AnonymousExprError
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.typing import CompliantExpr
 
@@ -399,12 +400,8 @@ class PandasLikeExpr(CompliantExpr[PandasLikeSeries]):
                     self._output_names is None or self._root_names is None
                 ):  # pragma: no cover
                     # Technically unreachable, but keep this for safety
-                    msg = (
-                        "Anonymous expressions are not supported in over.\n"
-                        "Instead of `nw.all()`, try using a named expression, such as "
-                        "`nw.col('a', 'b')`\n"
-                    )
-                    raise ValueError(msg)
+                    msg = "over"
+                    raise AnonymousExprError.from_expr_name(msg)
 
                 reverse = self._kwargs.get("reverse", False)
                 if reverse:
@@ -452,12 +449,8 @@ class PandasLikeExpr(CompliantExpr[PandasLikeSeries]):
 
             def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
                 if self._output_names is None:
-                    msg = (
-                        "Anonymous expressions are not supported in over.\n"
-                        "Instead of `nw.all()`, try using a named expression, such as "
-                        "`nw.col('a', 'b')`\n"
-                    )
-                    raise ValueError(msg)
+                    msg = "over"
+                    raise AnonymousExprError.from_expr_name(msg)
                 tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
                 tmp = df.select(*keys).join(
                     tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
diff --git a/narwhals/_pandas_like/expr_name.py b/narwhals/_pandas_like/expr_name.py
index 0050e698..fb56c944 100644
--- a/narwhals/_pandas_like/expr_name.py
+++ b/narwhals/_pandas_like/expr_name.py
@@ -3,6 +3,8 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 from typing import Callable
 
+from narwhals.exceptions import AnonymousExprError
+
 if TYPE_CHECKING:
     from typing_extensions import Self
 
@@ -17,12 +19,8 @@ class PandasLikeExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.keep`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.keep"
+            raise AnonymousExprError.from_expr_name(msg)
 
         return self._compliant_expr.__class__(
             lambda df: [
@@ -43,12 +41,8 @@ class PandasLikeExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.map`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.map"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [function(str(name)) for name in root_names]
 
@@ -70,12 +64,8 @@ class PandasLikeExprNameNamespace:
     def prefix(self: Self, prefix: str) -> PandasLikeExpr:
         root_names = self._compliant_expr._root_names
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.prefix`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.prefix"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [prefix + str(name) for name in root_names]
         return self._compliant_expr.__class__(
@@ -96,12 +86,8 @@ class PandasLikeExprNameNamespace:
     def suffix(self: Self, suffix: str) -> PandasLikeExpr:
         root_names = self._compliant_expr._root_names
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.suffix`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.suffix"
+            raise AnonymousExprError.from_expr_name(msg)
 
         output_names = [str(name) + suffix for name in root_names]
 
@@ -124,12 +110,9 @@ class PandasLikeExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.to_lowercase`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.to_lowercase"
+            raise AnonymousExprError.from_expr_name(msg)
+
         output_names = [str(name).lower() for name in root_names]
 
         return self._compliant_expr.__class__(
@@ -151,12 +134,9 @@ class PandasLikeExprNameNamespace:
         root_names = self._compliant_expr._root_names
 
         if root_names is None:
-            msg = (
-                "Anonymous expressions are not supported in `.name.to_uppercase`.\n"
-                "Instead of `nw.all()`, try using a named expression, such as "
-                "`nw.col('a', 'b')`\n"
-            )
-            raise ValueError(msg)
+            msg = ".name.to_uppercase"
+            raise AnonymousExprError.from_expr_name(msg)
+
         output_names = [str(name).upper() for name in root_names]
 
         return self._compliant_expr.__class__(
diff --git a/narwhals/_pandas_like/group_by.py b/narwhals/_pandas_like/group_by.py
index a1eca5b5..51efb1d8 100644
--- a/narwhals/_pandas_like/group_by.py
+++ b/narwhals/_pandas_like/group_by.py
@@ -15,6 +15,7 @@ from narwhals._pandas_like.utils import horizontal_concat
 from narwhals._pandas_like.utils import native_series_from_iterable
 from narwhals._pandas_like.utils import select_columns_by_name
 from narwhals._pandas_like.utils import set_columns
+from narwhals.exceptions import AnonymousExprError
 from narwhals.utils import Implementation
 from narwhals.utils import find_stacklevel
 from narwhals.utils import remove_prefix
@@ -92,12 +93,9 @@ class PandasLikeGroupBy:
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
             if expr._output_names is None:
-                msg = (
-                    "Anonymous expressions are not supported in group_by.agg.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
-                )
-                raise ValueError(msg)
+                msg = "group_by.agg"
+                raise AnonymousExprError.from_expr_name(msg)
+
             output_names.extend(expr._output_names)
 
         return agg_pandas(
diff --git a/narwhals/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index 52e56d34..4265a340 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -461,20 +461,17 @@ class PandasWhen:
 
         plx = df.__narwhals_namespace__()
         condition = parse_into_expr(self._condition, namespace=plx)(df)[0]
+
         try:
             value_series = parse_into_expr(self._then_value, namespace=plx)(df)[0]
         except TypeError:
-            # `self._otherwise_value` is a scalar and can't be converted to an expression
-            value_series = condition.__class__._from_iterable(
-                [self._then_value] * len(condition),
-                name="literal",
-                index=condition._native_series.index,
-                implementation=self._implementation,
-                backend_version=self._backend_version,
-                version=self._version,
+            # `self._then_value` is a scalar and can't be converted to an expression
+            value_series = plx._create_series_from_scalar(
+                self._then_value, reference_series=condition
             )
-        value_series_native, condition_native = broadcast_align_and_extract_native(
-            value_series, condition
+
+        condition_native, value_series_native = broadcast_align_and_extract_native(
+            condition, value_series
         )
 
         if self._otherwise_value is None:
@@ -494,7 +491,14 @@ class PandasWhen:
             ]
         else:
             otherwise_series = otherwise_expr(df)[0]
-            return [value_series.zip_with(condition, otherwise_series)]
+            _, otherwise_native = broadcast_align_and_extract_native(
+                condition, otherwise_series
+            )
+            return [
+                value_series._from_native_series(
+                    value_series_native.where(condition_native, otherwise_native)
+                )
+            ]
 
     def then(self, value: PandasLikeExpr | PandasLikeSeries | Any) -> PandasThen:
         self._then_value = value
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index 358041db..95a4eae8 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -13,6 +13,7 @@ from narwhals._pandas_like.series_dt import PandasLikeSeriesDateTimeNamespace
 from narwhals._pandas_like.series_list import PandasLikeSeriesListNamespace
 from narwhals._pandas_like.series_str import PandasLikeSeriesStringNamespace
 from narwhals._pandas_like.utils import broadcast_align_and_extract_native
+from narwhals._pandas_like.utils import get_dtype_backend
 from narwhals._pandas_like.utils import narwhals_to_native_dtype
 from narwhals._pandas_like.utils import native_series_from_iterable
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
@@ -232,8 +233,15 @@ class PandasLikeSeries(CompliantSeries):
         dtype: Any,
     ) -> Self:
         ser = self._native_series
+        dtype_backend = get_dtype_backend(
+            dtype=ser.dtype, implementation=self._implementation
+        )
         dtype = narwhals_to_native_dtype(
-            dtype, ser.dtype, self._implementation, self._backend_version, self._version
+            dtype,
+            dtype_backend=dtype_backend,
+            implementation=self._implementation,
+            backend_version=self._backend_version,
+            version=self._version,
         )
         return self._from_native_series(ser.astype(dtype))
 
diff --git a/narwhals/_pandas_like/series_list.py b/narwhals/_pandas_like/series_list.py
index 1daadab1..9d3b136b 100644
--- a/narwhals/_pandas_like/series_list.py
+++ b/narwhals/_pandas_like/series_list.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
+from narwhals._pandas_like.utils import get_dtype_backend
 from narwhals._pandas_like.utils import narwhals_to_native_dtype
 from narwhals._pandas_like.utils import rename
 from narwhals._pandas_like.utils import set_index
@@ -37,10 +38,15 @@ class PandasLikeSeriesListNamespace:
                 implementation=self._compliant_series._implementation,
                 backend_version=self._compliant_series._backend_version,
             )
+
+        implementation = self._compliant_series._implementation
+        dtype_backend = get_dtype_backend(
+            dtype=native_result.dtype, implementation=implementation
+        )
         dtype = narwhals_to_native_dtype(
             dtype=import_dtypes_module(self._compliant_series._version).UInt32(),
-            starting_dtype=native_result.dtype,
-            implementation=self._compliant_series._implementation,
+            dtype_backend=dtype_backend,
+            implementation=implementation,
             backend_version=self._compliant_series._backend_version,
             version=self._compliant_series._version,
         )
diff --git a/narwhals/_pandas_like/utils.py b/narwhals/_pandas_like/utils.py
index 08d49058..159a9f1c 100644
--- a/narwhals/_pandas_like/utils.py
+++ b/narwhals/_pandas_like/utils.py
@@ -17,6 +17,7 @@ from narwhals._arrow.utils import (
     native_to_narwhals_dtype as arrow_native_to_narwhals_dtype,
 )
 from narwhals.exceptions import ColumnNotFoundError
+from narwhals.exceptions import ShapeError
 from narwhals.utils import Implementation
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
@@ -152,7 +153,9 @@ def broadcast_align_and_extract_native(
     return lhs._native_series, rhs
 
 
-def validate_dataframe_comparand(index: Any, other: Any) -> Any:
+def validate_dataframe_comparand(
+    index: Any, other: Any, *, allow_broadcast: bool, method_name: str
+) -> Any:
     """Validate RHS of binary operation.
 
     If the comparison isn't supported, return `NotImplemented` so that the
@@ -164,10 +167,27 @@ def validate_dataframe_comparand(index: Any, other: Any) -> Any:
     if isinstance(other, PandasLikeDataFrame):
         return NotImplemented
     if isinstance(other, PandasLikeSeries):
-        if other.len() == 1:
+        len_index = len(index)
+        len_other = other.len()
+
+        if len_other == 1:
+            if len_index > 1 and not allow_broadcast:
+                msg = (
+                    f"{method_name}'s length: 1 differs from that of the series: "
+                    f"{len_index}"
+                )
+                raise ShapeError(msg)
             # broadcast
             s = other._native_series
             return s.__class__(s.iloc[0], index=index, dtype=s.dtype, name=s.name)
+
+        if len_index != len_other:
+            msg = (
+                f"{method_name}'s length: {len_other} differs from that of the series: "
+                f"{len_index}"
+            )
+            raise ShapeError(msg)
+
         if other._native_series.index is not index:
             return set_index(
                 other._native_series,
@@ -512,12 +532,11 @@ def get_dtype_backend(dtype: Any, implementation: Implementation) -> str:
 
 def narwhals_to_native_dtype(  # noqa: PLR0915
     dtype: DType | type[DType],
-    starting_dtype: Any,
+    dtype_backend: str | None,
     implementation: Implementation,
     backend_version: tuple[int, ...],
     version: Version,
 ) -> Any:
-    dtype_backend = get_dtype_backend(starting_dtype, implementation)
     dtypes = import_dtypes_module(version)
     if isinstance_or_issubclass(dtype, dtypes.Float64):
         if dtype_backend == "pyarrow-nullable":
diff --git a/narwhals/_spark_like/dataframe.py b/narwhals/_spark_like/dataframe.py
index 7e7f41dd..9d302014 100644
--- a/narwhals/_spark_like/dataframe.py
+++ b/narwhals/_spark_like/dataframe.py
@@ -108,7 +108,7 @@ class SparkLikeLazyFrame:
         new_columns_list = [col.alias(col_name) for col_name, col in new_columns.items()]
         return self._from_native_frame(self._native_frame.select(*new_columns_list))
 
-    def filter(self, *predicates: SparkLikeExpr, **constraints: Any) -> Self:
+    def filter(self: Self, *predicates: SparkLikeExpr, **constraints: Any) -> Self:
         plx = self.__narwhals_namespace__()
         expr = plx.all_horizontal(
             *chain(predicates, (plx.col(name) == v for name, v in constraints.items()))
@@ -183,7 +183,7 @@ class SparkLikeLazyFrame:
         sort_cols = [sort_f(col) for col, sort_f in zip(flat_by, sort_funcs)]
         return self._from_native_frame(self._native_frame.sort(*sort_cols))
 
-    def drop_nulls(self: Self, subset: str | list[str] | None) -> Self:
+    def drop_nulls(self: Self, subset: list[str] | None) -> Self:
         return self._from_native_frame(self._native_frame.dropna(subset=subset))
 
     def rename(self: Self, mapping: dict[str, str]) -> Self:
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index 07b68d26..98a1d2d5 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -7,9 +7,11 @@ from typing import Literal
 from typing import Sequence
 
 from narwhals._expression_parsing import infer_new_root_output_names
+from narwhals._spark_like.expr_name import SparkLikeExprNameNamespace
 from narwhals._spark_like.expr_str import SparkLikeExprStringNamespace
 from narwhals._spark_like.utils import get_column_name
 from narwhals._spark_like.utils import maybe_evaluate
+from narwhals._spark_like.utils import narwhals_to_native_dtype
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
 from narwhals.utils import parse_version
@@ -20,6 +22,7 @@ if TYPE_CHECKING:
 
     from narwhals._spark_like.dataframe import SparkLikeLazyFrame
     from narwhals._spark_like.namespace import SparkLikeNamespace
+    from narwhals.dtypes import DType
     from narwhals.utils import Version
 
 
@@ -284,6 +287,15 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
         return self._from_call(F.bool_or, "any", returns_scalar=True)
 
+    def cast(self: Self, dtype: DType | type[DType]) -> Self:
+        def _cast(_input: Column, dtype: DType | type[DType]) -> Column:
+            spark_dtype = narwhals_to_native_dtype(dtype, self._version)
+            return _input.cast(spark_dtype)
+
+        return self._from_call(
+            _cast, "cast", dtype=dtype, returns_scalar=self._returns_scalar
+        )
+
     def count(self) -> Self:
         from pyspark.sql import functions as F  # noqa: N812
 
@@ -491,6 +503,32 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
         return self._from_call(_n_unique, "n_unique", returns_scalar=True)
 
+    def over(self: Self, keys: list[str]) -> Self:
+        if self._output_names is None:
+            msg = (
+                "Anonymous expressions are not supported in over.\n"
+                "Instead of `nw.all()`, try using a named expression, such as "
+                "`nw.col('a', 'b')`\n"
+            )
+            raise ValueError(msg)
+
+        def func(df: SparkLikeLazyFrame) -> list[Column]:
+            from pyspark.sql import Window
+
+            return [expr.over(Window.partitionBy(*keys)) for expr in self._call(df)]
+
+        return self.__class__(
+            func,
+            depth=self._depth + 1,
+            function_name=self._function_name + "->over",
+            root_names=self._root_names,
+            output_names=self._output_names,
+            backend_version=self._backend_version,
+            version=self._version,
+            returns_scalar=False,
+            kwargs={**self._kwargs, "keys": keys},
+        )
+
     def is_null(self: Self) -> Self:
         from pyspark.sql import functions as F  # noqa: N812
 
@@ -499,3 +537,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
     @property
     def str(self: Self) -> SparkLikeExprStringNamespace:
         return SparkLikeExprStringNamespace(self)
+
+    @property
+    def name(self: Self) -> SparkLikeExprNameNamespace:
+        return SparkLikeExprNameNamespace(self)
diff --git a/narwhals/_spark_like/expr_name.py b/narwhals/_spark_like/expr_name.py
new file mode 100644
index 00000000..c3230527
--- /dev/null
+++ b/narwhals/_spark_like/expr_name.py
@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from typing import Callable
+
+from narwhals.exceptions import AnonymousExprError
+
+if TYPE_CHECKING:
+    from typing_extensions import Self
+
+    from narwhals._spark_like.expr import SparkLikeExpr
+
+
+class SparkLikeExprNameNamespace:
+    def __init__(self: Self, expr: SparkLikeExpr) -> None:
+        self._compliant_expr = expr
+
+    def keep(self: Self) -> SparkLikeExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.keep"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), root_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=root_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs=self._compliant_expr._kwargs,
+        )
+
+    def map(self: Self, function: Callable[[str], str]) -> SparkLikeExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.map"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        output_names = [function(str(name)) for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs={**self._compliant_expr._kwargs, "function": function},
+        )
+
+    def prefix(self: Self, prefix: str) -> SparkLikeExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.prefix"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        output_names = [prefix + str(name) for name in root_names]
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs={**self._compliant_expr._kwargs, "prefix": prefix},
+        )
+
+    def suffix(self: Self, suffix: str) -> SparkLikeExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.suffix"
+            raise AnonymousExprError.from_expr_name(msg)
+
+        output_names = [str(name) + suffix for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs={**self._compliant_expr._kwargs, "suffix": suffix},
+        )
+
+    def to_lowercase(self: Self) -> SparkLikeExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.to_lowercase"
+            raise AnonymousExprError.from_expr_name(msg)
+        output_names = [str(name).lower() for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs=self._compliant_expr._kwargs,
+        )
+
+    def to_uppercase(self: Self) -> SparkLikeExpr:
+        root_names = self._compliant_expr._root_names
+        if root_names is None:
+            msg = ".name.to_uppercase"
+            raise AnonymousExprError.from_expr_name(msg)
+        output_names = [str(name).upper() for name in root_names]
+
+        return self._compliant_expr.__class__(
+            lambda df: [
+                expr.alias(name)
+                for expr, name in zip(self._compliant_expr._call(df), output_names)
+            ],
+            depth=self._compliant_expr._depth,
+            function_name=self._compliant_expr._function_name,
+            root_names=root_names,
+            output_names=output_names,
+            returns_scalar=self._compliant_expr._returns_scalar,
+            backend_version=self._compliant_expr._backend_version,
+            version=self._compliant_expr._version,
+            kwargs=self._compliant_expr._kwargs,
+        )
diff --git a/narwhals/_spark_like/group_by.py b/narwhals/_spark_like/group_by.py
index 66f4bf2b..12433bbb 100644
--- a/narwhals/_spark_like/group_by.py
+++ b/narwhals/_spark_like/group_by.py
@@ -11,24 +11,23 @@ from narwhals._expression_parsing import is_simple_aggregation
 from narwhals._expression_parsing import parse_into_exprs
 from narwhals._spark_like.utils import _std
 from narwhals._spark_like.utils import _var
+from narwhals.exceptions import AnonymousExprError
 from narwhals.utils import parse_version
 from narwhals.utils import remove_prefix
 
 if TYPE_CHECKING:
     from pyspark.sql import Column
     from pyspark.sql import GroupedData
+    from typing_extensions import Self
 
     from narwhals._spark_like.dataframe import SparkLikeLazyFrame
     from narwhals._spark_like.typing import IntoSparkLikeExpr
     from narwhals.typing import CompliantExpr
 
 
-POLARS_TO_PYSPARK_AGGREGATIONS = {"len": "count"}
-
-
 class SparkLikeLazyGroupBy:
     def __init__(
-        self,
+        self: Self,
         df: SparkLikeLazyFrame,
         keys: list[str],
         drop_null_keys: bool,  # noqa: FBT001
@@ -43,7 +42,7 @@ class SparkLikeLazyGroupBy:
             self._grouped = self._df._native_frame.groupBy(*self._keys)
 
     def agg(
-        self,
+        self: Self,
         *aggs: IntoSparkLikeExpr,
         **named_aggs: IntoSparkLikeExpr,
     ) -> SparkLikeLazyFrame:
@@ -55,23 +54,20 @@ class SparkLikeLazyGroupBy:
         output_names: list[str] = copy(self._keys)
         for expr in exprs:
             if expr._output_names is None:  # pragma: no cover
-                msg = (
-                    "Anonymous expressions are not supported in group_by.agg.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
-                )
-                raise ValueError(msg)
+                msg = "group_by.agg"
+                raise AnonymousExprError.from_expr_name(msg)
 
             output_names.extend(expr._output_names)
 
         return agg_pyspark(
+            self._df,
             self._grouped,
             exprs,
             self._keys,
             self._from_native_frame,
         )
 
-    def _from_native_frame(self, df: SparkLikeLazyFrame) -> SparkLikeLazyFrame:
+    def _from_native_frame(self: Self, df: SparkLikeLazyFrame) -> SparkLikeLazyFrame:
         from narwhals._spark_like.dataframe import SparkLikeLazyFrame
 
         return SparkLikeLazyFrame(
@@ -90,6 +86,7 @@ def get_spark_function(function_name: str, **kwargs: Any) -> Column:
             ddof=kwargs["ddof"],
             np_version=parse_version(np.__version__),
         )
+
     elif function_name == "len":
         # Use count(*) to count all rows including nulls
         def _count(*_args: Any, **_kwargs: Any) -> Column:
@@ -97,16 +94,29 @@ def get_spark_function(function_name: str, **kwargs: Any) -> Column:
 
         return _count
 
+    elif function_name == "n_unique":
+        from pyspark.sql.types import IntegerType
+
+        def _n_unique(_input: Column) -> Column:
+            return F.count_distinct(_input) + F.max(F.isnull(_input).cast(IntegerType()))
+
+        return _n_unique
+
     else:
         return getattr(F, function_name)
 
 
 def agg_pyspark(
+    df: SparkLikeLazyFrame,
     grouped: GroupedData,
     exprs: Sequence[CompliantExpr[Column]],
     keys: list[str],
     from_dataframe: Callable[[Any], SparkLikeLazyFrame],
 ) -> SparkLikeLazyFrame:
+    if not exprs:
+        # No aggregation provided
+        return from_dataframe(df._native_frame.select(*keys).dropDuplicates(subset=keys))
+
     for expr in exprs:
         if not is_simple_aggregation(expr):  # pragma: no cover
             msg = (
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index a8c778b8..bf876a88 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -36,7 +36,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
             return [F.col(col_name) for col_name in df.columns]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=_all,
             depth=0,
             function_name="all",
@@ -63,7 +63,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
             return [F.lit(value).alias("literal")]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=_lit,
             depth=0,
             function_name="lit",
@@ -81,7 +81,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
             return [F.count("*").alias("len")]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             func,
             depth=0,
             function_name="len",
@@ -101,7 +101,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             col_name = get_column_name(df, cols[0])
             return [reduce(operator.and_, cols).alias(col_name)]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=func,
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="all_horizontal",
@@ -121,7 +121,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             col_name = get_column_name(df, cols[0])
             return [reduce(operator.or_, cols).alias(col_name)]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=func,
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="any_horizontal",
@@ -148,7 +148,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
                 ).alias(col_name)
             ]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=func,
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="sum_horizontal",
@@ -179,7 +179,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
                 ).alias(col_name)
             ]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=func,
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="mean_horizontal",
@@ -201,7 +201,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             col_name = get_column_name(df, cols[0])
             return [F.greatest(*cols).alias(col_name)]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=func,
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="max_horizontal",
@@ -223,7 +223,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
             col_name = get_column_name(df, cols[0])
             return [F.least(*cols).alias(col_name)]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=func,
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="min_horizontal",
@@ -320,7 +320,7 @@ class SparkLikeNamespace(CompliantNamespace["Column"]):
 
             return [result]
 
-        return SparkLikeExpr(  # type: ignore[abstract]
+        return SparkLikeExpr(
             call=func,
             depth=max(x._depth for x in parsed_exprs) + 1,
             function_name="concat_str",
@@ -392,7 +392,7 @@ class SparkLikeWhen:
     def then(self, value: SparkLikeExpr | Any) -> SparkLikeThen:
         self._then_value = value
 
-        return SparkLikeThen(  # type: ignore[abstract]
+        return SparkLikeThen(
             self,
             depth=0,
             function_name="whenthen",
diff --git a/narwhals/_spark_like/utils.py b/narwhals/_spark_like/utils.py
index fb3a3f3c..911b18ee 100644
--- a/narwhals/_spark_like/utils.py
+++ b/narwhals/_spark_like/utils.py
@@ -5,7 +5,9 @@ from typing import TYPE_CHECKING
 from typing import Any
 
 from narwhals.exceptions import InvalidIntoExprError
+from narwhals.exceptions import UnsupportedDTypeError
 from narwhals.utils import import_dtypes_module
+from narwhals.utils import isinstance_or_issubclass
 
 if TYPE_CHECKING:
     from pyspark.sql import Column
@@ -22,9 +24,10 @@ def native_to_narwhals_dtype(
     dtype: pyspark_types.DataType,
     version: Version,
 ) -> DType:  # pragma: no cover
-    dtypes = import_dtypes_module(version=version)
     from pyspark.sql import types as pyspark_types
 
+    dtypes = import_dtypes_module(version=version)
+
     if isinstance(dtype, pyspark_types.DoubleType):
         return dtypes.Float64()
     if isinstance(dtype, pyspark_types.FloatType):
@@ -35,6 +38,8 @@ def native_to_narwhals_dtype(
         return dtypes.Int32()
     if isinstance(dtype, pyspark_types.ShortType):
         return dtypes.Int16()
+    if isinstance(dtype, pyspark_types.ByteType):
+        return dtypes.Int8()
     string_types = [
         pyspark_types.StringType,
         pyspark_types.VarcharType,
@@ -58,6 +63,51 @@ def native_to_narwhals_dtype(
     return dtypes.Unknown()
 
 
+def narwhals_to_native_dtype(
+    dtype: DType | type[DType], version: Version
+) -> pyspark_types.DataType:
+    from pyspark.sql import types as pyspark_types
+
+    dtypes = import_dtypes_module(version)
+
+    if isinstance_or_issubclass(dtype, dtypes.Float64):
+        return pyspark_types.DoubleType()
+    if isinstance_or_issubclass(dtype, dtypes.Float32):
+        return pyspark_types.FloatType()
+    if isinstance_or_issubclass(dtype, dtypes.Int64):
+        return pyspark_types.LongType()
+    if isinstance_or_issubclass(dtype, dtypes.Int32):
+        return pyspark_types.IntegerType()
+    if isinstance_or_issubclass(dtype, dtypes.Int16):
+        return pyspark_types.ShortType()
+    if isinstance_or_issubclass(dtype, dtypes.Int8):
+        return pyspark_types.ByteType()
+    if isinstance_or_issubclass(dtype, dtypes.String):
+        return pyspark_types.StringType()
+    if isinstance_or_issubclass(dtype, dtypes.Boolean):
+        return pyspark_types.BooleanType()
+    if isinstance_or_issubclass(dtype, (dtypes.Date, dtypes.Datetime)):
+        msg = "Converting to Date or Datetime dtype is not supported yet"
+        raise NotImplementedError(msg)
+    if isinstance_or_issubclass(dtype, dtypes.List):  # pragma: no cover
+        msg = "Converting to List dtype is not supported yet"
+        raise NotImplementedError(msg)
+    if isinstance_or_issubclass(dtype, dtypes.Struct):  # pragma: no cover
+        msg = "Converting to Struct dtype is not supported yet"
+        raise NotImplementedError(msg)
+    if isinstance_or_issubclass(dtype, dtypes.Array):  # pragma: no cover
+        msg = "Converting to Array dtype is not supported yet"
+        raise NotImplementedError(msg)
+    if isinstance_or_issubclass(
+        dtype, (dtypes.UInt64, dtypes.UInt32, dtypes.UInt16, dtypes.UInt8)
+    ):  # pragma: no cover
+        msg = "Unsigned integer types are not supported by PySpark"
+        raise UnsupportedDTypeError(msg)
+
+    msg = f"Unknown dtype: {dtype}"  # pragma: no cover
+    raise AssertionError(msg)
+
+
 def get_column_name(df: SparkLikeLazyFrame, column: Column) -> str:
     return str(df._native_frame.select(column).columns[0])
 
diff --git a/narwhals/dataframe.py b/narwhals/dataframe.py
index 1ae43028..b055fed1 100644
--- a/narwhals/dataframe.py
+++ b/narwhals/dataframe.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from abc import abstractmethod
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
@@ -15,6 +16,7 @@ from warnings import warn
 
 from narwhals.dependencies import get_polars
 from narwhals.dependencies import is_numpy_array
+from narwhals.exceptions import OrderDependentExprError
 from narwhals.schema import Schema
 from narwhals.translate import to_native
 from narwhals.utils import find_stacklevel
@@ -70,25 +72,9 @@ class BaseFrame(Generic[FrameT]):
         kwargs = {k: self._extract_compliant(v) for k, v in kwargs.items()}
         return args, kwargs
 
+    @abstractmethod
     def _extract_compliant(self, arg: Any) -> Any:
-        from narwhals.expr import Expr
-        from narwhals.series import Series
-
-        if isinstance(arg, BaseFrame):
-            return arg._compliant_frame
-        if isinstance(arg, Series):
-            return arg._compliant_series
-        if isinstance(arg, Expr):
-            return arg._to_compliant_expr(self.__narwhals_namespace__())
-        if get_polars() is not None and "polars" in str(type(arg)):
-            msg = (
-                f"Expected Narwhals object, got: {type(arg)}.\n\n"
-                "Perhaps you:\n"
-                "- Forgot a `nw.from_native` somewhere?\n"
-                "- Used `pl.col` instead of `nw.col`?"
-            )
-            raise TypeError(msg)
-        return arg
+        raise NotImplementedError
 
     @property
     def schema(self) -> Schema:
@@ -107,7 +93,8 @@ class BaseFrame(Generic[FrameT]):
             self._compliant_frame.with_row_index(name),
         )
 
-    def drop_nulls(self: Self, subset: str | list[str] | None = None) -> Self:
+    def drop_nulls(self: Self, subset: str | list[str] | None) -> Self:
+        subset = [subset] if isinstance(subset, str) else subset
         return self._from_compliant_dataframe(
             self._compliant_frame.drop_nulls(subset=subset),
         )
@@ -360,6 +347,26 @@ class DataFrame(BaseFrame[DataFrameT]):
             ```
     """
 
+    def _extract_compliant(self, arg: Any) -> Any:
+        from narwhals.expr import Expr
+        from narwhals.series import Series
+
+        if isinstance(arg, BaseFrame):
+            return arg._compliant_frame
+        if isinstance(arg, Series):
+            return arg._compliant_series
+        if isinstance(arg, Expr):
+            return arg._to_compliant_expr(self.__narwhals_namespace__())
+        if get_polars() is not None and "polars" in str(type(arg)):
+            msg = (
+                f"Expected Narwhals object, got: {type(arg)}.\n\n"
+                "Perhaps you:\n"
+                "- Forgot a `nw.from_native` somewhere?\n"
+                "- Used `pl.col` instead of `nw.col`?"
+            )
+            raise TypeError(msg)
+        return arg
+
     @property
     def _series(self) -> type[Series[Any]]:
         from narwhals.series import Series
@@ -3620,6 +3627,40 @@ class LazyFrame(BaseFrame[FrameT]):
         ```
     """
 
+    def _extract_compliant(self, arg: Any) -> Any:
+        from narwhals.expr import Expr
+        from narwhals.series import Series
+
+        if isinstance(arg, BaseFrame):
+            return arg._compliant_frame
+        if isinstance(arg, Series):  # pragma: no cover
+            msg = "Binary operations between Series and LazyFrame are not supported."
+            raise TypeError(msg)
+        if isinstance(arg, Expr):
+            if arg._is_order_dependent:
+                msg = (
+                    "Order-dependent expressions are not supported for use in LazyFrame.\n\n"
+                    "Hints:\n"
+                    "- Instead of `lf.select(nw.col('a').sort())`, use `lf.select('a').sort()\n"
+                    "- Instead of `lf.select(nw.col('a').head())`, use `lf.select('a').head()\n"
+                    "- `Expr.cum_sum`, and other such expressions, are not currently supported.\n"
+                    "  In a future version of Narwhals, a `order_by` argument will be added and \n"
+                    "  they will be supported."
+                )
+                raise OrderDependentExprError(msg)
+            return arg._to_compliant_expr(self.__narwhals_namespace__())
+        if get_polars() is not None and "polars" in str(type(arg)):  # pragma: no cover
+            msg = (
+                f"Expected Narwhals object, got: {type(arg)}.\n\n"
+                "Perhaps you:\n"
+                "- Forgot a `nw.from_native` somewhere?\n"
+                "- Used `pl.col` instead of `nw.col`?"
+            )
+            raise TypeError(msg)
+        # TODO(unassigned): should this line even be reachable? Should we
+        # be raising here?
+        return arg  # pragma: no cover
+
     @property
     def _dataframe(self) -> type[DataFrame[Any]]:
         return DataFrame
diff --git a/narwhals/dependencies.py b/narwhals/dependencies.py
index 43904a0b..df671be3 100644
--- a/narwhals/dependencies.py
+++ b/narwhals/dependencies.py
@@ -92,7 +92,7 @@ def get_ibis() -> Any:
     return sys.modules.get("ibis", None)
 
 
-def get_dask_expr() -> Any:
+def get_dask_expr() -> Any:  # pragma: no cover
     """Get dask_expr module (if already imported - else return None)."""
     return sys.modules.get("dask_expr", None)
 
diff --git a/narwhals/exceptions.py b/narwhals/exceptions.py
index ee4b79b6..9b05e3ba 100644
--- a/narwhals/exceptions.py
+++ b/narwhals/exceptions.py
@@ -66,5 +66,34 @@ class InvalidIntoExprError(TypeError):
         return InvalidIntoExprError(message)
 
 
+class AnonymousExprError(ValueError):
+    """Exception raised when trying to perform operations on anonymous expressions."""
+
+    def __init__(self, message: str) -> None:
+        self.message = message
+        super().__init__(self.message)
+
+    @classmethod
+    def from_expr_name(cls, expr_name: str) -> AnonymousExprError:
+        message = (
+            f"Anonymous expressions are not supported in `{expr_name}`.\n"
+            "Instead of `nw.all()`, try using a named expression, such as "
+            "`nw.col('a', 'b')`"
+        )
+        return AnonymousExprError(message)
+
+
+class OrderDependentExprError(ValueError):
+    """Exception raised when trying to use an order-dependent expressions with LazyFrames."""
+
+    def __init__(self, message: str) -> None:
+        self.message = message
+        super().__init__(self.message)
+
+
+class UnsupportedDTypeError(ValueError):
+    """Exception raised when trying to convert to a DType which is not supported by the given backend."""
+
+
 class NarwhalsUnstableWarning(UserWarning):
     """Warning issued when a method or function is considered unstable in the stable api."""
diff --git a/narwhals/expr.py b/narwhals/expr.py
index 84646cf8..69c2e7dc 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -9,6 +9,7 @@ from typing import Mapping
 from typing import Sequence
 
 from narwhals._expression_parsing import extract_compliant
+from narwhals._expression_parsing import operation_is_order_dependent
 from narwhals.dtypes import _validate_dtype
 from narwhals.expr_cat import ExprCatNamespace
 from narwhals.expr_dt import ExprDateTimeNamespace
@@ -17,6 +18,7 @@ from narwhals.expr_name import ExprNameNamespace
 from narwhals.expr_str import ExprStringNamespace
 from narwhals.utils import _validate_rolling_arguments
 from narwhals.utils import flatten
+from narwhals.utils import issue_deprecation_warning
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -28,14 +30,21 @@ if TYPE_CHECKING:
 
 
 class Expr:
-    def __init__(self, to_compliant_expr: Callable[[Any], Any]) -> None:
+    def __init__(
+        self,
+        to_compliant_expr: Callable[[Any], Any],
+        is_order_dependent: bool,  # noqa: FBT001
+    ) -> None:
         # callable from CompliantNamespace to CompliantExpr
         self._to_compliant_expr = to_compliant_expr
+        self._is_order_dependent = is_order_dependent
 
     def _taxicab_norm(self) -> Self:
         # This is just used to test out the stable api feature in a realistic-ish way.
         # It's not intended to be used.
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).abs().sum())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).abs().sum(), self._is_order_dependent
+        )
 
     # --- convert ---
     def alias(self, name: str) -> Self:
@@ -91,7 +100,10 @@ class Expr:
             c: [[14,15]]
 
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).alias(name))
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).alias(name),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def pipe(self, function: Callable[[Any], Self], *args: Any, **kwargs: Any) -> Self:
         """Pipe function call.
@@ -212,24 +224,32 @@ class Expr:
         _validate_dtype(dtype)
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).cast(dtype),
+            is_order_dependent=self._is_order_dependent,
         )
 
     # --- binary ---
     def __eq__(self, other: object) -> Self:  # type: ignore[override]
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).__eq__(extract_compliant(plx, other))
+            lambda plx: self._to_compliant_expr(plx).__eq__(
+                extract_compliant(plx, other)
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __ne__(self, other: object) -> Self:  # type: ignore[override]
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).__ne__(extract_compliant(plx, other))
+            lambda plx: self._to_compliant_expr(plx).__ne__(
+                extract_compliant(plx, other)
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __and__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__and__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __rand__(self, other: Any) -> Self:
@@ -238,11 +258,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __or__(self, other: Any) -> Self:
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).__or__(extract_compliant(plx, other))
+            lambda plx: self._to_compliant_expr(plx).__or__(
+                extract_compliant(plx, other)
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __ror__(self, other: Any) -> Self:
@@ -251,13 +277,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __add__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__add__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __radd__(self, other: Any) -> Self:
@@ -266,13 +296,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __sub__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__sub__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __rsub__(self, other: Any) -> Self:
@@ -281,13 +315,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __truediv__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__truediv__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __rtruediv__(self, other: Any) -> Self:
@@ -296,13 +334,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __mul__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__mul__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __rmul__(self, other: Any) -> Self:
@@ -311,33 +353,49 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __le__(self, other: Any) -> Self:
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).__le__(extract_compliant(plx, other))
+            lambda plx: self._to_compliant_expr(plx).__le__(
+                extract_compliant(plx, other)
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __lt__(self, other: Any) -> Self:
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).__lt__(extract_compliant(plx, other))
+            lambda plx: self._to_compliant_expr(plx).__lt__(
+                extract_compliant(plx, other)
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __gt__(self, other: Any) -> Self:
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).__gt__(extract_compliant(plx, other))
+            lambda plx: self._to_compliant_expr(plx).__gt__(
+                extract_compliant(plx, other)
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __ge__(self, other: Any) -> Self:
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).__ge__(extract_compliant(plx, other))
+            lambda plx: self._to_compliant_expr(plx).__ge__(
+                extract_compliant(plx, other)
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __pow__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__pow__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __rpow__(self, other: Any) -> Self:
@@ -346,13 +404,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __floordiv__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__floordiv__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __rfloordiv__(self, other: Any) -> Self:
@@ -361,13 +423,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     def __mod__(self, other: Any) -> Self:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).__mod__(
                 extract_compliant(plx, other)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, other),
         )
 
     def __rmod__(self, other: Any) -> Self:
@@ -376,11 +442,17 @@ class Expr:
                 extract_compliant(plx, self)
             )
 
-        return self.__class__(func)
+        return self.__class__(
+            func,
+            is_order_dependent=operation_is_order_dependent(self, other),
+        )
 
     # --- unary ---
     def __invert__(self) -> Self:
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).__invert__())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).__invert__(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def any(self) -> Self:
         """Return whether any of the values in the column are `True`.
@@ -431,7 +503,10 @@ class Expr:
             a: [[true]]
             b: [[true]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).any())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).any(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def all(self) -> Self:
         """Return whether all values in the column are `True`.
@@ -482,7 +557,10 @@ class Expr:
             a: [[false]]
             b: [[true]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).all())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).all(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def ewm_mean(
         self: Self,
@@ -583,7 +661,8 @@ class Expr:
                 adjust=adjust,
                 min_periods=min_periods,
                 ignore_nulls=ignore_nulls,
-            )
+            ),
+            is_order_dependent=self._is_order_dependent,
         )
 
     def mean(self) -> Self:
@@ -635,7 +714,10 @@ class Expr:
             a: [[0]]
             b: [[4]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).mean())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).mean(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def median(self) -> Self:
         """Get median value.
@@ -689,7 +771,10 @@ class Expr:
             a: [[3]]
             b: [[4]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).median())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).median(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def std(self, *, ddof: int = 1) -> Self:
         """Get standard deviation.
@@ -743,7 +828,10 @@ class Expr:
             b: [[1.2657891697365016]]
 
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).std(ddof=ddof))
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).std(ddof=ddof),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def var(self, *, ddof: int = 1) -> Self:
         """Get variance.
@@ -798,7 +886,10 @@ class Expr:
             a: [[316.6666666666667]]
             b: [[1.6022222222222222]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).var(ddof=ddof))
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).var(ddof=ddof),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def map_batches(
         self,
@@ -872,7 +963,8 @@ class Expr:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).map_batches(
                 function=function, return_dtype=return_dtype
-            )
+            ),
+            is_order_dependent=True,  # safest assumption
         )
 
     def skew(self: Self) -> Self:
@@ -924,7 +1016,10 @@ class Expr:
             a: [[0]]
             b: [[1.4724267269058975]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).skew())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).skew(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def sum(self) -> Expr:
         """Return the sum value.
@@ -973,7 +1068,10 @@ class Expr:
             a: [[15]]
             b: [[150]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).sum())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).sum(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def min(self) -> Self:
         """Returns the minimum value(s) from a column(s).
@@ -1024,7 +1122,10 @@ class Expr:
             a: [[1]]
             b: [[3]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).min())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).min(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def max(self) -> Self:
         """Returns the maximum value(s) from a column(s).
@@ -1075,7 +1176,10 @@ class Expr:
             a: [[20]]
             b: [[100]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).max())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).max(),
+            is_order_dependent=self._is_order_dependent,
+        )
 
     def arg_min(self) -> Self:
         """Returns the index of the minimum value.
@@ -1128,7 +1232,9 @@ class Expr:
             a_arg_min: [[0]]
             b_arg_min: [[1]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).arg_min())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).arg_min(), is_order_dependent=True
+        )
 
     def arg_max(self) -> Self:
         """Returns the index of the maximum value.
@@ -1181,7 +1287,9 @@ class Expr:
             a_arg_max: [[1]]
             b_arg_max: [[0]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).arg_max())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).arg_max(), is_order_dependent=True
+        )
 
     def count(self) -> Self:
         """Returns the number of non-null elements in the column.
@@ -1232,7 +1340,9 @@ class Expr:
             a: [[3]]
             b: [[2]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).count())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).count(), self._is_order_dependent
+        )
 
     def n_unique(self) -> Self:
         """Returns count of unique values.
@@ -1281,7 +1391,9 @@ class Expr:
             a: [[5]]
             b: [[3]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).n_unique())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).n_unique(), self._is_order_dependent
+        )
 
     def unique(self, *, maintain_order: bool = False) -> Self:
         """Return unique values of this expression.
@@ -1342,7 +1454,10 @@ class Expr:
             b: [[2,4,6]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).unique(maintain_order=maintain_order)
+            lambda plx: self._to_compliant_expr(plx).unique(
+                maintain_order=maintain_order
+            ),
+            self._is_order_dependent,
         )
 
     def abs(self) -> Self:
@@ -1396,7 +1511,9 @@ class Expr:
             a: [[1,2]]
             b: [[3,4]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).abs())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).abs(), self._is_order_dependent
+        )
 
     def cum_sum(self: Self, *, reverse: bool = False) -> Self:
         """Return cumulative sum.
@@ -1457,7 +1574,8 @@ class Expr:
             b: [[2,6,10,16,22]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).cum_sum(reverse=reverse)
+            lambda plx: self._to_compliant_expr(plx).cum_sum(reverse=reverse),
+            is_order_dependent=True,
         )
 
     def diff(self) -> Self:
@@ -1524,7 +1642,9 @@ class Expr:
             ----
             a_diff: [[null,0,2,2,0]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).diff())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).diff(), is_order_dependent=True
+        )
 
     def shift(self, n: int) -> Self:
         """Shift values by `n` positions.
@@ -1593,7 +1713,9 @@ class Expr:
             ----
             a_shift: [[null,1,1,3,5]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).shift(n))
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).shift(n), is_order_dependent=True
+        )
 
     def replace_strict(
         self,
@@ -1684,100 +1806,39 @@ class Expr:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).replace_strict(
                 old, new, return_dtype=return_dtype
-            )
+            ),
+            self._is_order_dependent,
         )
 
     def sort(self, *, descending: bool = False, nulls_last: bool = False) -> Self:
         """Sort this column. Place null values first.
 
+        !!! warning
+            `Expr.sort` is deprecated and will be removed in a future version.
+            Hint: instead of `df.select(nw.col('a').sort())`, use
+            `df.select(nw.col('a')).sort()` instead.
+            Note: this will remain available in `narwhals.stable.v1`.
+            See [stable api](../backcompat.md/) for more information.
+
         Arguments:
             descending: Sort in descending order.
             nulls_last: Place null values last instead of first.
 
         Returns:
             A new expression.
-
-        Examples:
-            >>> import pandas as pd
-            >>> import polars as pl
-            >>> import pyarrow as pa
-            >>> import narwhals as nw
-            >>> from narwhals.typing import IntoFrameT
-            >>>
-            >>> data = {"a": [5, None, 1, 2]}
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
-            >>> df_pa = pa.table(data)
-
-            Let's define dataframe-agnostic functions:
-
-            >>> def agnostic_sort(df_native: IntoFrameT) -> IntoFrameT:
-            ...     df = nw.from_native(df_native)
-            ...     return df.select(nw.col("a").sort()).to_native()
-
-            >>> def agnostic_sort_descending(df_native: IntoFrameT) -> IntoFrameT:
-            ...     df = nw.from_native(df_native)
-            ...     return df.select(nw.col("a").sort(descending=True)).to_native()
-
-            We can then pass any supported library such as pandas, Polars, or
-            PyArrow to `agnostic_sort` and `agnostic_sort_descending`:
-
-            >>> agnostic_sort(df_pd)
-                 a
-            1  NaN
-            2  1.0
-            3  2.0
-            0  5.0
-
-            >>> agnostic_sort(df_pl)
-            shape: (4, 1)
-            ┌──────┐
-            │ a    │
-            │ ---  │
-            │ i64  │
-            ╞══════╡
-            │ null │
-            │ 1    │
-            │ 2    │
-            │ 5    │
-            └──────┘
-
-            >>> agnostic_sort(df_pa)
-            pyarrow.Table
-            a: int64
-            ----
-            a: [[null,1,2,5]]
-
-            >>> agnostic_sort_descending(df_pd)
-                 a
-            1  NaN
-            0  5.0
-            3  2.0
-            2  1.0
-
-            >>> agnostic_sort_descending(df_pl)
-            shape: (4, 1)
-            ┌──────┐
-            │ a    │
-            │ ---  │
-            │ i64  │
-            ╞══════╡
-            │ null │
-            │ 5    │
-            │ 2    │
-            │ 1    │
-            └──────┘
-
-            >>> agnostic_sort_descending(df_pa)
-            pyarrow.Table
-            a: int64
-            ----
-            a: [[null,5,2,1]]
         """
+        msg = (
+            "`Expr.sort` is deprecated and will be removed in a future version.\n\n"
+            "Hint: instead of `df.select(nw.col('a').sort())`, use `df.select(nw.col('a')).sort()`.\n\n"
+            "Note: this will remain available in `narwhals.stable.v1`.\n"
+            "See https://narwhals-dev.github.io/narwhals/backcompat/ for more information.\n"
+        )
+        issue_deprecation_warning(msg, _version="1.22.0")
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).sort(
                 descending=descending, nulls_last=nulls_last
-            )
+            ),
+            is_order_dependent=True,
         )
 
     # --- transform ---
@@ -1851,7 +1912,10 @@ class Expr:
                 extract_compliant(plx, lower_bound),
                 extract_compliant(plx, upper_bound),
                 closed,
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(
+                self, lower_bound, upper_bound
+            ),
         )
 
     def is_in(self, other: Any) -> Self:
@@ -1916,7 +1980,8 @@ class Expr:
             return self.__class__(
                 lambda plx: self._to_compliant_expr(plx).is_in(
                     extract_compliant(plx, other)
-                )
+                ),
+                self._is_order_dependent,
             )
         else:
             msg = "Narwhals `is_in` doesn't accept expressions as an argument, as opposed to Polars. You should provide an iterable instead."
@@ -1981,10 +2046,12 @@ class Expr:
             a: [[5,6,7]]
             b: [[10,11,12]]
         """
+        flat_predicates = flatten(predicates)
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).filter(
-                *[extract_compliant(plx, pred) for pred in flatten(predicates)],
-            )
+                *[extract_compliant(plx, pred) for pred in flat_predicates],
+            ),
+            is_order_dependent=operation_is_order_dependent(*flat_predicates),
         )
 
     def is_null(self) -> Self:
@@ -2063,7 +2130,9 @@ class Expr:
             a_is_null: [[false,false,true,false,false]]
             b_is_null: [[false,false,true,false,false]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).is_null())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).is_null(), self._is_order_dependent
+        )
 
     def is_nan(self) -> Self:
         """Indicate which values are NaN.
@@ -2128,7 +2197,9 @@ class Expr:
             divided: [[nan,null,1]]
             divided_is_nan: [[true,null,false]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).is_nan())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).is_nan(), self._is_order_dependent
+        )
 
     def arg_true(self) -> Self:
         """Find elements where boolean expression is True.
@@ -2179,7 +2250,9 @@ class Expr:
             ----
             a: [[1,2]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).arg_true())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).arg_true(), is_order_dependent=True
+        )
 
     def fill_null(
         self,
@@ -2321,7 +2394,8 @@ class Expr:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).fill_null(
                 value=value, strategy=strategy, limit=limit
-            )
+            ),
+            self._is_order_dependent,
         )
 
     # --- partial reduction ---
@@ -2382,7 +2456,10 @@ class Expr:
             ----
             a: [[2,4,3,5]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).drop_nulls())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).drop_nulls(),
+            self._is_order_dependent,
+        )
 
     def sample(
         self: Self,
@@ -2394,6 +2471,13 @@ class Expr:
     ) -> Self:
         """Sample randomly from this expression.
 
+        !!! warning
+            `Expr.sample` is deprecated and will be removed in a future version.
+            Hint: instead of `df.select(nw.col('a').sample())`, use
+            `df.select(nw.col('a')).sample()` instead.
+            Note: this will remain available in `narwhals.stable.v1`.
+            See [stable api](../backcompat.md/) for more information.
+
         Arguments:
             n: Number of items to return. Cannot be used with fraction.
             fraction: Fraction of items to return. Cannot be used with n.
@@ -2403,58 +2487,19 @@ class Expr:
 
         Returns:
             A new expression.
-
-        Examples:
-            >>> import pandas as pd
-            >>> import polars as pl
-            >>> import pyarrow as pa
-            >>> import narwhals as nw
-            >>> from narwhals.typing import IntoFrameT
-            >>>
-            >>> data = {"a": [1, 2, 3]}
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
-            >>> df_pa = pa.table(data)
-
-            Let's define a dataframe-agnostic function:
-
-            >>> def agnostic_sample(df_native: IntoFrameT) -> IntoFrameT:
-            ...     df = nw.from_native(df_native)
-            ...     return df.select(
-            ...         nw.col("a").sample(fraction=1.0, with_replacement=True)
-            ...     ).to_native()
-
-            We can then pass any supported library such as pandas, Polars, or
-            PyArrow to `agnostic_sample`:
-
-            >>> agnostic_sample(df_pd)  # doctest: +SKIP
-               a
-            2  3
-            0  1
-            2  3
-
-            >>> agnostic_sample(df_pl)  # doctest: +SKIP
-            shape: (3, 1)
-            ┌─────┐
-            │ a   │
-            │ --- │
-            │ f64 │
-            ╞═════╡
-            │ 2   │
-            │ 3   │
-            │ 3   │
-            └─────┘
-
-            >>> agnostic_sample(df_pa)  # doctest: +SKIP
-            pyarrow.Table
-            a: int64
-            ----
-            a: [[1,3,3]]
         """
+        msg = (
+            "`Expr.sample` is deprecated and will be removed in a future version.\n\n"
+            "Hint: instead of `df.select(nw.col('a').sample())`, use `df.select(nw.col('a')).sample()`.\n\n"
+            "Note: this will remain available in `narwhals.stable.v1`.\n"
+            "See https://narwhals-dev.github.io/narwhals/backcompat/ for more information.\n"
+        )
+        issue_deprecation_warning(msg, _version="1.22.0")
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).sample(
                 n, fraction=fraction, with_replacement=with_replacement, seed=seed
-            )
+            ),
+            self._is_order_dependent,
         )
 
     def over(self, *keys: str | Iterable[str]) -> Self:
@@ -2545,7 +2590,8 @@ class Expr:
             └─────┴─────┴─────┘
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).over(flatten(keys))
+            lambda plx: self._to_compliant_expr(plx).over(flatten(keys)),
+            self._is_order_dependent,
         )
 
     def is_duplicated(self) -> Self:
@@ -2603,7 +2649,10 @@ class Expr:
             a: [[true,false,false,true]]
             b: [[true,true,false,false]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).is_duplicated())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).is_duplicated(),
+            self._is_order_dependent,
+        )
 
     def is_unique(self) -> Self:
         r"""Return a boolean mask indicating unique values.
@@ -2660,7 +2709,9 @@ class Expr:
             a: [[false,true,true,false]]
             b: [[false,false,true,true]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).is_unique())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).is_unique(), self._is_order_dependent
+        )
 
     def null_count(self) -> Self:
         r"""Count null values.
@@ -2716,7 +2767,10 @@ class Expr:
             a: [[1]]
             b: [[2]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).null_count())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).null_count(),
+            self._is_order_dependent,
+        )
 
     def is_first_distinct(self) -> Self:
         r"""Return a boolean mask indicating the first occurrence of each distinct value.
@@ -2774,7 +2828,8 @@ class Expr:
             b: [[true,false,true,true]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).is_first_distinct()
+            lambda plx: self._to_compliant_expr(plx).is_first_distinct(),
+            is_order_dependent=True,
         )
 
     def is_last_distinct(self) -> Self:
@@ -2832,7 +2887,10 @@ class Expr:
             a: [[false,true,true,true]]
             b: [[false,true,true,true]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).is_last_distinct())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).is_last_distinct(),
+            is_order_dependent=True,
+        )
 
     def quantile(
         self,
@@ -2901,120 +2959,63 @@ class Expr:
             b: [[74.5]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).quantile(quantile, interpolation)
+            lambda plx: self._to_compliant_expr(plx).quantile(quantile, interpolation),
+            self._is_order_dependent,
         )
 
     def head(self, n: int = 10) -> Self:
         r"""Get the first `n` rows.
 
+        !!! warning
+            `Expr.head` is deprecated and will be removed in a future version.
+            Hint: instead of `df.select(nw.col('a').head())`, use
+            `df.select(nw.col('a')).head()` instead.
+            Note: this will remain available in `narwhals.stable.v1`.
+            See [stable api](../backcompat.md/) for more information.
+
         Arguments:
             n: Number of rows to return.
 
         Returns:
             A new expression.
-
-        Examples:
-            >>> import pandas as pd
-            >>> import polars as pl
-            >>> import pyarrow as pa
-            >>> import narwhals as nw
-            >>> from narwhals.typing import IntoFrameT
-            >>>
-            >>> data = {"a": list(range(10))}
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
-            >>> df_pa = pa.table(data)
-
-            Let's define a dataframe-agnostic function that returns the first 3 rows:
-
-            >>> def agnostic_head(df_native: IntoFrameT) -> IntoFrameT:
-            ...     df = nw.from_native(df_native)
-            ...     return df.select(nw.col("a").head(3)).to_native()
-
-            We can then pass any supported library such as pandas, Polars, or
-            PyArrow to `agnostic_head`:
-
-            >>> agnostic_head(df_pd)
-               a
-            0  0
-            1  1
-            2  2
-
-            >>> agnostic_head(df_pl)
-            shape: (3, 1)
-            ┌─────┐
-            │ a   │
-            │ --- │
-            │ i64 │
-            ╞═════╡
-            │ 0   │
-            │ 1   │
-            │ 2   │
-            └─────┘
-
-            >>> agnostic_head(df_pa)
-            pyarrow.Table
-            a: int64
-            ----
-            a: [[0,1,2]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).head(n))
+        msg = (
+            "`Expr.head` is deprecated and will be removed in a future version.\n\n"
+            "Hint: instead of `df.select(nw.col('a').head())`, use `df.select(nw.col('a')).head()`.\n\n"
+            "Note: this will remain available in `narwhals.stable.v1`.\n"
+            "See https://narwhals-dev.github.io/narwhals/backcompat/ for more information.\n"
+        )
+        issue_deprecation_warning(msg, _version="1.22.0")
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).head(n), is_order_dependent=True
+        )
 
     def tail(self, n: int = 10) -> Self:
         r"""Get the last `n` rows.
 
+        !!! warning
+            `Expr.tail` is deprecated and will be removed in a future version.
+            Hint: instead of `df.select(nw.col('a').tail())`, use
+            `df.select(nw.col('a')).tail()` instead.
+            Note: this will remain available in `narwhals.stable.v1`.
+            See [stable api](../backcompat.md/) for more information.
+
         Arguments:
             n: Number of rows to return.
 
         Returns:
             A new expression.
-
-        Examples:
-            >>> import pandas as pd
-            >>> import polars as pl
-            >>> import pyarrow as pa
-            >>> import narwhals as nw
-            >>> from narwhals.typing import IntoFrameT
-            >>>
-            >>> data = {"a": list(range(10))}
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
-            >>> df_pa = pa.table(data)
-
-            Let's define a dataframe-agnostic function that returns the last 3 rows:
-
-            >>> def agnostic_tail(df_native: IntoFrameT) -> IntoFrameT:
-            ...     df = nw.from_native(df_native)
-            ...     return df.select(nw.col("a").tail(3)).to_native()
-
-            We can then pass any supported library such as pandas, Polars, or
-            PyArrow to `agnostic_tail`:
-
-            >>> agnostic_tail(df_pd)
-               a
-            7  7
-            8  8
-            9  9
-
-            >>> agnostic_tail(df_pl)
-            shape: (3, 1)
-            ┌─────┐
-            │ a   │
-            │ --- │
-            │ i64 │
-            ╞═════╡
-            │ 7   │
-            │ 8   │
-            │ 9   │
-            └─────┘
-
-            >>> agnostic_tail(df_pa)
-            pyarrow.Table
-            a: int64
-            ----
-            a: [[7,8,9]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).tail(n))
+        msg = (
+            "`Expr.tail` is deprecated and will be removed in a future version.\n\n"
+            "Hint: instead of `df.select(nw.col('a').tail())`, use `df.select(nw.col('a')).tail()`.\n\n"
+            "Note: this will remain available in `narwhals.stable.v1`.\n"
+            "See https://narwhals-dev.github.io/narwhals/backcompat/ for more information.\n"
+        )
+        issue_deprecation_warning(msg, _version="1.22.0")
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).tail(n), is_order_dependent=True
+        )
 
     def round(self, decimals: int = 0) -> Self:
         r"""Round underlying floating point data by `decimals` digits.
@@ -3079,7 +3080,10 @@ class Expr:
             ----
             a: [[1.1,2.6,3.9]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).round(decimals))
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).round(decimals),
+            self._is_order_dependent,
+        )
 
     def len(self) -> Self:
         r"""Return the number of elements in the column.
@@ -3136,64 +3140,37 @@ class Expr:
             a1: [[2]]
             a2: [[1]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).len())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).len(), self._is_order_dependent
+        )
 
     def gather_every(self: Self, n: int, offset: int = 0) -> Self:
         r"""Take every nth value in the Series and return as new Series.
 
+        !!! warning
+            `Expr.gather_every` is deprecated and will be removed in a future version.
+            Hint: instead of `df.select(nw.col('a').gather_every())`, use
+            `df.select(nw.col('a')).gather_every()` instead.
+            Note: this will remain available in `narwhals.stable.v1`.
+            See [stable api](../backcompat.md/) for more information.
+
         Arguments:
             n: Gather every *n*-th row.
             offset: Starting index.
 
         Returns:
             A new expression.
-
-        Examples:
-            >>> import pandas as pd
-            >>> import polars as pl
-            >>> import pyarrow as pa
-            >>> import narwhals as nw
-            >>> from narwhals.typing import IntoFrameT
-            >>>
-            >>> data = {"a": [1, 2, 3, 4], "b": [5, 6, 7, 8]}
-            >>> df_pd = pd.DataFrame(data)
-            >>> df_pl = pl.DataFrame(data)
-            >>> df_pa = pa.table(data)
-
-            Let's define a dataframe-agnostic function in which 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.select(nw.col("a").gather_every(n=2, offset=1)).to_native()
-
-            We can then pass any supported library such as pandas, Polars, or
-            PyArrow to `agnostic_gather_every`:
-
-            >>> agnostic_gather_every(df_pd)
-               a
-            1  2
-            3  4
-
-            >>> agnostic_gather_every(df_pl)
-            shape: (2, 1)
-            ┌─────┐
-            │ a   │
-            │ --- │
-            │ i64 │
-            ╞═════╡
-            │ 2   │
-            │ 4   │
-            └─────┘
-
-            >>> agnostic_gather_every(df_pa)
-            pyarrow.Table
-            a: int64
-            ----
-            a: [[2,4]]
         """
+        msg = (
+            "`Expr.gather_every` is deprecated and will be removed in a future version.\n\n"
+            "Hint: instead of `df.select(nw.col('a').gather_every())`, use `df.select(nw.col('a')).gather_every()`.\n\n"
+            "Note: this will remain available in `narwhals.stable.v1`.\n"
+            "See https://narwhals-dev.github.io/narwhals/backcompat/ for more information.\n"
+        )
+        issue_deprecation_warning(msg, _version="1.22.0")
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).gather_every(n=n, offset=offset)
+            lambda plx: self._to_compliant_expr(plx).gather_every(n=n, offset=offset),
+            is_order_dependent=True,
         )
 
     # need to allow numeric typing
@@ -3340,7 +3317,10 @@ class Expr:
             lambda plx: self._to_compliant_expr(plx).clip(
                 extract_compliant(plx, lower_bound),
                 extract_compliant(plx, upper_bound),
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(
+                self, lower_bound, upper_bound
+            ),
         )
 
     def mode(self: Self) -> Self:
@@ -3395,7 +3375,9 @@ class Expr:
             ----
             a: [[1]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).mode())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).mode(), self._is_order_dependent
+        )
 
     def is_finite(self: Self) -> Self:
         """Returns boolean values indicating which original values are finite.
@@ -3455,7 +3437,9 @@ class Expr:
             ----
             a: [[false,false,true,null]]
         """
-        return self.__class__(lambda plx: self._to_compliant_expr(plx).is_finite())
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).is_finite(), self._is_order_dependent
+        )
 
     def cum_count(self: Self, *, reverse: bool = False) -> Self:
         r"""Return the cumulative count of the non-null values in the column.
@@ -3521,7 +3505,8 @@ class Expr:
             cum_count_reverse: [[3,2,1,1]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).cum_count(reverse=reverse)
+            lambda plx: self._to_compliant_expr(plx).cum_count(reverse=reverse),
+            is_order_dependent=True,
         )
 
     def cum_min(self: Self, *, reverse: bool = False) -> Self:
@@ -3588,7 +3573,8 @@ class Expr:
             cum_min_reverse: [[1,1,null,2]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).cum_min(reverse=reverse)
+            lambda plx: self._to_compliant_expr(plx).cum_min(reverse=reverse),
+            is_order_dependent=True,
         )
 
     def cum_max(self: Self, *, reverse: bool = False) -> Self:
@@ -3655,7 +3641,8 @@ class Expr:
             cum_max_reverse: [[3,3,null,2]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).cum_max(reverse=reverse)
+            lambda plx: self._to_compliant_expr(plx).cum_max(reverse=reverse),
+            is_order_dependent=True,
         )
 
     def cum_prod(self: Self, *, reverse: bool = False) -> Self:
@@ -3722,7 +3709,8 @@ class Expr:
             cum_prod_reverse: [[6,6,null,2]]
         """
         return self.__class__(
-            lambda plx: self._to_compliant_expr(plx).cum_prod(reverse=reverse)
+            lambda plx: self._to_compliant_expr(plx).cum_prod(reverse=reverse),
+            is_order_dependent=True,
         )
 
     def rolling_sum(
@@ -3816,7 +3804,8 @@ class Expr:
                 window_size=window_size,
                 min_periods=min_periods,
                 center=center,
-            )
+            ),
+            is_order_dependent=True,
         )
 
     def rolling_mean(
@@ -3910,7 +3899,8 @@ class Expr:
                 window_size=window_size,
                 min_periods=min_periods,
                 center=center,
-            )
+            ),
+            is_order_dependent=True,
         )
 
     def rolling_var(
@@ -4004,7 +3994,8 @@ class Expr:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).rolling_var(
                 window_size=window_size, min_periods=min_periods, center=center, ddof=ddof
-            )
+            ),
+            is_order_dependent=True,
         )
 
     def rolling_std(
@@ -4101,7 +4092,8 @@ class Expr:
                 min_periods=min_periods,
                 center=center,
                 ddof=ddof,
-            )
+            ),
+            is_order_dependent=True,
         )
 
     def rank(
@@ -4198,7 +4190,8 @@ class Expr:
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).rank(
                 method=method, descending=descending
-            )
+            ),
+            is_order_dependent=True,
         )
 
     @property
diff --git a/narwhals/expr_cat.py b/narwhals/expr_cat.py
index ada8e3a4..baf467df 100644
--- a/narwhals/expr_cat.py
+++ b/narwhals/expr_cat.py
@@ -62,5 +62,6 @@ class ExprCatNamespace(Generic[ExprT]):
             └────────┘
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).cat.get_categories()
+            lambda plx: self._expr._to_compliant_expr(plx).cat.get_categories(),
+            self._expr._is_order_dependent,
         )
diff --git a/narwhals/expr_dt.py b/narwhals/expr_dt.py
index d0676dd9..6b981315 100644
--- a/narwhals/expr_dt.py
+++ b/narwhals/expr_dt.py
@@ -71,7 +71,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a: [[2012-01-07,2023-03-10]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.date()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.date(),
+            self._expr._is_order_dependent,
         )
 
     def year(self: Self) -> ExprT:
@@ -139,7 +140,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             year: [[1978,2024,2065]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.year()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.year(),
+            self._expr._is_order_dependent,
         )
 
     def month(self: Self) -> ExprT:
@@ -207,7 +209,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             month: [[6,12,1]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.month()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.month(),
+            self._expr._is_order_dependent,
         )
 
     def day(self: Self) -> ExprT:
@@ -275,7 +278,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             day: [[1,13,1]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.day()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.day(),
+            self._expr._is_order_dependent,
         )
 
     def hour(self: Self) -> ExprT:
@@ -343,7 +347,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             hour: [[1,5,10]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.hour()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.hour(),
+            self._expr._is_order_dependent,
         )
 
     def minute(self: Self) -> ExprT:
@@ -411,7 +416,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             minute: [[1,30,20]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.minute()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.minute(),
+            self._expr._is_order_dependent,
         )
 
     def second(self: Self) -> ExprT:
@@ -477,7 +483,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             second: [[1,14,30]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.second()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.second(),
+            self._expr._is_order_dependent,
         )
 
     def millisecond(self: Self) -> ExprT:
@@ -543,7 +550,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             millisecond: [[0,505,67]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.millisecond()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.millisecond(),
+            self._expr._is_order_dependent,
         )
 
     def microsecond(self: Self) -> ExprT:
@@ -609,7 +617,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             microsecond: [[0,505000,67000]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.microsecond()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.microsecond(),
+            self._expr._is_order_dependent,
         )
 
     def nanosecond(self: Self) -> ExprT:
@@ -675,7 +684,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             nanosecond: [[0,500000000,60000000]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.nanosecond()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.nanosecond(),
+            self._expr._is_order_dependent,
         )
 
     def ordinal_day(self: Self) -> ExprT:
@@ -733,7 +743,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a_ordinal_day: [[1,216]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.ordinal_day()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.ordinal_day(),
+            self._expr._is_order_dependent,
         )
 
     def weekday(self: Self) -> ExprT:
@@ -789,7 +800,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a_weekday: [[3,1]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.weekday()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.weekday(),
+            self._expr._is_order_dependent,
         )
 
     def total_minutes(self: Self) -> ExprT:
@@ -852,7 +864,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a_total_minutes: [[10,20]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.total_minutes()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.total_minutes(),
+            self._expr._is_order_dependent,
         )
 
     def total_seconds(self: Self) -> ExprT:
@@ -915,7 +928,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a_total_seconds: [[10,20]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.total_seconds()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.total_seconds(),
+            self._expr._is_order_dependent,
         )
 
     def total_milliseconds(self: Self) -> ExprT:
@@ -983,7 +997,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a_total_milliseconds: [[10,20]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.total_milliseconds()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.total_milliseconds(),
+            self._expr._is_order_dependent,
         )
 
     def total_microseconds(self: Self) -> ExprT:
@@ -1051,7 +1066,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a_total_microseconds: [[10,1200]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.total_microseconds()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.total_microseconds(),
+            self._expr._is_order_dependent,
         )
 
     def total_nanoseconds(self: Self) -> ExprT:
@@ -1106,7 +1122,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             └───────────────────────────────┴──────────────────────────┘
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.total_nanoseconds()
+            lambda plx: self._expr._to_compliant_expr(plx).dt.total_nanoseconds(),
+            self._expr._is_order_dependent,
         )
 
     def to_string(self: Self, format: str) -> ExprT:  # noqa: A002
@@ -1204,7 +1221,8 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a: [["2020/03/01 00:00:00.000000","2020/04/01 00:00:00.000000","2020/05/01 00:00:00.000000"]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.to_string(format)
+            lambda plx: self._expr._to_compliant_expr(plx).dt.to_string(format),
+            self._expr._is_order_dependent,
         )
 
     def replace_time_zone(self: Self, time_zone: str | None) -> ExprT:
@@ -1268,7 +1286,10 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             a: [[2023-12-31 18:15:00.000000Z,2024-01-01 18:15:00.000000Z]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.replace_time_zone(time_zone)
+            lambda plx: self._expr._to_compliant_expr(plx).dt.replace_time_zone(
+                time_zone
+            ),
+            self._expr._is_order_dependent,
         )
 
     def convert_time_zone(self: Self, time_zone: str) -> ExprT:
@@ -1338,7 +1359,10 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             msg = "Target `time_zone` cannot be `None` in `convert_time_zone`. Please use `replace_time_zone(None)` if you want to remove the time zone."
             raise TypeError(msg)
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.convert_time_zone(time_zone)
+            lambda plx: self._expr._to_compliant_expr(plx).dt.convert_time_zone(
+                time_zone
+            ),
+            self._expr._is_order_dependent,
         )
 
     def timestamp(self: Self, time_unit: Literal["ns", "us", "ms"] = "us") -> ExprT:
@@ -1411,5 +1435,6 @@ class ExprDateTimeNamespace(Generic[ExprT]):
             )
             raise ValueError(msg)
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).dt.timestamp(time_unit)
+            lambda plx: self._expr._to_compliant_expr(plx).dt.timestamp(time_unit),
+            self._expr._is_order_dependent,
         )
diff --git a/narwhals/expr_list.py b/narwhals/expr_list.py
index 17efeaf2..c64defda 100644
--- a/narwhals/expr_list.py
+++ b/narwhals/expr_list.py
@@ -73,5 +73,6 @@ class ExprListNamespace(Generic[ExprT]):
             a_len: [[2,3,null,0]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).list.len()
+            lambda plx: self._expr._to_compliant_expr(plx).list.len(),
+            self._expr._is_order_dependent,
         )
diff --git a/narwhals/expr_name.py b/narwhals/expr_name.py
index 312a2bc9..0d428cea 100644
--- a/narwhals/expr_name.py
+++ b/narwhals/expr_name.py
@@ -59,7 +59,8 @@ class ExprNameNamespace(Generic[ExprT]):
             ['foo']
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).name.keep()
+            lambda plx: self._expr._to_compliant_expr(plx).name.keep(),
+            self._expr._is_order_dependent,
         )
 
     def map(self: Self, function: Callable[[str], str]) -> ExprT:
@@ -108,7 +109,8 @@ class ExprNameNamespace(Generic[ExprT]):
             ['oof', 'RAB']
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).name.map(function)
+            lambda plx: self._expr._to_compliant_expr(plx).name.map(function),
+            self._expr._is_order_dependent,
         )
 
     def prefix(self: Self, prefix: str) -> ExprT:
@@ -156,7 +158,8 @@ class ExprNameNamespace(Generic[ExprT]):
             ['with_prefix_foo', 'with_prefix_BAR']
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).name.prefix(prefix)
+            lambda plx: self._expr._to_compliant_expr(plx).name.prefix(prefix),
+            self._expr._is_order_dependent,
         )
 
     def suffix(self: Self, suffix: str) -> ExprT:
@@ -204,7 +207,8 @@ class ExprNameNamespace(Generic[ExprT]):
             ['foo_with_suffix', 'BAR_with_suffix']
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).name.suffix(suffix)
+            lambda plx: self._expr._to_compliant_expr(plx).name.suffix(suffix),
+            self._expr._is_order_dependent,
         )
 
     def to_lowercase(self: Self) -> ExprT:
@@ -249,7 +253,8 @@ class ExprNameNamespace(Generic[ExprT]):
             ['foo', 'bar']
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).name.to_lowercase()
+            lambda plx: self._expr._to_compliant_expr(plx).name.to_lowercase(),
+            self._expr._is_order_dependent,
         )
 
     def to_uppercase(self: Self) -> ExprT:
@@ -294,5 +299,6 @@ class ExprNameNamespace(Generic[ExprT]):
             ['FOO', 'BAR']
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).name.to_uppercase()
+            lambda plx: self._expr._to_compliant_expr(plx).name.to_uppercase(),
+            self._expr._is_order_dependent,
         )
diff --git a/narwhals/expr_str.py b/narwhals/expr_str.py
index 91d355c6..0ea89ceb 100644
--- a/narwhals/expr_str.py
+++ b/narwhals/expr_str.py
@@ -76,7 +76,8 @@ class ExprStringNamespace(Generic[ExprT]):
             words_len: [[3,4,3,2,null]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.len_chars()
+            lambda plx: self._expr._to_compliant_expr(plx).str.len_chars(),
+            self._expr._is_order_dependent,
         )
 
     def replace(
@@ -142,7 +143,8 @@ class ExprStringNamespace(Generic[ExprT]):
         return self._expr.__class__(
             lambda plx: self._expr._to_compliant_expr(plx).str.replace(
                 pattern, value, literal=literal, n=n
-            )
+            ),
+            self._expr._is_order_dependent,
         )
 
     def replace_all(
@@ -207,7 +209,8 @@ class ExprStringNamespace(Generic[ExprT]):
         return self._expr.__class__(
             lambda plx: self._expr._to_compliant_expr(plx).str.replace_all(
                 pattern, value, literal=literal
-            )
+            ),
+            self._expr._is_order_dependent,
         )
 
     def strip_chars(self: Self, characters: str | None = None) -> ExprT:
@@ -255,7 +258,8 @@ class ExprStringNamespace(Generic[ExprT]):
             {'fruits': ['apple', '\nmango'], 'stripped': ['apple', 'mango']}
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.strip_chars(characters)
+            lambda plx: self._expr._to_compliant_expr(plx).str.strip_chars(characters),
+            self._expr._is_order_dependent,
         )
 
     def starts_with(self: Self, prefix: str) -> ExprT:
@@ -317,7 +321,8 @@ class ExprStringNamespace(Generic[ExprT]):
             has_prefix: [[true,false,null]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.starts_with(prefix)
+            lambda plx: self._expr._to_compliant_expr(plx).str.starts_with(prefix),
+            self._expr._is_order_dependent,
         )
 
     def ends_with(self: Self, suffix: str) -> ExprT:
@@ -379,7 +384,8 @@ class ExprStringNamespace(Generic[ExprT]):
             has_suffix: [[false,true,null]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.ends_with(suffix)
+            lambda plx: self._expr._to_compliant_expr(plx).str.ends_with(suffix),
+            self._expr._is_order_dependent,
         )
 
     def contains(self: Self, pattern: str, *, literal: bool = False) -> ExprT:
@@ -457,7 +463,8 @@ class ExprStringNamespace(Generic[ExprT]):
         return self._expr.__class__(
             lambda plx: self._expr._to_compliant_expr(plx).str.contains(
                 pattern, literal=literal
-            )
+            ),
+            self._expr._is_order_dependent,
         )
 
     def slice(self: Self, offset: int, length: int | None = None) -> ExprT:
@@ -559,7 +566,8 @@ class ExprStringNamespace(Generic[ExprT]):
         return self._expr.__class__(
             lambda plx: self._expr._to_compliant_expr(plx).str.slice(
                 offset=offset, length=length
-            )
+            ),
+            self._expr._is_order_dependent,
         )
 
     def head(self: Self, n: int = 5) -> ExprT:
@@ -626,7 +634,8 @@ class ExprStringNamespace(Generic[ExprT]):
             lyrics_head: [["Atata","taata","taata","zukky"]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.slice(0, n)
+            lambda plx: self._expr._to_compliant_expr(plx).str.slice(0, n),
+            self._expr._is_order_dependent,
         )
 
     def tail(self: Self, n: int = 5) -> ExprT:
@@ -695,7 +704,8 @@ class ExprStringNamespace(Generic[ExprT]):
         return self._expr.__class__(
             lambda plx: self._expr._to_compliant_expr(plx).str.slice(
                 offset=-n, length=None
-            )
+            ),
+            self._expr._is_order_dependent,
         )
 
     def to_datetime(self: Self, format: str | None = None) -> ExprT:  # noqa: A002
@@ -764,7 +774,8 @@ class ExprStringNamespace(Generic[ExprT]):
             a: [[2020-01-01 00:00:00.000000,2020-01-02 00:00:00.000000]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.to_datetime(format=format)
+            lambda plx: self._expr._to_compliant_expr(plx).str.to_datetime(format=format),
+            self._expr._is_order_dependent,
         )
 
     def to_uppercase(self: Self) -> ExprT:
@@ -828,7 +839,8 @@ class ExprStringNamespace(Generic[ExprT]):
             upper_col: [["APPLE","MANGO",null]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.to_uppercase()
+            lambda plx: self._expr._to_compliant_expr(plx).str.to_uppercase(),
+            self._expr._is_order_dependent,
         )
 
     def to_lowercase(self: Self) -> ExprT:
@@ -887,5 +899,6 @@ class ExprStringNamespace(Generic[ExprT]):
             lower_col: [["apple","mango",null]]
         """
         return self._expr.__class__(
-            lambda plx: self._expr._to_compliant_expr(plx).str.to_lowercase()
+            lambda plx: self._expr._to_compliant_expr(plx).str.to_lowercase(),
+            self._expr._is_order_dependent,
         )
diff --git a/narwhals/functions.py b/narwhals/functions.py
index f10a397f..9b3a0609 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -13,6 +13,7 @@ from typing import Union
 from typing import overload
 
 from narwhals._expression_parsing import extract_compliant
+from narwhals._expression_parsing import operation_is_order_dependent
 from narwhals._pandas_like.utils import broadcast_align_and_extract_native
 from narwhals.dataframe import DataFrame
 from narwhals.dataframe import LazyFrame
@@ -351,7 +352,7 @@ def _new_series_impl(
             dtype = arrow_narwhals_to_native_dtype(dtype, version=version)
         native_series = native_namespace.chunked_array([values], type=dtype)
 
-    elif implementation is Implementation.DASK:
+    elif implementation is Implementation.DASK:  # pragma: no cover
         msg = "Dask support in Narwhals is lazy-only, so `new_series` is " "not supported"
         raise NotImplementedError(msg)
     else:  # pragma: no cover
@@ -499,6 +500,7 @@ def _from_dict_impl(
         native_frame = native_namespace.DataFrame.from_dict(aligned_data)
 
         if schema:
+            from narwhals._pandas_like.utils import get_dtype_backend
             from narwhals._pandas_like.utils import (
                 narwhals_to_native_dtype as pandas_like_narwhals_to_native_dtype,
             )
@@ -506,7 +508,11 @@ def _from_dict_impl(
             backend_version = parse_version(native_namespace.__version__)
             schema = {
                 name: pandas_like_narwhals_to_native_dtype(
-                    schema[name], native_type, implementation, backend_version, version
+                    dtype=schema[name],
+                    dtype_backend=get_dtype_backend(native_type, implementation),
+                    implementation=implementation,
+                    backend_version=backend_version,
+                    version=version,
                 )
                 for name, native_type in native_frame.dtypes.items()
             }
@@ -724,6 +730,7 @@ def _from_numpy_impl(
         Implementation.CUDF,
     }:
         if isinstance(schema, (dict, Schema)):
+            from narwhals._pandas_like.utils import get_dtype_backend
             from narwhals._pandas_like.utils import (
                 narwhals_to_native_dtype as pandas_like_narwhals_to_native_dtype,
             )
@@ -731,7 +738,11 @@ def _from_numpy_impl(
             backend_version = parse_version(native_namespace.__version__)
             schema = {
                 name: pandas_like_narwhals_to_native_dtype(
-                    schema[name], native_type, implementation, backend_version, version
+                    dtype=schema[name],
+                    dtype_backend=get_dtype_backend(native_type, implementation),
+                    implementation=implementation,
+                    backend_version=backend_version,
+                    version=version,
                 )
                 for name, native_type in schema.items()
             }
@@ -1382,7 +1393,7 @@ def col(*names: str | Iterable[str]) -> Expr:
     def func(plx: Any) -> Any:
         return plx.col(*flatten(names))
 
-    return Expr(func)
+    return Expr(func, is_order_dependent=False)
 
 
 def nth(*indices: int | Sequence[int]) -> Expr:
@@ -1444,7 +1455,7 @@ def nth(*indices: int | Sequence[int]) -> Expr:
     def func(plx: Any) -> Any:
         return plx.nth(*flatten(indices))
 
-    return Expr(func)
+    return Expr(func, is_order_dependent=False)
 
 
 # Add underscore so it doesn't conflict with builtin `all`
@@ -1501,7 +1512,7 @@ def all_() -> Expr:
         a: [[2,4,6]]
         b: [[8,10,12]]
     """
-    return Expr(lambda plx: plx.all())
+    return Expr(lambda plx: plx.all(), is_order_dependent=False)
 
 
 # Add underscore so it doesn't conflict with builtin `len`
@@ -1554,7 +1565,7 @@ def len_() -> Expr:
     def func(plx: Any) -> Any:
         return plx.len()
 
-    return Expr(func)
+    return Expr(func, is_order_dependent=False)
 
 
 def sum(*columns: str) -> Expr:
@@ -1610,7 +1621,7 @@ def sum(*columns: str) -> Expr:
         ----
         a: [[3]]
     """
-    return Expr(lambda plx: plx.col(*columns).sum())
+    return Expr(lambda plx: plx.col(*columns).sum(), is_order_dependent=False)
 
 
 def mean(*columns: str) -> Expr:
@@ -1666,7 +1677,7 @@ def mean(*columns: str) -> Expr:
         ----
         a: [[4]]
     """
-    return Expr(lambda plx: plx.col(*columns).mean())
+    return Expr(lambda plx: plx.col(*columns).mean(), is_order_dependent=False)
 
 
 def median(*columns: str) -> Expr:
@@ -1724,7 +1735,7 @@ def median(*columns: str) -> Expr:
         ----
         a: [[4]]
     """
-    return Expr(lambda plx: plx.col(*columns).median())
+    return Expr(lambda plx: plx.col(*columns).median(), is_order_dependent=False)
 
 
 def min(*columns: str) -> Expr:
@@ -1780,7 +1791,7 @@ def min(*columns: str) -> Expr:
         ----
         b: [[5]]
     """
-    return Expr(lambda plx: plx.col(*columns).min())
+    return Expr(lambda plx: plx.col(*columns).min(), is_order_dependent=False)
 
 
 def max(*columns: str) -> Expr:
@@ -1836,7 +1847,7 @@ def max(*columns: str) -> Expr:
         ----
         a: [[2]]
     """
-    return Expr(lambda plx: plx.col(*columns).max())
+    return Expr(lambda plx: plx.col(*columns).max(), is_order_dependent=False)
 
 
 def sum_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
@@ -1899,10 +1910,10 @@ def sum_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `sum_horizontal`"
         raise ValueError(msg)
+    flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.sum_horizontal(
-            *[extract_compliant(plx, v) for v in flatten(exprs)]
-        )
+        lambda plx: plx.sum_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        is_order_dependent=operation_is_order_dependent(*flat_exprs),
     )
 
 
@@ -1969,10 +1980,10 @@ def min_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `min_horizontal`"
         raise ValueError(msg)
+    flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.min_horizontal(
-            *[extract_compliant(plx, v) for v in flatten(exprs)]
-        )
+        lambda plx: plx.min_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        is_order_dependent=operation_is_order_dependent(*flat_exprs),
     )
 
 
@@ -2039,10 +2050,10 @@ def max_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `max_horizontal`"
         raise ValueError(msg)
+    flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.max_horizontal(
-            *[extract_compliant(plx, v) for v in flatten(exprs)]
-        )
+        lambda plx: plx.max_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        is_order_dependent=operation_is_order_dependent(*flat_exprs),
     )
 
 
@@ -2056,20 +2067,22 @@ class When:
     def _extract_predicates(self, plx: Any) -> Any:
         return [extract_compliant(plx, v) for v in self._predicates]
 
-    def then(self, value: Any) -> Then:
+    def then(self, value: IntoExpr | Any) -> Then:
         return Then(
             lambda plx: plx.when(*self._extract_predicates(plx)).then(
                 extract_compliant(plx, value)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(*self._predicates, value),
         )
 
 
 class Then(Expr):
-    def otherwise(self, value: Any) -> Expr:
+    def otherwise(self, value: IntoExpr | Any) -> Expr:
         return Expr(
             lambda plx: self._to_compliant_expr(plx).otherwise(
                 extract_compliant(plx, value)
-            )
+            ),
+            is_order_dependent=operation_is_order_dependent(self, value),
         )
 
 
@@ -2219,10 +2232,10 @@ def all_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `all_horizontal`"
         raise ValueError(msg)
+    flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.all_horizontal(
-            *[extract_compliant(plx, v) for v in flatten(exprs)]
-        )
+        lambda plx: plx.all_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        is_order_dependent=operation_is_order_dependent(*flat_exprs),
     )
 
 
@@ -2293,7 +2306,7 @@ def lit(value: Any, dtype: DType | type[DType] | None = None) -> Expr:
         msg = f"Nested datatypes are not supported yet. Got {value}"
         raise NotImplementedError(msg)
 
-    return Expr(lambda plx: plx.lit(value, dtype))
+    return Expr(lambda plx: plx.lit(value, dtype), is_order_dependent=False)
 
 
 def any_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
@@ -2367,10 +2380,10 @@ def any_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `any_horizontal`"
         raise ValueError(msg)
+    flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.any_horizontal(
-            *[extract_compliant(plx, v) for v in flatten(exprs)]
-        )
+        lambda plx: plx.any_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        is_order_dependent=operation_is_order_dependent(*flat_exprs),
     )
 
 
@@ -2437,10 +2450,10 @@ def mean_horizontal(*exprs: IntoExpr | Iterable[IntoExpr]) -> Expr:
     if not exprs:
         msg = "At least one expression must be passed to `mean_horizontal`"
         raise ValueError(msg)
+    flat_exprs = flatten(exprs)
     return Expr(
-        lambda plx: plx.mean_horizontal(
-            *[extract_compliant(plx, v) for v in flatten(exprs)]
-        )
+        lambda plx: plx.mean_horizontal(*[extract_compliant(plx, v) for v in flat_exprs]),
+        is_order_dependent=operation_is_order_dependent(*flat_exprs),
     )
 
 
@@ -2522,11 +2535,13 @@ def concat_str(
         ----
         full_sentence: [["2 dogs play","4 cats swim",null]]
     """
+    flat_exprs = flatten([exprs])
     return Expr(
         lambda plx: plx.concat_str(
-            [extract_compliant(plx, v) for v in flatten([exprs])],
+            [extract_compliant(plx, v) for v in flat_exprs],
             *[extract_compliant(plx, v) for v in more_exprs],
             separator=separator,
             ignore_nulls=ignore_nulls,
-        )
+        ),
+        is_order_dependent=operation_is_order_dependent(*flat_exprs, *more_exprs),
     )
diff --git a/narwhals/selectors.py b/narwhals/selectors.py
index 31a5f80e..664fdc9c 100644
--- a/narwhals/selectors.py
+++ b/narwhals/selectors.py
@@ -52,7 +52,9 @@ def by_dtype(*dtypes: Any) -> Expr:
         │ 4   ┆ 4.6 │
         └─────┴─────┘
     """
-    return Selector(lambda plx: plx.selectors.by_dtype(flatten(dtypes)))
+    return Selector(
+        lambda plx: plx.selectors.by_dtype(flatten(dtypes)), is_order_dependent=False
+    )
 
 
 def numeric() -> Expr:
@@ -95,7 +97,7 @@ def numeric() -> Expr:
         │ 4   ┆ 4.6 │
         └─────┴─────┘
     """
-    return Selector(lambda plx: plx.selectors.numeric())
+    return Selector(lambda plx: plx.selectors.numeric(), is_order_dependent=False)
 
 
 def boolean() -> Expr:
@@ -138,7 +140,7 @@ def boolean() -> Expr:
         │ true  │
         └───────┘
     """
-    return Selector(lambda plx: plx.selectors.boolean())
+    return Selector(lambda plx: plx.selectors.boolean(), is_order_dependent=False)
 
 
 def string() -> Expr:
@@ -181,7 +183,7 @@ def string() -> Expr:
         │ y   │
         └─────┘
     """
-    return Selector(lambda plx: plx.selectors.string())
+    return Selector(lambda plx: plx.selectors.string(), is_order_dependent=False)
 
 
 def categorical() -> Expr:
@@ -224,7 +226,7 @@ def categorical() -> Expr:
         │ y   │
         └─────┘
     """
-    return Selector(lambda plx: plx.selectors.categorical())
+    return Selector(lambda plx: plx.selectors.categorical(), is_order_dependent=False)
 
 
 def all() -> Expr:
@@ -267,7 +269,7 @@ def all() -> Expr:
         │ 2   ┆ y   ┆ true  │
         └─────┴─────┴───────┘
     """
-    return Selector(lambda plx: plx.selectors.all())
+    return Selector(lambda plx: plx.selectors.all(), is_order_dependent=False)
 
 
 __all__ = [
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index 171c5291..941f1637 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -17,6 +17,7 @@ from narwhals import exceptions
 from narwhals import selectors
 from narwhals.dataframe import DataFrame as NwDataFrame
 from narwhals.dataframe import LazyFrame as NwLazyFrame
+from narwhals.dependencies import get_polars
 from narwhals.expr import Expr as NwExpr
 from narwhals.functions import Then as NwThen
 from narwhals.functions import When as NwWhen
@@ -238,6 +239,31 @@ class LazyFrame(NwLazyFrame[IntoFrameT]):
     def _dataframe(self) -> type[DataFrame[Any]]:
         return DataFrame
 
+    def _extract_compliant(self, arg: Any) -> Any:
+        # After v1, we raise when passing order-dependent
+        # expressions to LazyFrame
+        from narwhals.dataframe import BaseFrame
+        from narwhals.expr import Expr
+        from narwhals.series import Series
+
+        if isinstance(arg, BaseFrame):
+            return arg._compliant_frame
+        if isinstance(arg, Series):  # pragma: no cover
+            msg = "Mixing Series with LazyFrame is not supported."
+            raise TypeError(msg)
+        if isinstance(arg, Expr):
+            # After stable.v1, we raise if arg._is_order_dependent
+            return arg._to_compliant_expr(self.__narwhals_namespace__())
+        if get_polars() is not None and "polars" in str(type(arg)):  # pragma: no cover
+            msg = (
+                f"Expected Narwhals object, got: {type(arg)}.\n\n"
+                "Perhaps you:\n"
+                "- Forgot a `nw.from_native` somewhere?\n"
+                "- Used `pl.col` instead of `nw.col`?"
+            )
+            raise TypeError(msg)
+        return arg
+
     def collect(self) -> DataFrame[Any]:
         r"""Materialize this LazyFrame into a DataFrame.
 
@@ -836,6 +862,98 @@ class Expr(NwExpr):
             ddof=ddof,
         )
 
+    def head(self, n: int = 10) -> Self:
+        r"""Get the first `n` rows.
+
+        Arguments:
+            n: Number of rows to return.
+
+        Returns:
+            A new expression.
+        """
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).head(n), is_order_dependent=True
+        )
+
+    def tail(self, n: int = 10) -> Self:
+        r"""Get the last `n` rows.
+
+        Arguments:
+            n: Number of rows to return.
+
+        Returns:
+            A new expression.
+        """
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).tail(n), is_order_dependent=True
+        )
+
+    def gather_every(self: Self, n: int, offset: int = 0) -> Self:
+        r"""Take every nth value in the Series and return as new Series.
+
+        Arguments:
+            n: Gather every *n*-th row.
+            offset: Starting index.
+
+        Returns:
+            A new expression.
+        """
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).gather_every(n=n, offset=offset),
+            is_order_dependent=True,
+        )
+
+    def sort(self, *, descending: bool = False, nulls_last: bool = False) -> Self:
+        """Sort this column. Place null values first.
+
+        Arguments:
+            descending: Sort in descending order.
+            nulls_last: Place null values last instead of first.
+
+        Returns:
+            A new expression.
+        """
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).sort(
+                descending=descending, nulls_last=nulls_last
+            ),
+            is_order_dependent=True,
+        )
+
+    def sample(
+        self: Self,
+        n: int | None = None,
+        *,
+        fraction: float | None = None,
+        with_replacement: bool = False,
+        seed: int | None = None,
+    ) -> Self:
+        """Sample randomly from this expression.
+
+        !!! warning
+            `Expr.sample` is deprecated and will be removed in a future version.
+            Hint: instead of `df.select(nw.col('a').sample())`, use
+            `df.select(nw.col('a')).sample()` instead.
+            Note: this will remain available in `narwhals.stable.v1`.
+            See [stable api](../backcompat.md/) for more information.
+
+        Arguments:
+            n: Number of items to return. Cannot be used with fraction.
+            fraction: Fraction of items to return. Cannot be used with n.
+            with_replacement: Allow values to be sampled more than once.
+            seed: Seed for the random number generator. If set to None (default), a random
+                seed is generated for each sample operation.
+
+        Returns:
+            A new expression.
+        """
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).sample(
+                n, fraction=fraction, with_replacement=with_replacement, seed=seed
+            ),
+            is_order_dependent=True,
+        )
+
 
 class Schema(NwSchema):
     """Ordered mapping of column names to their data type.
@@ -878,7 +996,7 @@ def _stableify(
             level=obj._level,
         )
     if isinstance(obj, NwExpr):
-        return Expr(obj._to_compliant_expr)
+        return Expr(obj._to_compliant_expr, is_order_dependent=obj._is_order_dependent)
     return obj
 
 
@@ -1836,7 +1954,7 @@ class When(NwWhen):
 class Then(NwThen, Expr):
     @classmethod
     def from_then(cls, then: NwThen) -> Self:
-        return cls(then._to_compliant_expr)
+        return cls(then._to_compliant_expr, is_order_dependent=then._is_order_dependent)
 
     def otherwise(self, value: Any) -> Expr:
         return _stableify(super().otherwise(value))
diff --git a/narwhals/utils.py b/narwhals/utils.py
index 73f0e20a..c913fceb 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -388,12 +388,14 @@ def parse_version(version: str) -> tuple[int, ...]:
     return tuple(int(re.sub(r"\D", "", str(v))) for v in version.split("."))
 
 
-def isinstance_or_issubclass(obj: Any, cls: Any) -> bool:
+def isinstance_or_issubclass(obj_or_cls: object | type, cls_or_tuple: Any) -> bool:
     from narwhals.dtypes import DType
 
-    if isinstance(obj, DType):
-        return isinstance(obj, cls)
-    return isinstance(obj, cls) or (isinstance(obj, type) and issubclass(obj, cls))
+    if isinstance(obj_or_cls, DType):
+        return isinstance(obj_or_cls, cls_or_tuple)
+    return isinstance(obj_or_cls, cls_or_tuple) or (
+        isinstance(obj_or_cls, type) and issubclass(obj_or_cls, cls_or_tuple)
+    )
 
 
 def validate_laziness(items: Iterable[Any]) -> None:
@@ -981,7 +983,7 @@ def validate_strict_and_pass_though(
             msg = (
                 "`strict` in `from_native` is deprecated, please use `pass_through` instead.\n\n"
                 "Note: `strict` will remain available in `narwhals.stable.v1`.\n"
-                "See [stable api](../backcompat.md/) for more information.\n"
+                "See https://narwhals-dev.github.io/narwhals/backcompat/ for more information.\n"
             )
             issue_deprecation_warning(msg, _version="1.13.0")
         pass_through = not strict
diff --git a/utils/check_api_reference.py b/utils/check_api_reference.py
index 1b186774..8df16e88 100644
--- a/utils/check_api_reference.py
+++ b/utils/check_api_reference.py
@@ -160,7 +160,9 @@ for namespace in NAMESPACES.difference({"name"}):
 
 # Expr methods
 expr_methods = [
-    i for i in nw.Expr(lambda: 0).__dir__() if not i[0].isupper() and i[0] != "_"
+    i
+    for i in nw.Expr(lambda: 0, is_order_dependent=False).__dir__()
+    if not i[0].isupper() and i[0] != "_"
 ]
 with open("docs/api-reference/expr.md") as fd:
     content = fd.read()
@@ -182,7 +184,9 @@ if extra := set(documented).difference(expr_methods):
 for namespace in NAMESPACES:
     expr_methods = [
         i
-        for i in getattr(nw.Expr(lambda: 0), namespace).__dir__()
+        for i in getattr(
+            nw.Expr(lambda: 0, is_order_dependent=False), namespace
+        ).__dir__()
         if not i[0].isupper() and i[0] != "_"
     ]
     with open(f"docs/api-reference/expr_{namespace}.md") as fd:
@@ -222,7 +226,11 @@ if extra := set(documented).difference(dtypes):
     ret = 1
 
 # Check Expr vs Series
-expr = [i for i in nw.Expr(lambda: 0).__dir__() if not i[0].isupper() and i[0] != "_"]
+expr = [
+    i
+    for i in nw.Expr(lambda: 0, is_order_dependent=False).__dir__()
+    if not i[0].isupper() and i[0] != "_"
+]
 series = [
     i
     for i in nw.from_native(pl.Series(), series_only=True).__dir__()
@@ -241,7 +249,9 @@ if extra := set(series).difference(expr).difference(SERIES_ONLY_METHODS):
 for namespace in NAMESPACES.difference({"name"}):
     expr_internal = [
         i
-        for i in getattr(nw.Expr(lambda: 0), namespace).__dir__()
+        for i in getattr(
+            nw.Expr(lambda: 0, is_order_dependent=False), namespace
+        ).__dir__()
         if not i[0].isupper() and i[0] != "_"
     ]
     series_internal = [


Test patch:
diff --git a/.github/workflows/downstream_tests.yml b/.github/workflows/downstream_tests.yml
index 5ad95b6d..647640dd 100644
--- a/.github/workflows/downstream_tests.yml
+++ b/.github/workflows/downstream_tests.yml
@@ -1,6 +1,6 @@
 name: Test Downstream Libraries - Fast
 
-on: 
+on:
   pull_request:
   push:
     branches: [main]
@@ -26,31 +26,31 @@ jobs:
           cache-dependency-glob: "pyproject.toml"
       - name: clone-altair
         run: |
-            git clone https://github.com/vega/altair.git --depth=1
-            cd altair
-            git log
+          git clone https://github.com/vega/altair.git --depth=1
+          cd altair
+          git log
       - name: install-basics
         run: uv pip install --upgrade tox virtualenv setuptools --system
       - name: install-altair-dev
         run: |
-            cd altair
-            uv pip install -e ".[dev, all]" --system
-            # temporary pin to get ci green
-            uv pip install "numpy<2.2" --system
+          cd altair
+          uv pip install -e ".[dev, all]" --system
+          # temporary pin to get ci green
+          uv pip install "numpy<2.2" --system
       - name: install-narwhals-dev
         run: |
-            uv pip uninstall narwhals --system
-            uv pip install -e . --system
+          uv pip uninstall narwhals --system
+          uv pip install -e . --system
       - name: show-deps
         run: uv pip freeze
       - name: Run pytest
         run: |
-            cd altair
-            pytest tests
+          cd altair
+          pytest tests
       - name: Run mypy
         run: |
-            cd altair
-            mypy altair tests
+          cd altair
+          mypy altair tests
 
   marimo:
     strategy:
@@ -73,45 +73,45 @@ jobs:
           cache-dependency-glob: "pyproject.toml"
       - name: clone-marimo
         run: |
-            git clone https://github.com/marimo-team/marimo.git --depth=1
-            cd marimo
-            uv venv -p 3.12
-            git log
+          git clone https://github.com/marimo-team/marimo.git --depth=1
+          cd marimo
+          uv venv -p 3.12
+          git log
       - name: install-basics
         run: uv pip install --upgrade tox virtualenv setuptools hatch --system
       - name: install-marimo-dev
         run: |
-            cd marimo
-            . .venv/bin/activate
-            uv pip install -e ".[dev]"
-            which python
+          cd marimo
+          . .venv/bin/activate
+          uv pip install -e ".[dev]"
+          which python
       - name: install-narwhals-dev
         run: |
-            cd marimo
-            . .venv/bin/activate
-            uv pip uninstall narwhals
-            uv pip install -e ./..
+          cd marimo
+          . .venv/bin/activate
+          uv pip uninstall narwhals
+          uv pip install -e ./..
       - name: show-deps
         run: |
-            cd marimo
-            . .venv/bin/activate
-            uv pip freeze
+          cd marimo
+          . .venv/bin/activate
+          uv pip freeze
       - name: Create assets directory, copy over index.html
         continue-on-error: true
         run: |
-            mkdir -p marimo/marimo/_static/assets
-            cp marimo/frontend/index.html marimo/marimo/_static/index.html
-            cp marimo/frontend/public/favicon.ico marimo/marimo/_static/favicon.ico
+          mkdir -p marimo/marimo/_static/assets
+          cp marimo/frontend/index.html marimo/marimo/_static/index.html
+          cp marimo/frontend/public/favicon.ico marimo/marimo/_static/favicon.ico
       - name: Run tests with full dependencies
         if: ${{ matrix.dependencies == 'core,optional' }}
         run: |
-            cd marimo
-            . .venv/bin/activate
-            # make sure that we use the .venv when running tests, so that
-            # the local narwhals install is picked up
-            sed -i '/^\[tool.hatch.envs.default\]/a path = ".venv"' pyproject.toml
-            hatch run python -c "import narwhals; print(narwhals.__file__)"
-            hatch run test-optional:test-narwhals
+          cd marimo
+          . .venv/bin/activate
+          # make sure that we use the .venv when running tests, so that
+          # the local narwhals install is picked up
+          sed -i '/^\[tool.hatch.envs.default\]/a path = ".venv"' pyproject.toml
+          hatch run python -c "import narwhals; print(narwhals.__file__)"
+          hatch run test-optional:test-narwhals
         timeout-minutes: 15
 
   scikit-lego:
@@ -138,18 +138,18 @@ jobs:
         run: uv pip install --upgrade tox virtualenv setuptools --system
       - name: install-scikit-lego-dev
         run: |
-            cd scikit-lego
-            uv pip install -e ".[test]" --system
+          cd scikit-lego
+          uv pip install -e ".[test]" --system
       - name: install-narwhals-dev
         run: |
-            uv pip uninstall narwhals --system
-            uv pip install -e . --system
+          uv pip uninstall narwhals --system
+          uv pip install -e . --system
       - name: show-deps
         run: uv pip freeze
       - name: Run pytest
         run: |
-            cd scikit-lego
-            pytest -n auto --disable-warnings --cov=sklego -m "not cvxpy and not formulaic and not umap"
+          cd scikit-lego
+          pytest -n auto --disable-warnings --cov=sklego -m "not cvxpy and not formulaic and not umap"
 
   shiny:
     strategy:
@@ -171,28 +171,28 @@ jobs:
           cache-dependency-glob: "pyproject.toml"
       - name: clone-shiny
         run: |
-            git clone https://github.com/posit-dev/py-shiny.git
-            cd py-shiny
-            git log
+          git clone https://github.com/posit-dev/py-shiny.git
+          cd py-shiny
+          git log
       - name: install-basics
         run: uv pip install --upgrade tox virtualenv setuptools --system
       - name: install-shiny-dev
         env:
           UV_SYSTEM_PYTHON: 1
         run: |
-            cd py-shiny
-            make narwhals-install-shiny
+          cd py-shiny
+          make narwhals-install-shiny
       - name: install-narwhals-dev
         run: |
-            uv pip uninstall narwhals --system
-            uv pip install -e . --system
+          uv pip uninstall narwhals --system
+          uv pip install -e . --system
       - name: show-deps
         run: uv pip freeze
       - name: Run `make narwhals-test-integration`
         run: |
-            cd py-shiny
-            make narwhals-test-integration
-  
+          cd py-shiny
+          make narwhals-test-integration
+
   tea-tasting:
     strategy:
       matrix:
@@ -203,9 +203,9 @@ jobs:
       - uses: actions/checkout@v4
       - name: clone-tea-tasting
         run: |
-            git clone https://github.com/e10v/tea-tasting.git --depth=1
-            cd tea-tasting
-            git log
+          git clone https://github.com/e10v/tea-tasting.git --depth=1
+          cd tea-tasting
+          git log
       - name: install-pdm
         uses: pdm-project/setup-pdm@v4
         with:
@@ -214,22 +214,22 @@ jobs:
           cache-dependency-path: tea-tasting/pdm.lock
       - name: install-tea-tasting-test
         run: |
-            cd tea-tasting
-            pdm sync -G test
+          cd tea-tasting
+          pdm sync -G test
       - name: install-narwhals-dev
         run: |
-            cd tea-tasting
-            pdm remove narwhals
-            pdm add ./..[dev]
+          cd tea-tasting
+          pdm remove narwhals
+          pdm add ./..[dev]
       - name: show-deps
         run: |
-            cd tea-tasting
-            pdm list
+          cd tea-tasting
+          pdm list
       - name: run-pytest
         run: |
-            cd tea-tasting
-            pdm run test
-  
+          cd tea-tasting
+          pdm run test
+
   tubular:
     strategy:
       matrix:
@@ -250,25 +250,25 @@ jobs:
           cache-dependency-glob: "pyproject.toml"
       - name: clone-tubular
         run: |
-            git clone https://github.com/lvgig/tubular --depth=1
-            cd tubular
-            git log
+          git clone https://github.com/lvgig/tubular --depth=1
+          cd tubular
+          git log
       - name: install-basics
         run: uv pip install --upgrade tox virtualenv setuptools pytest-env --system
       - name: install-tubular-dev
         run: |
-            cd tubular
-            uv pip install -e ".[dev]" --system
+          cd tubular
+          uv pip install -e ".[dev]" --system
       - name: install-narwhals-dev
         run: |
-            uv pip uninstall narwhals --system
-            uv pip install -e . --system
+          uv pip uninstall narwhals --system
+          uv pip install -e . --system
       - name: show-deps
         run: uv pip freeze
       - name: Run pytest
         run: |
-            cd tubular
-            pytest tests --config-file=pyproject.toml
+          cd tubular
+          pytest tests --config-file=pyproject.toml
 
   plotly:
     strategy:
@@ -290,25 +290,25 @@ jobs:
           cache-dependency-glob: "pyproject.toml"
       - name: clone-plotly
         run: |
-            git clone https://github.com/plotly/plotly.py --depth=1
-            cd plotly.py
-            git log
+          git clone https://github.com/plotly/plotly.py --depth=1
+          cd plotly.py
+          git log
       - name: install-basics
         run: uv pip install --upgrade tox virtualenv setuptools --system
       - name: install-deps
         run: |
-            cd plotly.py
-            uv pip install -r packages/python/plotly/test_requirements/requirements_312_optional.txt --system
+          cd plotly.py
+          uv pip install -r packages/python/plotly/test_requirements/requirements_312_optional.txt --system
       - name: install-narwhals-dev
         run: |
-            uv pip uninstall narwhals --system
-            uv pip install -e . --system
+          uv pip uninstall narwhals --system
+          uv pip install -e . --system
       - name: show-deps
         run: uv pip freeze
       - name: Run pytest on plotly express
         run: |
-            cd plotly.py
-            pytest packages/python/plotly/plotly/tests/test_optional/test_px
+          cd plotly.py
+          pytest packages/python/plotly/plotly/tests/test_optional/test_px
 
   hierarchicalforecast:
     strategy:
@@ -330,22 +330,65 @@ jobs:
           cache-dependency-glob: "pyproject.toml"
       - name: clone-hierarchicalforecast
         run: |
-            git clone https://github.com/Nixtla/hierarchicalforecast --depth=1
-            cd hierarchicalforecast
-            git log
+          git clone https://github.com/Nixtla/hierarchicalforecast --depth=1
+          cd hierarchicalforecast
+          git log
       - name: install-basics
         run: uv pip install --upgrade tox virtualenv setuptools --system
       - name: install-deps
         run: |
-            cd hierarchicalforecast
-            uv pip install --system ".[dev]"
+          cd hierarchicalforecast
+          uv pip install --system ".[dev]"
       - name: install-narwhals-dev
         run: |
-            uv pip uninstall narwhals --system
-            uv pip install -e . --system
+          uv pip uninstall narwhals --system
+          uv pip install -e . --system
       - name: show-deps
         run: uv pip freeze
       - name: Run tests
         run: |
-            cd hierarchicalforecast
-            nbdev_test --do_print --timing --n_workers 0
+          cd hierarchicalforecast
+          nbdev_test --do_print --timing --n_workers 0
+
+  formulaic:
+    strategy:
+      matrix:
+        python-version: ["3.12"]
+        os: [ubuntu-latest]
+
+    runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+        with:
+          enable-cache: "true"
+          cache-suffix: ${{ matrix.python-version }}
+          cache-dependency-glob: "pyproject.toml"
+      - name: clone-formulaic
+        run: |
+          git clone https://github.com/matthewwardrop/formulaic.git --depth=1
+          cd formulaic
+          git log
+      - name: create-hatch-env
+        run: |
+          cd formulaic
+          uv pip install hatch hatch-vcs --system
+      - name: install-formulaic-dev
+        run: |
+          cd formulaic
+          hatch run uv pip install -e ".[arrow,calculus]"
+      - name: install-narwhals-dev
+        run: |
+          cd formulaic
+          hatch run uv pip uninstall narwhals
+          hatch run uv pip install -e ./..
+      - name: show-deps
+        run: hatch run uv pip freeze
+      - name: run test
+        run: |
+          cd formulaic
+          hatch run tests --config-file=pyproject.toml
diff --git a/tests/conftest.py b/tests/conftest.py
index 497103d6..95b969e9 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -152,6 +152,8 @@ def pyspark_lazy_constructor() -> Callable[[Any], IntoFrame]:  # pragma: no cove
             # executing one task at a time makes the tests faster
             .config("spark.default.parallelism", "1")
             .config("spark.sql.shuffle.partitions", "2")
+            # common timezone for all tests environments
+            .config("spark.sql.session.timeZone", "UTC")
             .getOrCreate()
         )
 
@@ -232,7 +234,7 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
         if (
             any(
                 x in str(metafunc.module)
-                for x in ("list", "name", "unpivot", "from_dict", "from_numpy", "tail")
+                for x in ("list", "unpivot", "from_dict", "from_numpy", "tail")
             )
             and LAZY_CONSTRUCTORS["duckdb"] in constructors
         ):
diff --git a/tests/expr_and_series/cast_test.py b/tests/expr_and_series/cast_test.py
index ba2b8249..6dbda790 100644
--- a/tests/expr_and_series/cast_test.py
+++ b/tests/expr_and_series/cast_test.py
@@ -17,7 +17,7 @@ from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 from tests.utils import is_windows
 
-data = {
+DATA = {
     "a": [1],
     "b": [1],
     "c": [1],
@@ -35,7 +35,7 @@ data = {
     "o": ["a"],
     "p": [1],
 }
-schema = {
+SCHEMA = {
     "a": nw.Int64,
     "b": nw.Int32,
     "c": nw.Int16,
@@ -54,13 +54,15 @@ schema = {
     "p": nw.Int64,
 }
 
+SPARK_INCOMPATIBLE_COLUMNS = {"e", "f", "g", "h", "l", "o", "p"}
+
 
 @pytest.mark.filterwarnings("ignore:casting period[M] values to int64:FutureWarning")
 def test_cast(
     constructor: Constructor,
     request: pytest.FixtureRequest,
 ) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
     if "pyarrow_table_constructor" in str(constructor) and PYARROW_VERSION <= (
         15,
@@ -69,28 +71,20 @@ def test_cast(
     if "modin_constructor" in str(constructor):
         # TODO(unassigned): in modin, we end up with `'<U0'` dtype
         request.applymarker(pytest.mark.xfail)
+
+    if "pyspark" in str(constructor):
+        incompatible_columns = SPARK_INCOMPATIBLE_COLUMNS  # pragma: no cover
+    else:
+        incompatible_columns = set()
+
+    data = {c: v for c, v in DATA.items() if c not in incompatible_columns}
+    schema = {c: t for c, t in SCHEMA.items() if c not in incompatible_columns}
+
     df = nw.from_native(constructor(data)).select(
-        nw.col(key).cast(value) for key, value in schema.items()
-    )
-    result = df.select(
-        nw.col("a").cast(nw.Int32),
-        nw.col("b").cast(nw.Int16),
-        nw.col("c").cast(nw.Int8),
-        nw.col("d").cast(nw.Int64),
-        nw.col("e").cast(nw.UInt32),
-        nw.col("f").cast(nw.UInt16),
-        nw.col("g").cast(nw.UInt8),
-        nw.col("h").cast(nw.UInt64),
-        nw.col("i").cast(nw.Float32),
-        nw.col("j").cast(nw.Float64),
-        nw.col("k").cast(nw.String),
-        nw.col("l").cast(nw.Datetime),
-        nw.col("m").cast(nw.Int8),
-        nw.col("n").cast(nw.Int8),
-        nw.col("o").cast(nw.String),
-        nw.col("p").cast(nw.Duration),
+        nw.col(col_).cast(dtype) for col_, dtype in schema.items()
     )
-    expected = {
+
+    cast_map = {
         "a": nw.Int32,
         "b": nw.Int16,
         "c": nw.Int8,
@@ -108,7 +102,10 @@ def test_cast(
         "o": nw.String,
         "p": nw.Duration,
     }
-    assert dict(result.collect_schema()) == expected
+    cast_map = {c: t for c, t in cast_map.items() if c not in incompatible_columns}
+
+    result = df.select(*[nw.col(col_).cast(dtype) for col_, dtype in cast_map.items()])
+    assert dict(result.collect_schema()) == cast_map
 
 
 def test_cast_series(
@@ -123,8 +120,8 @@ def test_cast_series(
         # TODO(unassigned): in modin, we end up with `'<U0'` dtype
         request.applymarker(pytest.mark.xfail)
     df = (
-        nw.from_native(constructor_eager(data))
-        .select(nw.col(key).cast(value) for key, value in schema.items())
+        nw.from_native(constructor_eager(DATA))
+        .select(nw.col(key).cast(value) for key, value in SCHEMA.items())
         .lazy()
         .collect()
     )
@@ -180,11 +177,20 @@ def test_cast_string() -> None:
 def test_cast_raises_for_unknown_dtype(
     constructor: Constructor, request: pytest.FixtureRequest
 ) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
     if "pyarrow_table" in str(constructor) and PYARROW_VERSION < (15,):
         # Unsupported cast from string to dictionary using function cast_dictionary
         request.applymarker(pytest.mark.xfail)
+
+    if "pyspark" in str(constructor):
+        incompatible_columns = SPARK_INCOMPATIBLE_COLUMNS  # pragma: no cover
+    else:
+        incompatible_columns = set()
+
+    data = {k: v for k, v in DATA.items() if k not in incompatible_columns}
+    schema = {k: v for k, v in SCHEMA.items() if k not in incompatible_columns}
+
     df = nw.from_native(constructor(data)).select(
         nw.col(key).cast(value) for key, value in schema.items()
     )
@@ -204,7 +210,7 @@ def test_cast_datetime_tz_aware(
         or "duckdb" in str(constructor)
         or "cudf" in str(constructor)  # https://github.com/rapidsai/cudf/issues/16973
         or ("pyarrow_table" in str(constructor) and is_windows())
-        or ("pyspark" in str(constructor))
+        or "pyspark" in str(constructor)
     ):
         request.applymarker(pytest.mark.xfail)
 
diff --git a/tests/expr_and_series/cum_count_test.py b/tests/expr_and_series/cum_count_test.py
index dab77ebb..22a98344 100644
--- a/tests/expr_and_series/cum_count_test.py
+++ b/tests/expr_and_series/cum_count_test.py
@@ -3,7 +3,6 @@ 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
 
@@ -16,16 +15,9 @@ expected = {
 
 
 @pytest.mark.parametrize("reverse", [True, False])
-def test_cum_count_expr(
-    request: pytest.FixtureRequest, constructor: Constructor, *, reverse: bool
-) -> None:
-    if "dask" in str(constructor) and reverse:
-        request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_cum_count_expr(constructor_eager: ConstructorEager, *, reverse: bool) -> None:
     name = "reverse_cum_count" if reverse else "cum_count"
-    df = nw.from_native(constructor(data))
+    df = nw.from_native(constructor_eager(data))
     result = df.select(
         nw.col("a").cum_count(reverse=reverse).alias(name),
     )
diff --git a/tests/expr_and_series/cum_max_test.py b/tests/expr_and_series/cum_max_test.py
index 3df5a6ad..c5ec38b4 100644
--- a/tests/expr_and_series/cum_max_test.py
+++ b/tests/expr_and_series/cum_max_test.py
@@ -5,7 +5,6 @@ import pytest
 import narwhals.stable.v1 as nw
 from tests.utils import PANDAS_VERSION
 from tests.utils import PYARROW_VERSION
-from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -19,23 +18,18 @@ expected = {
 
 @pytest.mark.parametrize("reverse", [True, False])
 def test_cum_max_expr(
-    request: pytest.FixtureRequest, constructor: Constructor, *, reverse: bool
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager, *, reverse: bool
 ) -> None:
-    if "dask" in str(constructor) and reverse:
-        request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
-    if PYARROW_VERSION < (13, 0, 0) and "pyarrow_table" in str(constructor):
+    if PYARROW_VERSION < (13, 0, 0) and "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
 
     if (PANDAS_VERSION < (2, 1) or PYARROW_VERSION < (13,)) and "pandas_pyarrow" in str(
-        constructor
+        constructor_eager
     ):
         request.applymarker(pytest.mark.xfail)
 
     name = "reverse_cum_max" if reverse else "cum_max"
-    df = nw.from_native(constructor(data))
+    df = nw.from_native(constructor_eager(data))
     result = df.select(
         nw.col("a").cum_max(reverse=reverse).alias(name),
     )
diff --git a/tests/expr_and_series/cum_min_test.py b/tests/expr_and_series/cum_min_test.py
index a758dc8b..13791480 100644
--- a/tests/expr_and_series/cum_min_test.py
+++ b/tests/expr_and_series/cum_min_test.py
@@ -5,7 +5,6 @@ import pytest
 import narwhals.stable.v1 as nw
 from tests.utils import PANDAS_VERSION
 from tests.utils import PYARROW_VERSION
-from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -19,23 +18,18 @@ expected = {
 
 @pytest.mark.parametrize("reverse", [True, False])
 def test_cum_min_expr(
-    request: pytest.FixtureRequest, constructor: Constructor, *, reverse: bool
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager, *, reverse: bool
 ) -> None:
-    if "dask" in str(constructor) and reverse:
-        request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
-    if PYARROW_VERSION < (13, 0, 0) and "pyarrow_table" in str(constructor):
+    if PYARROW_VERSION < (13, 0, 0) and "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
 
     if (PANDAS_VERSION < (2, 1) or PYARROW_VERSION < (13,)) and "pandas_pyarrow" in str(
-        constructor
+        constructor_eager
     ):
         request.applymarker(pytest.mark.xfail)
 
     name = "reverse_cum_min" if reverse else "cum_min"
-    df = nw.from_native(constructor(data))
+    df = nw.from_native(constructor_eager(data))
     result = df.select(
         nw.col("a").cum_min(reverse=reverse).alias(name),
     )
diff --git a/tests/expr_and_series/cum_prod_test.py b/tests/expr_and_series/cum_prod_test.py
index 2d6861b8..0257d928 100644
--- a/tests/expr_and_series/cum_prod_test.py
+++ b/tests/expr_and_series/cum_prod_test.py
@@ -5,7 +5,6 @@ import pytest
 import narwhals.stable.v1 as nw
 from tests.utils import PANDAS_VERSION
 from tests.utils import PYARROW_VERSION
-from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -19,23 +18,18 @@ expected = {
 
 @pytest.mark.parametrize("reverse", [True, False])
 def test_cum_prod_expr(
-    request: pytest.FixtureRequest, constructor: Constructor, *, reverse: bool
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager, *, reverse: bool
 ) -> None:
-    if "dask" in str(constructor) and reverse:
-        request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
-    if PYARROW_VERSION < (13, 0, 0) and "pyarrow_table" in str(constructor):
+    if PYARROW_VERSION < (13, 0, 0) and "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
 
     if (PANDAS_VERSION < (2, 1) or PYARROW_VERSION < (13,)) and "pandas_pyarrow" in str(
-        constructor
+        constructor_eager
     ):
         request.applymarker(pytest.mark.xfail)
 
     name = "reverse_cum_prod" if reverse else "cum_prod"
-    df = nw.from_native(constructor(data))
+    df = nw.from_native(constructor_eager(data))
     result = df.select(
         nw.col("a").cum_prod(reverse=reverse).alias(name),
     )
diff --git a/tests/expr_and_series/cum_sum_test.py b/tests/expr_and_series/cum_sum_test.py
index 8a419c9a..eb1b985b 100644
--- a/tests/expr_and_series/cum_sum_test.py
+++ b/tests/expr_and_series/cum_sum_test.py
@@ -3,7 +3,6 @@ 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
 
@@ -15,16 +14,9 @@ expected = {
 
 
 @pytest.mark.parametrize("reverse", [True, False])
-def test_cum_sum_expr(
-    request: pytest.FixtureRequest, constructor: Constructor, *, reverse: bool
-) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    if "dask" in str(constructor) and reverse:
-        request.applymarker(pytest.mark.xfail)
-
+def test_cum_sum_expr(constructor_eager: ConstructorEager, *, reverse: bool) -> None:
     name = "reverse_cum_sum" if reverse else "cum_sum"
-    df = nw.from_native(constructor(data))
+    df = nw.from_native(constructor_eager(data))
     result = df.select(
         nw.col("a").cum_sum(reverse=reverse).alias(name),
     )
diff --git a/tests/expr_and_series/diff_test.py b/tests/expr_and_series/diff_test.py
index f7730a2d..0114ba19 100644
--- a/tests/expr_and_series/diff_test.py
+++ b/tests/expr_and_series/diff_test.py
@@ -4,7 +4,6 @@ import pytest
 
 import narwhals.stable.v1 as nw
 from tests.utils import PYARROW_VERSION
-from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -16,15 +15,15 @@ data = {
 
 
 def test_diff(
-    constructor: Constructor,
+    constructor_eager: ConstructorEager,
     request: pytest.FixtureRequest,
 ) -> None:
-    if "pyarrow_table_constructor" in str(constructor) and PYARROW_VERSION < (13,):
+    if "pyarrow_table_constructor" in str(constructor_eager) and PYARROW_VERSION < (13,):
         # pc.pairwisediff is available since pyarrow 13.0.0
         request.applymarker(pytest.mark.xfail)
-    if any(x in str(constructor) for x in ("duckdb", "pyspark")):
+    if any(x in str(constructor_eager) for x in ("duckdb", "pyspark")):
         request.applymarker(pytest.mark.xfail)
-    df = nw.from_native(constructor(data))
+    df = nw.from_native(constructor_eager(data))
     result = df.with_columns(c_diff=nw.col("c").diff()).filter(nw.col("i") > 0)
     expected = {
         "i": [1, 2, 3, 4],
diff --git a/tests/expr_and_series/gather_every_test.py b/tests/expr_and_series/gather_every_test.py
index fdaaac08..bcda13d8 100644
--- a/tests/expr_and_series/gather_every_test.py
+++ b/tests/expr_and_series/gather_every_test.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import pytest
 
+import narwhals as nw_main
 import narwhals.stable.v1 as nw
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
@@ -21,6 +22,9 @@ def test_gather_every_expr(
 
     assert_equal_data(result, expected)
 
+    with pytest.deprecated_call():
+        df.select(nw_main.col("a").gather_every(n=n, offset=offset))
+
 
 @pytest.mark.parametrize("n", [1, 2, 3])
 @pytest.mark.parametrize("offset", [1, 2, 3])
diff --git a/tests/expr_and_series/head_test.py b/tests/expr_and_series/head_test.py
index 49cc4124..952b3de1 100644
--- a/tests/expr_and_series/head_test.py
+++ b/tests/expr_and_series/head_test.py
@@ -2,7 +2,8 @@ from __future__ import annotations
 
 import pytest
 
-import narwhals as nw
+import narwhals as nw_main
+import narwhals.stable.v1 as nw
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -18,6 +19,9 @@ def test_head(
     expected = {"a": [1, 2]}
     assert_equal_data(result, expected)
 
+    with pytest.deprecated_call():
+        df.select(nw_main.col("a").head(5))
+
 
 @pytest.mark.parametrize("n", [2, -1])
 def test_head_series(constructor_eager: ConstructorEager, n: int) -> None:
diff --git a/tests/expr_and_series/is_first_distinct_test.py b/tests/expr_and_series/is_first_distinct_test.py
index 6870c339..21ded6aa 100644
--- a/tests/expr_and_series/is_first_distinct_test.py
+++ b/tests/expr_and_series/is_first_distinct_test.py
@@ -1,9 +1,6 @@
 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
 
@@ -13,12 +10,8 @@ data = {
 }
 
 
-def test_is_first_distinct_expr(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    df = nw.from_native(constructor(data))
+def test_is_first_distinct_expr(constructor_eager: ConstructorEager) -> None:
+    df = nw.from_native(constructor_eager(data))
     result = df.select(nw.all().is_first_distinct())
     expected = {
         "a": [True, False, True, True, False],
diff --git a/tests/expr_and_series/is_last_distinct_test.py b/tests/expr_and_series/is_last_distinct_test.py
index 9362cd02..4bfa7e13 100644
--- a/tests/expr_and_series/is_last_distinct_test.py
+++ b/tests/expr_and_series/is_last_distinct_test.py
@@ -1,9 +1,6 @@
 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
 
@@ -13,12 +10,8 @@ data = {
 }
 
 
-def test_is_last_distinct_expr(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    df = nw.from_native(constructor(data))
+def test_is_last_distinct_expr(constructor_eager: ConstructorEager) -> None:
+    df = nw.from_native(constructor_eager(data))
     result = df.select(nw.all().is_last_distinct())
     expected = {
         "a": [False, True, False, True, True],
diff --git a/tests/expr_and_series/mode_test.py b/tests/expr_and_series/mode_test.py
index a78a442f..3a9c1eee 100644
--- a/tests/expr_and_series/mode_test.py
+++ b/tests/expr_and_series/mode_test.py
@@ -4,7 +4,6 @@ import pytest
 
 import narwhals.stable.v1 as nw
 from tests.utils import POLARS_VERSION
-from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
@@ -14,7 +13,7 @@ data = {
 }
 
 
-def test_mode_single_expr(constructor_eager: Constructor) -> None:
+def test_mode_single_expr(constructor_eager: ConstructorEager) -> None:
     df = nw.from_native(constructor_eager(data))
     result = df.select(nw.col("a").mode()).sort("a")
     expected = {"a": [1, 2]}
diff --git a/tests/expr_and_series/name/keep_test.py b/tests/expr_and_series/name/keep_test.py
index e382db73..9c172dba 100644
--- a/tests/expr_and_series/name/keep_test.py
+++ b/tests/expr_and_series/name/keep_test.py
@@ -6,40 +6,28 @@ import polars as pl
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import AnonymousExprError
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
 data = {"foo": [1, 2, 3], "BAR": [4, 5, 6]}
 
 
-def test_keep(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_keep(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo", "BAR") * 2).name.keep())
     expected = {k: [e * 2 for e in v] for k, v in data.items()}
     assert_equal_data(result, expected)
 
 
-def test_keep_after_alias(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_keep_after_alias(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo")).alias("alias_for_foo").name.keep())
     expected = {"foo": data["foo"]}
     assert_equal_data(result, expected)
 
 
-def test_keep_raise_anonymous(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_keep_raise_anonymous(constructor: Constructor) -> None:
     df_raw = constructor(data)
     df = nw.from_native(df_raw)
 
@@ -47,7 +35,7 @@ def test_keep_raise_anonymous(
         does_not_raise()
         if isinstance(df_raw, (pl.LazyFrame, pl.DataFrame))
         else pytest.raises(
-            ValueError,
+            AnonymousExprError,
             match="Anonymous expressions are not supported in `.name.keep`.",
         )
     )
diff --git a/tests/expr_and_series/name/map_test.py b/tests/expr_and_series/name/map_test.py
index 276138ef..5b93de21 100644
--- a/tests/expr_and_series/name/map_test.py
+++ b/tests/expr_and_series/name/map_test.py
@@ -6,6 +6,7 @@ import polars as pl
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import AnonymousExprError
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
@@ -16,34 +17,21 @@ def map_func(s: str | None) -> str:
     return str(s)[::-1].lower()
 
 
-def test_map(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_map(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo", "BAR") * 2).name.map(function=map_func))
     expected = {map_func(k): [e * 2 for e in v] for k, v in data.items()}
     assert_equal_data(result, expected)
 
 
-def test_map_after_alias(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_map_after_alias(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo")).alias("alias_for_foo").name.map(function=map_func))
     expected = {map_func("foo"): data["foo"]}
     assert_equal_data(result, expected)
 
 
-def test_map_raise_anonymous(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_map_raise_anonymous(constructor: Constructor) -> None:
     df_raw = constructor(data)
     df = nw.from_native(df_raw)
 
@@ -51,7 +39,7 @@ def test_map_raise_anonymous(
         does_not_raise()
         if isinstance(df_raw, (pl.LazyFrame, pl.DataFrame))
         else pytest.raises(
-            ValueError,
+            AnonymousExprError,
             match="Anonymous expressions are not supported in `.name.map`.",
         )
     )
diff --git a/tests/expr_and_series/name/prefix_test.py b/tests/expr_and_series/name/prefix_test.py
index 934d1d66..5894153b 100644
--- a/tests/expr_and_series/name/prefix_test.py
+++ b/tests/expr_and_series/name/prefix_test.py
@@ -6,6 +6,7 @@ import polars as pl
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import AnonymousExprError
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
@@ -13,34 +14,21 @@ data = {"foo": [1, 2, 3], "BAR": [4, 5, 6]}
 prefix = "with_prefix_"
 
 
-def test_prefix(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_prefix(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo", "BAR") * 2).name.prefix(prefix))
     expected = {prefix + str(k): [e * 2 for e in v] for k, v in data.items()}
     assert_equal_data(result, expected)
 
 
-def test_suffix_after_alias(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_suffix_after_alias(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo")).alias("alias_for_foo").name.prefix(prefix))
     expected = {prefix + "foo": data["foo"]}
     assert_equal_data(result, expected)
 
 
-def test_prefix_raise_anonymous(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_prefix_raise_anonymous(constructor: Constructor) -> None:
     df_raw = constructor(data)
     df = nw.from_native(df_raw)
 
@@ -48,7 +36,7 @@ def test_prefix_raise_anonymous(
         does_not_raise()
         if isinstance(df_raw, (pl.LazyFrame, pl.DataFrame))
         else pytest.raises(
-            ValueError,
+            AnonymousExprError,
             match="Anonymous expressions are not supported in `.name.prefix`.",
         )
     )
diff --git a/tests/expr_and_series/name/suffix_test.py b/tests/expr_and_series/name/suffix_test.py
index 47954663..1c581615 100644
--- a/tests/expr_and_series/name/suffix_test.py
+++ b/tests/expr_and_series/name/suffix_test.py
@@ -13,34 +13,21 @@ data = {"foo": [1, 2, 3], "BAR": [4, 5, 6]}
 suffix = "_with_suffix"
 
 
-def test_suffix(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_suffix(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo", "BAR") * 2).name.suffix(suffix))
     expected = {str(k) + suffix: [e * 2 for e in v] for k, v in data.items()}
     assert_equal_data(result, expected)
 
 
-def test_suffix_after_alias(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_suffix_after_alias(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo")).alias("alias_for_foo").name.suffix(suffix))
     expected = {"foo" + suffix: data["foo"]}
     assert_equal_data(result, expected)
 
 
-def test_suffix_raise_anonymous(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_suffix_raise_anonymous(constructor: Constructor) -> None:
     df_raw = constructor(data)
     df = nw.from_native(df_raw)
 
diff --git a/tests/expr_and_series/name/to_lowercase_test.py b/tests/expr_and_series/name/to_lowercase_test.py
index 1b39fc72..7acf9af5 100644
--- a/tests/expr_and_series/name/to_lowercase_test.py
+++ b/tests/expr_and_series/name/to_lowercase_test.py
@@ -6,40 +6,28 @@ import polars as pl
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import AnonymousExprError
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
 data = {"foo": [1, 2, 3], "BAR": [4, 5, 6]}
 
 
-def test_to_lowercase(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_to_lowercase(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo", "BAR") * 2).name.to_lowercase())
     expected = {k.lower(): [e * 2 for e in v] for k, v in data.items()}
     assert_equal_data(result, expected)
 
 
-def test_to_lowercase_after_alias(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_to_lowercase_after_alias(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("BAR")).alias("ALIAS_FOR_BAR").name.to_lowercase())
     expected = {"bar": data["BAR"]}
     assert_equal_data(result, expected)
 
 
-def test_to_lowercase_raise_anonymous(
-    request: pytest.FixtureRequest, constructor: Constructor
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
+def test_to_lowercase_raise_anonymous(constructor: Constructor) -> None:
     df_raw = constructor(data)
     df = nw.from_native(df_raw)
 
@@ -47,7 +35,7 @@ def test_to_lowercase_raise_anonymous(
         does_not_raise()
         if isinstance(df_raw, (pl.LazyFrame, pl.DataFrame))
         else pytest.raises(
-            ValueError,
+            AnonymousExprError,
             match="Anonymous expressions are not supported in `.name.to_lowercase`.",
         )
     )
diff --git a/tests/expr_and_series/name/to_uppercase_test.py b/tests/expr_and_series/name/to_uppercase_test.py
index e6703212..7d0bd7e5 100644
--- a/tests/expr_and_series/name/to_uppercase_test.py
+++ b/tests/expr_and_series/name/to_uppercase_test.py
@@ -6,37 +6,28 @@ import polars as pl
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import AnonymousExprError
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
 data = {"foo": [1, 2, 3], "BAR": [4, 5, 6]}
 
 
-def test_to_uppercase(constructor: Constructor, request: pytest.FixtureRequest) -> None:
-    if any(x in str(constructor) for x in ("duckdb", "pyspark")):
-        request.applymarker(pytest.mark.xfail)
+def test_to_uppercase(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo", "BAR") * 2).name.to_uppercase())
     expected = {k.upper(): [e * 2 for e in v] for k, v in data.items()}
     assert_equal_data(result, expected)
 
 
-def test_to_uppercase_after_alias(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if any(x in str(constructor) for x in ("duckdb", "pyspark")):
-        request.applymarker(pytest.mark.xfail)
+def test_to_uppercase_after_alias(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     result = df.select((nw.col("foo")).alias("alias_for_foo").name.to_uppercase())
     expected = {"FOO": data["foo"]}
     assert_equal_data(result, expected)
 
 
-def test_to_uppercase_raise_anonymous(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if any(x in str(constructor) for x in ("duckdb", "pyspark")):
-        request.applymarker(pytest.mark.xfail)
+def test_to_uppercase_raise_anonymous(constructor: Constructor) -> None:
     df_raw = constructor(data)
     df = nw.from_native(df_raw)
 
@@ -44,7 +35,7 @@ def test_to_uppercase_raise_anonymous(
         does_not_raise()
         if isinstance(df_raw, (pl.LazyFrame, pl.DataFrame))
         else pytest.raises(
-            ValueError,
+            AnonymousExprError,
             match="Anonymous expressions are not supported in `.name.to_uppercase`.",
         )
     )
diff --git a/tests/expr_and_series/order_dependent_lazy_test.py b/tests/expr_and_series/order_dependent_lazy_test.py
new file mode 100644
index 00000000..769f508a
--- /dev/null
+++ b/tests/expr_and_series/order_dependent_lazy_test.py
@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pandas as pd
+import pytest
+
+import narwhals as nw
+import narwhals.stable.v1 as nw_v1
+from narwhals.exceptions import OrderDependentExprError
+from tests.utils import assert_equal_data
+
+if TYPE_CHECKING:
+    from tests.utils import Constructor
+
+
+def test_order_dependent_raises_in_lazy(constructor: Constructor) -> None:
+    lf = nw.from_native(constructor({"a": [1, 2, 3]})).lazy()
+    with pytest.raises(OrderDependentExprError, match="Order-dependent expressions"):
+        lf.select(nw.col("a").diff())
+
+
+def test_dask_order_dependent_ops() -> None:
+    # Preserve these for narwhals.stable.v1, even though they
+    # raise after stable.v1.
+    pytest.importorskip("dask")
+    import dask.dataframe as dd
+
+    df = nw_v1.from_native(dd.from_pandas(pd.DataFrame({"a": [1, 2, 3]})))
+    result = df.select(
+        a=nw.col("a").cum_sum(),
+        b=nw.col("a").cum_count(),
+        c=nw.col("a").cum_prod(),
+        d=nw.col("a").cum_max(),
+        e=nw.col("a").cum_min(),
+        f=nw.col("a").shift(1),
+        g=nw.col("a").diff(),
+        h=nw.col("a").is_first_distinct(),
+        i=nw.col("a").is_last_distinct(),
+    )
+    expected = {
+        "a": [1, 3, 6],
+        "b": [1, 2, 3],
+        "c": [1, 2, 6],
+        "d": [1, 2, 3],
+        "e": [1, 1, 1],
+        "f": [None, 1.0, 2.0],
+        "g": [None, 1.0, 1.0],
+        "h": [True, True, True],
+        "i": [True, True, True],
+    }
+    assert_equal_data(result, expected)
diff --git a/tests/expr_and_series/over_test.py b/tests/expr_and_series/over_test.py
index 45b64eba..2ccbe070 100644
--- a/tests/expr_and_series/over_test.py
+++ b/tests/expr_and_series/over_test.py
@@ -6,6 +6,7 @@ import pytest
 import narwhals.stable.v1 as nw
 from tests.utils import PANDAS_VERSION
 from tests.utils import Constructor
+from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
 data = {
@@ -24,7 +25,7 @@ data_cum = {
 def test_over_single(request: pytest.FixtureRequest, constructor: Constructor) -> None:
     if "dask_lazy_p2" in str(constructor):
         request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor(data))
@@ -42,25 +43,25 @@ def test_over_single(request: pytest.FixtureRequest, constructor: Constructor) -
 def test_over_multiple(request: pytest.FixtureRequest, constructor: Constructor) -> None:
     if "dask_lazy_p2" in str(constructor):
         request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor(data))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
-        "b": [1, 2, 3, 5, 3],
-        "c": [5, 4, 3, 2, 1],
-        "c_min": [5, 4, 1, 2, 1],
+        "b": [1, 2, 3, 3, 5],
+        "c": [5, 4, 3, 1, 2],
+        "c_min": [5, 4, 1, 1, 2],
     }
 
-    result = df.with_columns(c_min=nw.col("c").min().over("a", "b"))
+    result = df.with_columns(c_min=nw.col("c").min().over("a", "b")).sort("a", "b")
     assert_equal_data(result, expected)
 
 
 def test_over_invalid(request: pytest.FixtureRequest, constructor: Constructor) -> None:
     if "polars" in str(constructor):
         request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor(data))
@@ -68,15 +69,15 @@ def test_over_invalid(request: pytest.FixtureRequest, constructor: Constructor)
         df.with_columns(c_min=nw.all().min().over("a", "b"))
 
 
-def test_over_cumsum(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
+def test_over_cumsum(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
-    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
-        request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
 
-    df = nw.from_native(constructor(data_cum))
+    df = nw.from_native(constructor_eager(data_cum))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
         "b": [1, 2, None, 5, 3],
@@ -89,13 +90,13 @@ def test_over_cumsum(request: pytest.FixtureRequest, constructor: Constructor) -
     assert_equal_data(result, expected)
 
 
-def test_over_cumcount(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+def test_over_cumcount(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
 
-    df = nw.from_native(constructor(data_cum))
+    df = nw.from_native(constructor_eager(data_cum))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
         "b": [1, 2, None, 5, 3],
@@ -110,14 +111,14 @@ def test_over_cumcount(request: pytest.FixtureRequest, constructor: Constructor)
     assert_equal_data(result, expected)
 
 
-def test_over_cummax(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if any(x in str(constructor) for x in ("pyarrow_table", "dask_lazy_p2", "duckdb")):
+def test_over_cummax(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table",)):
         request.applymarker(pytest.mark.xfail)
-    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
+    if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    df = nw.from_native(constructor(data_cum))
+    df = nw.from_native(constructor_eager(data_cum))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
         "b": [1, 2, None, 5, 3],
@@ -129,15 +130,15 @@ def test_over_cummax(request: pytest.FixtureRequest, constructor: Constructor) -
     assert_equal_data(result, expected)
 
 
-def test_over_cummin(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
+def test_over_cummin(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
 
-    df = nw.from_native(constructor(data_cum))
+    df = nw.from_native(constructor_eager(data_cum))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
         "b": [1, 2, None, 5, 3],
@@ -150,15 +151,15 @@ def test_over_cummin(request: pytest.FixtureRequest, constructor: Constructor) -
     assert_equal_data(result, expected)
 
 
-def test_over_cumprod(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if any(x in str(constructor) for x in ("pyarrow_table", "dask_lazy_p2", "duckdb")):
+def test_over_cumprod(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if any(x in str(constructor_eager) for x in ("pyarrow_table",)):
         request.applymarker(pytest.mark.xfail)
-    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
-        request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
 
-    df = nw.from_native(constructor(data_cum))
+    df = nw.from_native(constructor_eager(data_cum))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
         "b": [1, 2, None, 5, 3],
@@ -179,15 +180,15 @@ def test_over_anonymous() -> None:
         nw.from_native(df).select(nw.all().cum_max().over("a"))
 
 
-def test_over_shift(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyarrow_table_constructor" in str(
-        constructor
-    ) or "dask_lazy_p2_constructor" in str(constructor):
+def test_over_shift(
+    request: pytest.FixtureRequest, constructor_eager: ConstructorEager
+) -> None:
+    if "pyarrow_table_constructor" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if ("pyspark" in str(constructor_eager)) or "duckdb" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
 
-    df = nw.from_native(constructor(data))
+    df = nw.from_native(constructor_eager(data))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
         "b": [1, 2, 3, 5, 3],
diff --git a/tests/expr_and_series/sample_test.py b/tests/expr_and_series/sample_test.py
index 8e88ba7c..af1f5665 100644
--- a/tests/expr_and_series/sample_test.py
+++ b/tests/expr_and_series/sample_test.py
@@ -2,44 +2,46 @@ from __future__ import annotations
 
 import pytest
 
+import narwhals as nw_main
 import narwhals.stable.v1 as nw
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
 
 def test_expr_sample(constructor_eager: ConstructorEager) -> None:
-    df = nw.from_native(constructor_eager({"a": [1, 2, 3], "b": [4, 5, 6]})).lazy()
+    df = nw.from_native(
+        constructor_eager({"a": [1, 2, 3], "b": [4, 5, 6]}), eager_only=True
+    )
 
-    result_expr = df.select(nw.col("a").sample(n=2)).collect().shape
+    result_expr = df.select(nw.col("a").sample(n=2)).shape
     expected_expr = (2, 1)
     assert result_expr == expected_expr
 
-    result_series = df.collect()["a"].sample(n=2).shape
+    result_series = df["a"].sample(n=2).shape
     expected_series = (2,)
     assert result_series == expected_series
 
+    with pytest.deprecated_call():
+        df.select(nw_main.col("a").sample(n=2))
 
-def test_expr_sample_fraction(
-    constructor_eager: ConstructorEager, request: pytest.FixtureRequest
-) -> None:
-    if "dask" in str(constructor_eager):
-        request.applymarker(pytest.mark.xfail)
+
+def test_expr_sample_fraction(constructor_eager: ConstructorEager) -> None:
     df = nw.from_native(
-        constructor_eager({"a": [1, 2, 3] * 10, "b": [4, 5, 6] * 10})
-    ).lazy()
+        constructor_eager({"a": [1, 2, 3] * 10, "b": [4, 5, 6] * 10}), eager_only=True
+    )
 
-    result_expr = df.select(nw.col("a").sample(fraction=0.1)).collect().shape
+    result_expr = df.select(nw.col("a").sample(fraction=0.1)).shape
     expected_expr = (3, 1)
     assert result_expr == expected_expr
 
-    result_series = df.collect()["a"].sample(fraction=0.1).shape
+    result_series = df["a"].sample(fraction=0.1).shape
     expected_series = (3,)
     assert result_series == expected_series
 
 
 def test_sample_with_seed(constructor_eager: ConstructorEager) -> None:
     size, n = 100, 10
-    df = nw.from_native(constructor_eager({"a": list(range(size))})).lazy()
+    df = nw.from_native(constructor_eager({"a": list(range(size))}))
     expected = {"res1": [True], "res2": [False]}
     result = df.select(
         seed1=nw.col("a").sample(n=n, seed=123),
@@ -52,7 +54,7 @@ def test_sample_with_seed(constructor_eager: ConstructorEager) -> None:
 
     assert_equal_data(result, expected)
 
-    series = df.collect()["a"]
+    series = df["a"]
     seed1 = series.sample(n=n, seed=123)
     seed2 = series.sample(n=n, seed=123)
     seed3 = series.sample(n=n, seed=42)
diff --git a/tests/expr_and_series/shift_test.py b/tests/expr_and_series/shift_test.py
index 4f789493..d6f0274f 100644
--- a/tests/expr_and_series/shift_test.py
+++ b/tests/expr_and_series/shift_test.py
@@ -1,10 +1,8 @@
 from __future__ import annotations
 
 import pyarrow as pa
-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
 
@@ -16,10 +14,8 @@ data = {
 }
 
 
-def test_shift(constructor: Constructor, request: pytest.FixtureRequest) -> None:
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-    df = nw.from_native(constructor(data))
+def test_shift(constructor_eager: ConstructorEager) -> None:
+    df = nw.from_native(constructor_eager(data))
     result = df.with_columns(nw.col("a", "b", "c").shift(2)).filter(nw.col("i") > 1)
     expected = {
         "i": [2, 3, 4],
diff --git a/tests/expr_and_series/sort_test.py b/tests/expr_and_series/sort_test.py
index 0d95722d..e35514cd 100644
--- a/tests/expr_and_series/sort_test.py
+++ b/tests/expr_and_series/sort_test.py
@@ -4,6 +4,7 @@ from typing import Any
 
 import pytest
 
+import narwhals as nw_main
 import narwhals.stable.v1 as nw
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
@@ -29,6 +30,11 @@ def test_sort_expr(
         nw.col("b").sort(descending=descending, nulls_last=nulls_last),
     )
     assert_equal_data(result, expected)
+    with pytest.deprecated_call():
+        df.select(
+            "a",
+            nw_main.col("b").sort(descending=descending, nulls_last=nulls_last),
+        )
 
 
 @pytest.mark.parametrize(
diff --git a/tests/expr_and_series/tail_test.py b/tests/expr_and_series/tail_test.py
index e0481381..2baabfd9 100644
--- a/tests/expr_and_series/tail_test.py
+++ b/tests/expr_and_series/tail_test.py
@@ -2,13 +2,14 @@ from __future__ import annotations
 
 import pytest
 
-import narwhals as nw
+import narwhals as nw_main
+import narwhals.stable.v1 as nw
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
 
 @pytest.mark.parametrize("n", [2, -1])
-def test_head(
+def test_tail(
     constructor_eager: ConstructorEager, n: int, request: pytest.FixtureRequest
 ) -> None:
     if "polars" in str(constructor_eager) and n < 0:
@@ -18,9 +19,12 @@ def test_head(
     expected = {"a": [2, 3]}
     assert_equal_data(result, expected)
 
+    with pytest.deprecated_call():
+        df.select(nw_main.col("a").tail(5))
+
 
 @pytest.mark.parametrize("n", [2, -1])
-def test_head_series(constructor_eager: ConstructorEager, n: int) -> None:
+def test_tail_series(constructor_eager: ConstructorEager, n: int) -> None:
     df = nw.from_native(constructor_eager({"a": [1, 2, 3]}), eager_only=True)
     result = df.select(df["a"].tail(n))
     expected = {"a": [2, 3]}
diff --git a/tests/expr_and_series/when_test.py b/tests/expr_and_series/when_test.py
index 94e37aaa..4f768db0 100644
--- a/tests/expr_and_series/when_test.py
+++ b/tests/expr_and_series/when_test.py
@@ -141,3 +141,13 @@ def test_when_then_otherwise_lit_str(
     result = df.select(nw.when(nw.col("a") > 1).then(nw.col("b")).otherwise(nw.lit("z")))
     expected = {"b": ["z", "b", "c"]}
     assert_equal_data(result, expected)
+
+
+def test_when_then_otherwise_both_lit(constructor: Constructor) -> None:
+    df = nw.from_native(constructor(data))
+    result = df.select(
+        x1=nw.when(nw.col("a") > 1).then(nw.lit(42)).otherwise(nw.lit(-1)),
+        x2=nw.when(nw.col("a") > 2).then(nw.lit(42)).otherwise(nw.lit(-1)),
+    )
+    expected = {"x1": [-1, 42, 42], "x2": [-1, -1, 42]}
+    assert_equal_data(result, expected)
diff --git a/tests/frame/drop_nulls_test.py b/tests/frame/drop_nulls_test.py
index 368ad6ba..c49b1712 100644
--- a/tests/frame/drop_nulls_test.py
+++ b/tests/frame/drop_nulls_test.py
@@ -7,18 +7,16 @@ from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
 data = {
-    "a": [1.0, 2.0, None, 4.0],
-    "b": [None, 3.0, None, 5.0],
+    "alpha": [1.0, 2.0, None, 4.0],
+    "beta gamma": [None, 3.0, None, 5.0],
 }
 
 
-def test_drop_nulls(constructor: Constructor, request: pytest.FixtureRequest) -> None:
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_drop_nulls(constructor: Constructor) -> None:
     result = nw.from_native(constructor(data)).drop_nulls()
     expected = {
-        "a": [2.0, 4.0],
-        "b": [3.0, 5.0],
+        "alpha": [2.0, 4.0],
+        "beta gamma": [3.0, 5.0],
     }
     assert_equal_data(result, expected)
 
@@ -26,18 +24,15 @@ def test_drop_nulls(constructor: Constructor, request: pytest.FixtureRequest) ->
 @pytest.mark.parametrize(
     ("subset", "expected"),
     [
-        ("a", {"a": [1, 2.0, 4.0], "b": [None, 3.0, 5.0]}),
-        (["a"], {"a": [1, 2.0, 4.0], "b": [None, 3.0, 5.0]}),
-        (["a", "b"], {"a": [2.0, 4.0], "b": [3.0, 5.0]}),
+        ("alpha", {"alpha": [1, 2.0, 4.0], "beta gamma": [None, 3.0, 5.0]}),
+        (["alpha"], {"alpha": [1, 2.0, 4.0], "beta gamma": [None, 3.0, 5.0]}),
+        (["alpha", "beta gamma"], {"alpha": [2.0, 4.0], "beta gamma": [3.0, 5.0]}),
     ],
 )
 def test_drop_nulls_subset(
     constructor: Constructor,
     subset: str | list[str],
     expected: dict[str, float],
-    request: pytest.FixtureRequest,
 ) -> None:
-    if "duckdb" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
     result = nw.from_native(constructor(data)).drop_nulls(subset=subset)
     assert_equal_data(result, expected)
diff --git a/tests/frame/filter_test.py b/tests/frame/filter_test.py
index 759d175c..922af0e6 100644
--- a/tests/frame/filter_test.py
+++ b/tests/frame/filter_test.py
@@ -3,8 +3,10 @@ from __future__ import annotations
 from contextlib import nullcontext as does_not_raise
 
 import pytest
+from polars.exceptions import ShapeError as PlShapeError
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import ShapeError
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
@@ -33,3 +35,40 @@ def test_filter_with_boolean_list(
         result = df.filter([False, True, True])
         expected = {"a": [3, 2], "b": [4, 6], "z": [8.0, 9.0]}
         assert_equal_data(result, expected)
+
+
+def test_filter_raise_on_agg_predicate(constructor: Constructor) -> None:
+    data = {"a": [1, 3, 2], "b": [4, 4, 6], "z": [7.0, 8, 9]}
+    df = nw.from_native(constructor(data))
+
+    context = (
+        pytest.raises(
+            ShapeError,
+            match="filter's length: 1 differs from that of the series: 3",
+        )
+        if any(x in str(constructor) for x in ("pandas", "pyarrow", "modin"))
+        else does_not_raise()
+        if "polars" in str(constructor)
+        else pytest.raises(Exception)  # type: ignore[arg-type] # noqa: PT011
+    )
+    with context:
+        df.filter(nw.col("a").max() > 2).lazy().collect()
+
+
+def test_filter_raise_on_shape_mismatch(constructor: Constructor) -> None:
+    data = {"a": [1, 3, 2], "b": [4, 4, 6], "z": [7.0, 8, 9]}
+    df = nw.from_native(constructor(data))
+
+    context = (
+        pytest.raises(
+            ShapeError, match="filter's length: 2 differs from that of the series: 3"
+        )
+        if any(x in str(constructor) for x in ("pandas", "pyarrow", "modin"))
+        else pytest.raises(
+            PlShapeError, match="filter's length: 2 differs from that of the series: 3"
+        )
+        if "polars" in str(constructor)
+        else pytest.raises(Exception)  # type: ignore[arg-type] # noqa: PT011
+    )
+    with context:
+        df.filter(nw.col("b").unique() > 2).lazy().collect()
diff --git a/tests/frame/join_test.py b/tests/frame/join_test.py
index 5ff112f3..4c3c1419 100644
--- a/tests/frame/join_test.py
+++ b/tests/frame/join_test.py
@@ -8,6 +8,7 @@ from typing import Literal
 import pandas as pd
 import pytest
 
+import narwhals as nw_main  # use nw_main in some tests for coverage
 import narwhals.stable.v1 as nw
 from narwhals.utils import Implementation
 from tests.utils import DUCKDB_VERSION
@@ -23,7 +24,7 @@ def test_inner_join_two_keys(constructor: Constructor) -> None:
         "zor ro": [7.0, 8, 9],
         "idx": [0, 1, 2],
     }
-    df = nw.from_native(constructor(data))
+    df = nw_main.from_native(constructor(data))
     df_right = df
     result = df.join(
         df_right,  # type: ignore[arg-type]
diff --git a/tests/frame/sample_test.py b/tests/frame/sample_test.py
index ff3591fd..19cbf0cb 100644
--- a/tests/frame/sample_test.py
+++ b/tests/frame/sample_test.py
@@ -5,10 +5,10 @@ from typing import TYPE_CHECKING
 import narwhals.stable.v1 as nw
 
 if TYPE_CHECKING:
-    from tests.utils import Constructor
+    from tests.utils import ConstructorEager
 
 
-def test_sample_n(constructor_eager: Constructor) -> None:
+def test_sample_n(constructor_eager: ConstructorEager) -> None:
     df = nw.from_native(
         constructor_eager({"a": [1, 2, 3, 4], "b": ["x", "y", "x", "y"]}), eager_only=True
     )
@@ -18,7 +18,7 @@ def test_sample_n(constructor_eager: Constructor) -> None:
     assert result_expr == expected_expr
 
 
-def test_sample_fraction(constructor_eager: Constructor) -> None:
+def test_sample_fraction(constructor_eager: ConstructorEager) -> None:
     df = nw.from_native(
         constructor_eager({"a": [1, 2, 3, 4], "b": ["x", "y", "x", "y"]}), eager_only=True
     )
@@ -28,7 +28,7 @@ def test_sample_fraction(constructor_eager: Constructor) -> None:
     assert result_expr == expected_expr
 
 
-def test_sample_with_seed(constructor_eager: Constructor) -> None:
+def test_sample_with_seed(constructor_eager: ConstructorEager) -> None:
     size, n = 100, 10
     df = nw.from_native(constructor_eager({"a": range(size)}), eager_only=True)
 
diff --git a/tests/frame/schema_test.py b/tests/frame/schema_test.py
index 9de39774..565bf015 100644
--- a/tests/frame/schema_test.py
+++ b/tests/frame/schema_test.py
@@ -274,7 +274,6 @@ def test_nested_dtypes_ibis(request: pytest.FixtureRequest) -> None:  # pragma:
 )
 def test_nested_dtypes_dask() -> None:
     pytest.importorskip("dask")
-    pytest.importorskip("dask_expr", exc_type=ImportError)
     import dask.dataframe as dd
 
     df = dd.from_pandas(
diff --git a/tests/group_by_test.py b/tests/group_by_test.py
index 911cc473..07ab9574 100644
--- a/tests/group_by_test.py
+++ b/tests/group_by_test.py
@@ -8,6 +8,7 @@ import pyarrow as pa
 import pytest
 
 import narwhals.stable.v1 as nw
+from narwhals.exceptions import AnonymousExprError
 from tests.utils import PANDAS_VERSION
 from tests.utils import PYARROW_VERSION
 from tests.utils import Constructor
@@ -37,7 +38,6 @@ def test_group_by_complex() -> None:
 
 def test_invalid_group_by_dask() -> None:
     pytest.importorskip("dask")
-    pytest.importorskip("dask_expr", exc_type=ImportError)
     import dask.dataframe as dd
 
     df_dask = dd.from_pandas(df_pandas)
@@ -49,7 +49,8 @@ def test_invalid_group_by_dask() -> None:
         nw.from_native(df_dask).group_by("a").agg(nw.col("b"))
 
     with pytest.raises(
-        ValueError, match=r"Anonymous expressions are not supported in group_by\.agg"
+        AnonymousExprError,
+        match=r"Anonymous expressions are not supported in `group_by\.agg`",
     ):
         nw.from_native(df_dask).group_by("a").agg(nw.all().mean())
 
@@ -60,11 +61,13 @@ def test_invalid_group_by() -> None:
     with pytest.raises(ValueError, match="does your"):
         df.group_by("a").agg(nw.col("b"))
     with pytest.raises(
-        ValueError, match=r"Anonymous expressions are not supported in group_by\.agg"
+        AnonymousExprError,
+        match=r"Anonymous expressions are not supported in `group_by\.agg`",
     ):
         df.group_by("a").agg(nw.all().mean())
     with pytest.raises(
-        ValueError, match=r"Anonymous expressions are not supported in group_by\.agg"
+        AnonymousExprError,
+        match=r"Anonymous expressions are not supported in `group_by\.agg`",
     ):
         nw.from_native(pa.table({"a": [1, 2, 3]})).group_by("a").agg(nw.all().mean())
     with pytest.raises(ValueError, match=r"Non-trivial complex aggregation found"):
@@ -115,8 +118,6 @@ def test_group_by_depth_1_agg(
     expected: dict[str, list[int | float]],
     request: pytest.FixtureRequest,
 ) -> None:
-    if "pyspark" in str(constructor) and attr == "n_unique":
-        request.applymarker(pytest.mark.xfail)
     if "pandas_pyarrow" in str(constructor) and attr == "var" and PANDAS_VERSION < (2, 1):
         # Known issue with variance calculation in pandas 2.0.x with pyarrow backend in groupby operations"
         request.applymarker(pytest.mark.xfail)
@@ -166,11 +167,7 @@ def test_group_by_median(constructor: Constructor) -> None:
     assert_equal_data(result, expected)
 
 
-def test_group_by_n_unique_w_missing(
-    constructor: Constructor, request: pytest.FixtureRequest
-) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_group_by_n_unique_w_missing(constructor: Constructor) -> None:
     data = {"a": [1, 1, 2], "b": [4, None, 5], "c": [None, None, 7], "d": [1, 1, 3]}
     result = (
         nw.from_native(constructor(data))
@@ -276,9 +273,6 @@ def test_key_with_nulls(
         # TODO(unassigned): Modin flaky here?
         request.applymarker(pytest.mark.skip)
 
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
     context = (
         pytest.raises(NotImplementedError, match="null values")
         if ("pandas_constructor" in str(constructor) and PANDAS_VERSION < (1, 1, 0))
@@ -303,9 +297,6 @@ def test_key_with_nulls_ignored(
     if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
 
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
-
     data = {"b": [4, 5, None], "a": [1, 2, 3]}
     result = (
         nw.from_native(constructor(data))
@@ -346,9 +337,7 @@ def test_key_with_nulls_iter(
     assert len(result) == 4
 
 
-def test_no_agg(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "pyspark" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_no_agg(constructor: Constructor) -> None:
     result = nw.from_native(constructor(data)).group_by(["a", "b"]).agg().sort("a", "b")
 
     expected = {"a": [1, 3], "b": [4, 6]}
@@ -428,7 +417,7 @@ def test_all_kind_of_aggs(
         # and modin lol https://github.com/modin-project/modin/issues/7414
         # and cudf https://github.com/rapidsai/cudf/issues/17649
         request.applymarker(pytest.mark.xfail)
-    if ("pyspark" in str(constructor)) or "duckdb" in str(constructor):
+    if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
     if "pandas" in str(constructor) and PANDAS_VERSION < (1, 4):
         # Bug in old pandas, can't do DataFrameGroupBy[['b', 'b']]
diff --git a/tests/new_series_test.py b/tests/new_series_test.py
index 0d635c85..416972ff 100644
--- a/tests/new_series_test.py
+++ b/tests/new_series_test.py
@@ -47,7 +47,6 @@ def test_new_series_v1(constructor_eager: ConstructorEager) -> None:
 
 def test_new_series_dask() -> None:
     pytest.importorskip("dask")
-    pytest.importorskip("dask_expr", exc_type=ImportError)
     import dask.dataframe as dd
 
     df = nw.from_native(dd.from_pandas(pd.DataFrame({"a": [1, 2, 3]})))
diff --git a/tests/no_imports_test.py b/tests/no_imports_test.py
index c89a9256..4f2e75e6 100644
--- a/tests/no_imports_test.py
+++ b/tests/no_imports_test.py
@@ -51,7 +51,6 @@ def test_pandas(monkeypatch: pytest.MonkeyPatch) -> None:
 
 def test_dask(monkeypatch: pytest.MonkeyPatch) -> None:
     pytest.importorskip("dask")
-    pytest.importorskip("dask_expr", exc_type=ImportError)
     import dask.dataframe as dd
 
     monkeypatch.delitem(sys.modules, "polars")
diff --git a/tests/series_only/alias_rename_test.py b/tests/series_only/alias_rename_test.py
index 87143a57..8b02b389 100644
--- a/tests/series_only/alias_rename_test.py
+++ b/tests/series_only/alias_rename_test.py
@@ -1,11 +1,11 @@
 from __future__ import annotations
 
 import narwhals as nw
-from tests.utils import Constructor
+from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
 
-def test_alias_rename(constructor_eager: Constructor) -> None:
+def test_alias_rename(constructor_eager: ConstructorEager) -> None:
     data = [1, 2, 3]
     expected = {"bar": data}
     series = nw.from_native(constructor_eager({"foo": data}), eager_only=True)["foo"]
diff --git a/tests/tpch_q1_test.py b/tests/tpch_q1_test.py
index cb6d4854..498ed234 100644
--- a/tests/tpch_q1_test.py
+++ b/tests/tpch_q1_test.py
@@ -31,7 +31,6 @@ def test_q1(library: str, request: pytest.FixtureRequest) -> None:
         df_raw = pl.scan_csv("tests/data/lineitem.csv")
     elif library == "dask":
         pytest.importorskip("dask")
-        pytest.importorskip("dask_expr", exc_type=ImportError)
         import dask.dataframe as dd
 
         df_raw = dd.from_pandas(
diff --git a/tests/translate/from_native_test.py b/tests/translate/from_native_test.py
index a9567445..c9928795 100644
--- a/tests/translate/from_native_test.py
+++ b/tests/translate/from_native_test.py
@@ -202,7 +202,6 @@ def test_init_already_narwhals_unstable() -> None:
 
 def test_series_only_dask() -> None:
     pytest.importorskip("dask")
-    pytest.importorskip("dask_expr", exc_type=ImportError)
     import dask.dataframe as dd
 
     dframe = dd.from_pandas(df_pd)
@@ -221,7 +220,6 @@ def test_series_only_dask() -> None:
 )
 def test_eager_only_lazy_dask(eager_only: Any, context: Any) -> None:
     pytest.importorskip("dask")
-    pytest.importorskip("dask_expr", exc_type=ImportError)
     import dask.dataframe as dd
 
     dframe = dd.from_pandas(df_pd)