Repository: dask/dask
Pull request #10496 (2023-09-07 09:11:48)

`get_default_shuffle_method` raises if `pyarrow` is outdated
- [x] Sibling to and blocked by dask/distributed#8157
- [x] Tests added / passed
- [x] Passes `pre-commit run --all-files`


jrbourbeau: Marking as `DNM` for now since the `distributed` PR needs to go in first

@hendrikmakait we're getting a `FAILED dask/tests/test_distributed.py::test_bag_groupby_default - RuntimeError: P2P shuffling requires pyarrow>=7.0.0` failure here now. Maybe this will go away once the `distributed` PR is merged (I've not looked into it at all). Just wanted to double check whether additional changes are still needed or not 
hendrikmakait: @jrbourbeau: I checked `test_distributed.py` locally and the following tests will start failing because the installed `pyarrow` version does not match the minimum requirement of `p2p`:

* `test_fused_blockwise_dataframe_merge`
* `test_map_partitions_df_input`

There are two ways to fix this:

1. Increase the installed version to `pyarrow>=12.0.0`
2. Force the tests to use task-based shuffling

From what I understand, we _want_ to test those lower version, so I'll take a quick stab at (2.).
hendrikmakait: When using the sibling branch on CI, tests are green: https://github.com/dask/dask/pull/10496/commits/f9a11b29d5a5dbf4b00e6520bd5a629abe4d18c1
fjetter: merged main again. This should also show that the changes are compatible with dask/distributed after merging https://github.com/dask/distributed/pull/8157

Once build is done (and green) we can merge
fjetter: There are plenty of failures on gpu CI but I think this is unrelated cc @dask/gpu 

Patch:
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 5863454f9..4603ae57c 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1526,10 +1526,14 @@ class Bag(DaskMethodsMixin):
         if method is not None:
             raise Exception("The method= keyword has been moved to shuffle=")
         if shuffle is None:
-            shuffle = get_default_shuffle_method()
-            if shuffle == "p2p":
-                # Not implemented for Bags
+            try:
+                shuffle = get_default_shuffle_method()
+            except ImportError:
                 shuffle = "tasks"
+            else:
+                if shuffle == "p2p":
+                    # Not implemented for Bags
+                    shuffle = "tasks"
         if shuffle == "disk":
             return groupby_disk(
                 self, grouper, npartitions=npartitions, blocksize=blocksize
diff --git a/dask/config.py b/dask/config.py
index 1ba48faee..e1ea81e83 100644
--- a/dask/config.py
+++ b/dask/config.py
@@ -126,7 +126,7 @@ def update(
         k = canonical_name(k, old)
 
         if isinstance(v, Mapping):
-            if k not in old or old[k] is None:
+            if k not in old or old[k] is None or not isinstance(old[k], dict):
                 old[k] = {}
             update(
                 old[k],
@@ -673,6 +673,7 @@ deprecations: dict[str, str | None] = {
     "array.rechunk-threshold": "array.rechunk.threshold",
     "dataframe.shuffle.algorithm": "dataframe.shuffle.method",
     "dataframe.shuffle-compression": "dataframe.shuffle.compression",
+    "admin.traceback.shorten.what": "admin.traceback.shorten",  # changed in 2023.9.0
 }
 
 
diff --git a/dask/utils.py b/dask/utils.py
index 81e5d4c6b..6713fb3b8 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -2116,17 +2116,17 @@ def get_default_shuffle_method() -> str:
         from distributed import default_client
 
         default_client()
-        try:
-            from distributed.shuffle import check_minimal_arrow_version
-
-            check_minimal_arrow_version()
-            return "p2p"
-        except RuntimeError:
-            pass
-        return "tasks"
     except (ImportError, ValueError):
         return "disk"
 
+    try:
+        from distributed.shuffle import check_minimal_arrow_version
+
+        check_minimal_arrow_version()
+    except ModuleNotFoundError:
+        return "tasks"
+    return "p2p"
+
 
 def get_meta_library(like):
     if hasattr(like, "_meta"):


Test patch:
diff --git a/dask/tests/test_config.py b/dask/tests/test_config.py
index 16339a8b6..aa44ffaed 100644
--- a/dask/tests/test_config.py
+++ b/dask/tests/test_config.py
@@ -91,6 +91,20 @@ def test_update_defaults():
     assert current == {"a": 0, "b": {"c": 0, "d": 3}, "extra": 0, "new-extra": 0}
 
 
+def test_update_list_to_dict():
+    a = {"x": {"y": 1, "z": 2}}
+    b = {"x": [1, 2], "w": 3}
+    update(b, a)
+    assert b == {"x": {"y": 1, "z": 2}, "w": 3}
+
+
+def test_update_dict_to_list():
+    a = {"x": [1, 2]}
+    b = {"x": {"y": 1, "z": 2}, "w": 3}
+    update(b, a)
+    assert b == {"x": [1, 2], "w": 3}
+
+
 def test_merge():
     a = {"x": 1, "y": {"a": 1}}
     b = {"x": 2, "z": 3, "y": {"b": 2}}
@@ -565,16 +579,29 @@ def test_rename():
         "x.y": "foo.y",
         "a.b": "ab",
         "not-found": "not.found",
+        "super.old": "super",
+    }
+    config = {
+        "foo_bar": 1,
+        "x": {"y": 2, "z": 3},
+        "a": {"b": None},
+        "super": {"old": 4},
     }
-    config = {"foo_bar": 1, "x": {"y": 2, "z": 3}, "a": {"b": None}}
     with pytest.warns(FutureWarning) as w:
         rename(aliases, config=config)
     assert [str(wi.message) for wi in w.list] == [
         "Dask configuration key 'foo-bar' has been deprecated; please use 'foo.bar' instead",
         "Dask configuration key 'x.y' has been deprecated; please use 'foo.y' instead",
         "Dask configuration key 'a.b' has been deprecated; please use 'ab' instead",
+        "Dask configuration key 'super.old' has been deprecated; please use 'super' instead",
     ]
-    assert config == {"foo": {"bar": 1, "y": 2}, "x": {"z": 3}, "a": {}, "ab": None}
+    assert config == {
+        "foo": {"bar": 1, "y": 2},
+        "x": {"z": 3},
+        "a": {},
+        "ab": None,
+        "super": 4,
+    }
 
 
 @pytest.mark.parametrize(
diff --git a/dask/tests/test_distributed.py b/dask/tests/test_distributed.py
index cad1c9835..fa7b29d6f 100644
--- a/dask/tests/test_distributed.py
+++ b/dask/tests/test_distributed.py
@@ -147,7 +147,7 @@ def test_fused_blockwise_dataframe_merge(c, fuse):
     df2 += 10
 
     with dask.config.set({"optimization.fuse.active": fuse}):
-        ddfm = ddf1.merge(ddf2, on=["x"], how="left")
+        ddfm = ddf1.merge(ddf2, on=["x"], how="left", shuffle="tasks")
         ddfm.head()  # https://github.com/dask/dask/issues/7178
         dfm = ddfm.compute().sort_values("x")
         # We call compute above since `sort_values` is not
@@ -765,7 +765,7 @@ def test_map_partitions_df_input():
         merged_df = dd.from_pandas(pd.DataFrame({"b": range(10)}), npartitions=1)
 
         # Notice, we include a shuffle in order to trigger a complex culling
-        merged_df = merged_df.shuffle(on="b")
+        merged_df = merged_df.shuffle(on="b", shuffle="tasks")
 
         merged_df.map_partitions(
             f, ddf, meta=merged_df, enforce_metadata=False
diff --git a/dask/tests/test_traceback.py b/dask/tests/test_traceback.py
index 034986b2c..cf64a06cc 100644
--- a/dask/tests/test_traceback.py
+++ b/dask/tests/test_traceback.py
@@ -132,3 +132,30 @@ def test_distributed_shorten_traceback():
         expect = [TEST_NAME, "compute", "compute", "f3", "f2", "f1"]
         with assert_tb_levels(expect):
             dp2.compute()
+
+
+def test_deprecated_config(tmp_path):
+    """Test config override in the format between 2023.6.1 and 2023.8.1"""
+    d = {}
+    dask.config.refresh(config=d)
+    actual = dask.config.get("admin.traceback.shorten", config=d)
+    assert isinstance(actual, list) and len(actual) > 2
+
+    d = {}
+    with open(tmp_path / "dask.yaml", "w") as fh:
+        fh.write(
+            """
+            admin:
+              traceback:
+                shorten:
+                  when:
+                    - dask/base.py
+                  what:
+                    - dask/core.py
+            """
+        )
+
+    with pytest.warns(FutureWarning):
+        dask.config.refresh(config=d, paths=[tmp_path])
+    actual = dask.config.get("admin.traceback.shorten", config=d)
+    assert actual == ["dask/core.py"]