Merge branch 'main' into fix-memory-erred

.github/workflows/conda.yml

@@ -30,7 +30,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Set up Python
-        uses: conda-incubator/[email protected].2
+        uses: conda-incubator/[email protected].3
         with:
           miniforge-variant: Mambaforge
           use-mamba: true

.github/workflows/test-report.yaml

@@ -21,7 +21,7 @@ jobs:
       - uses: actions/[email protected]
 
       - name: Setup Conda Environment
-        uses: conda-incubator/[email protected].2
+        uses: conda-incubator/[email protected].3
         with:
           miniforge-variant: Mambaforge
           miniforge-version: latest

.github/workflows/tests.yaml

@@ -115,7 +115,7 @@ jobs:
           fetch-depth: 0
 
       - name: Setup Conda Environment
-        uses: conda-incubator/[email protected].2
+        uses: conda-incubator/[email protected].3
         with:
           miniforge-variant: Mambaforge
           miniforge-version: latest

distributed/dashboard/tests/test_scheduler_bokeh.py

@@ -1346,7 +1346,7 @@ async def test_shuffling(c, s, a, b):
         freq="10 s",
     )
     with dask.config.set({"dataframe.shuffle.method": "p2p"}):
-        df2 = dd.shuffle.shuffle(df, "x").persist()
+        df2 = df.shuffle("x").persist()
     start = time()
     while not ss.source.data["comm_written"]:
         ss.update()

distributed/protocol/tests/test_highlevelgraph.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import contextlib
+
 import pytest
 
 np = pytest.importorskip("numpy")
@@ -173,7 +175,13 @@ async def test_dataframe_annotations(c, s, a, b):
     acol = df["a"]
     bcol = df["b"]
 
-    with dask.annotate(retries=retries):
+    ctx = contextlib.nullcontext()
+    if dd._dask_expr_enabled():
+        ctx = pytest.warns(
+            UserWarning, match="Annotations will be ignored when using query-planning"
+        )
+
+    with dask.annotate(retries=retries), ctx:
         df = acol + bcol
 
     with dask.config.set(optimization__fuse__active=False):
@@ -182,5 +190,6 @@ async def test_dataframe_annotations(c, s, a, b):
     assert rdf.dtypes == np.float64
     assert (rdf == 10.0).all()
 
-    # There is an annotation match per partition (i.e. task)
-    assert plugin.retry_matches == df.npartitions
+    if not dd._dask_expr_enabled():
+        # There is an annotation match per partition (i.e. task)
+        assert plugin.retry_matches == df.npartitions