fix(dask): don't call compute when executing cov/corr

BREAKING CHANGE: the dask backend no longer supports `cov`/`corr` with `how="pop"`.

ibis/backends/dask/executor.py

@@ -27,7 +27,7 @@
     PandasWindowFunction,
     plan,
 )
-from ibis.common.exceptions import UnboundExpressionError
+from ibis.common.exceptions import UnboundExpressionError, UnsupportedOperationError
 from ibis.formats.pandas import PandasData, PandasType
 from ibis.util import gen_name
 
@@ -213,37 +213,31 @@ def agg(df):
 
     @classmethod
     def visit(cls, op: ops.Correlation, left, right, where, how):
-        if where is None:
+        if how == "pop":
+            raise UnsupportedOperationError(
+                "Dask doesn't support `corr` with `how='pop'`"
+            )
 
-            def agg(df):
-                return df[left.name].corr(df[right.name])
-        else:
+        def agg(df):
+            if where is not None:
+                df = df.where(df[where.name])
 
-            def agg(df):
-                mask = df[where.name]
-                lhs = df[left.name][mask].compute()
-                rhs = df[right.name][mask].compute()
-                return lhs.corr(rhs)
+            return df[left.name].corr(df[right.name])
 
         return agg
 
     @classmethod
     def visit(cls, op: ops.Covariance, left, right, where, how):
-        # TODO(kszucs): raise a warning about triggering compute()?
-        ddof = {"pop": 0, "sample": 1}[how]
-        if where is None:
+        if how == "pop":
+            raise UnsupportedOperationError(
+                "Dask doesn't support `cov` with `how='pop'`"
+            )
 
-            def agg(df):
-                lhs = df[left.name].compute()
-                rhs = df[right.name].compute()
-                return lhs.cov(rhs, ddof=ddof)
-        else:
+        def agg(df):
+            if where is not None:
+                df = df.where(df[where.name])
 
-            def agg(df):
-                mask = df[where.name]
-                lhs = df[left.name][mask].compute()
-                rhs = df[right.name][mask].compute()
-                return lhs.cov(rhs, ddof=ddof)
+            return df[left.name].cov(df[right.name])
 
         return agg
 

ibis/backends/tests/test_aggregation.py

@@ -874,6 +874,11 @@ def test_quantile(
             lambda t, where: t.G[where].cov(t.RBI[where], ddof=0),
             id="covar_pop",
             marks=[
+                pytest.mark.notyet(
+                    ["dask"],
+                    reason="dask doesn't support `cov(ddof=0)` yet",
+                    raises=com.UnsupportedOperationError,
+                ),
                 pytest.mark.notimpl(
                     ["polars", "druid"],
                     raises=com.OperationNotDefinedError,
@@ -914,6 +919,11 @@ def test_quantile(
             lambda t, where: t.G[where].corr(t.RBI[where]),
             id="corr_pop",
             marks=[
+                pytest.mark.notyet(
+                    ["dask"],
+                    raises=com.UnsupportedOperationError,
+                    reason="dask doesn't support `corr(ddof=0)` yet",
+                ),
                 pytest.mark.notimpl(
                     ["druid"],
                     raises=com.OperationNotDefinedError,
@@ -978,6 +988,11 @@ def test_quantile(
             lambda t, where: (t.G[where] > 34.0).cov(t.G[where] <= 34.0, ddof=0),
             id="covar_pop_bool",
             marks=[
+                pytest.mark.notyet(
+                    ["dask"],
+                    raises=com.UnsupportedOperationError,
+                    reason="dask doesn't support `cov(ddof=0)` yet",
+                ),
                 pytest.mark.notimpl(
                     ["polars", "druid"],
                     raises=com.OperationNotDefinedError,
@@ -1002,6 +1017,11 @@ def test_quantile(
             lambda t, where: (t.G[where] > 34.0).corr(t.G[where] <= 34.0),
             id="corr_pop_bool",
             marks=[
+                pytest.mark.notyet(
+                    ["dask"],
+                    raises=com.UnsupportedOperationError,
+                    reason="dask doesn't support `corr(ddof=0)` yet",
+                ),
                 pytest.mark.notimpl(
                     ["druid"],
                     raises=com.OperationNotDefinedError,