Repository: dask/dask
Pull request: Add series is_monotonic*
- [x] Closes #8297
- [x] Tests added / passed
- [x] Passes `pre-commit run --all-files`


Problem statement:
Series is_monotonic, is_monotonic_decreasing, is_monotonic_increasing
I'd like to be able to use is_monotonic, is_monotonic_decreasing, is_monotonic_increasing. As of today, I can do this in pandas but not dask

```python

import pandas as pd
import dask.dataframe as dd

pdf = pd.DataFrame({
    "a": [0,1,0,1,1,0],
    "b": range(6),
    "c": ["a","b","c","d","e","f"]
})
ddf = dd.from_pandas(pdf, 2)

print(pdf["b"].is_monotonic)
print(ddf["b"].is_monotonic)
```
**Output**
```
True
Traceback (most recent call last):
  File "path/to/file", line 12, in <module>
    print(ddf["b"].is_monotonic)
AttributeError: 'Series' object has no attribute 'is_monotonic'
```

The same applies to `is_monotonic_decreasing`, `is_monotonic_increasing`. The dask version is:
```
dask.__version__
Out[3]: '2021.10.0'
```

Hints from issue discussion:
Thanks for the feature request @mesejo. Adding these `is_monotonic*` methods is certainly in scope for Dask DataFrame. Is this something you're interested in working on? (no obligation though)
@jrbourbeau  You are welcome! Yes, I'm interested in working on this. 

I believe this should not hard to implement, my idea was basically to apply a reduction to each overlapped block and then take the logical and of all these reductions. For example assume we are computing `is_monotonic_increasing`  for a Series containing `[0, 1, 2, 3, 4, 5, 6, 7, 8]`, the idea is as follows:

```
0 1 2 3 4 5 6 7 8 

0 1 2 3  --> is_monotonic_increasing -> True
3 4 5 6  --> is_monotonic_increasing -> True 	
6 7 8    --> is_monotonic_increasing -> True		

logical_and([True, True, True]) -> True
```

I was planning on using [`Series.map_overlap`](https://docs.dask.org/en/latest/generated/dask.dataframe.Series.map_overlap.html#dask.dataframe.Series.map_overlap) but I think is not suited for computing this reduction-like operations (I was coding the solution now I was not able to get it to work). Do you know what I could use instead? Also if you have any other ideas please let me know. 



Applying `is_monotonic_increasing` to each partition (with some overlap) and then using `all` to reduce the result seems like a sensible approach. Would you mind pushing up your existing branch as a WIP PR? It will probably be easier to discuss over concrete code changes

Commit message: Add is_monotonic* properties to Index

Patch to review:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index a76cba3c7f7..2a0661c64ab 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -3766,9 +3766,35 @@ def __rdivmod__(self, other):
         res2 = other % self
         return res1, res2
 
+    @property
+    @derived_from(pd.Series)
+    def is_monotonic(self):
+        return self.is_monotonic_increasing
 
-class Index(Series):
+    @property
+    @derived_from(pd.Series)
+    def is_monotonic_increasing(self):
+        return aca(
+            self,
+            chunk=_monotonic_increasing_chunk,
+            aggregate=_monotonic_increasing_aggregate,
+            meta=bool,
+            token="monotonic_increasing",
+        )
 
+    @property
+    @derived_from(pd.Series)
+    def is_monotonic_decreasing(self):
+        return aca(
+            self,
+            chunk=_monotonic_decreasing_chunk,
+            aggregate=_monotonic_decreasing_aggregate,
+            meta=bool,
+            token="monotonic_increasing",
+        )
+
+
+class Index(Series):
     _partition_type = pd.Index
     _is_partition_type = staticmethod(is_index_like)
     _token_prefix = "index-"
@@ -3809,11 +3835,19 @@ class Index(Series):
         "rename_categories",
     }
 
+    _monotonic_attributes = {
+        "is_monotonic",
+        "is_monotonic_increasing",
+        "is_monotonic_decreasing",
+    }
+
     def __getattr__(self, key):
         if is_categorical_dtype(self.dtype) and key in self._cat_attributes:
             return getattr(self.cat, key)
         elif key in self._dt_attributes:
             return getattr(self.dt, key)
+        elif key in self._monotonic_attributes:
+            return getattr(self, key)
         raise AttributeError("'Index' object has no attribute %r" % key)
 
     def __dir__(self):
@@ -3939,6 +3973,33 @@ def map(self, arg, na_action=None, meta=no_default, is_monotonic=False):
             applied = applied.clear_divisions()
         return applied
 
+    @property
+    @derived_from(pd.Index)
+    def is_monotonic(self):
+        return self.is_monotonic_increasing
+
+    @property
+    @derived_from(pd.Index)
+    def is_monotonic_increasing(self):
+        return aca(
+            self,
+            chunk=_index_monotonic_increasing_chunk,
+            aggregate=_monotonic_increasing_aggregate,
+            meta=bool,
+            token="monotonic_increasing",
+        )
+
+    @property
+    @derived_from(pd.Index)
+    def is_monotonic_decreasing(self):
+        return aca(
+            self,
+            chunk=_index_monotonic_decreasing_chunk,
+            aggregate=_monotonic_decreasing_aggregate,
+            meta=bool,
+            token="monotonic_increasing",
+        )
+
 
 class DataFrame(_Frame):
     """
@@ -6332,7 +6393,6 @@ def cov_corr_chunk(df, corr=False):
 
 
 def cov_corr_combine(data_in, corr=False):
-
     data = {"sum": None, "count": None, "cov": None}
     if corr:
         data["m"] = None
@@ -7349,3 +7409,45 @@ def series_map(base_series, map_series):
     divisions = list(base_series.divisions)
 
     return new_dd_object(graph, final_prefix, meta, divisions)
+
+
+def _monotonic_increasing_chunk(x):
+    return pd.DataFrame(
+        data=[[x.is_monotonic_increasing, x.iloc[0], x.iloc[-1]]],
+        columns=["monotonic", "first", "last"],
+    )
+
+
+def _index_monotonic_increasing_chunk(x):
+    return pd.DataFrame(
+        data=[[x.is_monotonic_increasing, x[0], x[-1]]],
+        columns=["monotonic", "first", "last"],
+    )
+
+
+def _monotonic_increasing_aggregate(concatenated):
+    bounds_are_monotonic = pd.Series(
+        concatenated[["first", "last"]].to_numpy().ravel()
+    ).is_monotonic_increasing
+    return concatenated["monotonic"].all() and bounds_are_monotonic
+
+
+def _monotonic_decreasing_chunk(x):
+    return pd.DataFrame(
+        data=[[x.is_monotonic_decreasing, x.iloc[0], x.iloc[-1]]],
+        columns=["monotonic", "first", "last"],
+    )
+
+
+def _index_monotonic_decreasing_chunk(x):
+    return pd.DataFrame(
+        data=[[x.is_monotonic_decreasing, x[0], x[-1]]],
+        columns=["monotonic", "first", "last"],
+    )
+
+
+def _monotonic_decreasing_aggregate(concatenated):
+    bounds_are_monotonic = pd.Series(
+        concatenated[["first", "last"]].to_numpy().ravel()
+    ).is_monotonic_decreasing
+    return concatenated["monotonic"].all() and bounds_are_monotonic
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 8261fea220f..31a8a378d62 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -4919,3 +4919,55 @@ def f(x):
     pxy = weakref.proxy(a)
     res = pxy["b"].groupby(pxy["g"]).sum()
     isinstance(res.compute(), pd.Series)
+
+
+def test_is_monotonic_numeric():
+    s = pd.Series(range(20))
+    ds = dd.from_pandas(s, npartitions=5)
+    assert_eq(s.is_monotonic_increasing, ds.is_monotonic_increasing)
+    assert_eq(s.is_monotonic, ds.is_monotonic)
+
+    s_2 = pd.Series(range(20, 0, -1))
+    ds_2 = dd.from_pandas(s_2, npartitions=5)
+    assert_eq(s_2.is_monotonic_decreasing, ds_2.is_monotonic_decreasing)
+
+    s_3 = pd.Series(list(range(0, 5)) + list(range(0, 20)))
+    ds_3 = dd.from_pandas(s_3, npartitions=5)
+    assert_eq(s_3.is_monotonic_increasing, ds_3.is_monotonic_increasing)
+    assert_eq(s_3.is_monotonic_decreasing, ds_3.is_monotonic_decreasing)
+
+
+def test_is_monotonic_dt64():
+    s = pd.Series(pd.date_range("20130101", periods=10))
+    ds = dd.from_pandas(s, npartitions=5)
+    assert_eq(s.is_monotonic_increasing, ds.is_monotonic_increasing)
+
+    s_2 = pd.Series(list(reversed(s)))
+    ds_2 = dd.from_pandas(s_2, npartitions=5)
+    assert_eq(s_2.is_monotonic_decreasing, ds_2.is_monotonic_decreasing)
+
+
+def test_index_is_monotonic_numeric():
+    s = pd.Series(1, index=range(20))
+    ds = dd.from_pandas(s, npartitions=5, sort=False)
+    assert_eq(s.index.is_monotonic_increasing, ds.index.is_monotonic_increasing)
+    assert_eq(s.index.is_monotonic, ds.index.is_monotonic)
+
+    s_2 = pd.Series(1, index=range(20, 0, -1))
+    ds_2 = dd.from_pandas(s_2, npartitions=5, sort=False)
+    assert_eq(s_2.index.is_monotonic_decreasing, ds_2.index.is_monotonic_decreasing)
+
+    s_3 = pd.Series(1, index=list(range(0, 5)) + list(range(0, 20)))
+    ds_3 = dd.from_pandas(s_3, npartitions=5, sort=False)
+    assert_eq(s_3.index.is_monotonic_increasing, ds_3.index.is_monotonic_increasing)
+    assert_eq(s_3.index.is_monotonic_decreasing, ds_3.index.is_monotonic_decreasing)
+
+
+def test_index_is_monotonic_dt64():
+    s = pd.Series(1, index=pd.date_range("20130101", periods=10))
+    ds = dd.from_pandas(s, npartitions=5, sort=False)
+    assert_eq(s.index.is_monotonic_increasing, ds.index.is_monotonic_increasing)
+
+    s_2 = pd.Series(1, index=list(reversed(s)))
+    ds_2 = dd.from_pandas(s_2, npartitions=5, sort=False)
+    assert_eq(s_2.index.is_monotonic_decreasing, ds_2.index.is_monotonic_decreasing)


Code review comments:
diff_hunk: @@ -7349,3 +7409,45 @@ def series_map(base_series, map_series):
     divisions = list(base_series.divisions)
 
     return new_dd_object(graph, final_prefix, meta, divisions)
+
+
+def _monotonic_increasing_chunk(x):
+    return pd.DataFrame(
+        data=[[x.is_monotonic_increasing, x.iloc[0], x.iloc[-1]]],
+        columns=["monotonic", "first", "last"],
+    )
+
+
+def _index_monotonic_increasing_chunk(x):
+    return pd.DataFrame(
path: dask/dataframe/core.py
text: @user1:
It might be simpler to move this logic into `_monotonic_increasing_chunk` and just include a type check in that function then the `Index` class can just inherit the property from the `Series` class. You can see examples of that pattern in functions that use `is_index_like` in https://github.com/dask/dask/blob/main/dask/dataframe/methods.py