Repository: narwhals-dev/narwhals

Issue:
[Bug]: cum_sum expression raises an exception with pandas when 'over' is used
### Describe the bug

The `cum_sum` expression raises an exception when used with the `over` expression.

I did a quick review of the code and I believe this issue can be resolved by simply adding the cum_sum expression name to the dictionary that maps pandas method names to polars method names. I would like to open a pull request to solve this c: 

### Steps or code to reproduce the bug

```
def replicating_error_cum_sum() -> None:
    data = {
        "cat": ["a", "a", "a", "b", "b", "b", "c", "c", "c"],
        'values': [1, 2, 3, 4, 5, 6, 7, 8, None]
    }
    original_data = pd.DataFrame(data)
    df_nw = nw.from_native(original_data)
    df_nw = (
        df_nw
        .with_columns(
            nw.col("values").cum_sum().over("cat")
        )
    )
    print(nw.to_native(df_nw))
```

### Expected results

```
┌─────┬────────┐
│ cat ┆ values │
│ --- ┆ ---    │
│ str ┆ i64    │
╞═════╪════════╡
│ a   ┆ 1      │
│ a   ┆ 3      │
│ a   ┆ 6      │
│ b   ┆ 4      │
│ b   ┆ 9      │
│ b   ┆ 15     │
│ c   ┆ 7      │
│ c   ┆ 15     │
│ c   ┆ null   │
└─────┴────────┘
```

### Actual results

RuntimeError: Failed to aggregated - does your aggregation function return a scalar?

### Please run narwhals.show_version() and enter the output below.

```shell
System:
    python: 3.12.5 (main, Aug 14 2024, 04:32:18) [Clang 18.1.8 ]
executable: /Users/abelisaiasgutierrezcruz/Documents/Proyects/test_narwhals/.venv/bin/python
   machine: macOS-15.0.1-arm64-arm-64bit

Python dependencies:
     narwhals: 1.9.3
       pandas: 2.2.3
       polars: 1.8.2
         cudf: 
        modin: 
      pyarrow: 17.0.0
        numpy: 2.1.1
```


### Relevant log output

```shell
File "/Users/abelisaiasgutierrezcruz/Documents/Proyects/test_narwhals/.venv/lib/python3.12/site-packages/narwhals/_expression_parsing.py", line 98, in <genexpr>
    for sublist in (evaluate_into_expr(df, into_expr) for into_expr in exprs)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/abelisaiasgutierrezcruz/Documents/Proyects/test_narwhals/.venv/lib/python3.12/site-packages/narwhals/_expression_parsing.py", line 63, in evaluate_into_expr
    return expr._call(df)  # type: ignore[arg-type]
           ^^^^^^^^^^^^^^
  File "/Users/abelisaiasgutierrezcruz/Documents/Proyects/test_narwhals/.venv/lib/python3.12/site-packages/narwhals/_pandas_like/expr.py", line 334, in func
    tmp = df.group_by(*keys).agg(self)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/abelisaiasgutierrezcruz/Documents/Proyects/test_narwhals/.venv/lib/python3.12/site-packages/narwhals/_pandas_like/group_by.py", line 76, in agg
    return agg_pandas(
           ^^^^^^^^^^^
  File "/Users/abelisaiasgutierrezcruz/Documents/Proyects/test_narwhals/.venv/lib/python3.12/site-packages/narwhals/_pandas_like/group_by.py", line 184, in agg_pandas
    raise RuntimeError(msg) from exc
RuntimeError: Failed to aggregated - does your aggregation function return a scalar?
```


Discussion:
ClaudioSalvatoreArcidiacono: Thanks for the help @MarcoGorelli, I don't think I could have managed it by myself so quickly without your help.

I have submitted a draft to fix the issue following your suggestion. I had to change it a little because pandas had some inconsistent behaviours among cumulative functions, e.g. `transform("cumsum")` returns a Dataframe whether `transform("cumcount")` returns a Series 🤨. Perhaps it's worth to raise an issue about it on the pandas project.

I have left the pyarrow part unchanged as I am not familiar with the project yet let alone pyarrow itself. I also believe it is better to solve it in a separate PR eventually, but curious to hear your thoughts on it.

I would like to know your feedback on this change [here](https://github.com/ClaudioSalvatoreArcidiacono/narwhals/blob/1177-add-support-for-pandas-cumulative-operations-with-over/narwhals/_pandas_like/expr.py#L408) I left the message on the raised exception a bit vague mainly because I am not sure in which case `self._root_names` can be `None`.


MarcoGorelli: wow, really solid effort! i'll take a closer look later, but if you've not seen it already, you may want to take a look at https://github.com/narwhals-dev/narwhals/blob/a2088f495a4725a1901b03621cc59a831af4c656/narwhals/_pandas_like/group_by.py#L128-L298
ClaudioSalvatoreArcidiacono: Opened an issue on pandas about the strange cumcount behavior (https://github.com/pandas-dev/pandas/issues/60551)
MarcoGorelli: another difference with Polars is that Polars cumcount counts non-null values, whereas pandas counts all values
ClaudioSalvatoreArcidiacono: Good one! I missed that detail. I will add a test to check consistent behavior with null values and improve the code.
MarcoGorelli: you may need to use `isna` and `cumcount`, where's what we do for `cumcount` outside of over

https://github.com/narwhals-dev/narwhals/blob/d788eece6f1bee9e21d3840e1ec9e25799504331/narwhals/_pandas_like/series.py#L985-L991
ClaudioSalvatoreArcidiacono: Thanks for the pointer! I also noticed that the current implementation does not take into consideration `reverse`
ClaudioSalvatoreArcidiacono: Thanks guys for your comments! I have addressed them in my most recent commits.
FBruzzesi: @ClaudioSalvatoreArcidiacono I am happy to use non-specific error and just add:

```py
if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
        request.applymarker(pytest.mark.xfail)
```

to mark that pyarrow and dask will fail in those cases. Eventually maybe we will support those operations for pyarrow.
I don't think it is worth to break it down to the minimal detail why it is failing
ClaudioSalvatoreArcidiacono: @FBruzzesi agree, I also noticed that it was failing sometimes for pandas_pyarrow for older python versions
FBruzzesi: I am having such a hard time replicating these failures with nox :| 
ClaudioSalvatoreArcidiacono: Yeah, it's pretty difficult indeed.

To get it to work I think we also need to check the pyarrow version.
MarcoGorelli: awesome, thanks both - if we can get the xfails right we can ship it
FBruzzesi: @ClaudioSalvatoreArcidiacono turned out to be pandas 2.0.x issue with pyarrow backend!
I am simply xfailing it, and all checks are green now 🙌🏼

Patch:
diff --git a/docs/api-reference/expr.md b/docs/api-reference/expr.md
index 61901a99..523a45f0 100644
--- a/docs/api-reference/expr.md
+++ b/docs/api-reference/expr.md
@@ -58,5 +58,6 @@
         - sum
         - tail
         - unique
+        - var
       show_source: false
       show_bases: false
diff --git a/docs/api-reference/series.md b/docs/api-reference/series.md
index 79e7be8d..818e71b5 100644
--- a/docs/api-reference/series.md
+++ b/docs/api-reference/series.md
@@ -76,6 +76,7 @@
         - to_native
         - unique
         - value_counts
+        - var
         - zip_with
       show_source: false
       show_bases: false
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index ae8f2d03..845181bd 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -242,9 +242,12 @@ class ArrowExpr(CompliantExpr[ArrowSeries]):
     def n_unique(self: Self) -> Self:
         return reuse_series_implementation(self, "n_unique", returns_scalar=True)
 
-    def std(self: Self, ddof: int = 1) -> Self:
+    def std(self: Self, ddof: int) -> Self:
         return reuse_series_implementation(self, "std", ddof=ddof, returns_scalar=True)
 
+    def var(self: Self, ddof: int) -> Self:
+        return reuse_series_implementation(self, "var", ddof=ddof, returns_scalar=True)
+
     def skew(self: Self) -> Self:
         return reuse_series_implementation(self, "skew", returns_scalar=True)
 
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index 900a6036..f97bc604 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -330,6 +330,13 @@ class ArrowSeries(CompliantSeries):
             pc.stddev(self._native_series, ddof=ddof), _return_py_scalar
         )
 
+    def var(self: Self, ddof: int, *, _return_py_scalar: bool = True) -> float:
+        import pyarrow.compute as pc
+
+        return maybe_extract_py_scalar(  # type: ignore[no-any-return]
+            pc.variance(self._native_series, ddof=ddof), _return_py_scalar
+        )
+
     def skew(self: Self, *, _return_py_scalar: bool = True) -> float | None:
         import pyarrow.compute as pc
 
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 748d332b..ac3b3c5b 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -432,7 +432,7 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
             returns_scalar=True,
         )
 
-    def std(self, ddof: int = 1) -> Self:
+    def std(self, ddof: int) -> Self:
         return self._from_call(
             lambda _input, ddof: _input.std(ddof=ddof),
             "std",
@@ -440,6 +440,14 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
             returns_scalar=True,
         )
 
+    def var(self, ddof: int) -> Self:
+        return self._from_call(
+            lambda _input, ddof: _input.var(ddof=ddof),
+            "var",
+            ddof,
+            returns_scalar=True,
+        )
+
     def skew(self: Self) -> Self:
         return self._from_call(
             lambda _input: _input.skew(),
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index a83d83c4..96af851e 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -9,6 +9,7 @@ from typing import Sequence
 from narwhals._expression_parsing import reuse_series_implementation
 from narwhals._expression_parsing import reuse_series_namespace_implementation
 from narwhals._pandas_like.series import PandasLikeSeries
+from narwhals._pandas_like.utils import rename
 from narwhals.dependencies import get_numpy
 from narwhals.dependencies import is_numpy_array
 from narwhals.exceptions import ColumnNotFoundError
@@ -23,6 +24,17 @@ if TYPE_CHECKING:
     from narwhals.utils import Implementation
     from narwhals.utils import Version
 
+CUMULATIVE_FUNCTIONS_TO_PANDAS_EQUIVALENT = {
+    "col->cum_sum": "cumsum",
+    "col->cum_min": "cummin",
+    "col->cum_max": "cummax",
+    "col->cum_prod": "cumprod",
+    # Pandas cumcount starts counting from 0 while Polars starts from 1
+    # Pandas cumcount counts nulls while Polars does not
+    # So, instead of using "cumcount" we use "cumsum" on notna() to get the same result
+    "col->cum_count": "cumsum",
+}
+
 
 class PandasLikeExpr(CompliantExpr[PandasLikeSeries]):
     def __init__(
@@ -246,9 +258,12 @@ class PandasLikeExpr(CompliantExpr[PandasLikeSeries]):
     def median(self) -> Self:
         return reuse_series_implementation(self, "median", returns_scalar=True)
 
-    def std(self, *, ddof: int = 1) -> Self:
+    def std(self, *, ddof: int) -> Self:
         return reuse_series_implementation(self, "std", ddof=ddof, returns_scalar=True)
 
+    def var(self, *, ddof: int) -> Self:
+        return reuse_series_implementation(self, "var", ddof=ddof, returns_scalar=True)
+
     def skew(self: Self) -> Self:
         return reuse_series_implementation(self, "skew", returns_scalar=True)
 
@@ -397,19 +412,54 @@ class PandasLikeExpr(CompliantExpr[PandasLikeSeries]):
         )
 
     def over(self, keys: list[str]) -> Self:
-        def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
-            if self._output_names is None:
-                msg = (
-                    "Anonymous expressions are not supported in over.\n"
-                    "Instead of `nw.all()`, try using a named expression, such as "
-                    "`nw.col('a', 'b')`\n"
+        if self._function_name in CUMULATIVE_FUNCTIONS_TO_PANDAS_EQUIVALENT:
+
+            def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
+                if (
+                    self._output_names is None or self._root_names is None
+                ):  # pragma: no cover
+                    # Technically unreachable, but keep this for safety
+                    msg = (
+                        "Anonymous expressions are not supported in over.\n"
+                        "Instead of `nw.all()`, try using a named expression, such as "
+                        "`nw.col('a', 'b')`\n"
+                    )
+                    raise ValueError(msg)
+
+                if self._function_name == "col->cum_count":
+                    plx = self.__narwhals_namespace__()
+                    df = df.with_columns(~plx.col(*self._root_names).is_null())
+
+                res_native = df._native_frame.groupby(list(keys), as_index=False)[
+                    self._root_names
+                ].transform(
+                    CUMULATIVE_FUNCTIONS_TO_PANDAS_EQUIVALENT[self._function_name]
                 )
-                raise ValueError(msg)
-            tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
-            tmp = df.select(*keys).join(
-                tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
-            )
-            return [tmp[name] for name in self._output_names]
+                result_frame = df._from_native_frame(
+                    rename(
+                        res_native,
+                        columns=dict(zip(self._root_names, self._output_names)),
+                        implementation=self._implementation,
+                        backend_version=self._backend_version,
+                    )
+                )
+                return [result_frame[name] for name in self._output_names]
+
+        else:
+
+            def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
+                if self._output_names is None:
+                    msg = (
+                        "Anonymous expressions are not supported in over.\n"
+                        "Instead of `nw.all()`, try using a named expression, such as "
+                        "`nw.col('a', 'b')`\n"
+                    )
+                    raise ValueError(msg)
+                tmp = df.group_by(*keys, drop_null_keys=False).agg(self)
+                tmp = df.select(*keys).join(
+                    tmp, how="left", left_on=keys, right_on=keys, suffix="_right"
+                )
+                return [tmp[name] for name in self._output_names]
 
         return self.__class__(
             func,
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index bd799039..f8d2e2cf 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -622,10 +622,14 @@ class PandasLikeSeries(CompliantSeries):
         ser = self._native_series
         return ser.median()
 
-    def std(self: Self, *, ddof: int = 1) -> float:
+    def std(self: Self, *, ddof: int) -> float:
         ser = self._native_series
         return ser.std(ddof=ddof)  # type: ignore[no-any-return]
 
+    def var(self: Self, *, ddof: int) -> float:
+        ser = self._native_series
+        return ser.var(ddof=ddof)  # type: ignore[no-any-return]
+
     def skew(self: Self) -> float | None:
         ser = self._native_series
         ser_not_null = ser.dropna()
diff --git a/narwhals/_spark_like/expr.py b/narwhals/_spark_like/expr.py
index 5695faf5..6074dff2 100644
--- a/narwhals/_spark_like/expr.py
+++ b/narwhals/_spark_like/expr.py
@@ -204,7 +204,7 @@ class SparkLikeExpr(CompliantExpr["Column"]):
 
         return self._from_call(_min, "min", returns_scalar=True)
 
-    def std(self, ddof: int = 1) -> Self:
+    def std(self, ddof: int) -> Self:
         import numpy as np  # ignore-banned-import
 
         def _std(_input: Column) -> Column:  # pragma: no cover
diff --git a/narwhals/expr.py b/narwhals/expr.py
index 674bb19a..41038670 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -706,6 +706,56 @@ class Expr:
         """
         return self.__class__(lambda plx: self._to_compliant_expr(plx).std(ddof=ddof))
 
+    def var(self, *, ddof: int = 1) -> Self:
+        """Get variance.
+
+        Arguments:
+            ddof: "Delta Degrees of Freedom": the divisor used in the calculation is N - ddof,
+                     where N represents the number of elements. By default ddof is 1.
+
+        Returns:
+            A new expression.
+
+        Examples:
+            >>> import polars as pl
+            >>> import pandas as pd
+            >>> import pyarrow as pa
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoFrameT
+            >>> df_pd = pd.DataFrame({"a": [20, 25, 60], "b": [1.5, 1, -1.4]})
+            >>> df_pl = pl.DataFrame({"a": [20, 25, 60], "b": [1.5, 1, -1.4]})
+            >>> df_pa = pa.table({"a": [20, 25, 60], "b": [1.5, 1, -1.4]})
+
+            Let's define a dataframe-agnostic function:
+
+            >>> def agnostic_var(df_native: IntoFrameT) -> IntoFrameT:
+            ...     df = nw.from_native(df_native)
+            ...     return df.select(nw.col("a", "b").var(ddof=0)).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_var(df_pd)
+                        a         b
+            0  316.666667  1.602222
+            >>> agnostic_var(df_pl)
+            shape: (1, 2)
+            ┌────────────┬──────────┐
+            │ a          ┆ b        │
+            │ ---        ┆ ---      │
+            │ f64        ┆ f64      │
+            ╞════════════╪══════════╡
+            │ 316.666667 ┆ 1.602222 │
+            └────────────┴──────────┘
+            >>> agnostic_var(df_pa)
+            pyarrow.Table
+            a: double
+            b: double
+            ----
+            a: [[316.6666666666667]]
+            b: [[1.6022222222222222]]
+        """
+        return self.__class__(lambda plx: self._to_compliant_expr(plx).var(ddof=ddof))
+
     def map_batches(
         self,
         function: Callable[[Any], Self],
@@ -2212,7 +2262,7 @@ class Expr:
 
             Let's define a dataframe-agnostic function:
 
-            >>> def my_library_agnostic_function(df_native: IntoFrameT) -> IntoFrameT:
+            >>> def agnostic_min_over_b(df_native: IntoFrameT) -> IntoFrameT:
             ...     df = nw.from_native(df_native)
             ...     return df.with_columns(
             ...         a_min_per_group=nw.col("a").min().over("b")
@@ -2220,12 +2270,12 @@ class Expr:
 
             We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
 
-            >>> my_library_agnostic_function(df_pd)
+            >>> agnostic_min_over_b(df_pd)
                a  b  a_min_per_group
             0  1  1                1
             1  2  1                1
             2  3  2                3
-            >>> my_library_agnostic_function(df_pl)
+            >>> agnostic_min_over_b(df_pl)
             shape: (3, 3)
             ┌─────┬─────┬─────────────────┐
             │ a   ┆ b   ┆ a_min_per_group │
@@ -2236,7 +2286,7 @@ class Expr:
             │ 2   ┆ 1   ┆ 1               │
             │ 3   ┆ 2   ┆ 3               │
             └─────┴─────┴─────────────────┘
-            >>> my_library_agnostic_function(df_pa)
+            >>> agnostic_min_over_b(df_pa)
             pyarrow.Table
             a: int64
             b: int64
@@ -2245,6 +2295,30 @@ class Expr:
             a: [[1,2,3]]
             b: [[1,1,2]]
             a_min_per_group: [[1,1,3]]
+
+            Cumulative operations are also supported, but (currently) only for
+            pandas and Polars:
+
+            >>> def agnostic_cum_sum(df_native: IntoFrameT) -> IntoFrameT:
+            ...     df = nw.from_native(df_native)
+            ...     return df.with_columns(c=nw.col("a").cum_sum().over("b")).to_native()
+
+            >>> agnostic_cum_sum(df_pd)
+               a  b  c
+            0  1  1  1
+            1  2  1  3
+            2  3  2  3
+            >>> agnostic_cum_sum(df_pl)
+            shape: (3, 3)
+            ┌─────┬─────┬─────┐
+            │ a   ┆ b   ┆ c   │
+            │ --- ┆ --- ┆ --- │
+            │ i64 ┆ i64 ┆ i64 │
+            ╞═════╪═════╪═════╡
+            │ 1   ┆ 1   ┆ 1   │
+            │ 2   ┆ 1   ┆ 3   │
+            │ 3   ┆ 2   ┆ 3   │
+            └─────┴─────┴─────┘
         """
         return self.__class__(
             lambda plx: self._to_compliant_expr(plx).over(flatten(keys))
diff --git a/narwhals/series.py b/narwhals/series.py
index 2ef2efc7..8601d72c 100644
--- a/narwhals/series.py
+++ b/narwhals/series.py
@@ -1029,7 +1029,7 @@ class Series(Generic[IntoSeriesT]):
         """Get the standard deviation of this Series.
 
         Arguments:
-            ddof: “Delta Degrees of Freedom”: the divisor used in the calculation is N - ddof,
+            ddof: "Delta Degrees of Freedom": the divisor used in the calculation is N - ddof,
                      where N represents the number of elements.
 
         Examples:
@@ -1056,6 +1056,37 @@ class Series(Generic[IntoSeriesT]):
         """
         return self._compliant_series.std(ddof=ddof)
 
+    def var(self, *, ddof: int = 1) -> Any:
+        """Get the variance of this Series.
+
+        Arguments:
+            ddof: "Delta Degrees of Freedom": the divisor used in the calculation is N - ddof,
+                     where N represents the number of elements.
+
+        Examples:
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoSeries
+            >>> s = [1, 2, 3]
+            >>> s_pd = pd.Series(s)
+            >>> s_pl = pl.Series(s)
+
+            We define a library agnostic function:
+
+            >>> def agnostic_var(s_native: IntoSeries):
+            ...     s = nw.from_native(s_native, series_only=True)
+            ...     return s.var()
+
+            We can then pass either pandas or Polars to `func`:
+
+            >>> agnostic_var(s_pd)
+            np.float64(1.0)
+            >>> agnostic_var(s_pl)
+            1.0
+        """
+        return self._compliant_series.var(ddof=ddof)
+
     def clip(
         self, lower_bound: Any | None = None, upper_bound: Any | None = None
     ) -> Self:
diff --git a/pyproject.toml b/pyproject.toml
index f6aaaa7a..cc92d1b7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -152,7 +152,7 @@ exclude_also = [
   "if sys.version_info() <",
   "if (:?self._)?implementation is Implementation.MODIN",
   "if (:?self._)?implementation is Implementation.CUDF",
-  'request.applymarker\(pytest.mark.xfail\)',
+  'request.applymarker\(pytest.mark.xfail',
   '\w+._backend_version < ',
   'backend_version <',
   'if "cudf" in str\(constructor'


Test patch:
diff --git a/tests/expr_and_series/over_test.py b/tests/expr_and_series/over_test.py
index e4ab273c..54593421 100644
--- a/tests/expr_and_series/over_test.py
+++ b/tests/expr_and_series/over_test.py
@@ -1,10 +1,10 @@
 from __future__ import annotations
 
-from contextlib import nullcontext as does_not_raise
-
+import pandas as pd
 import pytest
 
 import narwhals.stable.v1 as nw
+from tests.utils import PANDAS_VERSION
 from tests.utils import Constructor
 from tests.utils import assert_equal_data
 
@@ -15,7 +15,10 @@ data = {
 }
 
 
-def test_over_single(constructor: Constructor) -> None:
+def test_over_single(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+
     df = nw.from_native(constructor(data))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
@@ -24,21 +27,14 @@ def test_over_single(constructor: Constructor) -> None:
         "c_max": [5, 5, 3, 3, 3],
     }
 
-    context = (
-        pytest.raises(
-            NotImplementedError,
-            match="`Expr.over` is not supported for Dask backend with multiple partitions.",
-        )
-        if "dask_lazy_p2" in str(constructor)
-        else does_not_raise()
-    )
+    result = df.with_columns(c_max=nw.col("c").max().over("a"))
+    assert_equal_data(result, expected)
 
-    with context:
-        result = df.with_columns(c_max=nw.col("c").max().over("a"))
-        assert_equal_data(result, expected)
 
+def test_over_multiple(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
 
-def test_over_multiple(constructor: Constructor) -> None:
     df = nw.from_native(constructor(data))
     expected = {
         "a": ["a", "a", "b", "b", "b"],
@@ -47,18 +43,8 @@ def test_over_multiple(constructor: Constructor) -> None:
         "c_min": [5, 4, 1, 2, 1],
     }
 
-    context = (
-        pytest.raises(
-            NotImplementedError,
-            match="`Expr.over` is not supported for Dask backend with multiple partitions.",
-        )
-        if "dask_lazy_p2" in str(constructor)
-        else does_not_raise()
-    )
-
-    with context:
-        result = df.with_columns(c_min=nw.col("c").min().over("a", "b"))
-        assert_equal_data(result, expected)
+    result = df.with_columns(c_min=nw.col("c").min().over("a", "b"))
+    assert_equal_data(result, expected)
 
 
 def test_over_invalid(request: pytest.FixtureRequest, constructor: Constructor) -> None:
@@ -68,3 +54,121 @@ def test_over_invalid(request: pytest.FixtureRequest, constructor: Constructor)
     df = nw.from_native(constructor(data))
     with pytest.raises(ValueError, match="Anonymous expressions"):
         df.with_columns(c_min=nw.all().min().over("a", "b"))
+
+
+def test_over_cumsum(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
+        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],
+        "b_cumsum": [1, 3, 3, 8, 11],
+    }
+
+    result = df.with_columns(b_cumsum=nw.col("b").cum_sum().over("a"))
+    assert_equal_data(result, expected)
+
+
+def test_over_cumcount(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+
+    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],
+        "b_cumcount": [1, 2, 1, 2, 3],
+    }
+
+    result = df.with_columns(b_cumcount=nw.col("b").cum_count().over("a"))
+    assert_equal_data(result, expected)
+
+
+def test_over_cumcount_missing_values(
+    request: pytest.FixtureRequest, constructor: Constructor
+) -> None:
+    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+
+    data_with_missing_value = {
+        "a": ["a", "a", "b", "b", "b"],
+        "b": [1, 2, 3, 5, None],
+        "c": [5, 4, 3, 2, 1],
+    }
+
+    df = nw.from_native(constructor(data_with_missing_value))
+    expected = {
+        "a": ["a", "a", "b", "b", "b"],
+        "b": [1, 2, 3, 5, None],
+        "c": [5, 4, 3, 2, 1],
+        "b_cumcount": [1, 2, 1, 2, 2],
+    }
+
+    result = df.with_columns(b_cumcount=nw.col("b").cum_count().over("a"))
+    assert_equal_data(result, expected)
+
+
+def test_over_cummax(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
+        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],
+        "b_cummax": [1, 2, 3, 5, 5],
+    }
+    result = df.with_columns(b_cummax=nw.col("b").cum_max().over("a"))
+    assert_equal_data(result, expected)
+
+
+def test_over_cummin(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+
+    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
+        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],
+        "b_cummin": [1, 1, 3, 3, 3],
+    }
+
+    result = df.with_columns(b_cummin=nw.col("b").cum_min().over("a"))
+    assert_equal_data(result, expected)
+
+
+def test_over_cumprod(request: pytest.FixtureRequest, constructor: Constructor) -> None:
+    if "pyarrow_table" in str(constructor) or "dask_lazy_p2" in str(constructor):
+        request.applymarker(pytest.mark.xfail)
+
+    if "pandas_pyarrow" in str(constructor) and PANDAS_VERSION < (2, 1):
+        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],
+        "b_cumprod": [1, 2, 3, 15, 45],
+    }
+
+    result = df.with_columns(b_cumprod=nw.col("b").cum_prod().over("a"))
+    assert_equal_data(result, expected)
+
+
+def test_over_anonymous() -> None:
+    df = pd.DataFrame({"a": [1, 1, 2], "b": [4, 5, 6]})
+    with pytest.raises(ValueError, match="Anonymous expressions"):
+        nw.from_native(df).select(nw.all().cum_max().over("a"))
diff --git a/tests/expr_and_series/std_test.py b/tests/expr_and_series/std_test.py
index db51c657..b8310080 100644
--- a/tests/expr_and_series/std_test.py
+++ b/tests/expr_and_series/std_test.py
@@ -1,46 +1,44 @@
 from __future__ import annotations
 
+import pytest
+
 import narwhals.stable.v1 as nw
 from tests.utils import Constructor
 from tests.utils import ConstructorEager
 from tests.utils import assert_equal_data
 
 data = {"a": [1, 3, 2], "b": [4, 4, 6], "z": [7.0, 8, 9]}
+data_with_nulls = {"a": [1, 3, 2, None], "b": [4, 4, 6, None], "z": [7.0, 8, 9, None]}
+
+expected_results = {
+    "a_ddof_1": [1.0],
+    "a_ddof_0": [0.816497],
+    "b_ddof_2": [1.632993],
+    "z_ddof_0": [0.816497],
+}
 
 
-def test_std(constructor: Constructor) -> None:
-    df = nw.from_native(constructor(data))
+@pytest.mark.parametrize("input_data", [data, data_with_nulls])
+def test_std(constructor: Constructor, input_data: dict[str, list[float | None]]) -> None:
+    df = nw.from_native(constructor(input_data))
     result = df.select(
-        nw.col("a").std().alias("a_ddof_default"),
         nw.col("a").std(ddof=1).alias("a_ddof_1"),
         nw.col("a").std(ddof=0).alias("a_ddof_0"),
         nw.col("b").std(ddof=2).alias("b_ddof_2"),
         nw.col("z").std(ddof=0).alias("z_ddof_0"),
     )
-    expected = {
-        "a_ddof_default": [1.0],
-        "a_ddof_1": [1.0],
-        "a_ddof_0": [0.816497],
-        "b_ddof_2": [1.632993],
-        "z_ddof_0": [0.816497],
-    }
-    assert_equal_data(result, expected)
+    assert_equal_data(result, expected_results)
 
 
-def test_std_series(constructor_eager: ConstructorEager) -> None:
-    df = nw.from_native(constructor_eager(data), eager_only=True)
+@pytest.mark.parametrize("input_data", [data, data_with_nulls])
+def test_std_series(
+    constructor_eager: ConstructorEager, input_data: dict[str, list[float | None]]
+) -> None:
+    df = nw.from_native(constructor_eager(input_data), eager_only=True)
     result = {
-        "a_ddof_default": [df["a"].std()],
         "a_ddof_1": [df["a"].std(ddof=1)],
         "a_ddof_0": [df["a"].std(ddof=0)],
         "b_ddof_2": [df["b"].std(ddof=2)],
         "z_ddof_0": [df["z"].std(ddof=0)],
     }
-    expected = {
-        "a_ddof_default": [1.0],
-        "a_ddof_1": [1.0],
-        "a_ddof_0": [0.816497],
-        "b_ddof_2": [1.632993],
-        "z_ddof_0": [0.816497],
-    }
-    assert_equal_data(result, expected)
+    assert_equal_data(result, expected_results)
diff --git a/tests/expr_and_series/var_test.py b/tests/expr_and_series/var_test.py
new file mode 100644
index 00000000..bab97d38
--- /dev/null
+++ b/tests/expr_and_series/var_test.py
@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import pytest
+
+import narwhals.stable.v1 as nw
+from tests.utils import Constructor
+from tests.utils import ConstructorEager
+from tests.utils import assert_equal_data
+
+data = {"a": [1, 3, 2], "b": [4, 4, 6], "z": [7.0, 8, 9]}
+data_with_nulls = {"a": [1, 3, 2, None], "b": [4, 4, 6, None], "z": [7.0, 8, 9, None]}
+
+expected_results = {
+    "a_ddof_1": [1.0],
+    "a_ddof_0": [0.6666666666666666],
+    "b_ddof_2": [2.666666666666667],
+    "z_ddof_0": [0.6666666666666666],
+}
+
+
+@pytest.mark.parametrize("input_data", [data, data_with_nulls])
+def test_var(constructor: Constructor, input_data: dict[str, list[float | None]]) -> None:
+    df = nw.from_native(constructor(input_data))
+    result = df.select(
+        nw.col("a").var(ddof=1).alias("a_ddof_1"),
+        nw.col("a").var(ddof=0).alias("a_ddof_0"),
+        nw.col("b").var(ddof=2).alias("b_ddof_2"),
+        nw.col("z").var(ddof=0).alias("z_ddof_0"),
+    )
+    assert_equal_data(result, expected_results)
+
+
+@pytest.mark.parametrize("input_data", [data, data_with_nulls])
+def test_var_series(
+    constructor_eager: ConstructorEager, input_data: dict[str, list[float | None]]
+) -> None:
+    df = nw.from_native(constructor_eager(input_data), eager_only=True)
+    result = {
+        "a_ddof_1": [df["a"].var(ddof=1)],
+        "a_ddof_0": [df["a"].var(ddof=0)],
+        "b_ddof_2": [df["b"].var(ddof=2)],
+        "z_ddof_0": [df["z"].var(ddof=0)],
+    }
+    assert_equal_data(result, expected_results)
diff --git a/tests/group_by_test.py b/tests/group_by_test.py
index e51cbac6..61e63e74 100644
--- a/tests/group_by_test.py
+++ b/tests/group_by_test.py
@@ -105,6 +105,7 @@ def test_group_by_iter(constructor_eager: ConstructorEager) -> None:
         ("max", {"a": [1, 2], "b": [2, 3]}),
         ("min", {"a": [1, 2], "b": [1, 3]}),
         ("std", {"a": [1, 2], "b": [0.707107, None]}),
+        ("var", {"a": [1, 2], "b": [0.5, None]}),
         ("len", {"a": [1, 2], "b": [3, 1]}),
         ("n_unique", {"a": [1, 2], "b": [3, 1]}),
         ("count", {"a": [1, 2], "b": [2, 1]}),
@@ -118,6 +119,9 @@ def test_group_by_depth_1_agg(
 ) -> None:
     if "cudf" in str(constructor) and attr == "n_unique":
         request.applymarker(pytest.mark.xfail)
+    if "pandas_pyarrow" in str(constructor) and attr == "var" and PANDAS_VERSION < (2, 1):
+        # Known issue with variance calculation in pandas 2.0.x with pyarrow backend in groupby operations"
+        request.applymarker(pytest.mark.xfail)
     data = {"a": [1, 1, 1, 2], "b": [1, None, 2, 3]}
     expr = getattr(nw.col("b"), attr)()
     result = nw.from_native(constructor(data)).group_by("a").agg(expr).sort("a")