Repository: dask/dask
Pull request: Make ``is_monotonic`` work when some partitions are empty
- [x] Closes #8880
- [x] Tests added / passed
- [x] Passes `pre-commit run --all-files`


Problem statement:
is_monotonic fails with empty partitions
**What happened**: if a dataframe has empty partitions the method `is_monotonic` crashes, both for the index and for columns.

**What you expected to happen**: empty partitions should be ignored while checking for sortedness and monotonicity. In the particular case of a 0-rows dataframe, Pandas equivalent behaves as expected (i.e., yielding `True`).

**Minimal Complete Verifiable Example**:

```python
>>> import dask.dataframe as dd
>>> import pandas as pd

# Simple 2-partitioned frame
>>> ddf = dd.from_pandas(pd.DataFrame({'a': [1, 2, 3, 4], 'b': [4, 3, 2, 1]}), npartitions=2)
>>> ddf.npartitions
2
>>> ddf.divisions
(0, 2, 3)
>>> for p in ddf.partitions:
...     print(p.compute())
   a  b
0  1  4
1  2  3
   a  b
2  3  2
3  4  1

# Non-empty partitions work fine, for index/columns and for whole frame or individual partitions
>>> ddf.partitions[0]['a'].is_monotonic.compute()
True
>>> ddf.partitions[1].index.is_monotonic_increasing.compute()           # is_monotonic_increasing is alias for is_monotonic
True
>>> ddf['b'].is_monotonic_decreasing.compute()           # decreasing for checking descending order instead of ascending
True
>>> ddf.index.is_monotonic_decreasing.compute()           # it is increasing, not decreasing
False

# In Pandas everything is fine
>>> ddf[ddf['a'] >= 3].index.compute().is_monotonic_increasing       # 'a' >= 3 to get rid of 1st partition
True
>>> ddf[ddf['a'] >= 3]['a'].compute().is_monotonic
True
>>> ddf[ddf['a'] >= 3]['b'].compute().is_monotonic_decreasing
True
>>> ddf[ddf['a'] >= 5].index.compute().is_monotonic                # 'a' >= 5 to yield an empty frame
True
>>> ddf[ddf['a'] >= 5]['b'].compute().is_monotonic_decreasing
True

# In Dask empty partitions fail
>>> ddf[ddf['a'] >= 3]['a'].is_monotonic.compute()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/base.py", line 292, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/base.py", line 575, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/threaded.py", line 81, in get
    results = get_async(
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 508, in get_async
    raise_exception(exc, tb)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 316, in reraise
    raise exc
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 221, in execute_task
    result = _execute_task(task, data)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/optimization.py", line 990, in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 149, in get
    result = _execute_task(task, cache)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/dataframe/methods.py", line 448, in monotonic_increasing_chunk
    data=[[x.is_monotonic_increasing, data[0], data[-1]]],
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/pandas/core/indexing.py", line 967, in __getitem__
    return self._getitem_axis(maybe_callable, axis=axis)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/pandas/core/indexing.py", line 1520, in _getitem_axis
    self._validate_integer(key, axis)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/pandas/core/indexing.py", line 1452, in _validate_integer
    raise IndexError("single positional indexer is out-of-bounds")
    
# With index it fails with a different message, but the underlying problem is the same
>>> ddf[ddf['a'] >= 3].index.is_monotonic.compute()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/base.py", line 292, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/base.py", line 575, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/threaded.py", line 81, in get
    results = get_async(
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 508, in get_async
    raise_exception(exc, tb)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 316, in reraise
    raise exc
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 221, in execute_task
    result = _execute_task(task, data)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/optimization.py", line 990, in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 149, in get
    result = _execute_task(task, cache)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/dataframe/methods.py", line 448, in monotonic_increasing_chunk
    data=[[x.is_monotonic_increasing, data[0], data[-1]]],
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/pandas/core/indexes/base.py", line 5039, in __getitem__
    return getitem(key)
IndexError: index 0 is out of bounds for axis 0 with size 0

# Same happens with a totally empty frame
>>> ddf[ddf['a'] >= 5]['b'].is_monotonic_decreasing.compute()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/base.py", line 292, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/base.py", line 575, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/threaded.py", line 81, in get
    results = get_async(
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 508, in get_async
    raise_exception(exc, tb)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 316, in reraise
    raise exc
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/local.py", line 221, in execute_task
    result = _execute_task(task, data)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/optimization.py", line 990, in __call__
    return core.get(self.dsk, self.outkey, dict(zip(self.inkeys, args)))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 149, in get
    result = _execute_task(task, cache)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/core.py", line 119, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/dask/dataframe/methods.py", line 463, in monotonic_decreasing_chunk
    data=[[x.is_monotonic_decreasing, data[0], data[-1]]],
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/pandas/core/indexing.py", line 967, in __getitem__
    return self._getitem_axis(maybe_callable, axis=axis)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/pandas/core/indexing.py", line 1520, in _getitem_axis
    self._validate_integer(key, axis)
  File "/home/jorge.lopez/anaconda3/envs/AML/lib/python3.9/site-packages/pandas/core/indexing.py", line 1452, in _validate_integer
    raise IndexError("single positional indexer is out-of-bounds")
IndexError: single positional indexer is out-of-bounds
```

**Anything else we need to know?**:

The error seems clear. The conflicting lines are `data=[[x.is_monotonic_increasing, data[0], data[-1]]]` and `data=[[x.is_monotonic_decreasing, data[0], data[-1]]]`, which crash for any empty partition, as there is no `data[0]` nor `data[-1]`. 

For those particular partitions, both methods should give `True` without accessing data, as Pandas does.

**Environment**:

- Dask version: 2022.04.0
- Python version: 3.9.12
- Operating System: Ubuntu
- Install method (conda, pip, source): Pip

Hints from issue discussion:
Thanks for opening this one as well! I am just writing up a little test, then I'll open a PR.

Commit message: Make ``is_monotonic`` work when some partitions are empty

Patch to review:
diff --git a/dask/dataframe/methods.py b/dask/dataframe/methods.py
index 3cf026481fe..ee982342c4b 100644
--- a/dask/dataframe/methods.py
+++ b/dask/dataframe/methods.py
@@ -443,11 +443,13 @@ def assign_index(df, ind):
 
 
 def monotonic_increasing_chunk(x):
-    data = x if is_index_like(x) else x.iloc
-    return pd.DataFrame(
-        data=[[x.is_monotonic_increasing, data[0], data[-1]]],
-        columns=["monotonic", "first", "last"],
-    )
+    if len(x) == 0:
+        # if input is empty, return empty df for chunk
+        data = None
+    else:
+        data = x if is_index_like(x) else x.iloc
+        data = [[x.is_monotonic_increasing, data[0], data[-1]]]
+    return pd.DataFrame(data=data, columns=["monotonic", "first", "last"])
 
 
 def monotonic_increasing_aggregate(concatenated):
@@ -458,11 +460,13 @@ def monotonic_increasing_aggregate(concatenated):
 
 
 def monotonic_decreasing_chunk(x):
-    data = x if is_index_like(x) else x.iloc
-    return pd.DataFrame(
-        data=[[x.is_monotonic_decreasing, data[0], data[-1]]],
-        columns=["monotonic", "first", "last"],
-    )
+    if len(x) == 0:
+        # if input is empty, return empty df for chunk
+        data = None
+    else:
+        data = x if is_index_like(x) else x.iloc
+        data = [[x.is_monotonic_decreasing, data[0], data[-1]]]
+    return pd.DataFrame(data=data, columns=["monotonic", "first", "last"])
 
 
 def monotonic_decreasing_aggregate(concatenated):
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 0cce11323b1..0611d623690 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -5108,6 +5108,15 @@ def test_index_is_monotonic_dt64():
     assert_eq(s_2.index.is_monotonic_decreasing, ds_2.index.is_monotonic_decreasing)
 
 
+def test_is_monotonic_empty_partitions():
+    df = pd.DataFrame({"a": [1, 2, 3, 4], "b": [4, 3, 2, 1]})
+    ddf = dd.from_pandas(df, npartitions=2)
+    assert ddf[ddf["a"] >= 3]["a"].is_monotonic.compute()
+    assert ddf[ddf["a"] >= 3].index.is_monotonic.compute()
+    assert not ddf[ddf["a"] >= 3]["a"].is_monotonic_decreasing.compute()
+    assert not ddf[ddf["a"] >= 3].index.is_monotonic_decreasing.compute()
+
+
 def test_custom_map_reduce():
     # Make sure custom map-reduce workflows can use
     # the universal ACA code path with metadata


Code review comments:
diff_hunk: @@ -458,11 +460,13 @@ def monotonic_increasing_aggregate(concatenated):
 
 
 def monotonic_decreasing_chunk(x):
-    data = x if is_index_like(x) else x.iloc
-    return pd.DataFrame(
-        data=[[x.is_monotonic_decreasing, data[0], data[-1]]],
-        columns=["monotonic", "first", "last"],
-    )
+    if len(x) == 0:
path: dask/dataframe/methods.py
text: @user1:
Similar comment here

```suggestion
    if x.empty:
```