Repository: narwhals-dev/narwhals
Pull request #2118 (2025-02-28 19:24:28)

ci: Add darts to downstream tests
## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [x] 🐳 Other

## Related issues

[darts#2661](https://github.com/unit8co/darts/pull/2661) just got merged

## Checklist

- [ ] Code follows style guide (ruff)
- [ ] Tests added
- [ ] Documented the changes

## If you have comments or can explain your changes, please do so below

Darts entire test suite takes a bit too much for our use case - Narwhals is used in `TimeSeries` class, that's what I am adding as test here

FBruzzesi: Hey @dennisbader and @authierj, we try to run test suites for downstream libraries that use narwhals in our CI, to make sure that we don't break anything by accident.

As mentioned in the description, darts test suite takes ~30mins, but narwhals is used just in `TimeSeries` test. Do you think that's enough for now to test?
dennisbader: Hi @FBruzzesi, that's very nice 🚀 

I'd also add the following tests (they're very lightweight):
- `test_timeseries_multivariate.py`, contains narwhals related checks for multivariate DataFrames.
- `test_static_covariates.py`, contains narwhals related checks for DataFrames and static covariates.
- we'll be adding checks using polars as backend soon. Polars will only be an optional dependency and not be part of our `core.txt` requirements file (nor `dev.txt`). But it looks like you already install it for the tests, so all good 👍 
FBruzzesi: > I'd also add the following tests (they're very lightweight):
> 
> * `test_timeseries_multivariate.py`, contains narwhals related checks for multivariate DataFrames.
> * `test_static_covariates.py`, contains narwhals related checks for DataFrames and static covariates.

Thanks @dennisbader , I just added these two tests as well

> * we'll be adding checks using polars as backend soon. Polars will only be an optional dependency and not be part of our `core.txt` requirements file (nor `dev.txt`). But it looks like you already install it for the tests, so all good 👍

Actually we don't install polars if not needed - I added it now, with a TODO note to use darts specification when you are going to introduce it in CI/some requirements

Patch:
diff --git a/README.md b/README.md
index a2a0198a..913b4463 100644
--- a/README.md
+++ b/README.md
@@ -110,6 +110,7 @@ for an up-to-date plan of future work.
 Join the party!
 
 - [altair](https://github.com/vega/altair/)
+- [darts](https://github.com/unit8co/darts)
 - [hierarchicalforecast](https://github.com/Nixtla/hierarchicalforecast)
 - [marimo](https://github.com/marimo-team/marimo)
 - [panel-graphic-walker](https://github.com/panel-extensions/panel-graphic-walker)
diff --git a/docs/ecosystem.md b/docs/ecosystem.md
index 7c9cbb6c..acc7ff0c 100644
--- a/docs/ecosystem.md
+++ b/docs/ecosystem.md
@@ -6,6 +6,7 @@ The following is a non-exhaustive list of libraries and tools that choose to use
 for their dataframe interoperability needs:
 
 * [altair](https://github.com/vega/altair/)
+* [darts](https://github.com/unit8co/darts)
 * [hierarchicalforecast](https://github.com/Nixtla/hierarchicalforecast)
 * [marimo](https://github.com/marimo-team/marimo)
 * [panel-graphic-walker](https://github.com/panel-extensions/panel-graphic-walker)
diff --git a/docs/installation.md b/docs/installation.md
index 11ad75b6..24ad5504 100644
--- a/docs/installation.md
+++ b/docs/installation.md
@@ -30,7 +30,7 @@ To verify the installation, start the Python REPL and execute:
 ```python
 >>> import narwhals
 >>> narwhals.__version__
-'1.29.0'
+'1.29.1'
 ```
 
 If you see the version number, then the installation was successful!
diff --git a/narwhals/__init__.py b/narwhals/__init__.py
index 70bcfb26..feb5e240 100644
--- a/narwhals/__init__.py
+++ b/narwhals/__init__.py
@@ -80,7 +80,7 @@ from narwhals.utils import maybe_get_index
 from narwhals.utils import maybe_reset_index
 from narwhals.utils import maybe_set_index
 
-__version__ = "1.29.0"
+__version__ = "1.29.1"
 
 __all__ = [
     "Array",
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index ab1e0e54..f36a43e9 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
 from typing import Literal
+from typing import Mapping
 from typing import Sequence
 
 from narwhals._arrow.expr_cat import ArrowExprCatNamespace
@@ -21,6 +22,7 @@ from narwhals.dependencies import is_numpy_array
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
+from narwhals.utils import not_implemented
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -262,7 +264,7 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
     def skew(self: Self) -> Self:
         return reuse_series_implementation(self, "skew", returns_scalar=True)
 
-    def cast(self: Self, dtype: DType) -> Self:
+    def cast(self: Self, dtype: DType | type[DType]) -> Self:
         return reuse_series_implementation(self, "cast", dtype=dtype)
 
     def abs(self: Self) -> Self:
@@ -385,7 +387,11 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
         return reuse_series_implementation(self, "unique", maintain_order=False)
 
     def replace_strict(
-        self: Self, old: Sequence[Any], new: Sequence[Any], *, return_dtype: DType | None
+        self: Self,
+        old: Sequence[Any] | Mapping[Any, Any],
+        new: Sequence[Any],
+        *,
+        return_dtype: DType | type[DType] | None,
     ) -> Self:
         return reuse_series_implementation(
             self, "replace_strict", old=old, new=new, return_dtype=return_dtype
@@ -417,7 +423,7 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
             self, "clip", lower_bound=lower_bound, upper_bound=upper_bound
         )
 
-    def over(self: Self, keys: list[str], kind: ExprKind) -> Self:
+    def over(self: Self, keys: Sequence[str], kind: ExprKind) -> Self:
         if not is_scalar_like(kind):
             msg = "Only aggregation or literal operations are supported in `over` context for PyArrow."
             raise NotImplementedError(msg)
@@ -434,8 +440,9 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
                 raise NotImplementedError(msg)
 
             tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
+            on = list(keys)
             tmp = df.simple_select(*keys).join(
-                tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
+                tmp, how="left", left_on=on, right_on=on, suffix="_right"
             )
             return [tmp[alias] for alias in aliases]
 
@@ -455,7 +462,7 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
     def map_batches(
         self: Self,
         function: Callable[[Any], Any],
-        return_dtype: DType | None,
+        return_dtype: DType | type[DType] | None,
     ) -> Self:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             input_series_list = self._call(df)
@@ -579,6 +586,8 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
             self, "rank", method=method, descending=descending
         )
 
+    ewm_mean = not_implemented()
+
     @property
     def dt(self: Self) -> ArrowExprDateTimeNamespace:
         return ArrowExprDateTimeNamespace(self)
diff --git a/narwhals/_arrow/group_by.py b/narwhals/_arrow/group_by.py
index 564354ed..aefade0d 100644
--- a/narwhals/_arrow/group_by.py
+++ b/narwhals/_arrow/group_by.py
@@ -12,7 +12,7 @@ import pyarrow.compute as pc
 
 from narwhals._arrow.utils import extract_py_scalar
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
-from narwhals._expression_parsing import is_simple_aggregation
+from narwhals._expression_parsing import is_elementary_expression
 from narwhals.utils import generate_temporary_column_name
 
 if TYPE_CHECKING:
@@ -51,7 +51,7 @@ class ArrowGroupBy:
         all_simple_aggs = True
         for expr in exprs:
             if not (
-                is_simple_aggregation(expr)
+                is_elementary_expression(expr)
                 and re.sub(r"(\w+->)", "", expr._function_name)
                 in POLARS_TO_ARROW_AGGREGATIONS
             ):
diff --git a/narwhals/_arrow/namespace.py b/narwhals/_arrow/namespace.py
index ac499980..7d803103 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -3,6 +3,7 @@ from __future__ import annotations
 import operator
 from functools import partial
 from functools import reduce
+from itertools import chain
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
@@ -193,7 +194,7 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
 
     def all_horizontal(self: Self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            series = (s for _expr in exprs for s in _expr(df))
+            series = chain.from_iterable(expr(df) for expr in exprs)
             return [reduce(operator.and_, align_series_full_broadcast(*series))]
 
         return self._create_expr_from_callable(
@@ -206,7 +207,7 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
 
     def any_horizontal(self: Self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            series = (s for _expr in exprs for s in _expr(df))
+            series = chain.from_iterable(expr(df) for expr in exprs)
             return [reduce(operator.or_, align_series_full_broadcast(*series))]
 
         return self._create_expr_from_callable(
@@ -219,11 +220,8 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
 
     def sum_horizontal(self: Self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            series = [
-                s.fill_null(0, strategy=None, limit=None)
-                for _expr in exprs
-                for s in _expr(df)
-            ]
+            it = chain.from_iterable(expr(df) for expr in exprs)
+            series = (s.fill_null(0, strategy=None, limit=None) for s in it)
             return [reduce(operator.add, align_series_full_broadcast(*series))]
 
         return self._create_expr_from_callable(
@@ -238,7 +236,7 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
         dtypes = import_dtypes_module(self._version)
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            expr_results = [s for _expr in exprs for s in _expr(df)]
+            expr_results = list(chain.from_iterable(expr(df) for expr in exprs))
             series = align_series_full_broadcast(
                 *(s.fill_null(0, strategy=None, limit=None) for s in expr_results)
             )
@@ -257,7 +255,7 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
 
     def min_horizontal(self: Self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            init_series, *series = [s for _expr in exprs for s in _expr(df)]
+            init_series, *series = list(chain.from_iterable(expr(df) for expr in exprs))
             init_series, *series = align_series_full_broadcast(init_series, *series)
             # NOTE: Stubs copy the wrong signature https://github.com/zen-xu/pyarrow-stubs/blob/d97063876720e6a5edda7eb15f4efe07c31b8296/pyarrow-stubs/compute.pyi#L963
             min_element_wise: Incomplete = pc.min_element_wise
@@ -285,7 +283,7 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
 
     def max_horizontal(self: Self, *exprs: ArrowExpr) -> ArrowExpr:
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            init_series, *series = [s for _expr in exprs for s in _expr(df)]
+            init_series, *series = list(chain.from_iterable(expr(df) for expr in exprs))
             init_series, *series = align_series_full_broadcast(init_series, *series)
             # NOTE: stubs are missing `ChunkedArray` support
             # https://github.com/zen-xu/pyarrow-stubs/blob/d97063876720e6a5edda7eb15f4efe07c31b8296/pyarrow-stubs/compute.pyi#L948-L954
@@ -357,7 +355,7 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
 
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             compliant_series_list = align_series_full_broadcast(
-                *(s for _expr in exprs for s in _expr.cast(dtypes.String())(df))
+                *(chain.from_iterable(expr.cast(dtypes.String())(df) for expr in exprs))
             )
             null_handling: Literal["skip", "emit_null"] = (
                 "skip" if ignore_nulls else "emit_null"
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index 57e8f098..141c98fb 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -403,18 +403,28 @@ class ArrowSeries(CompliantSeries):
     def scatter(self: Self, indices: int | Sequence[int], values: Any) -> Self:
         import numpy as np  # ignore-banned-import
 
-        mask: _1DArray = np.zeros(self.len(), dtype=bool)
-        mask[indices] = True
-        if isinstance(values, self.__class__):
-            ser, values = extract_native(self, values)
+        if isinstance(indices, int):
+            indices_native = pa.array([indices])
+            values_native = pa.array([values])
         else:
-            ser = self._native_series
-        if isinstance(values, pa.ChunkedArray):
-            values = values.combine_chunks()
-        if not isinstance(values, pa.Array):
-            values = pa.array(values)
+            # TODO(unassigned): we may also want to let `indices` be a Series.
+            # https://github.com/narwhals-dev/narwhals/issues/2155
+            indices_native = pa.array(indices)
+            if isinstance(values, self.__class__):
+                values_native = values._native_series.combine_chunks()
+            else:
+                values_native = pa.array(values)
+
+        sorting_indices = pc.sort_indices(indices_native)  # type: ignore[call-overload]
+        indices_native = pc.take(indices_native, sorting_indices)
+        values_native = pc.take(values_native, sorting_indices)
+
+        mask: _1DArray = np.zeros(self.len(), dtype=bool)
+        mask[indices_native] = True
         result = pc.replace_with_mask(
-            ser, cast("list[bool]", mask), values.take(cast("Indices", indices))
+            self._native_series,
+            cast("list[bool]", mask),
+            values_native.take(cast("Indices", indices_native)),
         )
         return self._from_native_series(result)
 
@@ -519,7 +529,7 @@ class ArrowSeries(CompliantSeries):
     def is_nan(self: Self) -> Self:
         return self._from_native_series(pc.is_nan(self._native_series))
 
-    def cast(self: Self, dtype: DType) -> Self:
+    def cast(self: Self, dtype: DType | type[DType]) -> Self:
         ser = self._native_series
         data_type = narwhals_to_native_dtype(dtype, self._version)
         return self._from_native_series(pc.cast(ser, data_type))
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index b1930e56..5cb47377 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import re
+import warnings
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
@@ -14,12 +16,14 @@ from narwhals._dask.utils import maybe_evaluate_expr
 from narwhals._dask.utils import narwhals_to_native_dtype
 from narwhals._expression_parsing import ExprKind
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
+from narwhals._expression_parsing import is_elementary_expression
 from narwhals._pandas_like.utils import native_to_narwhals_dtype
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.exceptions import InvalidOperationError
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
 from narwhals.utils import generate_temporary_column_name
+from narwhals.utils import not_implemented
 
 if TYPE_CHECKING:
     try:
@@ -389,12 +393,6 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[
     def drop_nulls(self: Self) -> Self:
         return self._from_call(lambda _input: _input.dropna(), "drop_nulls")
 
-    def replace_strict(
-        self: Self, old: Sequence[Any], new: Sequence[Any], *, return_dtype: DType | None
-    ) -> Self:
-        msg = "`replace_strict` is not yet supported for Dask expressions"
-        raise NotImplementedError(msg)
-
     def abs(self: Self) -> Self:
         return self._from_call(lambda _input: _input.abs(), "abs")
 
@@ -535,30 +533,41 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[
             lambda _input: _input.isna().sum().to_series(), "null_count"
         )
 
-    def over(self: Self, keys: list[str], kind: ExprKind) -> Self:
-        def func(df: DaskLazyFrame) -> list[Any]:
-            output_names, aliases = evaluate_output_names_and_aliases(self, df, [])
-            if overlap := set(output_names).intersection(keys):
-                # E.g. `df.select(nw.all().sum().over('a'))`. This is well-defined,
-                # we just don't support it yet.
-                msg = (
-                    f"Column names {overlap} appear in both expression output names and in `over` keys.\n"
-                    "This is not yet supported."
-                )
-                raise NotImplementedError(msg)
-            if df._native_frame.npartitions == 1:  # pragma: no cover
-                tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
-                tmp_native = (
-                    df.simple_select(*keys)
-                    .join(tmp, how="left", left_on=keys, right_on=keys, suffix="_right")
-                    ._native_frame
-                )
-                return [tmp_native[name] for name in aliases]
-            # https://github.com/dask/dask/issues/6659
+    def over(self: Self, keys: Sequence[str], kind: ExprKind) -> Self:
+        # pandas is a required dependency of dask so it's safe to import this
+        from narwhals._pandas_like.group_by import AGGREGATIONS_TO_PANDAS_EQUIVALENT
+
+        if not is_elementary_expression(self):  # pragma: no cover
             msg = (
-                "`Expr.over` is not supported for Dask backend with multiple partitions."
+                "Only elementary expressions are supported for `.over` in dask.\n\n"
+                "Please see: "
+                "https://narwhals-dev.github.io/narwhals/pandas_like_concepts/improve_group_by_operation/"
             )
             raise NotImplementedError(msg)
+        function_name = re.sub(r"(\w+->)", "", self._function_name)
+        try:
+            dask_function_name = AGGREGATIONS_TO_PANDAS_EQUIVALENT[function_name]
+        except KeyError:
+            msg = (
+                f"Unsupported function: {function_name} in `over` context.\n\n."
+                f"Supported functions are {', '.join(AGGREGATIONS_TO_PANDAS_EQUIVALENT)}\n"
+            )
+            raise NotImplementedError(msg) from None
+
+        def func(df: DaskLazyFrame) -> list[dx.Series]:
+            output_names, aliases = evaluate_output_names_and_aliases(self, df, [])
+
+            with warnings.catch_warnings():
+                warnings.filterwarnings(
+                    "ignore", message=".*`meta` is not specified", category=UserWarning
+                )
+                res_native = df._native_frame.groupby(keys)[list(output_names)].transform(
+                    dask_function_name, **self._call_kwargs
+                )
+            result_frame = df._from_native_frame(
+                res_native.rename(columns=dict(zip(output_names, aliases)))
+            )._native_frame
+            return [result_frame[name] for name in aliases]
 
         return self.__class__(
             func,
@@ -568,7 +577,6 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[
             alias_output_names=self._alias_output_names,
             backend_version=self._backend_version,
             version=self._version,
-            call_kwargs=self._call_kwargs,
         )
 
     def cast(self: Self, dtype: DType | type[DType]) -> Self:
@@ -594,3 +602,24 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[
     @property
     def name(self: Self) -> DaskExprNameNamespace:
         return DaskExprNameNamespace(self)
+
+    arg_min = not_implemented()
+    arg_max = not_implemented()
+    arg_true = not_implemented()
+    head = not_implemented()
+    tail = not_implemented()
+    mode = not_implemented()
+    sort = not_implemented()
+    rank = not_implemented()
+    sample = not_implemented()
+    map_batches = not_implemented()
+    ewm_mean = not_implemented()
+    rolling_sum = not_implemented()
+    rolling_mean = not_implemented()
+    rolling_var = not_implemented()
+    rolling_std = not_implemented()
+    gather_every = not_implemented()
+    replace_strict = not_implemented()
+
+    cat = not_implemented()  # pyright: ignore[reportAssignmentType]
+    list = not_implemented()  # pyright: ignore[reportAssignmentType]
diff --git a/narwhals/_dask/group_by.py b/narwhals/_dask/group_by.py
index cc678051..f2e1ddbe 100644
--- a/narwhals/_dask/group_by.py
+++ b/narwhals/_dask/group_by.py
@@ -11,7 +11,7 @@ from typing import Sequence
 import dask.dataframe as dd
 
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
-from narwhals._expression_parsing import is_simple_aggregation
+from narwhals._expression_parsing import is_elementary_expression
 
 try:
     import dask.dataframe.dask_expr as dx
@@ -121,7 +121,7 @@ def agg_dask(
     all_simple_aggs = True
     for expr in exprs:
         if not (
-            is_simple_aggregation(expr)
+            is_elementary_expression(expr)
             and re.sub(r"(\w+->)", "", expr._function_name) in POLARS_TO_DASK_AGGREGATIONS
         ):
             all_simple_aggs = False
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 2f47942b..c47fa8bf 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -24,6 +24,7 @@ from narwhals._duckdb.utils import narwhals_to_native_dtype
 from narwhals._expression_parsing import ExprKind
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
+from narwhals.utils import not_implemented
 
 if TYPE_CHECKING:
     import duckdb
@@ -482,3 +483,36 @@ class DuckDBExpr(CompliantExpr["DuckDBLazyFrame", "duckdb.Expression"]):  # type
     @property
     def list(self: Self) -> DuckDBExprListNamespace:
         return DuckDBExprListNamespace(self)
+
+    arg_min = not_implemented()
+    arg_max = not_implemented()
+    arg_true = not_implemented()
+    head = not_implemented()
+    tail = not_implemented()
+    mode = not_implemented()
+    sort = not_implemented()
+    rank = not_implemented()
+    sample = not_implemented()
+    map_batches = not_implemented()
+    ewm_mean = not_implemented()
+    rolling_sum = not_implemented()
+    rolling_mean = not_implemented()
+    rolling_var = not_implemented()
+    rolling_std = not_implemented()
+    gather_every = not_implemented()
+    drop_nulls = not_implemented()
+    diff = not_implemented()
+    unique = not_implemented()
+    shift = not_implemented()
+    is_unique = not_implemented()
+    is_first_distinct = not_implemented()
+    is_last_distinct = not_implemented()
+    cum_sum = not_implemented()
+    cum_count = not_implemented()
+    cum_min = not_implemented()
+    cum_max = not_implemented()
+    cum_prod = not_implemented()
+    replace_strict = not_implemented()
+    over = not_implemented()
+
+    cat = not_implemented()  # pyright: ignore[reportAssignmentType]
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index cb3f132a..96a6f914 100644
--- a/narwhals/_expression_parsing.py
+++ b/narwhals/_expression_parsing.py
@@ -77,11 +77,7 @@ def evaluate_into_exprs(
     *exprs: CompliantExpr[CompliantFrameT, CompliantSeriesT_co],
 ) -> list[CompliantSeriesT_co]:
     """Evaluate each expr into Series."""
-    return [
-        item
-        for sublist in (evaluate_into_expr(df, into_expr) for into_expr in exprs)
-        for item in sublist
-    ]
+    return list(chain.from_iterable(evaluate_into_expr(df, expr) for expr in exprs))
 
 
 @overload
@@ -145,7 +141,6 @@ def reuse_series_implementation(
         attr: name of method.
         returns_scalar: whether the Series version returns a scalar. In this case,
             the expression version should return a 1-row Series.
-        args: arguments to pass to function.
         call_kwargs: non-expressifiable args which we may need to reuse in `agg` or `over`,
             such as `ddof` for `std` and `var`.
         expressifiable_args: keyword arguments to pass to function, which may
@@ -222,7 +217,6 @@ def reuse_series_namespace_implementation(
         expr: expression object.
         series_namespace: The Series namespace (e.g. `dt`, `cat`, `str`, `list`, `name`)
         attr: name of method.
-        args: arguments to pass to function.
         kwargs: keyword arguments to pass to function.
     """
     plx = expr.__narwhals_namespace__()
@@ -238,8 +232,8 @@ def reuse_series_namespace_implementation(
     )
 
 
-def is_simple_aggregation(expr: CompliantExpr[Any, Any]) -> bool:
-    """Check if expr is a very simple one.
+def is_elementary_expression(expr: CompliantExpr[Any, Any]) -> bool:
+    """Check if expr is elementary.
 
     Examples:
         - nw.col('a').mean()  # depth 1
@@ -250,7 +244,8 @@ def is_simple_aggregation(expr: CompliantExpr[Any, Any]) -> bool:
 
         - nw.col('a').filter(nw.col('b')>nw.col('c')).max()
 
-    because then, we can use a fastpath in pandas.
+    Elementary expressions are the only ones supported properly in
+    pandas, PyArrow, and Dask.
     """
     return expr._depth < 2
 
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index 3caf6c98..14b4625a 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -5,18 +5,19 @@ from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
 from typing import Literal
+from typing import Mapping
 from typing import Sequence
 
 from narwhals._expression_parsing import ExprKind
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
-from narwhals._expression_parsing import is_scalar_like
-from narwhals._expression_parsing import is_simple_aggregation
+from narwhals._expression_parsing import is_elementary_expression
 from narwhals._expression_parsing import reuse_series_implementation
 from narwhals._pandas_like.expr_cat import PandasLikeExprCatNamespace
 from narwhals._pandas_like.expr_dt import PandasLikeExprDateTimeNamespace
 from narwhals._pandas_like.expr_list import PandasLikeExprListNamespace
 from narwhals._pandas_like.expr_name import PandasLikeExprNameNamespace
 from narwhals._pandas_like.expr_str import PandasLikeExprStringNamespace
+from narwhals._pandas_like.group_by import AGGREGATIONS_TO_PANDAS_EQUIVALENT
 from narwhals._pandas_like.series import PandasLikeSeries
 from narwhals._pandas_like.utils import rename
 from narwhals.dependencies import get_numpy
@@ -33,7 +34,7 @@ if TYPE_CHECKING:
     from narwhals.utils import Implementation
     from narwhals.utils import Version
 
-MANY_TO_MANY_AGG_FUNCTIONS_TO_PANDAS_EQUIVALENT = {
+WINDOW_FUNCTIONS_TO_PANDAS_EQUIVALENT = {
     "cum_sum": "cumsum",
     "cum_min": "cummin",
     "cum_max": "cummax",
@@ -48,6 +49,26 @@ MANY_TO_MANY_AGG_FUNCTIONS_TO_PANDAS_EQUIVALENT = {
 }
 
 
+def window_kwargs_to_pandas_equivalent(
+    function_name: str, kwargs: dict[str, object]
+) -> dict[str, object]:
+    if function_name == "shift":
+        pandas_kwargs: dict[str, object] = {"periods": kwargs["n"]}
+    elif function_name == "rank":
+        _method = kwargs["method"]
+        pandas_kwargs = {
+            "method": "first" if _method == "ordinal" else _method,
+            "ascending": not kwargs["descending"],
+            "na_option": "keep",
+            "pct": False,
+        }
+    elif function_name.startswith("cum_"):  # Cumulative operation
+        pandas_kwargs = {"skipna": True}
+    else:  # e.g. std, var
+        pandas_kwargs = kwargs
+    return pandas_kwargs
+
+
 class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
     def __init__(
         self: Self,
@@ -366,7 +387,11 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
         return reuse_series_implementation(self, "drop_nulls")
 
     def replace_strict(
-        self: Self, old: Sequence[Any], new: Sequence[Any], *, return_dtype: DType | None
+        self: Self,
+        old: Sequence[Any] | Mapping[Any, Any],
+        new: Sequence[Any],
+        *,
+        return_dtype: DType | type[DType] | None,
     ) -> Self:
         return reuse_series_implementation(
             self, "replace_strict", old=old, new=new, return_dtype=return_dtype
@@ -432,81 +457,63 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
             call_kwargs=self._call_kwargs,
         )
 
-    def over(self: Self, keys: list[str], kind: ExprKind) -> Self:
-        if (
-            is_simple_aggregation(self)
-            and (function_name := re.sub(r"(\w+->)", "", self._function_name))
-            in MANY_TO_MANY_AGG_FUNCTIONS_TO_PANDAS_EQUIVALENT
-        ):
-
-            def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-                output_names, aliases = evaluate_output_names_and_aliases(self, df, [])
-
-                unsupported_reverse_msg = (
-                    "Cumulative operation with `reverse=True` is not supported in "
-                    "over context for pandas-like backend."
-                )
-                if function_name == "cum_count":
-                    if self._call_kwargs["reverse"]:
-                        raise NotImplementedError(unsupported_reverse_msg)
-                    plx = self.__narwhals_namespace__()
-                    df = df.with_columns(~plx.col(*output_names).is_null())
-
-                if function_name == "shift":
-                    kwargs = {"periods": self._call_kwargs["n"]}
-                elif function_name == "rank":
-                    _method = self._call_kwargs["method"]
-                    kwargs = {
-                        "method": "first" if _method == "ordinal" else _method,
-                        "ascending": not self._call_kwargs["descending"],
-                        "na_option": "keep",
-                        "pct": False,
-                    }
-                elif function_name.startswith("cum_"):  # Cumulative operation
-                    if self._call_kwargs["reverse"]:
-                        raise NotImplementedError(unsupported_reverse_msg)
-                    kwargs = {"skipna": True}
-                else:
-                    kwargs = {}
-
-                res_native = getattr(
-                    df._native_frame.groupby(keys)[list(output_names)],
-                    MANY_TO_MANY_AGG_FUNCTIONS_TO_PANDAS_EQUIVALENT[function_name],
-                )(**kwargs)
-                result_frame = df._from_native_frame(
-                    rename(
-                        res_native,
-                        columns=dict(zip(output_names, aliases)),
-                        implementation=self._implementation,
-                        backend_version=self._backend_version,
-                    )
-                )
-                return [result_frame[name] for name in aliases]
-        elif not is_scalar_like(kind):
+    def over(self: Self, partition_by: Sequence[str], kind: ExprKind) -> Self:
+        if not is_elementary_expression(self):
             msg = (
-                "Length-preserving operations are only supported in `over` context "
-                "for pandas if they are elementary "
-                "(e.g. `nw.col('a').cum_sum().over('b'))`)."
+                "Only elementary expressions are supported for `.over` in pandas-like backends.\n\n"
+                "Please see: "
+                "https://narwhals-dev.github.io/narwhals/pandas_like_concepts/improve_group_by_operation/"
             )
             raise NotImplementedError(msg)
-        else:
-
-            def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-                output_names, aliases = evaluate_output_names_and_aliases(self, df, [])
-                if overlap := set(output_names).intersection(keys):
-                    # E.g. `df.select(nw.all().sum().over('a'))`. This is well-defined,
-                    # we just don't support it yet.
-                    msg = (
-                        f"Column names {overlap} appear in both expression output names and in `over` keys.\n"
-                        "This is not yet supported."
-                    )
-                    raise NotImplementedError(msg)
+        function_name = re.sub(r"(\w+->)", "", self._function_name)
+        try:
+            pandas_function_name = WINDOW_FUNCTIONS_TO_PANDAS_EQUIVALENT[function_name]
+        except KeyError:
+            try:
+                pandas_function_name = AGGREGATIONS_TO_PANDAS_EQUIVALENT[function_name]
+            except KeyError:
+                msg = (
+                    f"Unsupported function: {function_name} in `over` context.\n\n"
+                    f"Supported functions are {', '.join(WINDOW_FUNCTIONS_TO_PANDAS_EQUIVALENT)}\n"
+                    f"and {', '.join(AGGREGATIONS_TO_PANDAS_EQUIVALENT)}."
+                )
+                raise NotImplementedError(msg) from None
+
+        def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
+            output_names, aliases = evaluate_output_names_and_aliases(self, df, [])
+            pandas_kwargs = window_kwargs_to_pandas_equivalent(
+                function_name, self._call_kwargs
+            )
 
-                tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
-                tmp = df.simple_select(*keys).join(
-                    tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
+            if function_name == "cum_count":
+                plx = self.__narwhals_namespace__()
+                df = df.with_columns(~plx.col(*output_names).is_null())
+            if function_name.startswith("cum_"):
+                reverse = self._call_kwargs["reverse"]
+            else:
+                assert "reverse" not in self._call_kwargs  # debug assertion  # noqa: S101
+                reverse = False
+            if reverse:
+                # Only select the columns we need to avoid reversing columns
+                # unnecessarily
+                columns = list(set(partition_by).union(output_names))
+                native_frame = df[columns]._native_frame[::-1]
+            else:
+                native_frame = df._native_frame
+            res_native = native_frame.groupby(partition_by)[list(output_names)].transform(
+                pandas_function_name, **pandas_kwargs
+            )
+            result_frame = df._from_native_frame(
+                rename(
+                    res_native,
+                    columns=dict(zip(output_names, aliases)),
+                    implementation=self._implementation,
+                    backend_version=self._backend_version,
                 )
-                return [tmp[name] for name in aliases]
+            )
+            if reverse:
+                return [result_frame[name][::-1] for name in aliases]
+            return [result_frame[name] for name in aliases]
 
         return self.__class__(
             func,
@@ -562,7 +569,7 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
     def map_batches(
         self: Self,
         function: Callable[[Any], Any],
-        return_dtype: DType | None,
+        return_dtype: DType | type[DType] | None,
     ) -> Self:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
             input_series_list = self._call(df)
diff --git a/narwhals/_pandas_like/group_by.py b/narwhals/_pandas_like/group_by.py
index 3bcb1ff9..0da57297 100644
--- a/narwhals/_pandas_like/group_by.py
+++ b/narwhals/_pandas_like/group_by.py
@@ -8,7 +8,7 @@ from typing import Any
 from typing import Iterator
 
 from narwhals._expression_parsing import evaluate_output_names_and_aliases
-from narwhals._expression_parsing import is_simple_aggregation
+from narwhals._expression_parsing import is_elementary_expression
 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
@@ -22,7 +22,7 @@ if TYPE_CHECKING:
     from narwhals._pandas_like.dataframe import PandasLikeDataFrame
     from narwhals._pandas_like.expr import PandasLikeExpr
 
-POLARS_TO_PANDAS_AGGREGATIONS = {
+AGGREGATIONS_TO_PANDAS_EQUIVALENT = {
     "sum": "sum",
     "mean": "mean",
     "median": "median",
@@ -84,9 +84,9 @@ class PandasLikeGroupBy:
             new_names.extend(aliases)
 
             if not (
-                is_simple_aggregation(expr)
+                is_elementary_expression(expr)
                 and re.sub(r"(\w+->)", "", expr._function_name)
-                in POLARS_TO_PANDAS_AGGREGATIONS
+                in AGGREGATIONS_TO_PANDAS_EQUIVALENT
             ):
                 all_aggs_are_simple = False
 
@@ -115,7 +115,7 @@ class PandasLikeGroupBy:
                 )
                 if expr._depth == 0:
                     # e.g. agg(nw.len()) # noqa: ERA001
-                    function_name = POLARS_TO_PANDAS_AGGREGATIONS.get(
+                    function_name = AGGREGATIONS_TO_PANDAS_EQUIVALENT.get(
                         expr._function_name, expr._function_name
                     )
                     simple_aggs_functions.add(function_name)
@@ -128,7 +128,7 @@ class PandasLikeGroupBy:
 
                 # e.g. agg(nw.mean('a')) # noqa: ERA001
                 function_name = re.sub(r"(\w+->)", "", expr._function_name)
-                function_name = POLARS_TO_PANDAS_AGGREGATIONS.get(
+                function_name = AGGREGATIONS_TO_PANDAS_EQUIVALENT.get(
                     function_name, function_name
                 )
 
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index f2c09ea8..7c487314 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -19,6 +19,7 @@ from narwhals._spark_like.utils import narwhals_to_native_dtype
 from narwhals.dependencies import get_pyspark
 from narwhals.typing import CompliantExpr
 from narwhals.utils import Implementation
+from narwhals.utils import not_implemented
 from narwhals.utils import parse_version
 
 if TYPE_CHECKING:
@@ -384,7 +385,13 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):  # type: ign
 
         from narwhals._spark_like.utils import _std
 
-        func = partial(_std, ddof=ddof, np_version=parse_version(np), functions=self._F)
+        func = partial(
+            _std,
+            ddof=ddof,
+            np_version=parse_version(np),
+            functions=self._F,
+            implementation=self._implementation,
+        )
 
         return self._from_call(func, "std")
 
@@ -395,7 +402,13 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):  # type: ign
 
         from narwhals._spark_like.utils import _var
 
-        func = partial(_var, ddof=ddof, np_version=parse_version(np), functions=self._F)
+        func = partial(
+            _var,
+            ddof=ddof,
+            np_version=parse_version(np),
+            functions=self._F,
+            implementation=self._implementation,
+        )
 
         return self._from_call(func, "var")
 
@@ -474,7 +487,7 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):  # type: ign
 
         return self._from_call(_n_unique, "n_unique")
 
-    def over(self: Self, keys: list[str], kind: ExprKind) -> Self:
+    def over(self: Self, keys: Sequence[str], kind: ExprKind) -> Self:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
             return [expr.over(self._Window.partitionBy(*keys)) for expr in self._call(df)]
 
@@ -514,3 +527,36 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):  # type: ign
     @property
     def list(self: Self) -> SparkLikeExprListNamespace:
         return SparkLikeExprListNamespace(self)
+
+    arg_min = not_implemented()
+    arg_max = not_implemented()
+    arg_true = not_implemented()
+    head = not_implemented()
+    tail = not_implemented()
+    mode = not_implemented()
+    sort = not_implemented()
+    rank = not_implemented()
+    sample = not_implemented()
+    map_batches = not_implemented()
+    ewm_mean = not_implemented()
+    rolling_sum = not_implemented()
+    rolling_mean = not_implemented()
+    rolling_var = not_implemented()
+    rolling_std = not_implemented()
+    gather_every = not_implemented()
+    drop_nulls = not_implemented()
+    diff = not_implemented()
+    unique = not_implemented()
+    shift = not_implemented()
+    is_first_distinct = not_implemented()
+    is_last_distinct = not_implemented()
+    cum_sum = not_implemented()
+    cum_count = not_implemented()
+    cum_min = not_implemented()
+    cum_max = not_implemented()
+    cum_prod = not_implemented()
+    replace_strict = not_implemented()
+    fill_null = not_implemented()
+    quantile = not_implemented()
+
+    cat = not_implemented()  # pyright: ignore[reportAssignmentType]
diff --git a/narwhals/_spark_like/utils.py b/narwhals/_spark_like/utils.py
index 29558c3b..6332f35d 100644
--- a/narwhals/_spark_like/utils.py
+++ b/narwhals/_spark_like/utils.py
@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
 from typing import Any
 
 from narwhals.exceptions import UnsupportedDTypeError
+from narwhals.utils import Implementation
 from narwhals.utils import import_dtypes_module
 from narwhals.utils import isinstance_or_issubclass
 
@@ -181,13 +182,17 @@ def maybe_evaluate_expr(df: SparkLikeLazyFrame, obj: SparkLikeExpr | object) ->
 
 
 def _std(
-    _input: Column | str, ddof: int, np_version: tuple[int, ...], functions: Any
+    _input: Column | str,
+    ddof: int,
+    np_version: tuple[int, ...],
+    functions: Any,
+    implementation: Implementation,
 ) -> Column:
-    if np_version > (2, 0):
-        if ddof == 1:
-            return functions.stddev_samp(_input)
+    if implementation is Implementation.SQLFRAME or np_version > (2, 0):
         if ddof == 0:
             return functions.stddev_pop(_input)
+        if ddof == 1:
+            return functions.stddev_samp(_input)
 
         n_rows = functions.count(_input)
         return functions.stddev_samp(_input) * functions.sqrt(
@@ -201,9 +206,15 @@ def _std(
 
 
 def _var(
-    _input: Column | str, ddof: int, np_version: tuple[int, ...], functions: Any
+    _input: Column | str,
+    ddof: int,
+    np_version: tuple[int, ...],
+    functions: Any,
+    implementation: Implementation,
 ) -> Column:
-    if np_version > (2, 0):
+    if implementation is Implementation.SQLFRAME or np_version > (2, 0):
+        if ddof == 0:
+            return functions.var_pop(_input)
         if ddof == 1:
             return functions.var_samp(_input)
 
diff --git a/narwhals/expr.py b/narwhals/expr.py
index ec3ad3b6..b829d378 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -31,13 +31,18 @@ if TYPE_CHECKING:
     from typing_extensions import Concatenate
     from typing_extensions import ParamSpec
     from typing_extensions import Self
+    from typing_extensions import TypeAlias
 
     from narwhals.dtypes import DType
     from narwhals.typing import CompliantExpr
+    from narwhals.typing import CompliantNamespace
     from narwhals.typing import IntoExpr
 
     PS = ParamSpec("PS")
     R = TypeVar("R")
+    _ToCompliant: TypeAlias = Callable[
+        [CompliantNamespace[Any, Any]], CompliantExpr[Any, Any]
+    ]
 
 
 class Expr:
@@ -47,7 +52,7 @@ class Expr:
         metadata: ExprMetadata,
     ) -> None:
         # callable from CompliantNamespace to CompliantExpr
-        self._to_compliant_expr = to_compliant_expr
+        self._to_compliant_expr: _ToCompliant = to_compliant_expr
         self._metadata = metadata
 
     def _from_callable(self, to_compliant_expr: Callable[[Any], Any]) -> Self:
@@ -608,7 +613,7 @@ class Expr:
 
     def map_batches(
         self: Self,
-        function: Callable[[Any], Self],
+        function: Callable[[Any], CompliantExpr[Any, Any]],
         return_dtype: DType | None = None,
     ) -> Self:
         """Apply a custom python function to a whole Series or sequence of Series.
diff --git a/narwhals/functions.py b/narwhals/functions.py
index e1f85f5d..a1fe424d 100644
--- a/narwhals/functions.py
+++ b/narwhals/functions.py
@@ -1490,7 +1490,7 @@ class Then(Expr):
             compliant_value = extract_compliant(plx, value, str_as_lit=False)
             if is_scalar_like(kind) and is_compliant_expr(compliant_value):
                 compliant_value = compliant_value.broadcast(kind)
-            return compliant_expr.otherwise(compliant_value)  # type: ignore[no-any-return]
+            return compliant_expr.otherwise(compliant_value)  # type: ignore[attr-defined, no-any-return]
 
         return Expr(
             func,
@@ -1502,14 +1502,15 @@ def when(*predicates: IntoExpr | Iterable[IntoExpr]) -> When:
     """Start a `when-then-otherwise` expression.
 
     Expression similar to an `if-else` statement in Python. Always initiated by a
-    `pl.when(<condition>).then(<value if condition>)`, and optionally followed by
-    chaining one or more `.when(<condition>).then(<value>)` statements.
-    Chained when-then operations should be read as Python `if, elif, ... elif`
-    blocks, not as `if, if, ... if`, i.e. the first condition that evaluates to
-    `True` will be picked.
-    If none of the conditions are `True`, an optional
-    `.otherwise(<value if all statements are false>)` can be appended at the end.
-    If not appended, and none of the conditions are `True`, `None` will be returned.
+    `pl.when(<condition>).then(<value if condition>)`, and optionally followed by a
+    `.otherwise(<value if condition is false>)` can be appended at the end. If not
+    appended, and the condition is not `True`, `None` will be returned.
+
+    !!! info
+
+        Chaining multiple `.when(<condition>).then(<value>)` statements is currently
+        not supported.
+        See [Narwhals#668](https://github.com/narwhals-dev/narwhals/issues/668).
 
     Arguments:
         predicates: Condition(s) that must be met in order to apply the subsequent
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index fbdc9351..da25bf60 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -2142,14 +2142,15 @@ def when(*predicates: IntoExpr | Iterable[IntoExpr]) -> When:
     """Start a `when-then-otherwise` expression.
 
     Expression similar to an `if-else` statement in Python. Always initiated by a
-    `pl.when(<condition>).then(<value if condition>)`, and optionally followed by
-    chaining one or more `.when(<condition>).then(<value>)` statements.
-    Chained when-then operations should be read as Python `if, elif, ... elif`
-    blocks, not as `if, if, ... if`, i.e. the first condition that evaluates to
-    `True` will be picked.
-    If none of the conditions are `True`, an optional
-    `.otherwise(<value if all statements are false>)` can be appended at the end.
-    If not appended, and none of the conditions are `True`, `None` will be returned.
+    `pl.when(<condition>).then(<value if condition>)`, and optionally followed by a
+    `.otherwise(<value if condition is false>)` can be appended at the end. If not
+    appended, and the condition is not `True`, `None` will be returned.
+
+    !!! info
+
+        Chaining multiple `.when(<condition>).then(<value>)` statements is currently
+        not supported.
+        See [Narwhals#668](https://github.com/narwhals-dev/narwhals/issues/668).
 
     Arguments:
         predicates: Condition(s) that must be met in order to apply the subsequent
diff --git a/narwhals/typing.py b/narwhals/typing.py
index d131a32e..b908e671 100644
--- a/narwhals/typing.py
+++ b/narwhals/typing.py
@@ -1,18 +1,21 @@
 from __future__ import annotations
 
+import sys
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
 from typing import Iterator
 from typing import Literal
+from typing import Mapping
 from typing import Protocol
 from typing import Sequence
 from typing import TypeVar
 from typing import Union
 
-if not TYPE_CHECKING:
-    import sys
+from narwhals.utils import deprecated
+from narwhals.utils import unstable
 
+if not TYPE_CHECKING:
     if sys.version_info >= (3, 9):
         from typing import Protocol as Protocol38
     else:
@@ -126,8 +129,149 @@ class CompliantExpr(Protocol38[CompliantFrameT, CompliantSeriesT_co]):
         self,
     ) -> CompliantNamespace[CompliantFrameT, CompliantSeriesT_co]: ...
     def is_null(self) -> Self: ...
+    def abs(self) -> Self: ...
+    def all(self) -> Self: ...
+    def any(self) -> Self: ...
     def alias(self, name: str) -> Self: ...
-    def cast(self, dtype: DType) -> Self: ...
+    def cast(self, dtype: DType | type[DType]) -> Self: ...
+    def count(self) -> Self: ...
+    def min(self) -> Self: ...
+    def max(self) -> Self: ...
+    def arg_min(self) -> Self: ...
+    def arg_max(self) -> Self: ...
+    def arg_true(self) -> Self: ...
+    def mean(self) -> Self: ...
+    def sum(self) -> Self: ...
+    def median(self) -> Self: ...
+    def skew(self) -> Self: ...
+    def std(self, *, ddof: int) -> Self: ...
+    def var(self, *, ddof: int) -> Self: ...
+    def n_unique(self) -> Self: ...
+    def null_count(self) -> Self: ...
+    def drop_nulls(self) -> Self: ...
+    def fill_null(
+        self,
+        value: Any | None,
+        strategy: Literal["forward", "backward"] | None,
+        limit: int | None,
+    ) -> Self: ...
+    def diff(self) -> Self: ...
+    def unique(self) -> Self: ...
+    def len(self) -> Self: ...
+    def round(self, decimals: int) -> Self: ...
+    def mode(self) -> Self: ...
+    def head(self, n: int) -> Self: ...
+    def tail(self, n: int) -> Self: ...
+    def shift(self, n: int) -> Self: ...
+    def is_finite(self) -> Self: ...
+    def is_nan(self) -> Self: ...
+    def is_unique(self) -> Self: ...
+    def is_first_distinct(self) -> Self: ...
+    def is_last_distinct(self) -> Self: ...
+    def cum_sum(self, *, reverse: bool) -> Self: ...
+    def cum_count(self, *, reverse: bool) -> Self: ...
+    def cum_min(self, *, reverse: bool) -> Self: ...
+    def cum_max(self, *, reverse: bool) -> Self: ...
+    def cum_prod(self, *, reverse: bool) -> Self: ...
+    def is_in(self, other: Any) -> Self: ...
+    def sort(self, *, descending: bool, nulls_last: bool) -> Self: ...
+    def rank(
+        self,
+        method: Literal["average", "min", "max", "dense", "ordinal"],
+        *,
+        descending: bool,
+    ) -> Self: ...
+    def replace_strict(
+        self,
+        old: Sequence[Any] | Mapping[Any, Any],
+        new: Sequence[Any],
+        *,
+        return_dtype: DType | type[DType] | None,
+    ) -> Self: ...
+    def over(self: Self, keys: Sequence[str], kind: ExprKind) -> Self: ...
+    def sample(
+        self,
+        n: int | None,
+        *,
+        fraction: float | None,
+        with_replacement: bool,
+        seed: int | None,
+    ) -> Self: ...
+    def quantile(
+        self,
+        quantile: float,
+        interpolation: Literal["nearest", "higher", "lower", "midpoint", "linear"],
+    ) -> Self: ...
+    def map_batches(
+        self,
+        function: Callable[[CompliantSeries], CompliantExpr[Any, Any]],
+        return_dtype: DType | type[DType] | None,
+    ) -> Self: ...
+
+    @property
+    def str(self) -> Any: ...
+    @property
+    def name(self) -> Any: ...
+    @property
+    def dt(self) -> Any: ...
+    @property
+    def cat(self) -> Any: ...
+    @property
+    def list(self) -> Any: ...
+
+    @unstable
+    def ewm_mean(
+        self,
+        *,
+        com: float | None,
+        span: float | None,
+        half_life: float | None,
+        alpha: float | None,
+        adjust: bool,
+        min_samples: int,
+        ignore_nulls: bool,
+    ) -> Self: ...
+
+    @unstable
+    def rolling_sum(
+        self,
+        window_size: int,
+        *,
+        min_samples: int | None,
+        center: bool,
+    ) -> Self: ...
+
+    @unstable
+    def rolling_mean(
+        self,
+        window_size: int,
+        *,
+        min_samples: int | None,
+        center: bool,
+    ) -> Self: ...
+
+    @unstable
+    def rolling_var(
+        self,
+        window_size: int,
+        *,
+        min_samples: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self: ...
+
+    @unstable
+    def rolling_std(
+        self,
+        window_size: int,
+        *,
+        min_samples: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self: ...
+
+    @deprecated("Since `1.22.0`")
+    def gather_every(self, n: int, offset: int) -> Self: ...
     def __and__(self, other: Any) -> Self: ...
     def __or__(self, other: Any) -> Self: ...
     def __add__(self, other: Any) -> Self: ...
@@ -141,6 +285,7 @@ class CompliantExpr(Protocol38[CompliantFrameT, CompliantSeriesT_co]):
     def __ge__(self, other: Any) -> Self: ...
     def __lt__(self, other: Any) -> Self: ...
     def __le__(self, other: Any) -> Self: ...
+    def __invert__(self) -> Self: ...
     def broadcast(
         self, kind: Literal[ExprKind.AGGREGATION, ExprKind.LITERAL]
     ) -> Self: ...
diff --git a/narwhals/utils.py b/narwhals/utils.py
index a75ead5f..43844171 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -76,12 +76,16 @@ if TYPE_CHECKING:
     _T1 = TypeVar("_T1")
     _T2 = TypeVar("_T2")
     _T3 = TypeVar("_T3")
+    _Fn = TypeVar("_Fn", bound="Callable[..., Any]")
 
     _TracksDepth: TypeAlias = "Literal[Implementation.DASK,Implementation.CUDF,Implementation.MODIN,Implementation.PANDAS,Implementation.PYSPARK]"
 
     class _SupportsVersion(Protocol):
         __version__: str
 
+    class _SupportsGet(Protocol):  # noqa: PYI046
+        def __get__(self, instance: Any, owner: Any | None = None, /) -> Any: ...
+
     class _StoresImplementation(Protocol):
         _implementation: Implementation
         """Implementation of native object (pandas, Polars, PyArrow, ...)."""
@@ -1405,3 +1409,124 @@ def _supports_dataframe_interchange(obj: Any) -> TypeIs[DataFrameLike]:
 def is_tracks_depth(obj: Implementation, /) -> TypeIs[_TracksDepth]:  # pragma: no cover
     # Return `True` for implementations that utilize `CompliantExpr._depth`.
     return obj.is_pandas_like() or obj in {Implementation.PYARROW, Implementation.DASK}
+
+
+# TODO @dangotbanned: Extend with runtime behavior for `v1.*`
+# See `narwhals.exceptions.NarwhalsUnstableWarning`
+def unstable(fn: _Fn, /) -> _Fn:
+    """Visual-only marker for unstable functionality.
+
+    Arguments:
+        fn: Function to decorate.
+
+    Returns:
+        Decorated function (unchanged).
+
+    Examples:
+        >>> from narwhals.utils import unstable
+        >>> @unstable
+        ... def a_work_in_progress_feature(*args):
+        ...     return args
+        >>>
+        >>> a_work_in_progress_feature.__name__
+        'a_work_in_progress_feature'
+        >>> a_work_in_progress_feature(1, 2, 3)
+        (1, 2, 3)
+    """
+    return fn
+
+
+class not_implemented:  # noqa: N801
+    """Mark some functionality as unsupported.
+
+    Arguments:
+        alias: optional name used instead of the data model hook [`__set_name__`].
+
+    Returns:
+        An exception-raising [descriptor].
+
+    Notes:
+        - Attribute/method name *doesn't* need to be declared twice
+        - Allows different behavior when looked up on the class vs instance
+        - Allows us to use `isinstance(...)` instead of monkeypatching an attribute to the function
+
+    Examples:
+        >>> from narwhals.utils import not_implemented
+        >>> class Thing:
+        ...     def totally_ready(self) -> str:
+        ...         return "I'm ready!"
+        ...
+        ...     not_ready_yet = not_implemented()
+        >>>
+        >>> thing = Thing()
+        >>> thing.totally_ready()
+        "I'm ready!"
+        >>> thing.not_ready_yet()
+        Traceback (most recent call last):
+            ...
+        NotImplementedError: 'not_ready_yet' is not implemented for: 'Thing'.
+        ...
+        >>> isinstance(Thing.not_ready_yet, not_implemented)
+        True
+
+    [`__set_name__`]: https://docs.python.org/3/reference/datamodel.html#object.__set_name__
+    [descriptor]: https://docs.python.org/3/howto/descriptor.html
+    """
+
+    def __init__(self, alias: str | None = None, /) -> None:
+        # NOTE: Don't like this
+        # Trying to workaround `mypy` requiring `@property` everywhere
+        self._alias: str | None = alias
+
+    def __repr__(self) -> str:
+        return f"<{type(self).__name__}>: {self._name_owner}.{self._name}"
+
+    def __set_name__(self, owner: type[_T], name: str) -> None:
+        # https://docs.python.org/3/howto/descriptor.html#customized-names
+        self._name_owner: str = owner.__name__
+        self._name: str = self._alias or name
+
+    def __get__(
+        self, instance: _T | Literal["raise"] | None, owner: type[_T] | None = None, /
+    ) -> Any:
+        if instance is None:
+            # NOTE: Branch for `cls._name`
+            # We can check that to see if an instance of `type(self)` for
+            # https://narwhals-dev.github.io/narwhals/api-completeness/expr/
+            return self
+        # NOTE: Prefer not exposing the actual class we're defining in
+        # `_implementation` may not be available everywhere
+        who = getattr(instance, "_implementation", self._name_owner)
+        raise _not_implemented_error(self._name, who)
+
+    def __call__(self, *args: Any, **kwds: Any) -> Any:
+        # NOTE: Purely to duck-type as assignable to **any** instance method
+        # Wouldn't be reachable through *regular* attribute access
+        return self.__get__("raise")
+
+
+def _not_implemented_error(what: str, who: str, /) -> NotImplementedError:
+    msg = (
+        f"{what!r} is not implemented for: {who!r}.\n\n"
+        "If you would like to see this functionality in `narwhals`, "
+        "please open an issue at: https://github.com/narwhals-dev/narwhals/issues"
+    )
+    return NotImplementedError(msg)
+
+
+if TYPE_CHECKING:
+    import sys
+
+    if sys.version_info >= (3, 13):
+        # NOTE: avoids `mypy`
+        #     error: Module "narwhals.utils" does not explicitly export attribute "deprecated"  [attr-defined]
+        from warnings import deprecated as deprecated  # noqa: PLC0414
+    else:
+        from typing_extensions import deprecated as deprecated  # noqa: PLC0414
+else:
+
+    def deprecated(message: str, /) -> Callable[[_Fn], _Fn]:  # noqa: ARG001
+        def wrapper(func: _Fn, /) -> _Fn:
+            return func
+
+        return wrapper
diff --git a/pyproject.toml b/pyproject.toml
index 8de1ec51..21a252f2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,7 +5,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "narwhals"
-version = "1.29.0"
+version = "1.29.1"
 dependencies = []
 requires-python = ">=3.8"
 authors = [
@@ -195,15 +195,16 @@ filterwarnings = [
   # This warning was temporarily raised by pandas but then reverted.
   'ignore:.*Passing a BlockManager to DataFrame:DeprecationWarning',
   # This warning was temporarily raised by Polars but then reverted.
-  'ignore:.*The default coalesce behavior of left join will change:DeprecationWarning',
+  'ignore:.*The default coalesce behavior of left join will change:DeprecationWarning:polars',
   'ignore: unclosed <socket.socket',
   'ignore:.*The distutils package is deprecated and slated for removal in Python 3.12:DeprecationWarning:pyspark',
   'ignore:.*distutils Version classes are deprecated. Use packaging.version instead.*:DeprecationWarning:pyspark',
   'ignore:.*is_datetime64tz_dtype is deprecated and will be removed in a future version.*:DeprecationWarning:pyspark',
   # Warning raised by PyArrow nightly just by importing pandas
   'ignore:.*Python binding for RankQuantileOptions not exposed:RuntimeWarning:pyarrow',
-  'ignore:.*pandas only supports SQLAlchemy:UserWarning',
+  'ignore:.*pandas only supports SQLAlchemy:UserWarning:sqlframe',
   'ignore:.*numpy.core is deprecated and has been renamed to numpy._core.*:DeprecationWarning:sqlframe',
+  "ignore:.*__array__ implementation doesn't accept a copy keyword, so passing copy=False failed:DeprecationWarning:modin"
 ]
 xfail_strict = true
 markers = ["slow: marks tests as slow (deselect with '-m \"not slow\"')"]
@@ -237,7 +238,8 @@ exclude_also = [
   'request.applymarker\(pytest.mark.xfail',
   'backend_version <',
   'if "cudf" in str\(constructor',
-  'if "pyspark" in str\(constructor'
+  'if "pyspark" in str\(constructor',
+  'pytest.skip\('
 ]
 
 [tool.mypy]
diff --git a/utils/bump_version.py b/utils/bump_version.py
index 59d1aa7f..048f82a0 100644
--- a/utils/bump_version.py
+++ b/utils/bump_version.py
@@ -77,3 +77,4 @@ with open("docs/installation.md", "w", encoding="utf-8") as f:
 subprocess.run(["git", "commit", "-a", "-m", f"release: Bump version to {version}"])
 subprocess.run(["git", "tag", "-a", f"v{version}", "-m", f"v{version}"])
 subprocess.run(["git", "push", "upstream", "HEAD", "--follow-tags"])
+subprocess.run(["git", "push", "upstream", "HEAD:stable"])
diff --git a/utils/generate_backend_completeness.py b/utils/generate_backend_completeness.py
index 06e47270..b269a94b 100644
--- a/utils/generate_backend_completeness.py
+++ b/utils/generate_backend_completeness.py
@@ -2,16 +2,21 @@ from __future__ import annotations
 
 import importlib
 import inspect
+from contextlib import suppress
 from enum import Enum
 from enum import auto
+from itertools import chain
 from pathlib import Path
 from typing import Any
 from typing import Final
+from typing import Iterator
 from typing import NamedTuple
 
 import polars as pl
 from jinja2 import Template
 
+from narwhals.utils import not_implemented
+
 TEMPLATE_PATH: Final[Path] = Path("utils") / "api-completeness.md.jinja"
 DESTINATION_PATH: Final[Path] = Path("docs") / "api-completeness"
 
@@ -62,8 +67,16 @@ def get_class_methods(kls: type[Any]) -> list[str]:
     return [m[0] for m in inspect.getmembers(kls) if not m[0].startswith("_")]
 
 
+def iter_implemented_methods(tp: type[Any], /) -> Iterator[str]:
+    """Variant of `get_class_methods` to exclude `not_implemented`."""
+    for name, member in inspect.getmembers(tp):
+        if not name.startswith("_") and not isinstance(member, not_implemented):
+            yield name
+
+
 def parse_module(module_name: str, backend: str, nw_class_name: str) -> list[str]:
-    try:
+    methods_ = []
+    with suppress(ModuleNotFoundError):
         module_ = importlib.import_module(f"narwhals.{backend}.{module_name}")
         class_ = inspect.getmembers(
             module_,
@@ -75,23 +88,20 @@ def parse_module(module_name: str, backend: str, nw_class_name: str) -> list[str
             ),
         )
 
-        methods_ = (
-            get_class_methods(class_[0][1]) + DIRECTLY_IMPLEMENTED_METHODS
-            if class_
-            else []
+        if not class_:
+            return methods_
+        methods_.extend(
+            chain(iter_implemented_methods(class_[0][1]), DIRECTLY_IMPLEMENTED_METHODS)
         )
-
-        if module_name == "expr_str" and class_:
-            methods_ += EXPR_STR_METHODS
-
-    except ModuleNotFoundError:
-        methods_ = []
-
+        if module_name == "expr_str":
+            methods_.extend(EXPR_STR_METHODS)
     return methods_
 
 
 def render_table_and_write_to_output(
-    results: list[pl.DataFrame], title: str, output_filename: str
+    results: list[pl.DataFrame],  # pyright: ignore[reportRedeclaration]
+    title: str,
+    output_filename: str,
 ) -> None:
     results: pl.DataFrame = (
         pl.concat(results)
@@ -125,7 +135,7 @@ def render_table_and_write_to_output(
     with (DESTINATION_PATH / f"{output_filename}.md").open(mode="w") as destination:
         destination.write(new_content)
 
-    return table
+    return table  # pyright: ignore[reportReturnType]
 
 
 def get_backend_completeness_table() -> None:
diff --git a/utils/generate_zen_content.py b/utils/generate_zen_content.py
index 31fe8af4..cb742d30 100644
--- a/utils/generate_zen_content.py
+++ b/utils/generate_zen_content.py
@@ -23,5 +23,5 @@ import narwhals.this
 ```
 """
 
-with DESTINATION_PATH.open(mode="w") as destination:
+with DESTINATION_PATH.open(mode="w", encoding="utf-8", newline="\n") as destination:
     destination.write(content)


Test patch:
diff --git a/.github/workflows/downstream_tests.yml b/.github/workflows/downstream_tests.yml
index 835fc658..fe6e04a2 100644
--- a/.github/workflows/downstream_tests.yml
+++ b/.github/workflows/downstream_tests.yml
@@ -306,7 +306,7 @@ jobs:
       - name: install-deps
         run: |
           cd plotly.py
-          uv pip install -r test_requirements/requirements_312_optional.txt --system
+          uv pip install -r test_requirements/requirements_core.txt -r test_requirements/requirements_optional.txt --system
           uv pip install -e . --system
       - name: install-narwhals-dev
         run: |
@@ -493,3 +493,43 @@ jobs:
           touch tests/utils/__init__.py
           pytest tests
         timeout-minutes: 15
+  
+  darts:
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+        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-darts
+        run: |
+          git clone https://github.com/unit8co/darts.git --depth=1
+          cd darts
+          git log
+      - name: install-deps
+        run: |
+          cd darts
+          uv pip install -r requirements/core.txt -r requirements/dev.txt --system
+      - name: install-narwhals-dev
+        run: |
+          uv pip uninstall narwhals --system
+          # TODO(FBruzzesi): Install polars from darts requirements when they introcude it
+          uv pip install -e ".[polars]" --system
+      - name: show-deps
+        run: uv pip freeze
+      - name: Run tests
+        run: |
+          cd darts
+          pytest darts/tests/test_timeseries.py
+          pytest darts/tests/test_timeseries_multivariate.py
+          pytest darts/tests/test_timeseries_static_covariates.py
diff --git a/tests/expr_and_series/over_test.py b/tests/expr_and_series/over_test.py
index ebbda168..5c72dc00 100644
--- a/tests/expr_and_series/over_test.py
+++ b/tests/expr_and_series/over_test.py
@@ -30,8 +30,6 @@ 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 "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
 
@@ -49,9 +47,44 @@ def test_over_single(request: pytest.FixtureRequest, constructor: Constructor) -
     assert_equal_data(result, expected)
 
 
-def test_over_multiple(request: pytest.FixtureRequest, constructor: Constructor) -> None:
-    if "dask_lazy_p2" in str(constructor):
+def test_over_std_var(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor):
+        # https://github.com/rapidsai/cudf/issues/18159
+        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_std0": [0.5, 0.5, 0.816496580927726, 0.816496580927726, 0.816496580927726],
+        "c_std1": [0.7071067811865476, 0.7071067811865476, 1.0, 1.0, 1.0],
+        "c_var0": [
+            0.25,
+            0.25,
+            0.6666666666666666,
+            0.6666666666666666,
+            0.6666666666666666,
+        ],
+        "c_var1": [0.5, 0.5, 1.0, 1.0, 1.0],
+    }
+
+    result = (
+        df.with_columns(
+            c_std0=nw.col("c").std(ddof=0).over("a"),
+            c_std1=nw.col("c").std(ddof=1).over("a"),
+            c_var0=nw.col("c").var(ddof=0).over("a"),
+            c_var1=nw.col("c").var(ddof=1).over("a"),
+        )
+        .sort("i")
+        .drop("i")
+    )
+    assert_equal_data(result, expected)
+
+
+def test_over_multiple(request: pytest.FixtureRequest, constructor: Constructor) -> None:
     if "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
 
@@ -79,6 +112,9 @@ def test_over_cumsum(
         request.applymarker(pytest.mark.xfail)
     if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor_eager(data_cum))
     expected = {
@@ -102,6 +138,9 @@ def test_over_cumcount(
 ) -> None:
     if "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor_eager(data_cum))
     expected = {
@@ -127,6 +166,9 @@ def test_over_cummax(
         request.applymarker(pytest.mark.xfail)
     if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
     df = nw.from_native(constructor_eager(data_cum))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
@@ -150,6 +192,9 @@ def test_over_cummin(
         request.applymarker(pytest.mark.xfail)
     if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor_eager(data_cum))
     expected = {
@@ -175,6 +220,9 @@ def test_over_cumprod(
         request.applymarker(pytest.mark.xfail)
     if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
         request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor_eager(data_cum))
     expected = {
@@ -193,7 +241,12 @@ def test_over_cumprod(
     assert_equal_data(result, expected)
 
 
-def test_over_anonymous_cumulative(constructor_eager: ConstructorEager) -> None:
+def test_over_anonymous_cumulative(
+    constructor_eager: ConstructorEager, request: pytest.FixtureRequest
+) -> None:
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
     df = nw.from_native(constructor_eager({"a": [1, 1, 2], "b": [4, 5, 6]}))
     context = (
         pytest.raises(NotImplementedError)
@@ -224,13 +277,14 @@ def test_over_anonymous_reduction(
     if "duckdb" in str(constructor):
         # TODO(unassigned): we should be able to support these
         request.applymarker(pytest.mark.xfail)
+    if "modin" in str(constructor):
+        # probably bugged
+        request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor({"a": [1, 1, 2], "b": [4, 5, 6]}))
     context = (
         pytest.raises(NotImplementedError)
         if df.implementation.is_pyarrow()
-        or df.implementation.is_pandas_like()
-        or df.implementation.is_dask()
         else does_not_raise()
     )
     with context:
@@ -248,11 +302,29 @@ def test_over_anonymous_reduction(
         assert_equal_data(result, expected)
 
 
+def test_over_unsupported() -> None:
+    dfpd = pd.DataFrame({"a": [1, 1, 2], "b": [4, 5, 6]})
+    with pytest.raises(NotImplementedError):
+        nw.from_native(dfpd).select(nw.col("a").round().over("a"))
+
+
+def test_over_unsupported_dask() -> None:
+    pytest.importorskip("dask")
+    import dask.dataframe as dd
+
+    df = dd.from_pandas(pd.DataFrame({"a": [1, 1, 2], "b": [4, 5, 6]}))
+    with pytest.raises(NotImplementedError):
+        nw.from_native(df).select(nw.col("a").round().over("a"))
+
+
 def test_over_shift(
     request: pytest.FixtureRequest, constructor_eager: ConstructorEager
 ) -> None:
     if "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor_eager(data))
     expected = {
@@ -270,6 +342,11 @@ def test_over_diff(
 ) -> None:
     if "pyarrow_table" in str(constructor_eager):
         request.applymarker(pytest.mark.xfail)
+    if "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1):
+        pytest.skip(reason="bug in old version")
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
 
     df = nw.from_native(constructor_eager(data))
     expected = {
@@ -282,14 +359,40 @@ def test_over_diff(
     assert_equal_data(result, expected)
 
 
-def test_over_cum_reverse() -> None:
-    df = pd.DataFrame({"a": [1, 1, 2], "b": [4, 5, 6]})
-
-    with pytest.raises(
-        NotImplementedError,
-        match=r"Cumulative operation with `reverse=True` is not supported",
-    ):
-        nw.from_native(df).select(nw.col("b").cum_max(reverse=True).over("a"))
+@pytest.mark.parametrize(
+    ("attr", "expected_b"),
+    [
+        ("cum_max", [5, 5, 9, None, 9]),
+        ("cum_min", [4, 5, 7, None, 9]),
+        ("cum_sum", [9, 5, 16, None, 9]),
+        ("cum_count", [2, 1, 2, 1, 1]),
+        ("cum_prod", [20, 5, 63, None, 9]),
+    ],
+)
+def test_over_cum_reverse(
+    constructor_eager: ConstructorEager,
+    request: pytest.FixtureRequest,
+    attr: str,
+    expected_b: list[object],
+) -> None:
+    if "pyarrow_table" in str(constructor_eager):
+        request.applymarker(pytest.mark.xfail)
+    if "pandas_nullable" in str(constructor_eager) and attr in {"cum_max", "cum_min"}:
+        # https://github.com/pandas-dev/pandas/issues/61031
+        request.applymarker(pytest.mark.xfail)
+    if "cudf" in str(constructor_eager):
+        # https://github.com/rapidsai/cudf/issues/18159
+        request.applymarker(pytest.mark.xfail)
+    df = constructor_eager(
+        {
+            "a": [1, 1, 2, 2, 2],
+            "b": [4, 5, 7, None, 9],
+        }
+    )
+    expr = getattr(nw.col("b"), attr)(reverse=True)
+    result = nw.from_native(df).with_columns(expr.over("a"))
+    expected = {"a": [1, 1, 2, 2, 2], "b": expected_b}
+    assert_equal_data(result, expected)
 
 
 def test_over_raise_len_change(constructor: Constructor) -> None:
diff --git a/tests/expr_and_series/rank_test.py b/tests/expr_and_series/rank_test.py
index 99a64371..90db9a84 100644
--- a/tests/expr_and_series/rank_test.py
+++ b/tests/expr_and_series/rank_test.py
@@ -102,14 +102,17 @@ def test_rank_expr_in_over_context(
     constructor_eager: ConstructorEager,
     method: Literal["average", "min", "max", "dense", "ordinal"],
 ) -> None:
-    if any(x in str(constructor_eager) for x in ("pyarrow_table", "dask")):
+    if any(x in str(constructor_eager) for x in ("pyarrow_table", "dask", "cudf")):
         # Pyarrow raises:
         # > pyarrow.lib.ArrowKeyError: No function registered with name: hash_rank
         # We can handle that to provide a better error message.
+        # cudf: https://github.com/rapidsai/cudf/issues/18159
         request.applymarker(pytest.mark.xfail)
 
     if "pandas_pyarrow" in str(constructor_eager) and PANDAS_VERSION < (2, 1):
-        request.applymarker(pytest.mark.xfail)
+        pytest.skip(reason="bug in old version")
+    if "pandas" in str(constructor_eager) and PANDAS_VERSION < (1, 1):
+        pytest.skip(reason="bug in old version")
 
     df = nw.from_native(constructor_eager(data_float))
 
diff --git a/tests/frame/select_test.py b/tests/frame/select_test.py
index 5e58b7a2..c490a9f0 100644
--- a/tests/frame/select_test.py
+++ b/tests/frame/select_test.py
@@ -154,6 +154,9 @@ def test_filtration_vs_aggregation(constructor_eager: ConstructorEager) -> None:
 
 
 def test_select_duplicates(constructor: Constructor) -> None:
+    if "cudf" in str(constructor):
+        # cudf already raises its own error
+        pytest.skip()
     df = nw.from_native(constructor({"a": [1, 2]})).lazy()
     with pytest.raises(ValueError, match="Expected unique|duplicate|more than one"):
         df.select("a", nw.col("a") + 1).collect()
diff --git a/tests/series_only/scatter_test.py b/tests/series_only/scatter_test.py
index 11065ec9..d7f0aa77 100644
--- a/tests/series_only/scatter_test.py
+++ b/tests/series_only/scatter_test.py
@@ -49,3 +49,28 @@ def test_single_series(constructor_eager: ConstructorEager) -> None:
     s.scatter([0, 1], [999, 888])
     expected = {"a": [1, 2, 3]}
     assert_equal_data({"a": s}, expected)
+
+
+def test_scatter_integer(constructor_eager: ConstructorEager) -> None:
+    df = nw.from_native(
+        constructor_eager({"a": [1, 2, 3], "b": [142, 124, 132]}), eager_only=True
+    )
+    s = df["a"]
+    result = s.scatter(0, 999)
+    expected = {"a": [999, 2, 3]}
+    assert_equal_data({"a": result}, expected)
+
+
+def test_scatter_unordered_indices(
+    constructor_eager: ConstructorEager, request: pytest.FixtureRequest
+) -> None:
+    if "modin" in str(constructor_eager):
+        # bugged
+        request.applymarker(pytest.mark.xfail)
+    data = {
+        "a": [16, 12, 10, 9, 6, 5, 2],
+    }
+    indices = [6, 1, 0, 5, 3, 2, 4]
+    df = nw.from_native(constructor_eager(data))
+    result = df["a"].scatter(indices, df["a"])
+    assert_equal_data({"a": result}, {"a": [10, 12, 5, 6, 2, 9, 16]})
diff --git a/tests/utils_test.py b/tests/utils_test.py
index 0459620d..b0494a04 100644
--- a/tests/utils_test.py
+++ b/tests/utils_test.py
@@ -5,17 +5,20 @@ import string
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Protocol
 from typing import cast
 
 import hypothesis.strategies as st
 import pandas as pd
 import polars as pl
+import pyarrow as pa
 import pytest
 from hypothesis import given
 from pandas.testing import assert_frame_equal
 from pandas.testing import assert_index_equal
 from pandas.testing import assert_series_equal
 
+import narwhals as unstable_nw
 import narwhals.stable.v1 as nw
 from narwhals.exceptions import ColumnNotFoundError
 from narwhals.utils import check_column_exists
@@ -24,6 +27,8 @@ from tests.utils import PANDAS_VERSION
 from tests.utils import get_module_version_as_tuple
 
 if TYPE_CHECKING:
+    from typing_extensions import Self
+
     from narwhals.series import Series
     from narwhals.typing import IntoSeries
     from narwhals.utils import _SupportsVersion
@@ -312,3 +317,113 @@ def test_check_column_exists() -> None:
         match=re.escape("Column(s) ['d', 'f'] not found in ['a', 'b', 'c']"),
     ):
         check_column_exists(columns, subset)
+
+
+def test_not_implemented() -> None:
+    from narwhals._arrow.expr import ArrowExpr
+    from narwhals._polars.expr import PolarsExpr
+    from narwhals._polars.expr import PolarsExprStringNamespace
+    from narwhals.utils import not_implemented
+
+    data: dict[str, Any] = {"foo": [1, 2], "bar": [6.0, 7.0]}
+    df = pa.table(data)
+    nw_df = unstable_nw.from_native(df)
+    ewm_mean = unstable_nw.col("foo").ewm_mean(com=1, ignore_nulls=False)
+    pattern = re.compile(
+        r".+ewm_mean.+ not implemented.+arrow", flags=re.DOTALL | re.IGNORECASE
+    )
+    with pytest.raises(NotImplementedError, match=pattern):
+        nw_df.with_columns(ewm_mean)
+
+    assert isinstance(ArrowExpr.ewm_mean, not_implemented)
+
+    if TYPE_CHECKING:
+        from narwhals.utils import _SupportsGet
+
+    class DummyCompliant(Protocol):
+        _implementation: nw.Implementation
+
+        def alias(self, name: str) -> str: ...
+        def unique(self) -> Self: ...
+
+        # NOTE property option (1)
+        str: _SupportsGet
+        dt: _SupportsGet
+
+        # NOTE property option (2)
+        @property
+        def cat(self) -> Any: ...
+        @property
+        def list(self) -> Any: ...
+
+    class DummyExpr(DummyCompliant):
+        def __init__(self) -> None:
+            self._implementation = nw.Implementation.POLARS
+
+        def alias(self, name: str) -> str:
+            return name
+
+        unique = not_implemented()
+
+        # NOTE: Only `mypy` has an issue with this?
+        # error: Cannot override writeable attribute with read-only property
+        @property
+        def str(self) -> PolarsExprStringNamespace:  # type: ignore[override]
+            pl_expr = cast("PolarsExpr", self)
+            return PolarsExprStringNamespace(pl_expr)
+
+        dt = not_implemented()
+
+        # NOTE: Typing is happy w/ double property
+        @property
+        def cat(self) -> PolarsExprStringNamespace:
+            pl_expr = cast("PolarsExpr", self)
+            return PolarsExprStringNamespace(pl_expr)
+
+        # NOTE: Typing still happy - but it complicates runtime (API completeness) access
+        _list = not_implemented("list")
+
+        @property
+        def list(self) -> Any:
+            return self._list
+
+    expr = DummyExpr()
+    # NOTE: Happy path
+    assert expr._implementation is nw.Implementation.POLARS
+    assert expr.alias("new name") == "new name"
+    assert isinstance(expr.str, PolarsExprStringNamespace)
+    assert isinstance(expr.cat, PolarsExprStringNamespace)
+
+    # NOTE: not implemented override
+    pattern = re.compile(
+        r".+unique.+ not implemented.+polars", flags=re.DOTALL | re.IGNORECASE
+    )
+    with pytest.raises(NotImplementedError, match=pattern):
+        expr.unique()
+
+    assert isinstance(DummyExpr.unique, not_implemented)
+    assert repr(DummyExpr.unique) == "<not_implemented>: DummyExpr.unique"
+
+    pattern = re.compile(
+        r".+unique.+ not implemented.+DummyExpr", flags=re.DOTALL | re.IGNORECASE
+    )
+    with pytest.raises(NotImplementedError, match=pattern):
+        DummyExpr.unique()
+
+    pattern = re.compile(
+        r".+dt.+ not implemented.+polars", flags=re.DOTALL | re.IGNORECASE
+    )
+    with pytest.raises(NotImplementedError, match=pattern):
+        expr.dt  # noqa: B018
+
+    assert isinstance(DummyExpr.dt, not_implemented)
+    assert repr(DummyExpr.dt) == "<not_implemented>: DummyExpr.dt"
+
+    pattern = re.compile(
+        r".+list.+ not implemented.+polars", flags=re.DOTALL | re.IGNORECASE
+    )
+    with pytest.raises(NotImplementedError, match=pattern):
+        expr.list  # noqa: B018
+
+    assert isinstance(DummyExpr._list, not_implemented)
+    assert repr(DummyExpr._list) == "<not_implemented>: DummyExpr.list"