Repository: dask/dask
Pull request #11094 (2024-05-03 18:11:03)

Clean up ``percentiles_summary`` logic
This is essentially a revision of https://github.com/dask/dask/pull/10551, which generalized the partition-wise `quantiles` logic used in `percentiles_summary` to work with both `cudf` and `pandas`-backed data. It turns out that those changes do not quite cover the case that a numerical column contains null values.

An additional problem with the current logic is that error handling for the `.values` operation is mixed with error handling for the `quantile` operation itself. This PR separates these steps a bit to make the logic less confusing.

Overview of the partition-wise `quantiles` logic after this PR:

1. We try to use the `Series.quantile` method (without calling `.values` yet)
2. If (1) fails, we use the more-robust `DataFrame.quantile(..., method="table")` method as if `pandas>=1.5`
3. If (1) fails and `pandas<1.5`, we fall back to `percentile` (this entire code path can be removed when pandas2+ is required)
4. After (1), (2) or (3) has succeeded, we try converting the result to an array with `.values`. Since `cudf->cupy` will fail to do this for string columns and/or null values, we simply `pass` if this step fails.
  - **NOTE**: It is not clear to me why we need to call  `.values` here at all. I believe NEP18 covers all necessary `Series` logic anyway. Perhaps we can remove this step?


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._

     15 files  ±0       15 suites  ±0   3h 24m 15s :stopwatch: +39s
 13 121 tests ±0   12 190 :white_check_mark: ±0     931 :zzz: ±0  0 :x: ±0 
162 468 runs  ±0  142 412 :white_check_mark: ±0  20 056 :zzz: ±0  0 :x: ±0 

Results for commit 4f297984. ± Comparison against base commit efb4a624.

[test-results]:data:application/gzip;base64,H4sIADgwNWYC/02Myw6DIBQFf8Ww7gIuF4T+TIMCCamPBmHV9N+LtIrLmXMyb+LD5DZy75i4dWTLITWyOZoU1mVnALGrsqa6cwbs4MeWx7GemKZNPsOrSM3bz5swFdU+LsY1/k3MSy1LQKn+fJYRkMEpf2WgVMjDXdKVr+VxneeQChD0oHutEKkw6J23XrqeD0iV4YN03KISDi2Qzxc4uImpGQEAAA==


Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 74c11f76d..f4420cfef 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -42,7 +42,7 @@ from dask.array.dispatch import (  # noqa: F401
     einsum_lookup,
     tensordot_lookup,
 )
-from dask.array.numpy_compat import _Recurser
+from dask.array.numpy_compat import NUMPY_GE_200, _Recurser
 from dask.array.slicing import replace_ellipsis, setitem_array, slice_array
 from dask.base import (
     DaskMethodsMixin,
@@ -3472,7 +3472,7 @@ def from_array(
     """
     if isinstance(x, Array):
         raise ValueError(
-            "Array is already a dask array. Use 'asarray' or " "'rechunk' instead."
+            "Array is already a dask array. Use 'asarray' or 'rechunk' instead."
         )
     elif is_dask_collection(x):
         warnings.warn(
@@ -5035,7 +5035,10 @@ def broadcast_arrays(*args, subok=False):
     shape = broadcast_shapes(*(e.shape for e in args))
     chunks = broadcast_chunks(*(e.chunks for e in args))
 
-    result = [broadcast_to(e, shape=shape, chunks=chunks) for e in args]
+    if NUMPY_GE_200:
+        result = tuple(broadcast_to(e, shape=shape, chunks=chunks) for e in args)
+    else:
+        result = [broadcast_to(e, shape=shape, chunks=chunks) for e in args]
 
     return result
 
diff --git a/dask/array/overlap.py b/dask/array/overlap.py
index 9a94065c3..76e8680a3 100644
--- a/dask/array/overlap.py
+++ b/dask/array/overlap.py
@@ -354,7 +354,7 @@ def ensure_minimum_chunksize(size, chunks):
         output[-1] += new
     else:
         raise ValueError(
-            f"The overlapping depth {size} is larger than your " f"array {sum(chunks)}."
+            f"The overlapping depth {size} is larger than your array {sum(chunks)}."
         )
 
     return tuple(output)
diff --git a/dask/array/slicing.py b/dask/array/slicing.py
index 9246c9cf1..29da273f9 100644
--- a/dask/array/slicing.py
+++ b/dask/array/slicing.py
@@ -2158,7 +2158,7 @@ def setitem(x, v, indices):
         x[tuple(indices)] = v
     except ValueError as e:
         raise ValueError(
-            "shape mismatch: value array could " "not be broadcast to indexing result"
+            "shape mismatch: value array could not be broadcast to indexing result"
         ) from e
 
     return x
diff --git a/dask/cache.py b/dask/cache.py
index b07adb886..ee2f06179 100644
--- a/dask/cache.py
+++ b/dask/cache.py
@@ -35,7 +35,7 @@ class Cache(Callback):
             import cachey
         except ImportError as ex:
             raise ImportError(
-                'Cache requires cachey, "{ex}" problem ' "importing".format(ex=str(ex))
+                f'Cache requires cachey, "{str(ex)}" problem importing'
             ) from ex
         self._nbytes = cachey.nbytes
         if isinstance(cache, Number):
diff --git a/dask/dataframe/io/json.py b/dask/dataframe/io/json.py
index 2f3e2790b..79df964c6 100644
--- a/dask/dataframe/io/json.py
+++ b/dask/dataframe/io/json.py
@@ -73,9 +73,7 @@ def to_json(
     if lines is None:
         lines = orient == "records"
     if orient != "records" and lines:
-        raise ValueError(
-            "Line-delimited JSON is only available with" 'orient="records".'
-        )
+        raise ValueError('Line-delimited JSON is only available with orient="records".')
     kwargs["orient"] = orient
     kwargs["lines"] = lines and orient == "records"
     outfiles = open_files(
@@ -200,9 +198,7 @@ def read_json(
     if lines is None:
         lines = orient == "records"
     if orient != "records" and lines:
-        raise ValueError(
-            "Line-delimited JSON is only available with" 'orient="records".'
-        )
+        raise ValueError('Line-delimited JSON is only available with orient="records".')
     if blocksize and (orient != "records" or not lines):
         raise ValueError(
             "JSON file chunking only allowed for JSON-lines"
diff --git a/dask/dataframe/partitionquantiles.py b/dask/dataframe/partitionquantiles.py
index 001c46637..77ccf58ba 100644
--- a/dask/dataframe/partitionquantiles.py
+++ b/dask/dataframe/partitionquantiles.py
@@ -78,8 +78,10 @@ from pandas.api.types import is_datetime64_dtype, is_integer_dtype
 from tlz import merge, merge_sorted, take
 
 from dask.base import tokenize
+from dask.dataframe._compat import PANDAS_GE_150
 from dask.dataframe.core import Series
 from dask.dataframe.dispatch import tolist_dispatch
+from dask.dataframe.utils import is_series_like
 from dask.utils import is_cupy_type, random_state_data
 
 
@@ -430,16 +432,13 @@ def percentiles_summary(df, num_old, num_new, upsample, state):
     elif is_datetime64_dtype(data.dtype) or is_integer_dtype(data.dtype):
         interpolation = "nearest"
 
-    # FIXME: pandas quantile doesn't work with some data types (e.g. strings).
-    # We fall back to an ndarray as a workaround.
     try:
-        vals = data.quantile(q=qs / 100, interpolation=interpolation).values
+        # Plan A: Use `Series.quantile`
+        vals = data.quantile(q=qs / 100, interpolation=interpolation)
     except (TypeError, NotImplementedError):
-        try:
-            vals, _ = _percentile(array_safe(data, like=data.values), qs, interpolation)
-        except (TypeError, NotImplementedError):
-            # `data.values` doesn't work for cudf, so we need to
-            # use `quantile(..., method="table")` as a fallback
+        # Series.quantile doesn't work with some data types (e.g. strings)
+        if PANDAS_GE_150:
+            # Fall back to DataFrame.quantile with "nearest" interpolation
             interpolation = "nearest"
             vals = (
                 data.to_frame()
@@ -451,6 +450,18 @@ def percentiles_summary(df, num_old, num_new, upsample, state):
                 )
                 .iloc[:, 0]
             )
+        else:
+            # Fall back to ndarray (Not supported for cuDF)
+            vals, _ = _percentile(array_safe(data, like=data.values), qs, interpolation)
+
+    # Convert to array if necessary (and possible)
+    if is_series_like(vals):
+        try:
+            vals = vals.values
+        except (ValueError, TypeError):
+            # cudf->cupy won't work if nulls are present,
+            # or if this is a string dtype
+            pass
 
     if (
         is_cupy_type(data)
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index cfa0d6701..24b89c97e 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -1,6 +1,38 @@
 Changelog
 =========
 
+.. _v2024.5.0:
+
+2024.5.0
+--------
+
+Highlights
+^^^^^^^^^^
+
+This release primarily contains minor bugfixes.
+
+.. dropdown:: Additional changes
+
+  - Don't link to ``click`` intersphinx dev version (:pr:`11091`) `M Bussonnier`_
+  - Fix API doc links for some ``dask-expr`` expressions (:pr:`11092`) `Patrick Hoefler`_
+  - Add ``dask-expr`` to upstream build (:pr:`11086`) `Patrick Hoefler`_
+  - Add ``melt`` support when ``query-planning`` is enabled (:pr:`11088`) `Richard (Rick) Zamora`_
+  - Skip dataframe/product when in ``numpy`` 2 envs (:pr:`11089`) `Benjamin Zaitlen`_
+  - Add plots to illustrate what the optimizer does (:pr:`11072`) `Patrick Hoefler`_
+  - Fixup ``pandas`` upstream tests (:pr:`11085`) `Patrick Hoefler`_
+  - Bump ``conda-incubator/setup-miniconda`` from 3.0.3 to 3.0.4 (:pr:`11084`)
+  - Bump ``actions/checkout`` from 4.1.3 to 4.1.4 (:pr:`11083`)
+  - Fix CI after ``pytest`` changes (:pr:`11082`) `Patrick Hoefler`_
+  - Fixup tests for more efficient ``dask-expr`` implementation (:pr:`11071`) `Patrick Hoefler`_
+  - Generalize ``clear_known_categories`` utility (:pr:`11059`) `Richard (Rick) Zamora`_
+  - Bump ``JamesIves/github-pages-deploy-action`` from 4.5.0 to 4.6.0 (:pr:`11062`)
+  - Bump ``release-drafter/release-drafter`` from 5 to 6 (:pr:`11063`)
+  - Bump ``actions/checkout`` from 4.1.2 to 4.1.3 (:pr:`11061`)
+  - Update GPU CI ``RAPIDS_VER`` to 24.06, disable query planning  (:pr:`11045`) `Charles Blackmon-Luca`_
+
+  - Move tests (:pr-distributed:`8631`) `Hendrik Makait`_
+  - Bump ``actions/checkout`` from 4.1.2 to 4.1.3 (:pr-distributed:`8628`)
+
 .. _v2024.4.2:
 
 2024.4.2
@@ -8119,3 +8151,4 @@ Other
 .. _`Jonas Lähnemann`: https://github.com/jlaehne
 .. _`Abel Aoun`: https://github.com/bzah
 .. _`Simon Høxbro Hansen`: https://github.com/Hoxbro
+.. _`M Bussonnier`: https://github.com/Carreau
diff --git a/pyproject.toml b/pyproject.toml
index eb43e1f1a..59381e957 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -55,9 +55,9 @@ dataframe = [
     # NOTE: dask-expr is pinning pandas>2. To use pandas<2, please install dask
     # without using the dataframe extra
     "pandas >= 1.3",
-    "dask-expr >= 1.0, <1.1",  # dask-expr pins the dask version
+    "dask-expr >= 1.1, <1.2",  # dask-expr pins the dask version
 ]
-distributed = ["distributed == 2024.4.2"]
+distributed = ["distributed == 2024.5.0"]
 diagnostics = [
     "bokeh >= 2.4.2",
     "jinja2 >= 2.10.3",


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index 68aa57410..842090546 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -1108,7 +1108,7 @@ def test_broadcast_arrays():
     a_r = np.broadcast_arrays(a_0, a_1)
     d_r = da.broadcast_arrays(d_a_0, d_a_1)
 
-    assert isinstance(d_r, list)
+    assert isinstance(d_r, (list, tuple))
     assert len(a_r) == len(d_r)
 
     for e_a_r, e_d_r in zip(a_r, d_r):
diff --git a/dask/bytes/tests/test_local.py b/dask/bytes/tests/test_local.py
index 85450f40c..009a5c084 100644
--- a/dask/bytes/tests/test_local.py
+++ b/dask/bytes/tests/test_local.py
@@ -38,9 +38,9 @@ files = {
 
 
 csv_files = {
-    ".test.fakedata.1.csv": (b"a,b\n" b"1,2\n"),
-    ".test.fakedata.2.csv": (b"a,b\n" b"3,4\n"),
-    "subdir/.test.fakedata.2.csv": (b"a,b\n" b"5,6\n"),
+    ".test.fakedata.1.csv": (b"a,b\n1,2\n"),
+    ".test.fakedata.2.csv": (b"a,b\n3,4\n"),
+    "subdir/.test.fakedata.2.csv": (b"a,b\n5,6\n"),
 }
 
 
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index d1e22427d..1cc22394b 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -102,13 +102,9 @@ Date,Open,High,Low,Close,Volume,Adj Close
 """.strip()
 
 csv_files = {
-    "2014-01-01.csv": (
-        b"name,amount,id\n" b"Alice,100,1\n" b"Bob,200,2\n" b"Charlie,300,3\n"
-    ),
+    "2014-01-01.csv": b"name,amount,id\nAlice,100,1\nBob,200,2\nCharlie,300,3\n",
     "2014-01-02.csv": b"name,amount,id\n",
-    "2014-01-03.csv": (
-        b"name,amount,id\n" b"Dennis,400,4\n" b"Edith,500,5\n" b"Frank,600,6\n"
-    ),
+    "2014-01-03.csv": b"name,amount,id\nDennis,400,4\nEdith,500,5\nFrank,600,6\n",
 }
 
 tsv_files = {k: v.replace(b",", b"\t") for (k, v) in csv_files.items()}
diff --git a/dask/dataframe/io/tests/test_sql.py b/dask/dataframe/io/tests/test_sql.py
index 9157372dc..705923870 100644
--- a/dask/dataframe/io/tests/test_sql.py
+++ b/dask/dataframe/io/tests/test_sql.py
@@ -64,7 +64,7 @@ def test_empty(db):
 
 
 @pytest.mark.filterwarnings(
-    "ignore:The default dtype for empty Series " "will be 'object' instead of 'float64'"
+    "ignore:The default dtype for empty Series will be 'object' instead of 'float64'"
 )
 @pytest.mark.parametrize("use_head", [True, False])
 def test_single_column(db, use_head):