Repository: dask/dask
Pull request: Add support for reductions on empty dataframes
Fixes #5802 
- modifications to handle calculation for empty dataframes

- adds test to check result is equivalent to pandas

- [ ] Tests added / passed
- [ ] Passes `black dask` / `flake8 dask`


Problem statement:
Dataframe operations not working on empty dataframes
```
>>> import dask.dataframe as dd
>>> import pandas as pd
>>> pdf = pd.DataFrame()
>>> ddf = dd.from_pandas(pdf, npartitions=1)
>>> pdf.sum()
Series([], dtype: float64)
>>> ddf.sum().compute()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/ec2-user/.local/lib/python3.6/site-packages/dask/dataframe/core.py", line 1522, in sum
    "sum", axis=axis, skipna=skipna, split_every=split_every, out=out
  File "/home/ec2-user/.local/lib/python3.6/site-packages/dask/dataframe/core.py", line 1490, in _reduction_agg
    result.divisions = (min(self.columns), max(self.columns))
ValueError: min() arg is an empty sequence
>>> pdf.count()
Series([], dtype: int64)
>>> ddf.count().compute()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/ec2-user/.local/lib/python3.6/site-packages/dask/dataframe/core.py", line 1648, in count
    result.divisions = (min(self.columns), max(self.columns))
ValueError: min() arg is an empty sequence
```
Same story for `mean`, `var` and `sem`.
The issue stems from the following line in `dask/dataframe/core.py`:
```
result.divisions = (min(self.columns), max(self.columns))
```

Hints from issue discussion:
Thanks for reporting @exemplary-citizen! I'm able to reproduce the `ValueError`. Perhaps this can be fixed by adding a check that `self.columns != 0`. Is this something you're interested in looking into? (no obligation though)
I was going to open a PR in a bit. I was thinking of replacing the above line with ```result.divisions = (self.columns.min(), self.columns.max())```

Commit message: add test

Patch to review:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index cccd4307a27..b308de3ebd2 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1487,7 +1487,7 @@ def _reduction_agg(self, name, axis=None, skipna=True, split_every=False, out=No
                 split_every=split_every,
             )
             if isinstance(self, DataFrame):
-                result.divisions = (min(self.columns), max(self.columns))
+                result.divisions = (self.columns.min(), self.columns.max())
             return handle_out(out, result)
 
     @derived_from(pd.DataFrame)
@@ -1645,7 +1645,9 @@ def count(self, axis=None, split_every=False):
                 split_every=split_every,
             )
             if isinstance(self, DataFrame):
-                result.divisions = (min(self.columns), max(self.columns))
+                result.divisions = (self.columns.min(), self.columns.max())
+                if len(result) == 0:
+                    result = result.astype("int64")
             return result
 
     @derived_from(pd.DataFrame)
@@ -1678,7 +1680,7 @@ def mean(self, axis=None, skipna=True, split_every=False, dtype=None, out=None):
                 enforce_metadata=False,
             )
             if isinstance(self, DataFrame):
-                result.divisions = (min(self.columns), max(self.columns))
+                result.divisions = (self.columns.min(), self.columns.max())
             return handle_out(out, result)
 
     @derived_from(pd.DataFrame)
@@ -1719,7 +1721,7 @@ def var(
                 result = self._var_numeric(skipna, ddof, split_every)
 
             if isinstance(self, DataFrame):
-                result.divisions = (min(self.columns), max(self.columns))
+                result.divisions = (self.columns.min(), self.columns.max())
             return handle_out(out, result)
 
     def _var_numeric(self, skipna=True, ddof=1, split_every=False):
@@ -1881,7 +1883,7 @@ def sem(self, axis=None, skipna=None, ddof=1, split_every=False):
             )
 
             if isinstance(self, DataFrame):
-                result.divisions = (min(self.columns), max(self.columns))
+                result.divisions = (self.columns.min(), self.columns.max())
             return result
 
     def quantile(self, q=0.5, axis=0, method="default"):
diff --git a/dask/dataframe/tests/test_arithmetics_reduction.py b/dask/dataframe/tests/test_arithmetics_reduction.py
index 6998afc032c..3257a77c0ee 100644
--- a/dask/dataframe/tests/test_arithmetics_reduction.py
+++ b/dask/dataframe/tests/test_arithmetics_reduction.py
@@ -1346,3 +1346,14 @@ def test_moment():
     ddf = dd.from_pandas(df, npartitions=2)
 
     assert_eq(stats.moment(ddf, 2, 0), scipy.stats.moment(df, 2, 0))
+
+
+@pytest.mark.parametrize("func", ["sum", "count", "mean", "var", "sem"])
+def test_empty_df_reductions(func):
+    pdf = pd.DataFrame()
+    ddf = dd.from_pandas(pdf, npartitions=1)
+
+    dsk_func = getattr(ddf.__class__, func)
+    pd_func = getattr(pdf.__class__, func)
+
+    assert_eq(dsk_func(ddf), pd_func(pdf))


Code review comments:
diff_hunk: @@ -1645,7 +1645,9 @@ def count(self, axis=None, split_every=False):
                 split_every=split_every,
             )
             if isinstance(self, DataFrame):
-                result.divisions = (min(self.columns), max(self.columns))
+                result.divisions = (self.columns.min(), self.columns.max())
+                if len(result) == 0:
path: dask/dataframe/core.py
text: @user1:
Why is this astype needed? Is this case tested?

@author:
pandas result `dtype` is `int64`
```
>>> import dask.dataframe as dd
>>> import pandas as pd
>>> pdf = pd.DataFrame()
>>> ddf = dd.from_pandas(pdf,1)
>>> pdf.count()
Series([], dtype: int64)
>>> ddf.count().compute()
Series([], dtype: float64)
```

@author:
it is tested by `test_empty_df_reductions("count")`

@user1:
Why does `self.reduction` give the wrong dtype so that it needs to be fixed here?

@author:
because of aggregation where it sums up the counts. `sum` return `dtype` is `float64`

@user1:
Right, my question is about why it's float64. Are we creating an empty array somewhere, and so the dtype is inferred to be float rather than int?

I'd prefer to fix this at the source rather than trying to patch things after the fact.

@author:
seems to be coming from here:
```
def _concat(args):
    if not args:
        return args
    if isinstance(first(core.flatten(args)), np.ndarray):
        return da.core.concatenate3(args)
    if not has_parallel_type(args[0]):
        try:
            return pd.Series(args)
        except Exception:
            return args
    # We filter out empty partitions here because pandas frequently has
    # inconsistent dtypes in results between empty and non-empty frames.
    # Ideally this would be handled locally for each operation, but in practice
    # this seems easier. TODO: don't do this.
    args2 = [i for i in args if len(i)]
    return args[0] if not args2 else methods.concat(args2, uniform=True)
```

@user1:
Mmm, I'm not sure this is the root problem, though I may be wrong. I can try taking a closer look later, but it may be a while.

@user1:
Sorry for the delay in this @user2-citizen. Still planning to dig into this, hopefully tomorrow.

@author:
Sounds good!

@user1:
Fixed in https://github.com/dask/dask/pull/5804/commits/42681a0bc3fb03de344f87ded4517f3502ba99ea. The problem is that `pd.DataFrame(index=[0]).sum()` returns a float dtype series.

We need to avoid the `if len(result) == 0`, since that triggers a computation of the task graph up to this point.