[REVIEW] Add support for percentile dispatch in dask_cudf

python/cudf/cudf/core/column/datetime.py

@@ -158,6 +158,15 @@ def to_pandas(
             index=index,
         )
 
+    @property
+    def values(self):
+        """
+        Return a CuPy representation of the DateTimeColumn.
+        """
+        raise NotImplementedError(
+            "DateTime Arrays is not yet implemented in cudf"
+        )
+
     def get_dt_field(self, field: str) -> ColumnBase:
         return libcudf.datetime.extract_datetime_component(self, field)
 

python/cudf/cudf/core/column/timedelta.py

@@ -92,6 +92,15 @@ def __contains__(self, item: DatetimeLikeScalar) -> bool:
             return False
         return item.view("int64") in self.as_numerical
 
+    @property
+    def values(self):
+        """
+        Return a CuPy representation of the TimeDeltaColumn.
+        """
+        raise NotImplementedError(
+            "TimeDelta Arrays is not yet implemented in cudf"
+        )
+
     def to_arrow(self) -> pa.Array:
         mask = None
         if self.nullable:

python/cudf/cudf/tests/test_datetime.py

@@ -1541,3 +1541,12 @@ def test_is_quarter_end(data, dtype):
     got = gs.dt.is_quarter_end
 
     assert_eq(expect, got)
+
+
+def test_error_values():
+    s = cudf.Series([1, 2, 3], dtype="datetime64[ns]")
+    with pytest.raises(
+        NotImplementedError,
+        match="DateTime Arrays is not yet implemented in cudf",
+    ):
+        s.values

python/cudf/cudf/tests/test_timedelta.py

@@ -1386,3 +1386,12 @@ def test_timedelta_reductions(data, op, dtype):
         assert True
     else:
         assert_eq(expected.to_numpy(), actual)
+
+
+def test_error_values():
+    s = cudf.Series([1, 2, 3], dtype="timedelta64[ns]")
+    with pytest.raises(
+        NotImplementedError,
+        match="TimeDelta Arrays is not yet implemented in cudf",
+    ):
+        s.values

python/dask_cudf/dask_cudf/backends.py

@@ -1,5 +1,7 @@
 # Copyright (c) 2020-2021, NVIDIA CORPORATION.
 
+from collections.abc import Iterator
+
 import cupy as cp
 import numpy as np
 import pandas as pd
@@ -256,6 +258,52 @@ def is_categorical_dtype_cudf(obj):
     return cudf.utils.dtypes.is_categorical_dtype(obj)
 
 
+try:
+    from dask.dataframe.dispatch import percentile_dispatch
+
+    @percentile_dispatch.register((cudf.Series, cp.ndarray, cudf.Index))
+    def percentile_cudf(a, q, interpolation="linear"):
+        # Cudf dispatch to the equivalent of `np.percentile`:
+        # https://numpy.org/doc/stable/reference/generated/numpy.percentile.html
+        a = cudf.Series(a)
+        # a is series.
+        n = len(a)
+        if not len(a):
+            return None, n
+        if isinstance(q, Iterator):
+            q = list(q)
+
+        if cudf.utils.dtypes.is_categorical_dtype(a.dtype):
+            result = cp.percentile(a.cat.codes, q, interpolation=interpolation)
+
+            return (
+                pd.Categorical.from_codes(
+                    result, a.dtype.categories, a.dtype.ordered
+                ),
+                n,
+            )
+        if np.issubdtype(a.dtype, np.datetime64):
+            result = a.quantile(
+                [i / 100.0 for i in q], interpolation=interpolation
+            )
+
+            if q[0] == 0:
+                # https://github.com/dask/dask/issues/6864
+                result[0] = min(result[0], a.min())
+            return result.to_pandas(), n
+        if not np.issubdtype(a.dtype, np.number):
+            interpolation = "nearest"
+        return (
+            a.quantile(
+                [i / 100.0 for i in q], interpolation=interpolation
+            ).to_pandas(),
+            n,
+        )
+
+
+except ImportError:
+    pass
+
 try:
     from dask.dataframe.dispatch import union_categoricals_dispatch