Repository: dask/dask
Pull request #11511 (2024-11-11 17:47:12)

Remove only_refs parsing option for TaskSpec
This removes a parsing option `only_refs` that is not only poorly documented but probably also poorly defined. In a nutshell, this should cover the way how `Fused` was implemented prior to https://github.com/dask/dask-expr/pull/1159 and can now be removed.

github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     13 files  ± 0       13 suites  ±0   3h 23m 39s ⏱️ - 8m 27s
 13 289 tests  -  2   12 227 ✅  -  2   1 062 💤 ±0  0 ❌ ±0 
138 676 runs   - 26  119 683 ✅  - 25  18 993 💤  - 1  0 ❌ ±0 

Results for commit d97a31b6. ± Comparison against base commit bfa88d2b.

<details>
  <summary>This pull request <b>removes</b> 2 tests.</summary>

```
dask.tests.test_task_spec ‑ test_convert_task_only_ref
dask.tests.test_task_spec ‑ test_convert_tasks_only_ref_lowlevel
```
</details>

[test-results]:data:application/gzip;base64,H4sIABZLMmcC/03MTQ6DIBCG4asY1l3wozDTyzSAkJCqNAgr07sXq0GX7zOTbyM+TG4lz46JR0fWEvJVY0k6h7jszTnDSvWajzuH1q+1WHs8cXXhO3x2pJI38zpM1WgDl1JMp6SyHNMglTy7TTOUIBqe04DY7Db97/uyjfMccg0yotKCGYmWWSocAJdIaT9qj9pZaoQZAHo/kO8PCVXSAxoBAAA=


Patch:
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index c4e7e8c92..d53298d7b 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -207,7 +207,6 @@ def convert_legacy_task(
     k: KeyType | None,
     task: _T,
     all_keys: Container,
-    only_refs: bool,
 ) -> GraphNode | _T:
     global SubgraphType
     if SubgraphType is None:
@@ -229,7 +228,7 @@ def convert_legacy_task(
                 subgraph.dsk, set(subgraph.inkeys), all_keys
             )
             sub_dsk = subgraph.dsk
-            converted_graph = convert_legacy_graph(sub_dsk, all_keys_inner, only_refs)
+            converted_graph = convert_legacy_graph(sub_dsk, all_keys_inner)
             external_deps = (
                 _get_dependencies(converted_graph)
                 - set(converted_graph)
@@ -244,20 +243,13 @@ def convert_legacy_task(
                 converted_sub_dsk,
                 func.outkey,
                 {
-                    k: convert_legacy_task(None, target, all_keys, only_refs)
+                    k: convert_legacy_task(None, target, all_keys)
                     for k, target in zip(subgraph.inkeys, args)
                 },
                 {ext_dep: Alias(ext_dep) for ext_dep in external_deps},
             )
         else:
-            new_args = tuple(
-                convert_legacy_task(None, a, all_keys, only_refs) for a in args
-            )
-            if only_refs:
-                if any(isinstance(a, GraphNode) for a in new_args):
-                    return Task(k, identity, func, *new_args)
-                else:
-                    return DataNode(k, (func, *new_args))
+            new_args = tuple(convert_legacy_task(None, a, all_keys) for a in args)
             return Task(k, func, *new_args)
     try:
         if isinstance(task, (bytes, int, float, str, tuple)):
@@ -275,14 +267,10 @@ def convert_legacy_task(
             return _wrap_namedtuple_task(
                 k,
                 task,
-                partial(
-                    convert_legacy_task, None, all_keys=all_keys, only_refs=only_refs
-                ),
+                partial(convert_legacy_task, None, all_keys=all_keys),
             )
         else:
-            parsed_args = tuple(
-                convert_legacy_task(None, t, all_keys, only_refs) for t in task
-            )
+            parsed_args = tuple(convert_legacy_task(None, t, all_keys) for t in task)
             if any(isinstance(a, GraphNode) for a in parsed_args):
                 return Task(
                     k, _identity_cast, *parsed_args, typ=DataNode(None, type(task))
@@ -291,10 +279,7 @@ def convert_legacy_task(
                 return cast(_T, type(task)(parsed_args))
     elif isinstance(task, dict):
         return _wrap_dict_in_task(
-            {
-                k: convert_legacy_task(k, v, all_keys, only_refs=True)
-                for k, v in task.items()
-            }
+            {k: convert_legacy_task(k, v, all_keys) for k, v in task.items()}
         )
 
     elif isinstance(task, TaskRef):
@@ -306,22 +291,13 @@ def convert_legacy_task(
         return task
 
 
-def convert_legacy_graph(
-    dsk: Mapping,
-    all_keys: Container | None = None,
-    only_refs: bool = False,
-):
+def convert_legacy_graph(dsk: Mapping, all_keys: Container | None = None):
     if not all_keys:
         all_keys = set(dsk)
     new_dsk = {}
     for k, arg in dsk.items():
-        t = convert_legacy_task(k, arg, all_keys, only_refs)
-        if (
-            not only_refs
-            and isinstance(t, Alias)
-            and isinstance(arg, TaskRef)
-            and t.key == arg.key
-        ):
+        t = convert_legacy_task(k, arg, all_keys)
+        if isinstance(t, Alias) and isinstance(arg, TaskRef) and t.key == arg.key:
             # This detects cycles?
             continue
         new_dsk[k] = t
diff --git a/dask/array/__init__.py b/dask/array/__init__.py
index 02a5fb3fb..e01bda51a 100644
--- a/dask/array/__init__.py
+++ b/dask/array/__init__.py
@@ -100,7 +100,7 @@ try:
     from dask.array.numpy_compat import moveaxis, rollaxis
     from dask.array.optimization import optimize
     from dask.array.overlap import map_overlap, push
-    from dask.array.percentile import percentile
+    from dask.array.percentile import nanpercentile, percentile
     from dask.array.rechunk import rechunk
     from dask.array.reductions import (
         all,
diff --git a/dask/array/percentile.py b/dask/array/percentile.py
index 9d7023862..b2710ed88 100644
--- a/dask/array/percentile.py
+++ b/dask/array/percentile.py
@@ -11,6 +11,7 @@ from tlz import merge
 from dask.array.core import Array
 from dask.base import tokenize
 from dask.highlevelgraph import HighLevelGraph
+from dask.utils import derived_from
 
 
 @wraps(np.percentile)
@@ -107,8 +108,14 @@ def percentile(a, q, method="linear", internal_method="default", **kwargs):
     numpy.percentile : Numpy's equivalent Percentile function
     """
     from dask.array.dispatch import percentile_lookup as _percentile
+    from dask.array.reductions import quantile
     from dask.array.utils import array_safe, meta_from_array
 
+    if a.ndim > 1:
+        q = np.true_divide(q, a.dtype.type(100) if a.dtype.kind == "f" else 100)
+
+        return quantile(a, q, method=method, **kwargs)
+
     allowed_internal_methods = ["default", "dask", "tdigest"]
 
     if method in allowed_internal_methods:
@@ -309,3 +316,12 @@ def merge_percentiles(finalq, qs, vals, method="lower", Ns=None, raise_on_nan=Tr
                 "'higher', 'midpoint', or 'nearest'"
             )
     return rv
+
+
+@derived_from(np)
+def nanpercentile(a, q, **kwargs):
+    from dask.array.reductions import nanquantile
+
+    q = np.true_divide(q, a.dtype.type(100) if a.dtype.kind == "f" else 100)
+
+    return nanquantile(a, q, **kwargs)
diff --git a/docs/source/array-api.rst b/docs/source/array-api.rst
index 30c279ac9..2e0143040 100644
--- a/docs/source/array-api.rst
+++ b/docs/source/array-api.rst
@@ -166,6 +166,7 @@ Top level functions
    nanmin
    nanprod
    nanquantile
+   nanpercentile
    nanstd
    nansum
    nanvar
diff --git a/docs/source/array-numpy-compatibility.rst b/docs/source/array-numpy-compatibility.rst
index db1e889d1..6dc930951 100644
--- a/docs/source/array-numpy-compatibility.rst
+++ b/docs/source/array-numpy-compatibility.rst
@@ -242,7 +242,7 @@ is available via the `Array API Comparison repository <https://github.com/data-a
    :obj:`numpy.nanmean`, :obj:`dask.array.nanmean` [#1]_, dask equivalent
    :obj:`numpy.nanmedian`, :obj:`dask.array.nanmedian` [#16]_, dask equivalent
    :obj:`numpy.nanmin`, :obj:`dask.array.nanmin` [#1]_ [#2]_, dask equivalent
-   :obj:`numpy.nanpercentile`, \-
+   :obj:`numpy.nanpercentile`, :obj:`dask.array.nanpercentile`
    :obj:`numpy.nanprod`, :obj:`dask.array.nanprod` [#1]_ [#2]_, dask equivalent
    :obj:`numpy.nanquantile`, :obj:`dask.array.nanquantile`
    :obj:`numpy.nanstd`, :obj:`dask.array.nanstd` [#1]_, dask equivalent


Test patch:
diff --git a/.github/workflows/test-report.yaml b/.github/workflows/test-report.yaml
index c84889062..1fff48b7d 100644
--- a/.github/workflows/test-report.yaml
+++ b/.github/workflows/test-report.yaml
@@ -60,7 +60,7 @@ jobs:
           mv test_report.html test_short_report.html deploy/
 
       - name: Deploy 🚀
-        uses: JamesIves/github-pages-deploy-action@v4.6.8
+        uses: JamesIves/github-pages-deploy-action@v4.6.9
         with:
           branch: gh-pages
           folder: deploy
diff --git a/dask/array/tests/test_reductions.py b/dask/array/tests/test_reductions.py
index 01625c0eb..d5cdb1d18 100644
--- a/dask/array/tests/test_reductions.py
+++ b/dask/array/tests/test_reductions.py
@@ -914,7 +914,7 @@ def test_weighted_reduction():
 @pytest.mark.parametrize("q", [0.75, [0.75], [0.75, 0.4]])
 @pytest.mark.parametrize("rechunk", [True, False])
 def test_nanquantile(rechunk, q, axis):
-    shape = 10, 15, 10, 15
+    shape = 7, 10, 7, 10
     arr = np.random.randn(*shape)
     indexer = np.random.randint(0, 10, size=shape)
     arr[indexer >= 8] = np.nan
@@ -925,6 +925,14 @@ def test_nanquantile(rechunk, q, axis):
         da.nanquantile(darr, q, axis=axis, keepdims=True),
         np.nanquantile(arr, q, axis=axis, keepdims=True),
     )
+    assert_eq(
+        da.nanpercentile(darr, q * 100, axis=axis),
+        np.nanpercentile(arr, q * 100, axis=axis),
+    )
+    assert_eq(
+        da.nanpercentile(darr, q * 100, axis=axis, keepdims=True),
+        np.nanpercentile(arr, q * 100, axis=axis, keepdims=True),
+    )
 
 
 @pytest.mark.parametrize("axis", [3, [1, 3]])
@@ -942,6 +950,11 @@ def test_quantile(rechunk, q, axis):
         da.quantile(darr, q, axis=axis, keepdims=True),
         np.quantile(arr, q, axis=axis, keepdims=True),
     )
+    assert_eq(da.percentile(darr, q, axis=axis), np.percentile(arr, q, axis=axis))
+    assert_eq(
+        da.percentile(darr, q, axis=axis, keepdims=True),
+        np.percentile(arr, q, axis=axis, keepdims=True),
+    )
 
 
 def test_nanquantile_all_nan():
@@ -954,6 +967,7 @@ def test_nanquantile_all_nan():
         assert_eq(
             da.nanquantile(darr, 0.75, axis=-1), np.nanquantile(arr, 0.75, axis=-1)
         )
+        assert_eq(da.percentile(darr, 0.75, axis=-1), np.percentile(arr, 0.75, axis=-1))
 
 
 def test_nanquantile_method():
@@ -966,6 +980,10 @@ def test_nanquantile_method():
         da.nanquantile(darr, 0.75, axis=-1, method="weibull"),
         np.nanquantile(arr, 0.75, axis=-1, method="weibull"),
     )
+    assert_eq(
+        da.nanpercentile(darr, 0.75, axis=-1, method="weibull"),
+        np.nanpercentile(arr, 0.75, axis=-1, method="weibull"),
+    )
 
 
 def test_nanquantile_one_dim():
@@ -978,3 +996,6 @@ def test_nanquantile_two_dims():
     arr = np.random.randn(10, 10)
     darr = da.from_array(arr, chunks=(2, -1))
     assert_eq(da.nanquantile(darr, 0.75, axis=-1), np.nanquantile(arr, 0.75, axis=-1))
+    assert_eq(
+        da.nanpercentile(darr, 0.75, axis=-1), np.nanpercentile(arr, 0.75, axis=-1)
+    )
diff --git a/dask/dataframe/tests/test_utils_dataframe.py b/dask/dataframe/tests/test_utils_dataframe.py
index 6b0af9935..cf31f6bf5 100644
--- a/dask/dataframe/tests/test_utils_dataframe.py
+++ b/dask/dataframe/tests/test_utils_dataframe.py
@@ -410,10 +410,11 @@ def test_check_meta():
     # Series metadata error
     with pytest.raises(ValueError) as err:
         check_meta(d, meta.d.astype("f8"), numeric_equal=False)
+    series = "pandas.core.series.Series" if not PANDAS_GE_300 else "pandas.Series"
     assert str(err.value) == (
         "Metadata mismatch found.\n"
         "\n"
-        "Partition type: `pandas.core.series.Series`\n"
+        f"Partition type: `{series}`\n"
         "+----------+---------+\n"
         "|          | dtype   |\n"
         "+----------+---------+\n"
@@ -449,7 +450,7 @@ def test_check_meta():
     assert str(err.value) == (
         "Metadata mismatch found.\n"
         "\n"
-        "Partition type: `pandas.core.series.Series`\n"
+        f"Partition type: `{series}`\n"
         "+----------+--------+\n"
         "|          | dtype  |\n"
         "+----------+--------+\n"
diff --git a/dask/tests/test_task_spec.py b/dask/tests/test_task_spec.py
index 27b31c967..e9b99cd6b 100644
--- a/dask/tests/test_task_spec.py
+++ b/dask/tests/test_task_spec.py
@@ -7,7 +7,6 @@ from collections import namedtuple
 
 import pytest
 
-import dask
 from dask._task_spec import (
     Alias,
     DataNode,
@@ -16,7 +15,6 @@ from dask._task_spec import (
     TaskRef,
     _get_dependencies,
     convert_legacy_graph,
-    convert_legacy_task,
     execute_graph,
     resolve_aliases,
 )
@@ -316,43 +314,6 @@ def test_avoid_cycles():
     assert not new_dsk
 
 
-def test_convert_tasks_only_ref_lowlevel():
-    t = convert_legacy_task(None, (func, "a", "b"), set(), only_refs=True)
-    assert t() == (func, "a", "b")
-
-    t = convert_legacy_task(None, (func, "a", "b"), {"a"}, only_refs=True)
-    assert t({"a": "c"}) == (func, "c", "b")
-
-
-def test_convert_task_only_ref():
-    # This is for support of legacy subgraphs
-    # We'll only want to insert pack/unpack dependencies but don't want to touch
-    # the task itself
-
-    def subgraph_func(dsk, key):
-        return dask.get(dsk, [key])[0]
-
-    dsk = {
-        "dep": "foo",
-        ("baz", 1): "baz",
-        "subgraph": (
-            subgraph_func,
-            {
-                "a": "dep",
-                "b": (func, "a", "bar"),
-                "c": (func, "a", ("baz", 1)),
-                "d": (func, "b", "c"),
-            },
-            "d",
-        ),
-    }
-    new_dsk = convert_legacy_graph(dsk)
-    _assert_dsk_conversion(new_dsk)
-    assert new_dsk["subgraph"].dependencies == {"dep", ("baz", 1)}
-
-    assert new_dsk["subgraph"]({"dep": "foo", ("baz", 1): "baz"}) == "foo-bar-foo-baz"
-
-
 def test_runnable_as_kwarg():
     def func_kwarg(a, b, c=""):
         return a + b + str(c)