Repository: narwhals-dev/narwhals (python)

Issue:
[Bug]: `nw.len().over("column")` not working for pandas DataFrames
### Describe the bug

```python
df.with_columns(a_len_per_group=nw.len().over("b"))
```

Raises an exception when df native DataFrame is a pandas DataFrame while it works as expected when df native DataFrame is a polars DataFrame.

### Steps or code to reproduce the bug

```python
import pandas as pd
import narwhals as nw
df_native = pd.DataFrame({"a": [1, 2, 4], "b": ["x", "x", "y"]})
df = nw.from_native(df_native)
df.with_columns(a_len_per_group=nw.len().over("b"))
```

### Expected results

```
┌───────────────────────────────┐
|      Narwhals DataFrame       |
|-------------------------------|
|shape: (3, 3)                  |
|┌─────┬─────┬─────────────────┐|
|│ a   ┆ b   ┆ a_len_per_group │|
|│ --- ┆ --- ┆ ---             │|
|│ i64 ┆ str ┆ u32             │|
|╞═════╪═════╪═════════════════╡|
|│ 1   ┆ x   ┆ 2               │|
|│ 2   ┆ x   ┆ 2               │|
|│ 4   ┆ y   ┆ 1               │|
|└─────┴─────┴─────────────────┘|
└───────────────────────────────┘
```

### Actual results

It raises `KeyError: "Columns not found: 'len'"`

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

```shell
System:
    python: 3.10.15 | packaged by conda-forge | (main, Sep 20 2024, 16:37:05) [GCC 13.3.0]
executable: /{redacted}/.venv/bin/python
   machine: Linux-5.10.0-34-cloud-amd64-x86_64-with-glibc2.31

Python dependencies:
     narwhals: 1.34.1
       pandas: 2.2.3
       polars: 1.24.0
         cudf: 
        modin: 
      pyarrow: 18.1.0
        numpy: 1.26.4
```

### Relevant log output

```shell

```

Pull request description:
fix: `nw.len().over` was unnecessarily raising for pandas-like
closes #2369 

<!--
# 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/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index 6fa8e7e3..a1f0abe4 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -644,9 +644,18 @@ class DaskExpr(
                         message=".*`meta` is not specified",
                         category=UserWarning,
                     )
-                    res_native = df.native.groupby(partition_by)[
-                        list(output_names)
-                    ].transform(dask_function_name, **self._call_kwargs)
+                    grouped = df.native.groupby(partition_by)
+                    if dask_function_name == "size":
+                        if len(output_names) != 1:  # pragma: no cover
+                            msg = "Safety check failed, please report a bug."
+                            raise AssertionError(msg)
+                        res_native = grouped.transform(
+                            dask_function_name, **self._call_kwargs
+                        ).to_frame(output_names[0])
+                    else:
+                        res_native = grouped[list(output_names)].transform(
+                            dask_function_name, **self._call_kwargs
+                        )
                 result_frame = df._with_native(
                     res_native.rename(columns=dict(zip(output_names, aliases)))
                 ).native
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index 0f4584a5..a812a386 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -198,7 +198,7 @@ class PandasLikeExpr(EagerExpr["PandasLikeDataFrame", PandasLikeSeries]):
     def shift(self: Self, n: int) -> Self:
         return self._reuse_series("shift", call_kwargs={"n": n})
 
-    def over(
+    def over(  # noqa: PLR0915
         self: Self,
         partition_by: Sequence[str],
         order_by: Sequence[str] | None,
@@ -265,10 +265,9 @@ class PandasLikeExpr(EagerExpr["PandasLikeDataFrame", PandasLikeSeries]):
                 elif reverse:
                     columns = list(set(partition_by).union(output_names))
                     df = df[columns][::-1]
+                grouped = df._native_frame.groupby(partition_by)
                 if function_name.startswith("rolling"):
-                    rolling = df._native_frame.groupby(partition_by)[
-                        list(output_names)
-                    ].rolling(**pandas_kwargs)
+                    rolling = grouped[list(output_names)].rolling(**pandas_kwargs)
                     assert pandas_function_name is not None  # help mypy  # noqa: S101
                     if pandas_function_name in {"std", "var"}:
                         res_native = getattr(rolling, pandas_function_name)(
@@ -276,10 +275,15 @@ class PandasLikeExpr(EagerExpr["PandasLikeDataFrame", PandasLikeSeries]):
                         )
                     else:
                         res_native = getattr(rolling, pandas_function_name)()
+                elif function_name == "len":
+                    if len(output_names) != 1:  # pragma: no cover
+                        msg = "Safety check failed, please report a bug."
+                        raise AssertionError(msg)
+                    res_native = grouped.transform("size").to_frame(aliases[0])
                 else:
-                    res_native = df._native_frame.groupby(partition_by)[
-                        list(output_names)
-                    ].transform(pandas_function_name, **pandas_kwargs)
+                    res_native = grouped[list(output_names)].transform(
+                        pandas_function_name, **pandas_kwargs
+                    )
                 result_frame = df._with_native(res_native).rename(
                     dict(zip(output_names, aliases))
                 )


Test patch:
diff --git a/tests/expr_and_series/over_test.py b/tests/expr_and_series/over_test.py
index be800852..2e53ab1d 100644
--- a/tests/expr_and_series/over_test.py
+++ b/tests/expr_and_series/over_test.py
@@ -428,3 +428,17 @@ def test_over_without_partition_by(
     )
     expected = {"a": [1, 2, -1], "b": [1, 3, 4], "i": [0, 1, 2]}
     assert_equal_data(result, expected)
+
+
+def test_len_over_2369(constructor: Constructor, request: pytest.FixtureRequest) -> None:
+    if "duckdb" in str(constructor) and DUCKDB_VERSION < (1, 3):
+        pytest.skip()
+    if "pandas" in str(constructor) and PANDAS_VERSION < (1, 5):
+        pytest.skip()
+    if any(x in str(constructor) for x in ("modin",)):
+        # https://github.com/modin-project/modin/issues/7508
+        request.applymarker(pytest.mark.xfail)
+    df = nw.from_native(constructor({"a": [1, 2, 4], "b": ["x", "x", "y"]}))
+    result = df.with_columns(a_len_per_group=nw.len().over("b")).sort("a")
+    expected = {"a": [1, 2, 4], "b": ["x", "x", "y"], "a_len_per_group": [2, 2, 1]}
+    assert_equal_data(result, expected)