feat: dask expr len method (#762)

Author: FBruzzesi

narwhals/_dask/expr.py

@@ -473,6 +473,13 @@ def clip(
             returns_scalar=False,
         )
 
+    def len(self: Self) -> Self:
+        return self._from_call(
+            lambda _input: _input.size,
+            "len",
+            returns_scalar=True,
+        )
+
     @property
     def str(self: Self) -> DaskExprStringNamespace:
         return DaskExprStringNamespace(self)

tests/expr_and_series/len_test.py

@@ -6,7 +6,18 @@
 from tests.utils import compare_dicts
 
 
-def test_len(constructor: Any, request: Any) -> None:
+def test_len_no_filter(constructor: Any) -> None:
+    data = {"a": list("xyz"), "b": [1, 2, 1]}
+    expected = {"l": [3], "l2": [6]}
+    df = nw.from_native(constructor(data)).select(
+        nw.col("a").len().alias("l"),
+        (nw.col("a").len() * 2).alias("l2"),
+    )
+
+    compare_dicts(df, expected)
+
+
+def test_len_len_chaining(constructor: Any, request: Any) -> None:
     data = {"a": list("xyz"), "b": [1, 2, 1]}
     expected = {"a1": [2], "a2": [1]}
     if "dask" in str(constructor):

tests/test_group_by.py

@@ -94,9 +94,7 @@ def test_group_by_iter(constructor_eager: Any) -> None:
     assert sorted(keys) == sorted(expected_keys)
 
 
-def test_group_by_len(constructor: Any, request: Any) -> None:
-    if "dask" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
+def test_group_by_len(constructor: Any) -> None:
     result = (
         nw.from_native(constructor(data)).group_by("a").agg(nw.col("b").len()).sort("a")
     )