Repository: narwhals-dev/narwhals
Pull request #2138 (2025-03-03 18:03:58)

feat: support anonymous expressions in `over` context for pandas-like, support `over` for multi-partitioned dask
This simplifies a lot on the pandas side, and unlocks `.over` for dask

On the downside, for some very old versions of pandas (pre 1.1), some tests would fail...but tbh those versions are so old i'm not overly bothered. pandas 1.0 is >5 years old...

<!--
# Thanks for contributing a pull request! 
## Please make sure you see our contribution guidelines: https://github.com/narwhals-dev/narwhals/blob/main/CONTRIBUTING.md
-->

## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [ ] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #\<issue number\>
- Closes #\<issue number\>

## Checklist

- [ ] Code follows style guide (ruff)
- [ ] Tests added
- [ ] Documented the changes

## If you have comments or can explain your changes, please do so below



Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index c4f2714c..424d4059 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -4,7 +4,7 @@ ci:
 repos:
 - repo: https://github.com/astral-sh/ruff-pre-commit
   # Ruff version.
-  rev: 'v0.9.4'
+  rev: 'v0.9.9'
   hooks:
     # Run the formatter.
     - id: ruff-format
@@ -22,7 +22,7 @@ repos:
       args: [--ignore-words-list=ser]
       exclude: ^docs/api-completeness.md$
 -   repo: https://github.com/pycqa/flake8
-    rev: '7.1.1'  # todo: remove once https://github.com/astral-sh/ruff/issues/458 is addressed
+    rev: '7.1.2'  # todo: remove once https://github.com/astral-sh/ruff/issues/458 is addressed
     hooks:
     -   id: flake8
         additional_dependencies: [darglint==1.8.1, Flake8-pyproject]
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index a712b458..ab1e0e54 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -87,12 +87,13 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
     @classmethod
     def from_column_names(
         cls: type[Self],
-        *column_names: str,
+        evaluate_column_names: Callable[[ArrowDataFrame], Sequence[str]],
+        /,
+        *,
+        function_name: str,
         backend_version: tuple[int, ...],
         version: Version,
     ) -> Self:
-        from narwhals._arrow.series import ArrowSeries
-
         def func(df: ArrowDataFrame) -> list[ArrowSeries]:
             try:
                 return [
@@ -102,10 +103,12 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
                         backend_version=df._backend_version,
                         version=df._version,
                     )
-                    for column_name in column_names
+                    for column_name in evaluate_column_names(df)
                 ]
             except KeyError as e:
-                missing_columns = [x for x in column_names if x not in df.columns]
+                missing_columns = [
+                    x for x in evaluate_column_names(df) if x not in df.columns
+                ]
                 raise ColumnNotFoundError.from_missing_and_available_column_names(
                     missing_columns=missing_columns, available_columns=df.columns
                 ) from e
@@ -113,8 +116,8 @@ class ArrowExpr(CompliantExpr["ArrowDataFrame", ArrowSeries]):
         return cls(
             func,
             depth=0,
-            function_name="col",
-            evaluate_output_names=lambda _df: column_names,
+            function_name=function_name,
+            evaluate_output_names=evaluate_column_names,
             alias_output_names=None,
             backend_version=backend_version,
             version=version,
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 e740b859..7d803103 100644
--- a/narwhals/_arrow/namespace.py
+++ b/narwhals/_arrow/namespace.py
@@ -1,7 +1,9 @@
 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
@@ -27,8 +29,10 @@ from narwhals._expression_parsing import combine_alias_output_names
 from narwhals._expression_parsing import combine_evaluate_output_names
 from narwhals.typing import CompliantNamespace
 from narwhals.utils import Implementation
+from narwhals.utils import exclude_column_names
 from narwhals.utils import get_column_names
 from narwhals.utils import import_dtypes_module
+from narwhals.utils import passthrough_column_names
 
 if TYPE_CHECKING:
     from typing import Callable
@@ -118,36 +122,18 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
         from narwhals._arrow.expr import ArrowExpr
 
         return ArrowExpr.from_column_names(
-            *column_names, backend_version=self._backend_version, version=self._version
+            passthrough_column_names(column_names),
+            function_name="col",
+            backend_version=self._backend_version,
+            version=self._version,
         )
 
     def exclude(self: Self, excluded_names: Container[str]) -> ArrowExpr:
-        from narwhals._arrow.series import ArrowSeries
-
-        def evaluate_output_names(df: ArrowDataFrame) -> Sequence[str]:
-            return [
-                column_name
-                for column_name in df.columns
-                if column_name not in excluded_names
-            ]
-
-        def func(df: ArrowDataFrame) -> list[ArrowSeries]:
-            return [
-                ArrowSeries(
-                    df._native_frame[column_name],
-                    name=column_name,
-                    backend_version=df._backend_version,
-                    version=df._version,
-                )
-                for column_name in evaluate_output_names(df)
-            ]
-
-        return self._create_expr_from_callable(
-            func,
-            depth=0,
+        return ArrowExpr.from_column_names(
+            partial(exclude_column_names, names=excluded_names),
             function_name="exclude",
-            evaluate_output_names=evaluate_output_names,
-            alias_output_names=None,
+            backend_version=self._backend_version,
+            version=self._version,
         )
 
     def nth(self: Self, *column_indices: int) -> ArrowExpr:
@@ -177,23 +163,9 @@ class ArrowNamespace(CompliantNamespace[ArrowDataFrame, ArrowSeries]):
         )
 
     def all(self: Self) -> ArrowExpr:
-        from narwhals._arrow.expr import ArrowExpr
-        from narwhals._arrow.series import ArrowSeries
-
-        return ArrowExpr(
-            lambda df: [
-                ArrowSeries(
-                    df._native_frame[column_name],
-                    name=column_name,
-                    backend_version=df._backend_version,
-                    version=df._version,
-                )
-                for column_name in df.columns
-            ],
-            depth=0,
+        return ArrowExpr.from_column_names(
+            get_column_names,
             function_name="all",
-            evaluate_output_names=get_column_names,
-            alias_output_names=None,
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -222,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(
@@ -235,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(
@@ -248,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(
@@ -267,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)
             )
@@ -286,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
@@ -314,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
@@ -386,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/_dask/expr.py b/narwhals/_dask/expr.py
index c6a861c3..46fdc5db 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,6 +16,7 @@ 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
@@ -90,15 +93,23 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[
     @classmethod
     def from_column_names(
         cls: type[Self],
-        *column_names: str,
+        evaluate_column_names: Callable[[DaskLazyFrame], Sequence[str]],
+        /,
+        *,
+        function_name: str,
         backend_version: tuple[int, ...],
         version: Version,
     ) -> Self:
         def func(df: DaskLazyFrame) -> list[dx.Series]:
             try:
-                return [df._native_frame[column_name] for column_name in column_names]
+                return [
+                    df._native_frame[column_name]
+                    for column_name in evaluate_column_names(df)
+                ]
             except KeyError as e:
-                missing_columns = [x for x in column_names if x not in df.columns]
+                missing_columns = [
+                    x for x in evaluate_column_names(df) if x not in df.columns
+                ]
                 raise ColumnNotFoundError.from_missing_and_available_column_names(
                     missing_columns=missing_columns,
                     available_columns=df.columns,
@@ -107,8 +118,8 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[
         return cls(
             func,
             depth=0,
-            function_name="col",
-            evaluate_output_names=lambda _df: column_names,
+            function_name=function_name,
+            evaluate_output_names=evaluate_column_names,
             alias_output_names=None,
             backend_version=backend_version,
             version=version,
@@ -528,29 +539,40 @@ class DaskExpr(CompliantExpr["DaskLazyFrame", "dx.Series"]):  # pyright: ignore[
         )
 
     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
+        # 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,
@@ -560,7 +582,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:
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/_dask/namespace.py b/narwhals/_dask/namespace.py
index 8d64cf9d..eddbe792 100644
--- a/narwhals/_dask/namespace.py
+++ b/narwhals/_dask/namespace.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import operator
+from functools import partial
 from functools import reduce
 from typing import TYPE_CHECKING
 from typing import Any
@@ -25,7 +26,9 @@ from narwhals._expression_parsing import combine_alias_output_names
 from narwhals._expression_parsing import combine_evaluate_output_names
 from narwhals.typing import CompliantNamespace
 from narwhals.utils import Implementation
+from narwhals.utils import exclude_column_names
 from narwhals.utils import get_column_names
+from narwhals.utils import passthrough_column_names
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -53,43 +56,25 @@ class DaskNamespace(CompliantNamespace[DaskLazyFrame, "dx.Series"]):  # pyright:
         self._version = version
 
     def all(self: Self) -> DaskExpr:
-        def func(df: DaskLazyFrame) -> list[dx.Series]:
-            return [df._native_frame[column_name] for column_name in df.columns]
-
-        return DaskExpr(
-            func,
-            depth=0,
+        return DaskExpr.from_column_names(
+            get_column_names,
             function_name="all",
-            evaluate_output_names=get_column_names,
-            alias_output_names=None,
             backend_version=self._backend_version,
             version=self._version,
         )
 
     def col(self: Self, *column_names: str) -> DaskExpr:
         return DaskExpr.from_column_names(
-            *column_names, backend_version=self._backend_version, version=self._version
+            passthrough_column_names(column_names),
+            function_name="col",
+            backend_version=self._backend_version,
+            version=self._version,
         )
 
     def exclude(self: Self, excluded_names: Container[str]) -> DaskExpr:
-        def evaluate_output_names(df: DaskLazyFrame) -> Sequence[str]:
-            return [
-                column_name
-                for column_name in df.columns
-                if column_name not in excluded_names
-            ]
-
-        def func(df: DaskLazyFrame) -> list[dx.Series]:
-            return [
-                df._native_frame[column_name] for column_name in evaluate_output_names(df)
-            ]
-
-        return DaskExpr(
-            func,
-            depth=0,
+        return DaskExpr.from_column_names(
+            partial(exclude_column_names, names=excluded_names),
             function_name="exclude",
-            evaluate_output_names=evaluate_output_names,
-            alias_output_names=None,
             backend_version=self._backend_version,
             version=self._version,
         )
diff --git a/narwhals/_duckdb/expr.py b/narwhals/_duckdb/expr.py
index 51f4ac6e..2f47942b 100644
--- a/narwhals/_duckdb/expr.py
+++ b/narwhals/_duckdb/expr.py
@@ -79,17 +79,20 @@ class DuckDBExpr(CompliantExpr["DuckDBLazyFrame", "duckdb.Expression"]):  # type
     @classmethod
     def from_column_names(
         cls: type[Self],
-        *column_names: str,
+        evaluate_column_names: Callable[[DuckDBLazyFrame], Sequence[str]],
+        /,
+        *,
+        function_name: str,
         backend_version: tuple[int, ...],
         version: Version,
     ) -> Self:
-        def func(_: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            return [ColumnExpression(col_name) for col_name in column_names]
+        def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
+            return [ColumnExpression(col_name) for col_name in evaluate_column_names(df)]
 
         return cls(
             func,
-            function_name="col",
-            evaluate_output_names=lambda _df: column_names,
+            function_name=function_name,
+            evaluate_output_names=evaluate_column_names,
             alias_output_names=None,
             backend_version=backend_version,
             version=version,
diff --git a/narwhals/_duckdb/namespace.py b/narwhals/_duckdb/namespace.py
index 45c87796..f8780fbc 100644
--- a/narwhals/_duckdb/namespace.py
+++ b/narwhals/_duckdb/namespace.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-import functools
 import operator
+from functools import partial
 from functools import reduce
 from typing import TYPE_CHECKING
 from typing import Any
@@ -12,7 +12,6 @@ from typing import Sequence
 
 from duckdb import CaseExpression
 from duckdb import CoalesceOperator
-from duckdb import ColumnExpression
 from duckdb import FunctionExpression
 from duckdb.typing import BIGINT
 from duckdb.typing import VARCHAR
@@ -26,7 +25,9 @@ from narwhals._expression_parsing import combine_alias_output_names
 from narwhals._expression_parsing import combine_evaluate_output_names
 from narwhals.typing import CompliantNamespace
 from narwhals.utils import Implementation
+from narwhals.utils import exclude_column_names
 from narwhals.utils import get_column_names
+from narwhals.utils import passthrough_column_names
 
 if TYPE_CHECKING:
     import duckdb
@@ -51,14 +52,9 @@ class DuckDBNamespace(CompliantNamespace["DuckDBLazyFrame", "duckdb.Expression"]
         return DuckDBSelectorNamespace(self)
 
     def all(self: Self) -> DuckDBExpr:
-        def _all(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            return [ColumnExpression(col_name) for col_name in df.columns]
-
-        return DuckDBExpr(
-            call=_all,
+        return DuckDBExpr.from_column_names(
+            get_column_names,
             function_name="all",
-            evaluate_output_names=get_column_names,
-            alias_output_names=None,
             backend_version=self._backend_version,
             version=self._version,
         )
@@ -80,9 +76,7 @@ class DuckDBNamespace(CompliantNamespace["DuckDBLazyFrame", "duckdb.Expression"]
         if how == "vertical" and not all(x.schema == schema for x in items[1:]):
             msg = "inputs should all have the same schema"
             raise TypeError(msg)
-        res = functools.reduce(
-            lambda x, y: x.union(y), (item._native_frame for item in items)
-        )
+        res = reduce(lambda x, y: x.union(y), (item._native_frame for item in items))
         return first._from_native_frame(res)
 
     def concat_str(
@@ -238,27 +232,16 @@ class DuckDBNamespace(CompliantNamespace["DuckDBLazyFrame", "duckdb.Expression"]
 
     def col(self: Self, *column_names: str) -> DuckDBExpr:
         return DuckDBExpr.from_column_names(
-            *column_names, backend_version=self._backend_version, version=self._version
+            passthrough_column_names(column_names),
+            function_name="col",
+            backend_version=self._backend_version,
+            version=self._version,
         )
 
     def exclude(self: Self, excluded_names: Container[str]) -> DuckDBExpr:
-        def evaluate_output_names(df: DuckDBLazyFrame) -> Sequence[str]:
-            return [
-                column_name
-                for column_name in df.columns
-                if column_name not in excluded_names
-            ]
-
-        def func(df: DuckDBLazyFrame) -> list[duckdb.Expression]:
-            return [
-                ColumnExpression(column_name) for column_name in evaluate_output_names(df)
-            ]
-
-        return DuckDBExpr(
-            func,
+        return DuckDBExpr.from_column_names(
+            partial(exclude_column_names, names=excluded_names),
             function_name="exclude",
-            evaluate_output_names=evaluate_output_names,
-            alias_output_names=None,
             backend_version=self._backend_version,
             version=self._version,
         )
diff --git a/narwhals/_expression_parsing.py b/narwhals/_expression_parsing.py
index cb3f132a..69f49aac 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
@@ -238,8 +234,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 +246,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 857f1afa..3d595520 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -9,14 +9,14 @@ 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 +33,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 +48,32 @@ MANY_TO_MANY_AGG_FUNCTIONS_TO_PANDAS_EQUIVALENT = {
 }
 
 
+def window_kwargs_to_pandas_equivalent(
+    function_name: str, kwargs: dict[str, object]
+) -> dict[str, object]:
+    unsupported_reverse_msg = (
+        "Cumulative operation with `reverse=True` is not supported in "
+        "over context for pandas-like backend."
+    )
+    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
+        if kwargs["reverse"]:
+            raise NotImplementedError(unsupported_reverse_msg)
+        pandas_kwargs = {"skipna": True}
+    else:
+        pandas_kwargs = {}
+    return pandas_kwargs
+
+
 class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
     def __init__(
         self: Self,
@@ -115,7 +141,10 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
     @classmethod
     def from_column_names(
         cls: type[Self],
-        *column_names: str,
+        evaluate_column_names: Callable[[PandasLikeDataFrame], Sequence[str]],
+        /,
+        *,
+        function_name: str,
         implementation: Implementation,
         backend_version: tuple[int, ...],
         version: Version,
@@ -129,10 +158,12 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
                         backend_version=df._backend_version,
                         version=df._version,
                     )
-                    for column_name in column_names
+                    for column_name in evaluate_column_names(df)
                 ]
             except KeyError as e:
-                missing_columns = [x for x in column_names if x not in df.columns]
+                missing_columns = [
+                    x for x in evaluate_column_names(df) if x not in df.columns
+                ]
                 raise ColumnNotFoundError.from_missing_and_available_column_names(
                     missing_columns=missing_columns,
                     available_columns=df.columns,
@@ -141,8 +172,8 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
         return cls(
             func,
             depth=0,
-            function_name="col",
-            evaluate_output_names=lambda _df: column_names,
+            function_name=function_name,
+            evaluate_output_names=evaluate_column_names,
             alias_output_names=None,
             implementation=implementation,
             backend_version=backend_version,
@@ -428,80 +459,49 @@ class PandasLikeExpr(CompliantExpr["PandasLikeDataFrame", PandasLikeSeries]):
         )
 
     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):
+        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
+            )
+
+            if function_name == "cum_count":
+                plx = self.__narwhals_namespace__()
+                df = df.with_columns(~plx.col(*output_names).is_null())
 
-                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"
+            res_native = df._native_frame.groupby(keys)[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]
+            )
+            return [result_frame[name] for name in aliases]
 
         return self.__class__(
             func,
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/_pandas_like/namespace.py b/narwhals/_pandas_like/namespace.py
index 75b5cec6..4b4994c6 100644
--- a/narwhals/_pandas_like/namespace.py
+++ b/narwhals/_pandas_like/namespace.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import operator
+from functools import partial
 from functools import reduce
 from typing import TYPE_CHECKING
 from typing import Any
@@ -23,8 +24,10 @@ from narwhals._pandas_like.utils import extract_dataframe_comparand
 from narwhals._pandas_like.utils import horizontal_concat
 from narwhals._pandas_like.utils import vertical_concat
 from narwhals.typing import CompliantNamespace
+from narwhals.utils import exclude_column_names
 from narwhals.utils import get_column_names
 from narwhals.utils import import_dtypes_module
+from narwhals.utils import passthrough_column_names
 
 if TYPE_CHECKING:
     from typing_extensions import Self
@@ -110,37 +113,20 @@ class PandasLikeNamespace(CompliantNamespace[PandasLikeDataFrame, PandasLikeSeri
     # --- selection ---
     def col(self: Self, *column_names: str) -> PandasLikeExpr:
         return PandasLikeExpr.from_column_names(
-            *column_names,
+            passthrough_column_names(column_names),
+            function_name="col",
             implementation=self._implementation,
             backend_version=self._backend_version,
             version=self._version,
         )
 
     def exclude(self: Self, excluded_names: Container[str]) -> PandasLikeExpr:
-        def evaluate_output_names(df: PandasLikeDataFrame) -> Sequence[str]:
-            return [
-                column_name
-                for column_name in df.columns
-                if column_name not in excluded_names
-            ]
-
-        def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            return [
-                PandasLikeSeries(
-                    df._native_frame[column_name],
-                    implementation=df._implementation,
-                    backend_version=df._backend_version,
-                    version=df._version,
-                )
-                for column_name in evaluate_output_names(df)
-            ]
-
-        return self._create_expr_from_callable(
-            func,
-            depth=0,
-            evaluate_output_names=evaluate_output_names,
+        return PandasLikeExpr.from_column_names(
+            partial(exclude_column_names, names=excluded_names),
             function_name="exclude",
-            alias_output_names=None,
+            implementation=self._implementation,
+            backend_version=self._backend_version,
+            version=self._version,
         )
 
     def nth(self: Self, *column_indices: int) -> PandasLikeExpr:
@@ -152,20 +138,9 @@ class PandasLikeNamespace(CompliantNamespace[PandasLikeDataFrame, PandasLikeSeri
         )
 
     def all(self: Self) -> PandasLikeExpr:
-        return PandasLikeExpr(
-            lambda df: [
-                PandasLikeSeries(
-                    df._native_frame[column_name],
-                    implementation=self._implementation,
-                    backend_version=self._backend_version,
-                    version=self._version,
-                )
-                for column_name in df.columns
-            ],
-            depth=0,
+        return PandasLikeExpr.from_column_names(
+            get_column_names,
             function_name="all",
-            evaluate_output_names=get_column_names,
-            alias_output_names=None,
             implementation=self._implementation,
             backend_version=self._backend_version,
             version=self._version,
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index 41bbe46c..f2c09ea8 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -131,18 +131,21 @@ class SparkLikeExpr(CompliantExpr["SparkLikeLazyFrame", "Column"]):  # type: ign
     @classmethod
     def from_column_names(
         cls: type[Self],
-        *column_names: str,
+        evaluate_column_names: Callable[[SparkLikeLazyFrame], Sequence[str]],
+        /,
+        *,
+        function_name: str,
+        implementation: Implementation,
         backend_version: tuple[int, ...],
         version: Version,
-        implementation: Implementation,
     ) -> Self:
         def func(df: SparkLikeLazyFrame) -> list[Column]:
-            return [df._F.col(col_name) for col_name in column_names]
+            return [df._F.col(col_name) for col_name in evaluate_column_names(df)]
 
         return cls(
             func,
-            function_name="col",
-            evaluate_output_names=lambda _df: column_names,
+            function_name=function_name,
+            evaluate_output_names=evaluate_column_names,
             alias_output_names=None,
             backend_version=backend_version,
             version=version,
diff --git a/narwhals/_spark_like/namespace.py b/narwhals/_spark_like/namespace.py
index 91cbd4f3..91cad2bb 100644
--- a/narwhals/_spark_like/namespace.py
+++ b/narwhals/_spark_like/namespace.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import operator
+from functools import partial
 from functools import reduce
 from typing import TYPE_CHECKING
 from typing import Any
@@ -19,7 +20,9 @@ from narwhals._spark_like.selectors import SparkLikeSelectorNamespace
 from narwhals._spark_like.utils import maybe_evaluate_expr
 from narwhals._spark_like.utils import narwhals_to_native_dtype
 from narwhals.typing import CompliantNamespace
+from narwhals.utils import exclude_column_names
 from narwhals.utils import get_column_names
+from narwhals.utils import passthrough_column_names
 
 if TYPE_CHECKING:
     from pyspark.sql import Column
@@ -48,46 +51,30 @@ class SparkLikeNamespace(CompliantNamespace["SparkLikeLazyFrame", "Column"]):  #
         return SparkLikeSelectorNamespace(self)
 
     def all(self: Self) -> SparkLikeExpr:
-        def _all(df: SparkLikeLazyFrame) -> list[Column]:
-            return [df._F.col(col_name) for col_name in df.columns]
-
-        return SparkLikeExpr(
-            call=_all,
+        return SparkLikeExpr.from_column_names(
+            get_column_names,
             function_name="all",
-            evaluate_output_names=get_column_names,
-            alias_output_names=None,
+            implementation=self._implementation,
             backend_version=self._backend_version,
             version=self._version,
-            implementation=self._implementation,
         )
 
     def col(self: Self, *column_names: str) -> SparkLikeExpr:
         return SparkLikeExpr.from_column_names(
-            *column_names,
+            passthrough_column_names(column_names),
+            function_name="col",
+            implementation=self._implementation,
             backend_version=self._backend_version,
             version=self._version,
-            implementation=self._implementation,
         )
 
     def exclude(self: Self, excluded_names: Container[str]) -> SparkLikeExpr:
-        def evaluate_output_names(df: SparkLikeLazyFrame) -> Sequence[str]:
-            return [
-                column_name
-                for column_name in df.columns
-                if column_name not in excluded_names
-            ]
-
-        def func(df: SparkLikeLazyFrame) -> list[Column]:
-            return [df._F.col(column_name) for column_name in evaluate_output_names(df)]
-
-        return SparkLikeExpr(
-            func,
+        return SparkLikeExpr.from_column_names(
+            partial(exclude_column_names, names=excluded_names),
             function_name="exclude",
-            evaluate_output_names=evaluate_output_names,
-            alias_output_names=None,
+            implementation=self._implementation,
             backend_version=self._backend_version,
             version=self._version,
-            implementation=self._implementation,
         )
 
     def nth(self: Self, *column_indices: int) -> SparkLikeExpr:
diff --git a/narwhals/utils.py b/narwhals/utils.py
index f5959bb5..a75ead5f 100644
--- a/narwhals/utils.py
+++ b/narwhals/utils.py
@@ -9,6 +9,8 @@ from inspect import getattr_static
 from secrets import token_hex
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Callable
+from typing import Container
 from typing import Iterable
 from typing import Literal
 from typing import Sequence
@@ -1356,6 +1358,17 @@ def get_column_names(frame: _StoresColumns, /) -> Sequence[str]:
     return frame.columns
 
 
+def exclude_column_names(frame: _StoresColumns, names: Container[str]) -> Sequence[str]:
+    return [col_name for col_name in frame.columns if col_name not in names]
+
+
+def passthrough_column_names(names: Sequence[str], /) -> Callable[[Any], Sequence[str]]:
+    def fn(_frame: Any, /) -> Sequence[str]:
+        return names
+
+    return fn
+
+
 def _hasattr_static(obj: Any, attr: str) -> bool:
     sentinel = object()
     return getattr_static(obj, attr, sentinel) is not sentinel
diff --git a/pyproject.toml b/pyproject.toml
index 8de1ec51..0ef18f69 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -237,7 +237,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]


Test patch:
diff --git a/tests/expr_and_series/over_test.py b/tests/expr_and_series/over_test.py
index ebbda168..4e9b2fe3 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)
 
@@ -50,8 +48,6 @@ 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 "duckdb" in str(constructor):
         request.applymarker(pytest.mark.xfail)
 
@@ -224,13 +220,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,6 +245,21 @@ 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:
@@ -270,6 +282,8 @@ 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")
 
     df = nw.from_native(constructor_eager(data))
     expected = {
diff --git a/tests/expr_and_series/rank_test.py b/tests/expr_and_series/rank_test.py
index 99a64371..dac77e67 100644
--- a/tests/expr_and_series/rank_test.py
+++ b/tests/expr_and_series/rank_test.py
@@ -109,7 +109,9 @@ def test_rank_expr_in_over_context(
         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))