feat: dask expr is_unique & is_duplicated

narwhals/_dask/expr.py

@@ -546,12 +546,30 @@ def func(_input: Any) -> Any:
         )
 
     def is_duplicated(self: Self) -> Self:
-        msg = "`Expr.is_duplicated` is not support since Dask currently has no native duplicated check"
-        raise NotImplementedError(msg)
+        def func(_input: Any) -> Any:
+            _name = _input.name
+            return (
+                _input.to_frame().groupby(_name).transform("size", meta=(_name, int)) > 1
+            )
+
+        return self._from_call(
+            func,
+            "is_duplicated",
+            returns_scalar=False,
+        )
 
     def is_unique(self: Self) -> Self:
-        msg = "`Expr.is_duplicated` is not support since Dask currently has no native duplicated check"
-        raise NotImplementedError(msg)
+        def func(_input: Any) -> Any:
+            _name = _input.name
+            return (
+                _input.to_frame().groupby(_name).transform("size", meta=(_name, int)) == 1
+            )
+
+        return self._from_call(
+            func,
+            "is_unique",
+            returns_scalar=False,
+        )
 
     def is_in(self: Self, other: Any) -> Self:
         return self._from_call(

tests/expr_and_series/is_duplicated_test.py

@@ -13,8 +13,6 @@
 
 
 def test_is_duplicated_expr(constructor: Any, request: Any) -> None:
-    if "dask" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
     if "modin" in str(constructor):
         # TODO(unassigned): why is Modin failing here?
         request.applymarker(pytest.mark.xfail)

tests/expr_and_series/is_unique_test.py

@@ -13,8 +13,6 @@
 
 
 def test_is_unique_expr(constructor: Any, request: Any) -> None:
-    if "dask" in str(constructor):
-        request.applymarker(pytest.mark.xfail)
     if "modin" in str(constructor):
         # TODO(unassigned): why is Modin failing here?
         request.applymarker(pytest.mark.xfail)