Repository: dask/dask

Issue:
Implement .str.cat
This should be doable:

```python
a = dd.from_pandas(pd.Series(["a"] * 100), 2)
b = dd.from_pandas(pd.Series(['b'] * 100), 2)
a.str.cat(b, sep=":")  # NotImplementedError
```

Some issues around alignment probably.

Discussion:
TomAugspurger: Just a small comment, if you could raise for `others=str`.

Sorry about the extra commits. I fixed the merge conflict in the changelog, but the CI services didn't like something about it.

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/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/docs/source/changelog.rst b/docs/source/changelog.rst
index 247d55663..349aabf7f 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -35,7 +35,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`
+- 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/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 1472dd9fc..f01784f16 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -1378,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():