Repository: dask/dask

Issue:
Dataframe.reset_index() fails when index is named
```
import numpy as np
import pandas as pd
import dask.dataframe as dd


def test(idx_name=None):
    index = pd.Series(list('abcde'), name=idx_name)
    df = pd.DataFrame({
        'A': np.arange(5, dtype=np.int32),
        'B': np.arange(10, 15, dtype=np.int32),
    }, index=index)
    df = dd.from_pandas(df, npartitions=2)
    df = df.map_partitions(lambda p: (1.0 / p.sum(axis=1)).astype(np.float32),
                           meta=('new', np.float32))
    df = df.reset_index()
    df = df.compute()
    return df


if __name__ == '__main__':
    for idx_name in None, 'index', 'foo':
        print("\nidx_name: {!r}".format(idx_name))
        try:
            print(test(idx_name))
        except ValueError as ex:
            print(ex)
```
```

idx_name: None
  index       new
0     a  0.100000
1     b  0.083333
2     c  0.071429
0     d  0.062500
1     e  0.055556

idx_name: 'index'
  index       new
0     a  0.100000
1     b  0.083333
2     c  0.071429
0     d  0.062500
1     e  0.055556

idx_name: 'foo'
The columns in the computed data do not match the columns in the provided metadata
```

Discussion:

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 61158f7c5..08ca79f7e 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2428,7 +2428,7 @@ def unify_chunks(*args, **kwargs):
     if not args:
         return {}, []
 
-    arginds = [(asarray(a) if ind is not None else a, ind)
+    arginds = [(asanyarray(a) if ind is not None else a, ind)
                for a, ind in partition(2, args)]  # [x, ij, y, jk]
     args = list(concat(arginds))  # [(x, ij), (y, jk)]
     warn = kwargs.get('warn', True)
@@ -2934,19 +2934,12 @@ def asarray(a, **kwargs):
     >>> da.asarray(y)
     dask.array<array, shape=(2, 3), dtype=int64, chunksize=(2, 3)>
     """
-    def frame_types():
-        try:
-            import dask.dataframe as dd
-            return (dd.Series, dd.DataFrame)
-        except ImportError:
-            return ()
-
     if isinstance(a, Array):
         return a
-    if isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
-        a = stack(a)
-    elif isinstance(a, frame_types()):
+    elif hasattr(a, 'to_dask_array'):
         return a.to_dask_array()
+    elif isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
+        a = stack(a)
     elif not isinstance(getattr(a, 'shape', None), Iterable):
         a = np.asarray(a)
     return from_array(a, chunks=a.shape, getitem=getter_inline, **kwargs)
diff --git a/dask/bag/core.py b/dask/bag/core.py
index 141ade7e2..16551d3f6 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -343,7 +343,7 @@ class Item(DaskMethodsMixin):
         self.dask, self.key = state
 
     def apply(self, func):
-        name = 'apply-{0}-{1}'.format(funcname(func), tokenize(self, func))
+        name = '{0}-{1}'.format(funcname(func), tokenize(self, func, 'apply'))
         dsk = {name: (func, self.key)}
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=[self])
         return Item(graph, name)
@@ -538,8 +538,8 @@ class Bag(DaskMethodsMixin):
         >>> b.starmap(myadd, z=max_second).compute()
         [13, 17, 21, 25, 29]
         """
-        name = 'starmap-{0}-{1}'.format(funcname(func),
-                                        tokenize(self, func, kwargs))
+        name = '{0}-{1}'.format(funcname(func),
+                                tokenize(self, func, 'starmap', **kwargs))
         dependencies = [self]
         if kwargs:
             kwargs, collections = unpack_scalar_dask_kwargs(kwargs)
@@ -1849,7 +1849,7 @@ def bag_map(func, *args, **kwargs):
     >>> db.map(myadd, b, b.max()).compute()
     [4, 5, 6, 7, 8]
     """
-    name = 'map-%s-%s' % (funcname(func), tokenize(func, args, kwargs))
+    name = '%s-%s' % (funcname(func), tokenize(func, 'map', *args, **kwargs))
     dsk = {}
     dependencies = []
 
@@ -1942,8 +1942,8 @@ def map_partitions(func, *args, **kwargs):
     single graph, and then computes everything at once, and in some cases
     may be more efficient.
     """
-    name = 'map-partitions-%s-%s' % (funcname(func),
-                                     tokenize(func, args, kwargs))
+    name = '%s-%s' % (funcname(func),
+                      tokenize(func, 'map-partitions', *args, **kwargs))
     dsk = {}
     dependencies = []
 
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index d1243ea68..0eb574da8 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -27,7 +27,7 @@ from ..compatibility import (apply, operator_div, bind_method, string_types,
 from ..context import globalmethod
 from ..utils import (random_state_data, pseudorandom, derived_from, funcname,
                      memory_repr, put_lines, M, key_split, OperatorMethodMixin,
-                     is_arraylike)
+                     is_arraylike, typename)
 from ..array.core import Array, normalize_arg
 from ..blockwise import blockwise, Blockwise
 from ..base import DaskMethodsMixin, tokenize, dont_optimize, is_dask_collection
@@ -90,7 +90,7 @@ class Scalar(DaskMethodsMixin, OperatorMethodMixin):
         meta = make_meta(meta)
         if is_dataframe_like(meta) or is_series_like(meta) or is_index_like(meta):
             raise TypeError("Expected meta to specify scalar, got "
-                            "{0}".format(type(meta).__name__))
+                            "{0}".format(typename(type(meta))))
         self._meta = meta
 
     def __dask_graph__(self):
@@ -253,8 +253,8 @@ class _Frame(DaskMethodsMixin, OperatorMethodMixin):
         meta = make_meta(meta)
         if not isinstance(meta, self._partition_type):
             raise TypeError("Expected meta to specify type {0}, got type "
-                            "{1}".format(self._partition_type.__name__,
-                                         type(meta).__name__))
+                            "{1}".format(typename(self._partition_type),
+                                         typename(type(meta))))
         self._meta = meta
         self.divisions = tuple(divisions)
 
@@ -385,7 +385,8 @@ Dask Name: {name}, {task} tasks""".format(klass=self.__class__.__name__,
     @property
     def index(self):
         """Return dask Index instance"""
-        return self.map_partitions(getattr, 'index', token=self._name + '-index')
+        return self.map_partitions(getattr, 'index', token=self._name + '-index',
+                                   meta=self._meta.index)
 
     def reset_index(self, drop=False):
         """Reset the index to the default index.
@@ -2128,7 +2129,7 @@ Dask Name: {name}, {task} tasks""".format(klass=self.__class__.__name__,
         if meta is no_default:
             meta = _emulate(M.map, self, arg, na_action=na_action, udf=True)
         else:
-            meta = make_meta(meta)
+            meta = make_meta(meta, index=getattr(make_meta(self), 'index', None))
 
         return Series(graph, name, meta, self.divisions)
 
@@ -3583,8 +3584,9 @@ def apply_concat_apply(args, chunk=None, aggregate=None, combine=None,
     if not isinstance(args, (tuple, list)):
         args = [args]
 
-    npartitions = set(arg.npartitions for arg in args
-                      if isinstance(arg, _Frame))
+    dfs = [arg for arg in args if isinstance(arg, _Frame)]
+
+    npartitions = set(arg.npartitions for arg in dfs)
     if len(npartitions) > 1:
         raise ValueError("All arguments must have same number of partitions")
     npartitions = npartitions.pop()
@@ -3610,13 +3612,13 @@ def apply_concat_apply(args, chunk=None, aggregate=None, combine=None,
         dsk = {(a, 0, i, 0): (apply, chunk,
                               [(x._name, i) if isinstance(x, _Frame)
                                else x for x in args], chunk_kwargs)
-               for i in range(args[0].npartitions)}
+               for i in range(npartitions)}
 
     # Split
     if split_out and split_out > 1:
         split_prefix = 'split-%s' % token_key
         shard_prefix = 'shard-%s' % token_key
-        for i in range(args[0].npartitions):
+        for i in range(npartitions):
             dsk[(split_prefix, i)] = (hash_shard, (a, 0, i, 0), split_out,
                                       split_out_setup, split_out_setup_kwargs)
             for j in range(split_out):
@@ -3654,11 +3656,10 @@ def apply_concat_apply(args, chunk=None, aggregate=None, combine=None,
         meta_chunk = _emulate(chunk, *args, udf=True, **chunk_kwargs)
         meta = _emulate(aggregate, _concat([meta_chunk]), udf=True,
                         **aggregate_kwargs)
-    meta = make_meta(meta)
-
-    dependencies = [arg for arg in args if isinstance(arg, _Frame)]
+    meta = make_meta(meta, index=(getattr(make_meta(dfs[0]), 'index', None)
+                                  if dfs else None))
 
-    graph = HighLevelGraph.from_collections(b, dsk, dependencies=dependencies)
+    graph = HighLevelGraph.from_collections(b, dsk, dependencies=dfs)
 
     divisions = [None] * (split_out + 1)
 
@@ -3713,9 +3714,6 @@ def map_partitions(func, *args, **kwargs):
     meta = kwargs.pop('meta', no_default)
     name = kwargs.pop('token', None)
 
-    if meta is not no_default:
-        meta = make_meta(meta)
-
     # Normalize keyword arguments
     kwargs2 = {k: normalize_arg(v) for k, v in kwargs.items()}
 
@@ -3730,9 +3728,13 @@ def map_partitions(func, *args, **kwargs):
     from .multi import _maybe_align_partitions
     args = _maybe_from_pandas(args)
     args = _maybe_align_partitions(args)
+    dfs = [df for df in args if isinstance(df, _Frame)]
+    meta_index = getattr(make_meta(dfs[0]), 'index', None) if dfs else None
 
     if meta is no_default:
         meta = _emulate(func, *args, udf=True, **kwargs2)
+    else:
+        meta = make_meta(meta, index=meta_index)
 
     if all(isinstance(arg, Scalar) for arg in args):
         layer = {(name, 0):
@@ -3742,7 +3744,9 @@ def map_partitions(func, *args, **kwargs):
     elif not (has_parallel_type(meta) or is_arraylike(meta)):
         # If `meta` is not a pandas object, the concatenated results will be a
         # different type
-        meta = _concat([meta])
+        meta = make_meta(_concat([meta]), index=meta_index)
+
+    # Ensure meta is empty series
     meta = make_meta(meta)
 
     args2 = []
@@ -3776,7 +3780,6 @@ def map_partitions(func, *args, **kwargs):
         _meta=meta,
         **kwargs3
     )
-    dfs = [df for df in args if isinstance(df, _Frame)]
 
     graph = HighLevelGraph.from_collections(name, dsk, dependencies=dependencies)
     return new_dd_object(graph, name, meta, dfs[0].divisions)
diff --git a/dask/dataframe/groupby.py b/dask/dataframe/groupby.py
index a0683cfc6..b115833b4 100644
--- a/dask/dataframe/groupby.py
+++ b/dask/dataframe/groupby.py
@@ -283,16 +283,21 @@ def _nunique_df_chunk(df, *index, **kwargs):
     name = kwargs.pop('name')
 
     g = _groupby_raise_unaligned(df, by=index)
-    grouped = g[[name]].apply(pd.DataFrame.drop_duplicates)
-
-    # we set the index here to force a possibly duplicate index
-    # for our reduce step
-    if isinstance(levels, list):
-        grouped.index = pd.MultiIndex.from_arrays([
-            grouped.index.get_level_values(level=level) for level in levels
-        ])
+    if len(df) > 0:
+        grouped = g[[name]].apply(pd.DataFrame.drop_duplicates)
+        # we set the index here to force a possibly duplicate index
+        # for our reduce step
+        if isinstance(levels, list):
+            grouped.index = pd.MultiIndex.from_arrays([
+                grouped.index.get_level_values(level=level) for level in levels
+            ])
+        else:
+            grouped.index = grouped.index.get_level_values(level=levels)
     else:
-        grouped.index = grouped.index.get_level_values(level=levels)
+        # Manually create empty version, since groupby-apply for empty frame
+        # results in df with no columns
+        grouped = g[[name]].nunique()
+        grouped = grouped.astype(df.dtypes[grouped.columns].to_dict())
 
     return grouped
 
diff --git a/dask/dataframe/rolling.py b/dask/dataframe/rolling.py
index d953997a3..79d2fff89 100644
--- a/dask/dataframe/rolling.py
+++ b/dask/dataframe/rolling.py
@@ -86,7 +86,7 @@ def map_overlap(func, df, before, after, *args, **kwargs):
         meta = kwargs.pop('meta')
     else:
         meta = _emulate(func, df, *args, **kwargs)
-    meta = make_meta(meta)
+    meta = make_meta(meta, index=df._meta.index)
 
     name = '{0}-{1}'.format(func_name, token)
     name_a = 'overlap-prepend-' + tokenize(df, before)
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index ca00d6260..681d070ba 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -264,13 +264,13 @@ make_meta = Dispatch('make_meta')
 
 
 @make_meta.register((pd.Series, pd.DataFrame))
-def make_meta_pandas(x):
+def make_meta_pandas(x, index=None):
     return x.iloc[:0]
 
 
 @make_meta.register(pd.Index)
-def make_meta_index(x):
-    return x[:0]
+def make_meta_index(x, index=None):
+    return x[0:0]
 
 
 @make_meta.register(object)
@@ -305,7 +305,8 @@ def make_meta_object(x, index=None):
     elif is_arraylike(x):
         return x[:0]
 
-    index = index if index is None else index[0:0]
+    if index is not None:
+        index = make_meta(index)
 
     if isinstance(x, dict):
         return pd.DataFrame({c: _empty_series(c, d, index=index)
@@ -398,13 +399,12 @@ def _nonempty_index(idx):
                                  name=idx.name)
     elif typ is pd.CategoricalIndex:
         if len(idx.categories) == 0:
-            data = _nonempty_index(idx.categories)
-            cats = None
+            data = pd.Categorical(_nonempty_index(idx.categories),
+                                  ordered=idx.ordered)
         else:
-            data = _nonempty_index(_nonempty_index(idx.categories))
-            cats = idx.categories
-        return pd.CategoricalIndex(data, categories=cats,
-                                   ordered=idx.ordered, name=idx.name)
+            data = pd.Categorical.from_codes(
+                [-1, 0], categories=idx.categories, ordered=idx.ordered)
+        return pd.CategoricalIndex(data, name=idx.name)
     elif typ is pd.MultiIndex:
         levels = [_nonempty_index(l) for l in idx.levels]
         labels = [[0, 0] for i in idx.levels]
diff --git a/dask/utils.py b/dask/utils.py
index ac58dfcb1..73aff6372 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -558,6 +558,25 @@ def funcname(func):
         return str(func)
 
 
+def typename(typ):
+    """
+    Return the name of a type
+
+    Examples
+    --------
+    >>> typename(int)
+    'int'
+
+    >>> from dask.core import literal
+    >>> typename(literal)
+    'dask.core.literal'
+    """
+    if not typ.__module__ or typ.__module__ == 'builtins':
+        return typ.__name__
+    else:
+        return typ.__module__ + '.' + typ.__name__
+
+
 def ensure_bytes(s):
     """ Turn string or bytes to bytes
 
diff --git a/docs/source/develop.rst b/docs/source/develop.rst
index 3e0ad25f7..d22755cb1 100644
--- a/docs/source/develop.rst
+++ b/docs/source/develop.rst
@@ -88,7 +88,7 @@ environment on your computer to compile them when installing with ``pip``::
 Install Dask and dependencies::
 
    cd dask
-   pip install -e .[complete]
+   pip install -e ".[complete]"
 
 For development, Dask uses the following additional dependencies::
 


Test patch:
diff --git a/dask/array/tests/test_masked.py b/dask/array/tests/test_masked.py
index 6608d2cc4..222e46c5c 100644
--- a/dask/array/tests/test_masked.py
+++ b/dask/array/tests/test_masked.py
@@ -348,3 +348,17 @@ def test_average_weights_with_masked_array():
     da_avg = da.ma.average(d_a, weights=d_weights, axis=1)
 
     assert_eq(np_avg, da_avg)
+
+
+def test_arithmetic_results_in_masked():
+    mask = np.array([[True, False],
+                     [True, True],
+                     [False, True]])
+    x = np.arange(6).reshape((3, 2))
+    masked = np.ma.array(x, mask=mask)
+    dx = da.from_array(x, chunks=(2, 2))
+
+    res = dx + masked
+    sol = x + masked
+    assert_eq(res, sol)
+    assert isinstance(res.compute(), np.ma.masked_array)
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index 116b41b99..c6a46be60 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -1314,3 +1314,11 @@ def test_bag_paths():
     assert b.to_textfiles('foo*') == ['foo0', 'foo1']
     os.remove('foo0')
     os.remove('foo1')
+
+
+def test_map_keynames():
+    b = db.from_sequence([1, 2, 3])
+    d = dict(b.map(inc).__dask_graph__())
+    assert 'inc' in map(dask.utils.key_split, d)
+
+    assert set(b.map(inc).__dask_graph__()) != set(b.map_partitions(inc).__dask_graph__())
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 0476bf8a3..83742dad0 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -615,6 +615,22 @@ def test_map_partitions_method_names():
     assert b.dtype == 'i8'
 
 
+def test_map_partitions_propagates_index_metadata():
+    index = pd.Series(list('abcde'), name='myindex')
+    df = pd.DataFrame({'A': np.arange(5, dtype=np.int32),
+                       'B': np.arange(10, 15, dtype=np.int32)},
+                      index=index)
+    ddf = dd.from_pandas(df, npartitions=2)
+    res = ddf.map_partitions(lambda df: df.assign(C=df.A + df.B),
+                             meta=[('A', 'i4'), ('B', 'i4'), ('C', 'i4')])
+    sol = df.assign(C=df.A + df.B)
+    assert_eq(res, sol)
+
+    res = ddf.map_partitions(lambda df: df.rename_axis("newindex"))
+    sol = df.rename_axis("newindex")
+    assert_eq(res, sol)
+
+
 @pytest.mark.xfail(reason='now we use SubgraphCallables')
 def test_map_partitions_keeps_kwargs_readable():
     df = pd.DataFrame({'x': [1, 2, 3, 4], 'y': [5, 6, 7, 8]})
@@ -946,16 +962,21 @@ def test_assign_callable():
 
 
 def test_map():
-    assert_eq(d.a.map(lambda x: x + 1), full.a.map(lambda x: x + 1))
-    lk = dict((v, v + 1) for v in full.a.values)
-    assert_eq(d.a.map(lk), full.a.map(lk))
-    assert_eq(d.b.map(lk), full.b.map(lk))
+    df = pd.DataFrame({'a': range(9),
+                       'b': [4, 5, 6, 1, 2, 3, 0, 0, 0]},
+                      index=pd.Index([0, 1, 3, 5, 6, 8, 9, 9, 9], name='myindex'))
+    ddf = dd.from_pandas(df, npartitions=3)
+
+    assert_eq(ddf.a.map(lambda x: x + 1), df.a.map(lambda x: x + 1))
+    lk = dict((v, v + 1) for v in df.a.values)
+    assert_eq(ddf.a.map(lk), df.a.map(lk))
+    assert_eq(ddf.b.map(lk), df.b.map(lk))
     lk = pd.Series(lk)
-    assert_eq(d.a.map(lk), full.a.map(lk))
-    assert_eq(d.b.map(lk), full.b.map(lk))
-    assert_eq(d.b.map(lk, meta=d.b), full.b.map(lk))
-    assert_eq(d.b.map(lk, meta=('b', 'i8')), full.b.map(lk))
-    pytest.raises(TypeError, lambda: d.a.map(d.b))
+    assert_eq(ddf.a.map(lk), df.a.map(lk))
+    assert_eq(ddf.b.map(lk), df.b.map(lk))
+    assert_eq(ddf.b.map(lk, meta=ddf.b), df.b.map(lk))
+    assert_eq(ddf.b.map(lk, meta=('b', 'i8')), df.b.map(lk))
+    pytest.raises(TypeError, lambda: ddf.a.map(d.b))
 
 
 def test_concat():
@@ -3380,3 +3401,12 @@ def test_has_parallel_type():
     assert has_parallel_type(pd.DataFrame())
     assert has_parallel_type(pd.Series())
     assert not has_parallel_type(123)
+
+
+def test_meta_error_message():
+    with pytest.raises(TypeError) as info:
+        dd.DataFrame({('x', 1): 123}, 'x', pd.Series(), [None, None])
+
+    assert 'Series' in str(info.value)
+    assert 'DataFrame' in str(info.value)
+    assert 'pandas' in str(info.value)
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 83309b00d..075fcec68 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -405,10 +405,22 @@ def test_groupby_set_index():
                   lambda: ddf.groupby(df.index.month, as_index=False))
 
 
-def test_split_apply_combine_on_series():
-    pdf = pd.DataFrame({'a': [1, 2, 6, 4, 4, 6, 4, 3, 7],
-                        'b': [4, 2, 7, 3, 3, 1, 1, 1, 2]},
-                       index=[0, 1, 3, 5, 6, 8, 9, 9, 9])
+@pytest.mark.parametrize('empty', [
+    pytest.param(True, marks=pytest.mark.skipif(PANDAS_VERSION < '0.21.0',
+                 reason="Empty groupby-reductions fail for older pandas")),
+    pytest.param(False)
+])
+def test_split_apply_combine_on_series(empty):
+    if empty:
+        pdf = pd.DataFrame({'a': [1.], 'b': [1.]}, index=[0]).iloc[:0]
+        # There's a bug in pandas where df.groupby(...).var(ddof=0) results in
+        # no columns. Just skip these checks for now.
+        ddofs = []
+    else:
+        ddofs = [0, 1, 2]
+        pdf = pd.DataFrame({'a': [1, 2, 6, 4, 4, 6, 4, 3, 7],
+                            'b': [4, 2, 7, 3, 3, 1, 1, 1, 2]},
+                           index=[0, 1, 3, 5, 6, 8, 9, 9, 9])
     ddf = dd.from_pandas(pdf, npartitions=3)
 
     for ddkey, pdkey in [('b', 'b'), (ddf.b, pdf.b), (ddf.b + 1, pdf.b + 1)]:
@@ -420,7 +432,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddkey).a.size(), pdf.groupby(pdkey).a.size())
         assert_eq(ddf.groupby(ddkey).a.first(), pdf.groupby(pdkey).a.first())
         assert_eq(ddf.groupby(ddkey).a.last(), pdf.groupby(pdkey).a.last())
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddkey).a.var(ddof),
                       pdf.groupby(pdkey).a.var(ddof))
             assert_eq(ddf.groupby(ddkey).a.std(ddof),
@@ -435,7 +447,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddkey).first(), pdf.groupby(pdkey).first())
         assert_eq(ddf.groupby(ddkey).last(), pdf.groupby(pdkey).last())
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddkey).var(ddof),
                       pdf.groupby(pdkey).var(ddof), check_dtype=False)
             assert_eq(ddf.groupby(ddkey).std(ddof),
@@ -450,7 +462,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.a.groupby(ddkey).first(), pdf.a.groupby(pdkey).first(), check_names=False)
         assert_eq(ddf.a.groupby(ddkey).last(), pdf.a.groupby(pdkey).last(), check_names=False)
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.a.groupby(ddkey).var(ddof),
                       pdf.a.groupby(pdkey).var(ddof))
             assert_eq(ddf.a.groupby(ddkey).std(ddof),
@@ -495,7 +507,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddf.a > i).first(), pdf.groupby(pdf.a > i).first())
         assert_eq(ddf.groupby(ddf.a > i).last(), pdf.groupby(pdf.a > i).last())
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddf.b > i).std(ddof),
                       pdf.groupby(pdf.b > i).std(ddof))
 
@@ -520,7 +532,7 @@ def test_split_apply_combine_on_series():
         assert_eq(ddf.groupby(ddkey).first(), pdf.groupby(pdkey).first())
         assert_eq(ddf.groupby(ddkey).last(), pdf.groupby(pdkey).last())
 
-        for ddof in [0, 1, 2]:
+        for ddof in ddofs:
             assert_eq(ddf.groupby(ddkey).b.std(ddof),
                       pdf.groupby(pdkey).b.std(ddof))
 
diff --git a/dask/dataframe/tests/test_rolling.py b/dask/dataframe/tests/test_rolling.py
index a37b2dbc7..5edf16612 100644
--- a/dask/dataframe/tests/test_rolling.py
+++ b/dask/dataframe/tests/test_rolling.py
@@ -52,7 +52,7 @@ def test_map_overlap(npartitions):
         assert_eq(res, sol)
 
 
-def test_map_partitions_names():
+def test_map_overlap_names():
     npartitions = 3
     ddf = dd.from_pandas(df, npartitions)
 
@@ -70,7 +70,7 @@ def test_map_partitions_names():
     assert res4._name != res._name
 
 
-def test_map_partitions_errors():
+def test_map_overlap_errors():
     # Non-integer
     with pytest.raises(ValueError):
         ddf.map_overlap(shifted_sum, 0.5, 3, 0, 2, c=2)
@@ -89,6 +89,18 @@ def test_map_partitions_errors():
                         0, 2, c=2)
 
 
+def test_map_overlap_provide_meta():
+    df = pd.DataFrame({'x': [1, 2, 4, 7, 11],
+                       'y': [1., 2., 3., 4., 5.]}).rename_axis('myindex')
+    ddf = dd.from_pandas(df, npartitions=2)
+
+    # Provide meta spec, but not full metadata
+    res = ddf.map_overlap(lambda df: df.rolling(2).sum(), 2, 0,
+                          meta={'x': 'i8', 'y': 'i8'})
+    sol = df.rolling(2).sum()
+    assert_eq(res, sol)
+
+
 def mad(x):
     return np.fabs(x - x.mean()).mean()
 
diff --git a/dask/dataframe/tests/test_utils_dataframe.py b/dask/dataframe/tests/test_utils_dataframe.py
index 603331329..081246014 100644
--- a/dask/dataframe/tests/test_utils_dataframe.py
+++ b/dask/dataframe/tests/test_utils_dataframe.py
@@ -209,7 +209,7 @@ def test_meta_nonempty_index():
     assert res.freq == idx.freq
     assert res.name == idx.name
 
-    idx = pd.CategoricalIndex(['a'], ['a', 'b'], ordered=True, name='foo')
+    idx = pd.CategoricalIndex(['xyx'], ['xyx', 'zzz'], ordered=True, name='foo')
     res = meta_nonempty(idx)
     assert type(res) is pd.CategoricalIndex
     assert (res.categories == idx.categories).all()
@@ -233,7 +233,7 @@ def test_meta_nonempty_index():
     assert res.names == idx.names
 
     levels = [pd.Int64Index([1], name='a'),
-              pd.CategoricalIndex(data=['b'], categories=['b'], name='b'),
+              pd.CategoricalIndex(data=['xyx'], categories=['xyx'], name='b'),
               pd.TimedeltaIndex([np.timedelta64(1, 'D')], name='timedelta')]
     idx = pd.MultiIndex(levels=levels, labels=[[0], [0], [0]], names=['a', 'b', 'timedelta'])
     res = meta_nonempty(idx)
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 5d80253a2..82fb21e96 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -511,17 +511,17 @@ def test_compute_no_opt():
     keys = []
     with Callback(pretask=lambda key, *args: keys.append(key)):
         o.compute(scheduler='single-threaded', optimize_graph=False)
-    assert len([k for k in keys if '-mul-' in k[0]]) == 4
-    assert len([k for k in keys if '-add-' in k[0]]) == 4
+    assert len([k for k in keys if 'mul' in k[0]]) == 4
+    assert len([k for k in keys if 'add' in k[0]]) == 4
     # Check that without the kwarg, the optimization does happen
     keys = []
     with Callback(pretask=lambda key, *args: keys.append(key)):
         o.compute(scheduler='single-threaded')
     # Names of fused tasks have been merged, and the original key is an alias.
     # Otherwise, the lengths below would be 4 and 0.
-    assert len([k for k in keys if '-mul-' in k[0]]) == 8
-    assert len([k for k in keys if '-add-' in k[0]]) == 4
-    assert len([k for k in keys if 'add-map-mul' in k[0]]) == 4  # See? Renamed
+    assert len([k for k in keys if 'mul' in k[0]]) == 8
+    assert len([k for k in keys if 'add' in k[0]]) == 4
+    assert len([k for k in keys if 'add-from_sequence-mul' in k[0]]) == 4  # See? Renamed
 
 
 @pytest.mark.skipif('not da')