Repository: dask/dask

Issue:
Sequential DataFrame joining error
I ran across a case where multiple DataFrame joins will sometimes fail, depending on how the DataFrames are partitioned.

The below example does sequential inner joins (i.e. `df1.join(df2, how='inner').join(df3, how='inner')`) using pandas DataFrames and then using dask DataFrames. 

```python

import pandas as pd
import dask.dataframe as dd
from dask.dataframe.utils import assert_eq

# Pandas version of multiple inner joins
df1 = pd.DataFrame({'key': list(range(6)),
                    'A': ['A0', 'A1', 'A2', 'A3', 'A4', 'A5']})
df2 = pd.DataFrame({'key': list(range(4)),
                    'B': ['B0', 'B1', 'B2', 'B3']})
df3 = pd.DataFrame({'key': list(range(1, 5)),
                    'C': ['C0', 'C1', 'C2', 'C3']})

join_pd = df1.join(df2, how='inner', lsuffix='_l', rsuffix='_r')
multi_join_pd = join_pd.join(df3, how='inner', lsuffix='_l', rsuffix='_r')
print(f'join_pd = {join_pd}')
print(f'multi_join_pd = {multi_join_pd}')

# Dask version of multiple inner joins
ddf1 = dd.from_pandas(df1, npartitions=3)
ddf2 = dd.from_pandas(df2, npartitions=2)
ddf3 = dd.from_pandas(df3, npartitions=2)

join_dd = ddf1.join(ddf2, how='inner', lsuffix='_l', rsuffix='_r')
multi_join_dd = join_dd.join(ddf3, how='inner', lsuffix='_l', rsuffix='_r')
print(f'join_dd.compute() = {join_dd.compute()}')
print(f'multi_join_dd.compute() = {multi_join_dd.compute()}')
```

The pandas case works as expected, however the dask case fails on `master`. 

<details>
<summary>Traceback details</summary>

```
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-1-b3483754a3b1> in <module>()
     24 multi_join_dd = join_dd.join(ddf3, how='inner', lsuffix='_l', rsuffix='_r')
     25 print(f'join_dd.compute() = {join_dd.compute()}')
---> 26 print(f'multi_join_dd.compute() = {multi_join_dd.compute()}')

~/github/dask/dask/dask/base.py in compute(self, **kwargs)
    154         dask.base.compute
    155         """
--> 156         (result,) = compute(self, traverse=False, **kwargs)
    157         return result
    158

~/github/dask/dask/dask/base.py in compute(*args, **kwargs)
    395     keys = [x.__dask_keys__() for x in collections]
    396     postcomputes = [x.__dask_postcompute__() for x in collections]
--> 397     results = schedule(dsk, keys, **kwargs)
    398     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    399

~/github/dask/dask/dask/threaded.py in get(dsk, result, cache, num_workers, **kwargs)
     74     results = get_async(pool.apply_async, len(pool._pool), dsk, result,
     75                         cache=cache, get_id=_thread_get_id,
---> 76                         pack_exception=pack_exception, **kwargs)
     77
     78     # Cleanup pools associated to dead threads

~/github/dask/dask/dask/local.py in get_async(apply_async, num_workers, dsk, result, cache, get_id, rerun_exceptions_locally, pack_exception, raise_exception, callbacks, dumps, loads, **kwargs)
    499                         _execute_task(task, data)  # Re-execute locally
    500                     else:
--> 501                         raise_exception(exc, tb)
    502                 res, worker_id = loads(res_info)
    503                 state['cache'][key] = res

~/github/dask/dask/dask/compatibility.py in reraise(exc, tb)
    110         if exc.__traceback__ is not tb:
    111             raise exc.with_traceback(tb)
--> 112         raise exc
    113
    114 else:

~/github/dask/dask/dask/local.py in execute_task(key, task_info, dumps, loads, get_id, pack_exception)
    270     try:
    271         task, data = loads(task_info)
--> 272         result = _execute_task(task, data)
    273         id = get_id()
    274         result = dumps((result, id))

~/github/dask/dask/dask/local.py in _execute_task(arg, cache, dsk)
    251         func, args = arg[0], arg[1:]
    252         args2 = [_execute_task(a, cache) for a in args]
--> 253         return func(*args2)
    254     elif not ishashable(arg):
    255         return arg

~/github/dask/dask/dask/dataframe/methods.py in boundary_slice(df, start, stop, right_boundary, left_boundary, kind)
    101         return df
    102     else:
--> 103         result = getattr(df, kind)[start:stop]
    104     if not right_boundary:
    105         right_index = result.index.get_slice_bound(stop, 'left', kind)

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexing.py in __getitem__(self, key)
   1476
   1477             maybe_callable = com._apply_if_callable(key, self.obj)
-> 1478             return self._getitem_axis(maybe_callable, axis=axis)
   1479
   1480     def _is_scalar_access(self, key):

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexing.py in _getitem_axis(self, key, axis)
   1864         if isinstance(key, slice):
   1865             self._validate_key(key, axis)
-> 1866             return self._get_slice_axis(key, axis=axis)
   1867         elif com.is_bool_indexer(key):
   1868             return self._getbool_axis(key, axis=axis)

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexing.py in _get_slice_axis(self, slice_obj, axis)
   1509         labels = obj._get_axis(axis)
   1510         indexer = labels.slice_indexer(slice_obj.start, slice_obj.stop,
-> 1511                                        slice_obj.step, kind=self.name)
   1512
   1513         if isinstance(indexer, slice):

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexes/base.py in slice_indexer(self, start, end, step, kind)
   4105         """
   4106         start_slice, end_slice = self.slice_locs(start, end, step=step,
-> 4107                                                  kind=kind)
   4108
   4109         # return a slice

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexes/base.py in slice_locs(self, start, end, step, kind)
   4306         start_slice = None
   4307         if start is not None:
-> 4308             start_slice = self.get_slice_bound(start, 'left', kind)
   4309         if start_slice is None:
   4310             start_slice = 0

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexes/base.py in get_slice_bound(self, label, side, kind)
   4232         # For datetime indices label may be a string that has to be converted
   4233         # to datetime boundary according to its resolution.
-> 4234         label = self._maybe_cast_slice_bound(label, side, kind)
   4235
   4236         # we need to look up the label

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexes/base.py in _maybe_cast_slice_bound(self, label, side, kind)
   4183         # this is rejected (generally .loc gets you here)
   4184         elif is_integer(label):
-> 4185             self._invalid_indexer('slice', label)
   4186
   4187         return label

~/miniconda/envs/dask_dev/lib/python3.6/site-packages/pandas/core/indexes/base.py in _invalid_indexer(self, form, key)
   1861                         "indexers [{key}] of {kind}".format(
   1862                             form=form, klass=type(self), key=key,
-> 1863                             kind=type(key)))
   1864
   1865     def get_duplicates(self):

TypeError: cannot do slice indexing on <class 'pandas.core.indexes.base.Index'> with these indexers [4] of <class 'int'>
```

</details>

<br>

Interestingly, the failure is partition-dependent. For instance in the above example, if you change the partitions for `ddf1` so that `ddf1 = dd.from_pandas(df1, npartitions=2)`, then the sequential joining of dask DataFrames works. 

Discussion:

Patch:
diff --git a/dask/array/top.py b/dask/array/top.py
index bbd601f5c..f71f588ae 100644
--- a/dask/array/top.py
+++ b/dask/array/top.py
@@ -601,20 +601,32 @@ def optimize_atop(full_graph, keys=()):
             deps = set(top_layers)
             while deps:  # we gather as many sub-layers as we can
                 dep = deps.pop()
-                if (dep in layers and
-                        isinstance(layers[dep], TOP) and
-                        not (dep != layer and dep in keep) and  # output layer
-                        layers[dep].concatenate == layers[layer].concatenate):  # punt on mixed concatenate
-                    top_layers.add(dep)
-
-                    # traverse further to this child's children
-                    for d in full_graph.dependencies.get(dep, ()):
-                        if len(dependents[d]) <= 1:
-                            deps.add(d)
-                        else:
-                            stack.append(d)
-                else:
+                if dep not in layers:
+                    stack.append(dep)
+                    continue
+                if not isinstance(layers[dep], TOP):
+                    stack.append(dep)
+                    continue
+                if (dep != layer and dep in keep):
+                    stack.append(dep)
+                    continue
+                if layers[dep].concatenate != layers[layer].concatenate:
                     stack.append(dep)
+                    continue
+
+                # passed everything, proceed
+                top_layers.add(dep)
+
+                # traverse further to this child's children
+                for d in full_graph.dependencies.get(dep, ()):
+                    # Don't allow reductions to proceed
+                    output_indices = set(layers[dep].output_indices)
+                    input_indices = {i for _, ind in layers[dep].indices if ind for i in ind}
+
+                    if len(dependents[d]) <= 1 and output_indices.issuperset(input_indices):
+                        deps.add(d)
+                    else:
+                        stack.append(d)
 
             # Merge these TOP layers into one
             new_layer = rewrite_atop([layers[l] for l in top_layers])
diff --git a/dask/dataframe/methods.py b/dask/dataframe/methods.py
index aa4f271f4..193bfbd60 100644
--- a/dask/dataframe/methods.py
+++ b/dask/dataframe/methods.py
@@ -51,6 +51,8 @@ def boundary_slice(df, start, stop, right_boundary=True, left_boundary=True,
                    kind='loc'):
     """Index slice start/stop. Can switch include/exclude boundaries.
 
+    Examples
+    --------
     >>> df = pd.DataFrame({'x': [10, 20, 30, 40, 50]}, index=[1, 2, 2, 3, 4])
     >>> boundary_slice(df, 2, None)
         x
@@ -69,7 +71,18 @@ def boundary_slice(df, start, stop, right_boundary=True, left_boundary=True,
     1  10
     2  20
     2  30
+
+    Empty input DataFrames are returned
+
+    >>> df_empty = pd.DataFrame()
+    >>> boundary_slice(df_empty, 1, 3)
+    Empty DataFrame
+    Columns: []
+    Index: []
     """
+    if df.empty:
+        return df
+
     if kind == 'loc' and not df.index.is_monotonic:
         # Pandas treats missing keys differently for label-slicing
         # on monotonic vs. non-monotonic indexes
diff --git a/docs/source/dataframe-create.rst b/docs/source/dataframe-create.rst
index c8476a03b..118990454 100644
--- a/docs/source/dataframe-create.rst
+++ b/docs/source/dataframe-create.rst
@@ -1,19 +1,19 @@
 Create and Store Dask DataFrames
 ================================
 
-Dask can create dataframes from various data storage formats like CSV, HDF,
-Apache Parquet, and others.  For most formats this data can live on various
+Dask can create DataFrames from various data storage formats like CSV, HDF,
+Apache Parquet, and others.  For most formats, this data can live on various
 storage systems including local disk, network file systems (NFS), the Hadoop
 File System (HDFS), and Amazon's S3 (excepting HDF, which is only available on
 POSIX like file systems).
 
 See the :doc:`Overview section <dataframe-overview>` for an in depth
-discussion of ``dask.dataframe`` scope, use, limitations.
+discussion of ``dask.dataframe`` scope, use, and limitations.
 
 API
 ---
 
-The following functions provide access to convert between Dask Dataframes,
+The following functions provide access to convert between Dask DataFrames,
 file formats, and other Dask or Python collections.
 
 .. currentmodule:: dask.dataframe
@@ -51,9 +51,9 @@ Pandas:
 Locations
 ---------
 
-For text, CSV, and Apache Parquet formats data can come from local disk, from
-the Hadoop File System, from S3FS, or others, by prepending the filenames with
-a protocol.
+For text, CSV, and Apache Parquet formats, data can come from local disk,
+the Hadoop File System, S3FS, or other sources, by prepending the filenames with
+a protocol:
 
 .. code-block:: python
 
@@ -61,10 +61,10 @@ a protocol.
    >>> df = dd.read_csv('hdfs:///path/to/my-data-*.csv')
    >>> df = dd.read_csv('s3://bucket-name/my-data-*.csv')
 
-For remote systems like HDFS or S3 credentials may be an issue.  Usually these
-are handled by configuration files on disk (such as a ``.boto`` file for S3)
+For remote systems like HDFS or S3, credentials may be an issue.  Usually, these
+are handled by configuration files on disk (such as a ``.boto`` file for S3),
 but in some cases you may want to pass storage-specific options through to the
-storage backend.  You can do this with the ``storage_options=`` keyword.
+storage backend.  You can do this with the ``storage_options=`` keyword:
 
 .. code-block:: python
 
@@ -75,11 +75,11 @@ storage backend.  You can do this with the ``storage_options=`` keyword.
 Dask Delayed
 ------------
 
-For more complex situations not covered by the functions above you may want to
-use :doc:`dask.delayed<delayed>` , which lets you construct
-Dask.dataframes out of arbitrary Python function calls that load dataframes.
-This can allow you to handle new formats easily, or bake in particular logic
-around loading data if, for example, your data is stored with some special
+For more complex situations not covered by the functions above, you may want to
+use :doc:`dask.delayed<delayed>`, which lets you construct Dask DataFrames out 
+of arbitrary Python function calls that load DataFrames.  This can allow you to 
+handle new formats easily or bake in particular logic around loading data if, 
+for example, your data is stored with some special format.
 
 See :doc:`documentation on using dask.delayed with
 collections<delayed-collections>` or an `example notebook
@@ -87,36 +87,36 @@ collections<delayed-collections>` or an `example notebook
 how to create a Dask DataFrame from a nested directory structure of Feather
 files (as a stand in for any custom file format).
 
-Dask.delayed is particularly useful when simple ``map`` operations aren't
+Dask delayed is particularly useful when simple ``map`` operations aren't
 sufficient to capture the complexity of your data layout.
 
 
 From Raw Dask Graphs
 --------------------
 
-This section is mainly for developers wishing to extend dask.dataframe.  It
+This section is mainly for developers wishing to extend ``dask.dataframe``.  It
 discusses internal API not normally needed by users.  Everything below can be
 done just as effectively with :doc:`dask.delayed<delayed>`  described
-just above.  You should never need to create a dataframe object by hand.
+just above.  You should never need to create a DataFrame object by hand.
 
 To construct a DataFrame manually from a dask graph you need the following
 information:
 
-1.  dask: a dask graph with keys like ``{(name, 0): ..., (name, 1): ...}`` as
+1.  Dask: a Dask graph with keys like ``{(name, 0): ..., (name, 1): ...}`` as
     well as any other tasks on which those tasks depend.  The tasks
     corresponding to ``(name, i)`` should produce ``pandas.DataFrame`` objects
-    that correspond to the columns and divisions information discussed below.
-2.  name: The special name used above
-3.  columns: A list of column names
-4.  divisions: A list of index values that separate the different partitions.
-    Alternatively, if you don't know the divisions (this is common) you can
+    that correspond to the columns and divisions information discussed below
+2.  Name: the special name used above
+3.  Columns: a list of column names
+4.  Divisions: a list of index values that separate the different partitions.
+    Alternatively, if you don't know the divisions (this is common), you can
     provide a list of ``[None, None, None, ...]`` with as many partitions as
-    you have plus one.  For more information see the Partitions section in the
-    :doc:`dataframe documentation <dataframe>`.
+    you have plus one.  For more information, see the Partitions section in the
+    :doc:`DataFrame documentation <dataframe>`
 
 As an example, we build a DataFrame manually that reads several CSV files that
-have a datetime index separated by day.  Note, you should never do this.  The
-``dd.read_csv`` function does this for you.
+have a datetime index separated by day.  Note that you should **never** do this.  
+The ``dd.read_csv`` function does this for you:
 
 .. code-block:: Python
 


Test patch:
diff --git a/dask/array/tests/test_atop.py b/dask/array/tests/test_atop.py
index db243f08b..3f5e75f63 100644
--- a/dask/array/tests/test_atop.py
+++ b/dask/array/tests/test_atop.py
@@ -269,7 +269,7 @@ def test_atop_stacked_new_axes_front(concatenate):
     assert z.chunks == ((7,), (7,), (2, 2, 1))
     assert_eq(z, np.ones((7, 7, 5)))
 
-    w = atop(lambda x: x[:, 0, 0], 'a', z, 'abc', dtype=x.dtype, concatenate=concatenate)
+    w = atop(lambda x: x[:, 0, 0], 'a', z, 'abc', dtype=x.dtype, concatenate=True)
     assert w.chunks == ((7,),)
     assert_eq(w, np.ones((7,)))
 
@@ -413,3 +413,16 @@ def test_gh_4176():
 
     D = da.Array(array_dsk, name, chunks, dtype=A.dtype)
     D.sum(axis=0).compute()
+
+
+def test_dont_merge_before_reductions():
+    x = da.ones(10, chunks=(5,))
+    y = da.atop(inc, 'i', x, 'i', dtype=x.dtype)
+    z = da.atop(sum, '', y, 'i', dtype=y.dtype)
+    w = da.atop(sum, '', z, '', dtype=y.dtype)
+
+    dsk = optimize_atop(w.dask)
+
+    assert len([d for d in dsk.dicts.values() if isinstance(d, TOP)]) == 2
+
+    z.compute()
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 7a42bb91e..832b0a015 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -3076,6 +3076,13 @@ def test_boundary_slice_nonmonotonic():
     tm.assert_frame_equal(result, expected)
 
 
+def test_boundary_slice_empty():
+    df = pd.DataFrame()
+    result = methods.boundary_slice(df, 1, 4)
+    expected = pd.DataFrame()
+    tm.assert_frame_equal(result, expected)
+
+
 @pytest.mark.parametrize('start, stop, right_boundary, left_boundary, drop', [
     (-1, None, False, False, [-1, -2]),
     (-1, None, False, True, [-2]),
diff --git a/dask/dataframe/tests/test_multi.py b/dask/dataframe/tests/test_multi.py
index 405359cb8..5065188df 100644
--- a/dask/dataframe/tests/test_multi.py
+++ b/dask/dataframe/tests/test_multi.py
@@ -239,6 +239,28 @@ def test_hash_join(how, shuffle):
             hash_join(a, 'y', b, 'y', 'outer', shuffle=shuffle)._name)
 
 
+def test_sequential_joins():
+    # Pandas version of multiple inner joins
+    df1 = pd.DataFrame({'key': list(range(6)),
+                        'A': ['A0', 'A1', 'A2', 'A3', 'A4', 'A5']})
+    df2 = pd.DataFrame({'key': list(range(4)),
+                        'B': ['B0', 'B1', 'B2', 'B3']})
+    df3 = pd.DataFrame({'key': list(range(1, 5)),
+                        'C': ['C0', 'C1', 'C2', 'C3']})
+
+    join_pd = df1.join(df2, how='inner', lsuffix='_l', rsuffix='_r')
+    multi_join_pd = join_pd.join(df3, how='inner', lsuffix='_l', rsuffix='_r')
+
+    # Dask version of multiple inner joins
+    ddf1 = dd.from_pandas(df1, npartitions=3)
+    ddf2 = dd.from_pandas(df2, npartitions=2)
+    ddf3 = dd.from_pandas(df3, npartitions=2)
+
+    join_dd = ddf1.join(ddf2, how='inner', lsuffix='_l', rsuffix='_r')
+    multi_join_dd = join_dd.join(ddf3, how='inner', lsuffix='_l', rsuffix='_r')
+    assert_eq(multi_join_pd, multi_join_dd)
+
+
 @pytest.mark.parametrize('join', ['inner', 'outer'])
 def test_indexed_concat(join):
     A = pd.DataFrame({'x': [1, 2, 3, 4, 6, 7], 'y': list('abcdef')},