Repository: dask/dask

Issue:
reduction "dtype" is required but docstring says optional
**What happened**:

I'm playing around with the dask.array.reduction function while working on a custom algorithm and was confused when I got a ValueError for not specifying the `dtype` when the docstring says it is optional and will default to `x.dtype` if not provided. See:

https://github.com/dask/dask/blob/1d8e1e920b5139bd434749f5877266293f6432a2/dask/array/reductions.py#L145-L146

**What you expected to happen**:

Either do the default that the docstring says or make dtype a required argument.

**Minimal Complete Verifiable Example**:

I can make one if you really need it, but this is more of a behavioral question about what is actually desired. Is defaulting to the input array's dtype going to cause issues in the future (has it happened in other dask functions)? If so, I'll update the docstring and make a PR. If not, I'll update the code to default to `x.dtype`.

Discussion:

Patch:
diff --git a/dask/array/percentile.py b/dask/array/percentile.py
index ef4e9c09f..e35c6f7a0 100644
--- a/dask/array/percentile.py
+++ b/dask/array/percentile.py
@@ -126,10 +126,12 @@ def percentile(a, q, interpolation="linear", method="default"):
 
     # Otherwise use the custom percentile algorithm
     else:
-
+        # Add 0 and 100 during calculation for more robust behavior (hopefully)
+        calc_q = np.pad(q, 1, mode="constant")
+        calc_q[-1] = 100
         name = "percentile_chunk-" + token
         dsk = dict(
-            ((name, i), (_percentile, key, q, interpolation))
+            ((name, i), (_percentile, key, calc_q, interpolation))
             for i, key in enumerate(a.__dask_keys__())
         )
 
@@ -138,7 +140,7 @@ def percentile(a, q, interpolation="linear", method="default"):
             (name2, 0): (
                 merge_percentiles,
                 q,
-                [q] * len(a.chunks[0]),
+                [calc_q] * len(a.chunks[0]),
                 sorted(dsk),
                 interpolation,
             )
diff --git a/dask/array/reductions.py b/dask/array/reductions.py
index 5138702f4..b307fee69 100644
--- a/dask/array/reductions.py
+++ b/dask/array/reductions.py
@@ -77,8 +77,9 @@ def reduction(
     keepdims: boolean, optional
         Whether the reduction function should preserve the reduced axes,
         leaving them at size ``output_size``, or remove them.
-    dtype: np.dtype, optional
-        Force output dtype. Defaults to x.dtype if omitted.
+    dtype: np.dtype
+        data type of output. This argument was previously optional, but
+        leaving as ``None`` will now raise an exception.
     split_every: int >= 2 or dict(axis: int), optional
         Determines the depth of the recursive aggregation. If set to or more
         than the number of input chunks, the aggregation will be performed in
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index b8a134c5f..be1051092 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -3667,17 +3667,17 @@ class DataFrame(_Frame):
     ):
         """Set the DataFrame index (row labels) using an existing column.
 
-        This realigns the dataset to be sorted by a new column.  This can have a
+        This realigns the dataset to be sorted by a new column. This can have a
         significant impact on performance, because joins, groupbys, lookups, etc.
-        are all much faster on that column.  However, this performance increase
+        are all much faster on that column. However, this performance increase
         comes with a cost, sorting a parallel dataset requires expensive shuffles.
         Often we ``set_index`` once directly after data ingest and filtering and
         then perform many cheap computations off of the sorted dataset.
 
         This function operates exactly like ``pandas.set_index`` except with
-        different performance costs (dask dataframe ``set_index`` is much more expensive).  Under normal
-        operation this function does an initial pass over the index column to
-        compute approximate qunatiles to serve as future divisions.  It then passes
+        different performance costs (dask dataframe ``set_index`` is much more expensive).
+        Under normal operation this function does an initial pass over the index column
+        to compute approximate qunatiles to serve as future divisions. It then passes
         over the data a second time, splitting up each input partition into several
         pieces and sharing those pieces to all of the output partitions now in
         sorted order.
@@ -3685,23 +3685,21 @@ class DataFrame(_Frame):
         In some cases we can alleviate those costs, for example if your dataset is
         sorted already then we can avoid making many small pieces or if you know
         good values to split the new index column then we can avoid the initial
-        pass over the data.  For example if your new index is a datetime index and
+        pass over the data. For example if your new index is a datetime index and
         your data is already sorted by day then this entire operation can be done
-        for free.  You can control these options with the following parameters.
+        for free. You can control these options with the following parameters.
 
         Parameters
         ----------
-        df: Dask DataFrame
-        index: string or Dask Series
-        npartitions: int, None, or 'auto'
-            The ideal number of output partitions.   If None use the same as
-            the input.  If 'auto' then decide by memory use.
-        shuffle: string, optional
-            Either ``'disk'`` for single-node operation or ``'tasks'`` for
-            distributed operation.  Will be inferred by your current scheduler.
+        other: string or Dask Series
+        drop: boolean, default True
+            Delete column to be used as the new index.
         sorted: bool, optional
             If the index column is already sorted in increasing order.
             Defaults to False
+        npartitions: int, None, or 'auto'
+            The ideal number of output partitions. If None, use the same as
+            the input. If 'auto' then decide by memory use.
         divisions: list, optional
             Known values on which to separate index values of the partitions.
             See https://docs.dask.org/en/latest/dataframe-design.html#partitions
@@ -3709,14 +3707,19 @@ class DataFrame(_Frame):
             that if ``sorted=True``, specified divisions are assumed to match
             the existing partitions in the data. If ``sorted=False``, you should
             leave divisions empty and call ``repartition`` after ``set_index``.
-        inplace : bool, optional
+        inplace: bool, optional
             Modifying the DataFrame in place is not supported by Dask.
             Defaults to False.
-        compute: bool
+        shuffle: string, 'disk' or 'tasks', optional
+            Either ``'disk'`` for single-node operation or ``'tasks'`` for
+            distributed operation.  Will be inferred by your current scheduler.
+        compute: bool, default False
             Whether or not to trigger an immediate computation. Defaults to False.
             Note, that even if you set ``compute=False``, an immediate computation
             will still be triggered if ``divisions`` is ``None``.
-
+        partition_size: int, optional
+            Desired size of each partitions in bytes.
+            Only used when ``npartition='auto'``
         Examples
         --------
         >>> df2 = df.set_index('x')  # doctest: +SKIP
@@ -5457,16 +5460,19 @@ def quantile(df, q, method="default"):
 
         from dask.array.percentile import _percentile, merge_percentiles
 
+        # Add 0 and 100 during calculation for more robust behavior (hopefully)
+        calc_qs = np.pad(qs, 1, mode="constant")
+        calc_qs[-1] = 100
         name = "quantiles-1-" + token
         val_dsk = {
-            (name, i): (_percentile, (getattr, key, "values"), qs)
+            (name, i): (_percentile, (getattr, key, "values"), calc_qs)
             for i, key in enumerate(df.__dask_keys__())
         }
 
         name2 = "quantiles-2-" + token
         merge_dsk = {
             (name2, 0): finalize_tsk(
-                (merge_percentiles, qs, [qs] * df.npartitions, sorted(val_dsk))
+                (merge_percentiles, qs, [calc_qs] * df.npartitions, sorted(val_dsk))
             )
         }
     dsk = merge(val_dsk, merge_dsk)


Test patch:
diff --git a/dask/array/tests/test_array_function.py b/dask/array/tests/test_array_function.py
index a76530bce..f016a6a53 100644
--- a/dask/array/tests/test_array_function.py
+++ b/dask/array/tests/test_array_function.py
@@ -179,7 +179,13 @@ def test_non_existent_func():
 
 @pytest.mark.skipif(missing_arrfunc_cond, reason=missing_arrfunc_reason)
 @pytest.mark.parametrize(
-    "func", [np.equal, np.matmul, np.dot, lambda x, y: np.stack([x, y])]
+    "func",
+    [
+        np.equal,
+        np.matmul,
+        np.dot,
+        lambda x, y: np.stack([x, y]),
+    ],
 )
 @pytest.mark.parametrize(
     "arr_upcast, arr_downcast",
diff --git a/dask/array/tests/test_percentiles.py b/dask/array/tests/test_percentiles.py
index c8669d6c2..fc118b426 100644
--- a/dask/array/tests/test_percentiles.py
+++ b/dask/array/tests/test_percentiles.py
@@ -80,6 +80,15 @@ def test_percentiles_with_empty_arrays(method):
     )
 
 
+@percentile_methods
+def test_percentiles_with_empty_q(method):
+    x = da.ones(10, chunks=((5, 0, 5),))
+    assert_eq(
+        da.percentile(x, [], method=method),
+        np.array([], dtype=x.dtype),
+    )
+
+
 @percentile_methods
 @pytest.mark.parametrize("q", [5, 5.0, np.int64(5), np.float64(5)])
 def test_percentiles_with_scaler_percentile(method, q):
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 8b2aca252..9287c4455 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -1190,7 +1190,7 @@ def test_nbytes():
 
 @pytest.mark.parametrize(
     "method,expected",
-    [("tdigest", (0.35, 3.80, 2.5, 6.5, 2.0)), ("dask", (0.0, 5.4, 1.2, 7.8, 5.0))],
+    [("tdigest", (0.35, 3.80, 2.5, 6.5, 2.0)), ("dask", (0.0, 4.0, 1.2, 6.2, 2.0))],
 )
 def test_quantile(method, expected):
     if method == "tdigest":
@@ -1278,9 +1278,9 @@ def test_empty_quantile(method):
         (
             "dask",
             (
-                pd.Series([16.5, 36.5, 26.5], index=["A", "X", "B"]),
+                pd.Series([7.0, 27.0, 17.0], index=["A", "X", "B"]),
                 pd.DataFrame(
-                    [[1.50, 21.50, 11.50], [17.75, 37.75, 27.75]],
+                    [[1.50, 21.50, 11.50], [14.0, 34.0, 24.0]],
                     index=[0.25, 0.75],
                     columns=["A", "X", "B"],
                 ),
@@ -1355,6 +1355,14 @@ def test_quantile_for_possibly_unsorted_q():
     assert_eq(r, 25.0)
 
 
+def test_quantile_tiny_partitions():
+    """ See https://github.com/dask/dask/issues/6551 """
+    df = pd.DataFrame({"a": [1, 2, 3]})
+    ddf = dd.from_pandas(df, npartitions=3)
+    r = ddf["a"].quantile(0.5).compute()
+    assert r == 2
+
+
 def test_index():
     assert_eq(d.index, full.index)