Repository: dask/dask

Issue:
repeated cumsum on dataframe returns the results of the first cumsum
reproduce with 
```python
import pandas as pd
import dask.dataframe as ddf

df = pd.DataFrame(dict(a=list('aabbcc')),
                  index=pd.date_range(start='20100101', periods=6))
df['ones']=1
df['twos']=2

dadf = ddf.from_pandas(df, npartitions=3)
dadf['onescs']=dadf.ones.cumsum()
dadf['twoscs']=dadf.twos.cumsum()
print dadf.compute()
```
which yields
```
            a  ones  twos  onescs  twoscs
2010-01-01  a     1     2       1       1
2010-01-02  a     1     2       2       2
2010-01-03  b     1     2       3       3
2010-01-04  b     1     2       4       4
2010-01-05  c     1     2       5       5
2010-01-06  c     1     2       6       6
```

Pandas itself works correctly:
```python
df['onescs']=df.ones.cumsum()
df['twoscs']=df.twos.cumsum()
print df
```
which yields
```
            a  ones  twos  onescs  twoscs
2010-01-01  a     1     2       1       2
2010-01-02  a     1     2       2       4
2010-01-03  b     1     2       3       6
2010-01-04  b     1     2       4       8
2010-01-05  c     1     2       5      10
2010-01-06  c     1     2       6      12
```

Discussion:
mrocklin: This is great.  Thank you for the bug report and the excellent fix.  

I suspect that this bug escaped notice for a while because of the ambiguous use of the name `token`, which we often use to mean "unique identifier" but which in the case of this method seems to mean "name of operation".  We might consider changing the name in this method as well from `token=` to `name=` or something similar.

That's just a small thought though.  I'm +1 on this change after tests pass.
mrocklin: The last failing test is due to https://github.com/dask/dask/issues/3043 and is unrelated to this PR. 

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index bfecfc992..206ac4c2d 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -18,12 +18,12 @@ import uuid
 import warnings
 
 try:
-    from cytoolz import (partition, concat, join, first,
+    from cytoolz import (partition, concat, concatv, join, first,
                          groupby, valmap, accumulate, assoc)
     from cytoolz.curried import filter, pluck
 
 except ImportError:
-    from toolz import (partition, concat, join, first,
+    from toolz import (partition, concat, concatv, join, first,
                        groupby, valmap, accumulate, assoc)
     from toolz.curried import filter, pluck
 from toolz import pipe, map, reduce
@@ -32,14 +32,15 @@ import numpy as np
 from . import chunk
 from .numpy_compat import _make_sliced_dtype
 from .slicing import slice_array, replace_ellipsis
-from ..base import Base, tokenize, dont_optimize, compute_as_if_collection
+from ..base import (Base, tokenize, dont_optimize, compute_as_if_collection,
+                    persist)
 from ..context import _globals, globalmethod
 from ..utils import (homogeneous_deepmap, ndeepmap, ignoring, concrete,
                      is_integer, IndexCallable, funcname, derived_from,
                      SerializableLock, ensure_dict, Dispatch)
 from ..compatibility import unicode, long, getargspec, zip_longest, apply
 from ..core import quote
-from ..delayed import to_task_dask
+from ..delayed import Delayed, to_task_dask
 from .. import threaded, core
 from .. import sharedict
 from ..sharedict import ShareDict
@@ -815,7 +816,8 @@ def broadcast_chunks(*chunkss):
     return tuple(result)
 
 
-def store(sources, targets, lock=True, regions=None, compute=True, **kwargs):
+def store(sources, targets, lock=True, regions=None, compute=True,
+          return_stored=False, **kwargs):
     """ Store dask arrays in array-like objects, overwrite data in target
 
     This stores dask arrays into object that supports numpy-style setitem
@@ -842,6 +844,8 @@ def store(sources, targets, lock=True, regions=None, compute=True, **kwargs):
         for the corresponding source and target in sources and targets, respectively.
     compute: boolean, optional
         If true compute immediately, return ``dask.delayed.Delayed`` otherwise
+    return_stored: boolean, optional
+        Optionally return the stored result (default False).
 
     Examples
     --------
@@ -859,6 +863,7 @@ def store(sources, targets, lock=True, regions=None, compute=True, **kwargs):
 
     >>> store([x, y, z], [dset1, dset2, dset3])  # doctest: +SKIP
     """
+
     if isinstance(sources, Array):
         sources = [sources]
         targets = [targets]
@@ -880,31 +885,73 @@ def store(sources, targets, lock=True, regions=None, compute=True, **kwargs):
         raise ValueError("Different number of sources [%d] and targets [%d] than regions [%d]"
                          % (len(sources), len(targets), len(regions)))
 
-    updates = {}
-    keys = []
+    # Optimize all sources together
+    sources_dsk = sharedict.merge(*[e.__dask_graph__() for e in sources])
+    sources_dsk = Array.__dask_optimize__(
+        sources_dsk,
+        [e.__dask_keys__() for e in sources]
+    )
+
+    tgt_dsks = []
+    store_keys = []
+    store_dsks = []
+    if return_stored:
+        load_names = []
+        load_dsks = []
     for tgt, src, reg in zip(targets, sources, regions):
         # if out is a delayed object update dictionary accordingly
         try:
-            dsk = {}
-            dsk.update(tgt.dask)
+            each_tgt_dsk = {}
+            each_tgt_dsk.update(tgt.dask)
             tgt = tgt.key
         except AttributeError:
-            dsk = {}
+            each_tgt_dsk = {}
+
+        src = Array(sources_dsk, src.name, src.chunks, src.dtype)
+
+        each_store_dsk = insert_to_ooc(
+            src, tgt, lock=lock, region=reg, return_stored=return_stored
+        )
+
+        if return_stored:
+            load_names.append('load-store-%s' % src.name)
+            load_dsks.append(retrieve_from_ooc(
+                each_store_dsk.keys(),
+                each_store_dsk
+            ))
+
+        tgt_dsks.append(each_tgt_dsk)
+
+        store_keys.extend(each_store_dsk.keys())
+        store_dsks.append(each_store_dsk)
 
-        update = insert_to_ooc(src, tgt, lock=lock, region=reg)
-        keys.extend(update)
+    store_dsks_mrg = sharedict.merge(*concatv(
+        store_dsks, tgt_dsks, [sources_dsk]
+    ))
 
-        update.update(dsk)
-        updates.update(update)
+    if return_stored:
+        if compute:
+            store_dlyds = [Delayed(k, store_dsks_mrg) for k in store_keys]
+            store_dlyds = persist(*store_dlyds)
+            store_dsks_mrg = sharedict.merge(*[e.dask for e in store_dlyds])
 
-    name = 'store-' + tokenize(*keys)
-    dsk = sharedict.merge((name, updates), *[src.dask for src in sources])
-    if compute:
-        compute_as_if_collection(Array, dsk, keys, **kwargs)
+        load_dsks_mrg = sharedict.merge(store_dsks_mrg, *load_dsks)
+
+        result = tuple(
+            Array(load_dsks_mrg, n, src.chunks, src.dtype) for n in load_names
+        )
+
+        return result
     else:
-        from ..delayed import Delayed
-        dsk.update({name: keys})
-        return Delayed(name, dsk)
+        name = 'store-' + tokenize(*store_keys)
+        dsk = sharedict.merge({name: store_keys}, store_dsks_mrg)
+        result = Delayed(name, dsk)
+
+        if compute:
+            result.compute()
+            return None
+        else:
+            return result
 
 
 def blockdims_from_blockshape(shape, chunks):
@@ -2157,9 +2204,7 @@ def atop(func, out_ind, *args, **kwargs):
     concatenate : bool, keyword only
         If true concatenate arrays along dummy indices, else provide lists
     adjust_chunks : dict
-        Dictionary mapping index to information to adjust chunk sizes.  Can
-        either be a constant chunksize, a tuple of all chunksizes, or a
-        function that converts old chunksize to new chunksize
+        Dictionary mapping index to function to be applied to chunk sizes
     new_axes : dict, keyword only
         New indexes and their dimension lengths
 
@@ -2568,7 +2613,7 @@ def concatenate(seq, axis=0, allow_unknown_chunksizes=False):
     return Array(dsk2, name, chunks, dtype=dt)
 
 
-def store_chunk(x, out, index, lock, region):
+def store_chunk(x, out, index, lock, region, return_stored):
     """
     A function inserted in a Dask graph for storing a chunk.
 
@@ -2584,15 +2629,21 @@ def store_chunk(x, out, index, lock, region):
         Lock to use before writing to ``out``.
     region: slice-like or None
         Where relative to ``out`` to store ``x``.
+    return_stored: bool
+        Whether to return ``out``.
 
     Examples
     --------
 
     >>> a = np.ones((5, 6))
     >>> b = np.empty(a.shape)
-    >>> store_chunk(a, b, (slice(None), slice(None)), False, None)
+    >>> store_chunk(a, b, (slice(None), slice(None)), False, None, False)
     """
 
+    result = None
+    if return_stored:
+        result = out
+
     subindex = index
     if region is not None:
         subindex = fuse_slice(region, index)
@@ -2605,10 +2656,10 @@ def store_chunk(x, out, index, lock, region):
         if lock:
             lock.release()
 
-    return None
+    return result
 
 
-def insert_to_ooc(arr, out, lock=True, region=None):
+def insert_to_ooc(arr, out, lock=True, region=None, return_stored=False):
     """
     Creates a Dask graph for storing chunks from ``arr`` in ``out``.
 
@@ -2624,6 +2675,9 @@ def insert_to_ooc(arr, out, lock=True, region=None):
     region: slice-like, optional
         Where in ``out`` to store ``arr``'s results
         (default is ``None``, meaning all of ``out``).
+    return_stored: bool, optional
+        Whether to return ``out``
+        (default is ``False``, meaning ``None`` is returned).
 
     Examples
     --------
@@ -2642,13 +2696,79 @@ def insert_to_ooc(arr, out, lock=True, region=None):
     dsk = dict()
     for t, slc in zip(core.flatten(arr.__dask_keys__()), slices):
         store_key = (name,) + t[1:]
-        dsk[store_key] = (
-            store_chunk, t, out, slc, lock, region
-        )
+        dsk[store_key] = (store_chunk, t, out, slc, lock, region, return_stored)
 
     return dsk
 
 
+def load_chunk(x, index, lock, region):
+    """
+    A function inserted in a Dask graph for loading a chunk.
+
+    Parameters
+    ----------
+    x: array-like
+        An array (potentially a NumPy one)
+    index: slice-like
+        Where to store result from ``x`` in ``out``.
+    lock: Lock-like or False
+        Lock to use before writing to ``out``.
+    region: slice-like or None
+        Where relative to ``out`` to store ``x``.
+
+    Examples
+    --------
+
+    >>> a = np.ones((5, 6))
+    >>> load_chunk(a, (slice(None), slice(None)), False, None)  # doctest: +SKIP
+    """
+
+    result = None
+
+    subindex = index
+    if region is not None:
+        subindex = fuse_slice(region, index)
+
+    if lock:
+        lock.acquire()
+    try:
+        result = x[subindex]
+    finally:
+        if lock:
+            lock.release()
+
+    return result
+
+
+def retrieve_from_ooc(keys, dsk):
+    """
+    Creates a Dask graph for loading stored ``keys`` from ``dsk``.
+
+    Parameters
+    ----------
+    keys: Sequence
+        A sequence containing Dask graph keys to load
+    dsk: Mapping
+        A Dask graph corresponding to a Dask Array
+
+    Examples
+    --------
+    >>> import dask.array as da
+    >>> d = da.ones((5, 6), chunks=(2, 3))
+    >>> a = np.empty(d.shape)
+    >>> g = insert_to_ooc(d, a)
+    >>> retrieve_from_ooc(g.keys(), g)  # doctest: +SKIP
+    """
+
+    load_dsk = dict()
+    for each_key in keys:
+        load_key = ('load-%s' % each_key[0],) + each_key[1:]
+        # Reuse the result and arguments from `store_chunk` in `load_chunk`.
+        load_dsk[load_key] = (load_chunk, each_key,) + dsk[each_key][3:-1]
+
+    return load_dsk
+
+
 def asarray(a):
     """Convert the input to a dask array.
 
diff --git a/dask/compatibility.py b/dask/compatibility.py
index 95a63efe2..fdadfa70e 100644
--- a/dask/compatibility.py
+++ b/dask/compatibility.py
@@ -55,6 +55,12 @@ if PY3:
     def _getargspec(func):
         return inspect.getfullargspec(func)
 
+    def get_named_args(func):
+        """Get all non ``*args/**kwargs`` arguments for a function"""
+        s = inspect.signature(func)
+        return [n for n, p in s.parameters.items()
+                if p.kind == p.POSITIONAL_OR_KEYWORD]
+
     def reraise(exc, tb=None):
         if exc.__traceback__ is not tb:
             raise exc.with_traceback(tb)
@@ -93,6 +99,14 @@ else:
     def _getargspec(func):
         return inspect.getargspec(func)
 
+    def get_named_args(func):
+        """Get all non ``*args/**kwargs`` arguments for a function"""
+        try:
+            return getargspec(func).args
+        except TypeError as e:
+            # Be consistent with py3
+            raise ValueError(*e.args)
+
     def gzip_decompress(b):
         f = gzip.GzipFile(fileobj=BytesIO(b))
         result = f.read()
diff --git a/dask/dataframe/accessor.py b/dask/dataframe/accessor.py
index 10ab0f3bc..5462f9a6e 100644
--- a/dask/dataframe/accessor.py
+++ b/dask/dataframe/accessor.py
@@ -119,6 +119,21 @@ class StringAccessor(Accessor):
     def split(self, pat=None, n=-1):
         return self._function_map('split', pat=pat, n=n)
 
+    @derived_from(pd.core.strings.StringMethods)
+    def cat(self, others=None, sep=None, na_rep=None):
+        from .core import Series, Index
+        if others is None:
+            raise NotImplementedError("x.str.cat() with `others == None`")
+
+        valid_types = (Series, Index, pd.Series, pd.Index)
+        if isinstance(others, valid_types):
+            others = [others]
+        elif not all(isinstance(a, valid_types) for a in others):
+            raise TypeError("others must be Series/Index")
+
+        return self._series.map_partitions(str_cat, *others, sep=sep,
+                                           na_rep=na_rep, meta=self._series._meta)
+
     def __getitem__(self, index):
         return self._series.map_partitions(str_get, index,
                                            meta=self._series._meta)
@@ -127,3 +142,7 @@ class StringAccessor(Accessor):
 def str_get(series, index):
     """ Implements series.str[index] """
     return series.str[index]
+
+
+def str_cat(self, *others, **kwargs):
+    return self.str.cat(others=others, **kwargs)
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index f31475ebc..e0e022a33 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1360,27 +1360,29 @@ Dask Name: {name}, {task} tasks""".format(klass=self.__class__.__name__,
 
         return new_dd_object(dsk, name, num._meta, divisions=[None, None])
 
-    def _cum_agg(self, token, chunk, aggregate, axis, skipna=True,
+    def _cum_agg(self, op_name, chunk, aggregate, axis, skipna=True,
                  chunk_kwargs=None):
         """ Wrapper for cumulative operation """
 
         axis = self._validate_axis(axis)
 
         if axis == 1:
-            name = '{0}{1}(axis=1)'.format(self._token_prefix, token)
+            name = '{0}{1}(axis=1)'.format(self._token_prefix, op_name)
             return self.map_partitions(chunk, token=name, **chunk_kwargs)
         else:
             # cumulate each partitions
-            name1 = '{0}{1}-map'.format(self._token_prefix, token)
+            name1 = '{0}{1}-map'.format(self._token_prefix, op_name)
             cumpart = map_partitions(chunk, self, token=name1, meta=self,
                                      **chunk_kwargs)
 
-            name2 = '{0}{1}-take-last'.format(self._token_prefix, token)
+            name2 = '{0}{1}-take-last'.format(self._token_prefix, op_name)
             cumlast = map_partitions(_take_last, cumpart, skipna,
                                      meta=pd.Series([]), token=name2)
 
-            name = '{0}{1}'.format(self._token_prefix, token)
-            cname = '{0}{1}-cum-last'.format(self._token_prefix, token)
+            suffix = tokenize(self)
+            name = '{0}{1}-{2}'.format(self._token_prefix, op_name, suffix)
+            cname = '{0}{1}-cum-last-{2}'.format(self._token_prefix, op_name,
+                                                 suffix)
 
             # aggregate cumulated partisions and its previous last element
             dask = {}
@@ -2966,7 +2968,11 @@ def elemwise(op, *args, **kwargs):
             # should not occur in current funcs
             msg = 'elemwise with 2 or more DataFrames and Scalar is not supported'
             raise NotImplementedError(msg)
-        meta = _emulate(op, *args, **kwargs)
+        # For broadcastable series, use no rows.
+        parts = [d._meta if _is_broadcastable(d)
+                 else d._meta_nonempty for d in dasks]
+        with raise_on_meta_error(funcname(op)):
+            meta = partial_by_order(*parts, function=op, other=other)
 
     return new_dd_object(dsk, _name, meta, divisions)
 
@@ -3152,9 +3158,9 @@ def apply_concat_apply(args, chunk=None, aggregate=None, combine=None,
             dsk[(b, j)] = (aggregate, conc)
 
     if meta is no_default:
-        meta_chunk = _emulate(apply, chunk, args, chunk_kwargs)
-        meta = _emulate(apply, aggregate, [_concat([meta_chunk])],
-                        aggregate_kwargs)
+        meta_chunk = _emulate(chunk, *args, **chunk_kwargs)
+        meta = _emulate(aggregate, _concat([meta_chunk]),
+                        **aggregate_kwargs)
     meta = make_meta(meta)
 
     for arg in args:
@@ -3173,15 +3179,8 @@ def _extract_meta(x, nonempty=False):
     """
     Extract internal cache data (``_meta``) from dd.DataFrame / dd.Series
     """
-    if isinstance(x, Scalar):
+    if isinstance(x, (Scalar, _Frame)):
         return x._meta_nonempty if nonempty else x._meta
-    elif isinstance(x, _Frame):
-        if (isinstance(x, Series) and
-                x.npartitions == 1 and
-                x.known_divisions):  # may be broadcastable
-            return x._meta
-        else:
-            return x._meta_nonempty if nonempty else x._meta
     elif isinstance(x, list):
         return [_extract_meta(_x, nonempty) for _x in x]
     elif isinstance(x, tuple):
diff --git a/dask/dataframe/groupby.py b/dask/dataframe/groupby.py
index ed7a1d8d5..a6c7752d3 100644
--- a/dask/dataframe/groupby.py
+++ b/dask/dataframe/groupby.py
@@ -289,8 +289,17 @@ def _nunique_df_chunk(df, *index, **kwargs):
     return grouped
 
 
+def _drop_duplicates_rename(df):
+    # Avoid duplicate index labels in a groupby().apply() context
+    # https://github.com/dask/dask/issues/3039
+    # https://github.com/pandas-dev/pandas/pull/18882
+    names = [None] * df.index.nlevels
+    return df.drop_duplicates().rename_axis(names, copy=False)
+
+
 def _nunique_df_combine(df, levels):
-    result = df.groupby(level=levels, sort=False).apply(pd.DataFrame.drop_duplicates)
+    result = df.groupby(level=levels,
+                        sort=False).apply(_drop_duplicates_rename)
 
     if isinstance(levels, list):
         result.index = pd.MultiIndex.from_arrays([
diff --git a/dask/utils.py b/dask/utils.py
index 89d3ac50b..10f23125a 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -16,7 +16,7 @@ import multiprocessing as mp
 import uuid
 from weakref import WeakValueDictionary
 
-from .compatibility import getargspec, PY3, unicode, bind_method
+from .compatibility import get_named_args, getargspec, PY3, unicode, bind_method
 from .core import get_deps
 from .context import _globals
 from .optimize import key_split    # noqa: F401
@@ -485,10 +485,10 @@ def derived_from(original_klass, version=None, ua_args=[]):
                 doc = ''
 
             try:
-                method_args = getargspec(method).args
-                original_args = getargspec(original_method).args
+                method_args = get_named_args(method)
+                original_args = get_named_args(original_method)
                 not_supported = [m for m in original_args if m not in method_args]
-            except TypeError:
+            except ValueError:
                 not_supported = []
 
             if len(ua_args) > 0:
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 1cbd01691..b6b532f86 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -26,6 +26,7 @@ bounds indexes (:pr:`2967`) `Stephan Hoyer`_
 DataFrame
 +++++++++
 
+- Fixed naming bug in cumulative aggregations (:issue:`3037`) `Martijn Arts`_
 - Fixed ``dd.read_csv`` when ``names`` is given but ``header`` is not set to ``None`` (:issue:`2976`) `Martijn Arts`_
 - Fixed ``dd.read_csv`` so that passing instances of ``CategoricalDtype`` in ``dtype`` will result in known categoricals (:pr:`2997`) `Tom Augspurger`_
 - Prevent ``bool()`` coercion from calling compute (:pr:`2958`) `Albert DeFusco`_
@@ -35,6 +36,8 @@ DataFrame
 - Fixed ``dd.concat`` losing the index dtype when the data contained a categorical (:issue:`2932`) `Tom Augspurger`_
 - ``DataFrame.merge()`` (:pr:`2960`) now supports merging on a combination of columns and the index `Jon Mease`_
 - Removed the deprecated ``dd.rolling*`` methods, in preperation for their removal in the next pandas release (:pr:`2995`) `Tom Augspurger`_
+- Fix metadata inference bug in which single-partition series were mistakenly special cased (:pr:`3035`) `Jim Crist`_
+- Add support for ``Series.str.cat`` (:pr:`3028`) `Jim Crist`_
 
 
 Core


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index 928206559..0eececc3c 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -19,7 +19,7 @@ from toolz.curried import identity
 import dask
 import dask.array as da
 from dask.base import tokenize, compute_as_if_collection
-from dask.delayed import delayed
+from dask.delayed import Delayed, delayed
 from dask.local import get_sync
 from dask.utils import ignoring, tmpfile, tmpdir
 from dask.utils_test import inc
@@ -1213,14 +1213,31 @@ def test_store_delayed_target():
     atd = delayed(make_target)('at')
     btd = delayed(make_target)('bt')
 
-    store([a, b], [atd, btd])
+    # test not keeping result
+    st = store([a, b], [atd, btd])
 
     at = targs['at']
     bt = targs['bt']
 
+    assert st is None
     assert_eq(at, a)
     assert_eq(bt, b)
 
+    # test keeping result
+    st = store([a, b], [atd, btd], return_stored=True, compute=False)
+    st = dask.compute(*st)
+
+    at = targs['at']
+    bt = targs['bt']
+
+    assert st is not None
+    assert isinstance(st, tuple)
+    assert all([isinstance(v, np.ndarray) for v in st])
+    assert_eq(at, a)
+    assert_eq(bt, b)
+    assert_eq(st[0], a)
+    assert_eq(st[1], b)
+
     pytest.raises(ValueError, lambda: store([a], [at, bt]))
     pytest.raises(ValueError, lambda: store(at, at))
     pytest.raises(ValueError, lambda: store([at, bt], [at, bt]))
@@ -1233,7 +1250,8 @@ def test_store():
     at = np.empty(shape=(4, 4))
     bt = np.empty(shape=(4, 4))
 
-    store([a, b], [at, bt])
+    st = store([a, b], [at, bt])
+    assert st is None
     assert (at == 2).all()
     assert (bt == 3).all()
 
@@ -1252,22 +1270,77 @@ def test_store_regions():
     at = np.zeros(shape=(8, 4, 6))
     bt = np.zeros(shape=(8, 4, 6))
     v = store([a, b], [at, bt], regions=region, compute=False)
+    assert isinstance(v, Delayed)
+    assert (at == 0).all() and (bt[region] == 0).all()
+    assert all([ev is None for ev in v.compute()])
+    assert (at[region] == 2).all() and (bt[region] == 3).all()
+    assert not (bt == 3).all() and not ( bt == 0 ).all()
+    assert not (at == 2).all() and not ( at == 0 ).all()
+
+    # Single region (keep result):
+    at = np.zeros(shape=(8, 4, 6))
+    bt = np.zeros(shape=(8, 4, 6))
+    v = store(
+        [a, b], [at, bt], regions=region, compute=False, return_stored=True
+    )
+    assert isinstance(v, tuple)
+    assert all([isinstance(e, da.Array) for e in v])
     assert (at == 0).all() and (bt[region] == 0).all()
-    v.compute()
+
+    ar, br = v
+    assert ar.dtype == a.dtype
+    assert br.dtype == b.dtype
+    assert ar.shape == a.shape
+    assert br.shape == b.shape
+    assert ar.chunks == a.chunks
+    assert br.chunks == b.chunks
+
+    ar, br = da.compute(ar, br)
     assert (at[region] == 2).all() and (bt[region] == 3).all()
     assert not (bt == 3).all() and not ( bt == 0 ).all()
     assert not (at == 2).all() and not ( at == 0 ).all()
+    assert (br == 3).all()
+    assert (ar == 2).all()
 
     # Multiple regions:
     at = np.zeros(shape=(8, 4, 6))
     bt = np.zeros(shape=(8, 4, 6))
     v = store([a, b], [at, bt], regions=[region, region], compute=False)
+    assert isinstance(v, Delayed)
     assert (at == 0).all() and (bt[region] == 0).all()
-    v.compute()
+    assert all([ev is None for ev in v.compute()])
     assert (at[region] == 2).all() and (bt[region] == 3).all()
     assert not (bt == 3).all() and not ( bt == 0 ).all()
     assert not (at == 2).all() and not ( at == 0 ).all()
 
+    # Multiple regions (keep result):
+    at = np.zeros(shape=(8, 4, 6))
+    bt = np.zeros(shape=(8, 4, 6))
+    v = store(
+        [a, b], [at, bt],
+        regions=[region, region],
+        compute=False,
+        return_stored=True
+    )
+    assert isinstance(v, tuple)
+    assert all([isinstance(e, da.Array) for e in v])
+    assert (at == 0).all() and (bt[region] == 0).all()
+
+    ar, br = v
+    assert ar.dtype == a.dtype
+    assert br.dtype == b.dtype
+    assert ar.shape == a.shape
+    assert br.shape == b.shape
+    assert ar.chunks == a.chunks
+    assert br.chunks == b.chunks
+
+    ar, br = da.compute(ar, br)
+    assert (at[region] == 2).all() and (bt[region] == 3).all()
+    assert not (bt == 3).all() and not ( bt == 0 ).all()
+    assert not (at == 2).all() and not ( at == 0 ).all()
+    assert (br == 3).all()
+    assert (ar == 2).all()
+
 
 def test_store_compute_false():
     d = da.ones((4, 4), chunks=(2, 2))
@@ -1277,8 +1350,18 @@ def test_store_compute_false():
     bt = np.zeros(shape=(4, 4))
 
     v = store([a, b], [at, bt], compute=False)
+    assert isinstance(v, Delayed)
+    assert (at == 0).all() and (bt == 0).all()
+    assert all([ev is None for ev in v.compute()])
+    assert (at == 2).all() and (bt == 3).all()
+
+    at = np.zeros(shape=(4, 4))
+    bt = np.zeros(shape=(4, 4))
+
+    dat, dbt = store([a, b], [at, bt], compute=False, return_stored=True)
+    assert isinstance(dat, Array) and isinstance(dbt, Array)
     assert (at == 0).all() and (bt == 0).all()
-    v.compute()
+    assert (dat.compute() == at).all() and (dbt.compute() == bt).all()
     assert (at == 2).all() and (bt == 3).all()
 
 
@@ -1320,6 +1403,7 @@ def test_store_locks():
 
     lock = Lock()
     v = store([a, b], [at, bt], compute=False, lock=lock)
+    assert isinstance(v, Delayed)
     dsk = v.dask
     locks = set(vv for v in dsk.values() for vv in v if isinstance(vv, _Lock))
     assert locks == set([lock])
@@ -1328,16 +1412,18 @@ def test_store_locks():
     at = NonthreadSafeStore()
     v = store([a, b], [at, at], lock=lock,
               get=dask.threaded.get, num_workers=10)
+    assert v is None
 
     # Don't assume thread safety by default
     at = NonthreadSafeStore()
-    store(a, at, get=dask.threaded.get, num_workers=10)
-    a.store(at, get=dask.threaded.get, num_workers=10)
+    assert store(a, at, get=dask.threaded.get, num_workers=10) is None
+    assert a.store(at, get=dask.threaded.get, num_workers=10) is None
 
     # Ensure locks can be removed
     at = ThreadSafeStore()
     for i in range(10):
-        a.store(at, lock=False, get=dask.threaded.get, num_workers=10)
+        st = a.store(at, lock=False, get=dask.threaded.get, num_workers=10)
+        assert st is None
         if at.max_concurrent_uses > 1:
             break
         if i == 9:
@@ -1350,7 +1436,8 @@ def test_store_multiprocessing_lock():
     a = d + 1
 
     at = np.zeros(shape=(10, 10))
-    a.store(at, get=dask.multiprocessing.get, num_workers=10)
+    st = a.store(at, get=dask.multiprocessing.get, num_workers=10)
+    assert st is None
 
 
 def test_to_hdf5():
diff --git a/dask/array/tests/test_chunk.py b/dask/array/tests/test_chunk.py
index c951fb2eb..4d9059932 100644
--- a/dask/array/tests/test_chunk.py
+++ b/dask/array/tests/test_chunk.py
@@ -16,7 +16,6 @@ def test_keepdims_wrapper_no_axis():
     summer_wrapped = keepdims_wrapper(summer)
 
     assert summer_wrapped != summer
-    assert summer_wrapped == keepdims_wrapper(summer_wrapped)
 
     a = np.arange(24).reshape(1, 2, 3, 4)
 
@@ -44,7 +43,6 @@ def test_keepdims_wrapper_one_axis():
     summer_wrapped = keepdims_wrapper(summer)
 
     assert summer_wrapped != summer
-    assert summer_wrapped == keepdims_wrapper(summer_wrapped)
 
     a = np.arange(24).reshape(1, 2, 3, 4)
 
@@ -72,7 +70,6 @@ def test_keepdims_wrapper_two_axes():
     summer_wrapped = keepdims_wrapper(summer)
 
     assert summer_wrapped != summer
-    assert summer_wrapped == keepdims_wrapper(summer_wrapped)
 
     a = np.arange(24).reshape(1, 2, 3, 4)
 
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 9c9aa323b..f4230fd3d 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -519,6 +519,22 @@ def test_map_partitions_keeps_kwargs_in_dict():
     assert a.x.map_partitions(f, x=5)._name != a.x.map_partitions(f, x=6)._name
 
 
+def test_metadata_inference_single_partition_aligned_args():
+    # https://github.com/dask/dask/issues/3034
+    # Previously broadcastable series functionality broke this
+
+    df = pd.DataFrame({'x': [1, 2, 3, 4, 5]})
+    ddf = dd.from_pandas(df, npartitions=1)
+
+    def check(df, df_x):
+        assert len(df) == len(df_x)
+        assert len(df) > 0
+        return df
+
+    res = dd.map_partitions(check, ddf, ddf.x)
+    assert_eq(res, ddf)
+
+
 def test_drop_duplicates():
     res = d.drop_duplicates()
     res2 = d.drop_duplicates(split_every=2)
@@ -1362,44 +1378,61 @@ def test_datetime_accessor():
 
 def test_str_accessor():
     df = pd.DataFrame({'x': ['abc', 'bcd', 'cdef', 'DEFG'], 'y': [1, 2, 3, 4]},
-                      index=['e', 'f', 'g', 'H'])
+                      index=['E', 'f', 'g', 'h'])
 
-    a = dd.from_pandas(df, 2, sort=False)
+    ddf = dd.from_pandas(df, 2)
 
     # Check that str not in dir/hasattr for non-object columns
-    assert 'str' not in dir(a.y)
-    assert not hasattr(a.y, 'str')
+    assert 'str' not in dir(ddf.y)
+    assert not hasattr(ddf.y, 'str')
 
     # not implemented methods don't show up
-    assert 'get_dummies' not in dir(a.x.str)
-    assert not hasattr(a.x.str, 'get_dummies')
+    assert 'get_dummies' not in dir(ddf.x.str)
+    assert not hasattr(ddf.x.str, 'get_dummies')
 
-    assert 'upper' in dir(a.x.str)
-    assert_eq(a.x.str.upper(), df.x.str.upper())
-    assert set(a.x.str.upper().dask) == set(a.x.str.upper().dask)
+    assert 'upper' in dir(ddf.x.str)
+    assert_eq(ddf.x.str.upper(), df.x.str.upper())
+    assert set(ddf.x.str.upper().dask) == set(ddf.x.str.upper().dask)
 
-    assert 'upper' in dir(a.index.str)
-    assert_eq(a.index.str.upper(), df.index.str.upper())
-    assert set(a.index.str.upper().dask) == set(a.index.str.upper().dask)
+    assert 'upper' in dir(ddf.index.str)
+    assert_eq(ddf.index.str.upper(), df.index.str.upper())
+    assert set(ddf.index.str.upper().dask) == set(ddf.index.str.upper().dask)
 
     # make sure to pass thru args & kwargs
-    assert 'contains' in dir(a.x.str)
-    assert_eq(a.x.str.contains('a'), df.x.str.contains('a'))
-    assert set(a.x.str.contains('a').dask) == set(a.x.str.contains('a').dask)
+    assert 'contains' in dir(ddf.x.str)
+    assert_eq(ddf.x.str.contains('a'), df.x.str.contains('a'))
+    assert set(ddf.x.str.contains('a').dask) == set(ddf.x.str.contains('a').dask)
 
-    assert_eq(a.x.str.contains('d', case=False), df.x.str.contains('d', case=False))
-    assert set(a.x.str.contains('d', case=False).dask) == set(a.x.str.contains('d', case=False).dask)
+    assert_eq(ddf.x.str.contains('d', case=False), df.x.str.contains('d', case=False))
+    assert (set(ddf.x.str.contains('d', case=False).dask) ==
+            set(ddf.x.str.contains('d', case=False).dask))
 
     for na in [True, False]:
-        assert_eq(a.x.str.contains('a', na=na), df.x.str.contains('a', na=na))
-        assert set(a.x.str.contains('a', na=na).dask) == set(a.x.str.contains('a', na=na).dask)
+        assert_eq(ddf.x.str.contains('a', na=na), df.x.str.contains('a', na=na))
+        assert (set(ddf.x.str.contains('a', na=na).dask) ==
+                set(ddf.x.str.contains('a', na=na).dask))
 
     for regex in [True, False]:
-        assert_eq(a.x.str.contains('a', regex=regex), df.x.str.contains('a', regex=regex))
-        assert set(a.x.str.contains('a', regex=regex).dask) == set(a.x.str.contains('a', regex=regex).dask)
+        assert_eq(ddf.x.str.contains('a', regex=regex), df.x.str.contains('a', regex=regex))
+        assert (set(ddf.x.str.contains('a', regex=regex).dask) ==
+                set(ddf.x.str.contains('a', regex=regex).dask))
+
+    assert_eq(ddf.x.str[:2], df.x.str[:2])
+    assert_eq(ddf.x.str[1], df.x.str[1])
+
+    # str.cat
+    sol = df.x.str.cat(df.x.str.upper(), sep=':')
+    assert_eq(ddf.x.str.cat(ddf.x.str.upper(), sep=':'), sol)
+    assert_eq(ddf.x.str.cat(df.x.str.upper(), sep=':'), sol)
+    assert_eq(ddf.x.str.cat([ddf.x.str.upper(), df.x.str.lower()], sep=':'),
+              df.x.str.cat([df.x.str.upper(), df.x.str.lower()], sep=':'))
 
-    assert_eq(df.x.str[:2], df.x.str[:2])
-    assert_eq(a.x.str[1], a.x.str[1])
+    for o in ['foo', ['foo']]:
+        with pytest.raises(TypeError):
+            ddf.x.str.cat(o)
+
+    with pytest.raises(NotImplementedError):
+        ddf.x.str.cat(sep=':')
 
 
 def test_empty_max():
@@ -2823,3 +2856,18 @@ def test_bool():
     for cond in conditions:
         with pytest.raises(ValueError):
             bool(cond)
+
+
+def test_cumulative_multiple_columns():
+    # GH 3037
+    df = pd.DataFrame(np.random.randn(100, 5), columns=list('abcde'))
+    ddf = dd.from_pandas(df, 5)
+
+    for d in [ddf, df]:
+        for c in df.columns:
+            d[c + 'cs'] = d[c].cumsum()
+            d[c + 'cmin'] = d[c].cummin()
+            d[c + 'cmax'] = d[c].cummax()
+            d[c + 'cp'] = d[c].cumprod()
+
+    assert_eq(ddf, df)
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 2c6a675df..de61dcd4d 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -160,6 +160,9 @@ def test_groupby_on_index(get):
     def func(df):
         return df.assign(b=df.b - df.b.mean())
 
+    def func2(df):
+        return df[['b']] - df[['b']].mean()
+
     with dask.set_options(get=get):
         with pytest.warns(None):
             assert_eq(ddf.groupby('a').apply(func),
@@ -168,8 +171,8 @@ def test_groupby_on_index(get):
             assert_eq(ddf.groupby('a').apply(func).set_index('a'),
                       pdf.groupby('a').apply(func).set_index('a'))
 
-            assert_eq(pdf2.groupby(pdf2.index).apply(func),
-                      ddf2.groupby(ddf2.index).apply(func))
+            assert_eq(pdf2.groupby(pdf2.index).apply(func2),
+                      ddf2.groupby(ddf2.index).apply(func2))
 
 
 def test_groupby_multilevel_getitem():