Repository: dask/dask

Issue:
Resample Dask Dataframe results in loss of index column name
I created this self containable example to demonstrate a bug I found when resampling a Dask Dataframe. As can be seen below, after resampling the Pandas Dataframe it still contains the name of the index, but after resampling the Dask Dataframe it loses the name of the index. It's interesting to note that when doing `df2.index.name` it shows the correct index name, but when doing `df2.head()` this doesn't happen. 

<details>
<summary>Code</summary>

```
import numpy as np
import pandas as pd
from numpy import sqrt
import matplotlib.pyplot as plt
import dask
import dask.dataframe as dd
from datetime import datetime, timedelta

# Create random time series
date_today = datetime.now()
days = pd.date_range(date_today, date_today + timedelta(20), freq='D')

np.random.seed(seed=1111)
data = np.random.randint(1, high=100, size=len(days))

# Create Pandas Dataframe
df = pd.DataFrame({'date': days, 'values': data})
df = df.set_index('date')
print(df.head())
print(df.resample('D').mean().head())

# Create Dask Dataframe
df2 = dd.from_pandas(df, npartitions=4)
print(df2.head())
print(df2.resample('D').mean().head())

df2.index.name

```

</details>

<details>
<summary>Output</summary>

```
                            values
date                              
2018-07-30 16:10:31.049569      29
2018-07-31 16:10:31.049569      56
2018-08-01 16:10:31.049569      82
2018-08-02 16:10:31.049569      13
2018-08-03 16:10:31.049569      35
            values
date              
2018-07-30      29
2018-07-31      56
2018-08-01      82
2018-08-02      13
2018-08-03      35
                            values
date                              
2018-07-30 16:10:31.049569      29
2018-07-31 16:10:31.049569      56
2018-08-01 16:10:31.049569      82
2018-08-02 16:10:31.049569      13
2018-08-03 16:10:31.049569      35
            values
2018-07-30      29
2018-07-31      56
2018-08-01      82
2018-08-02      13
2018-08-03      35
'date'

```

</details>

Discussion:
TomAugspurger: Does this also close https://github.com/dask/dask/issues/3660?

Patch:
diff --git a/dask/bytes/core.py b/dask/bytes/core.py
index 73b0eae33..aa395eead 100644
--- a/dask/bytes/core.py
+++ b/dask/bytes/core.py
@@ -263,6 +263,41 @@ def infer_options(urlpath):
     return urlpath, protocol, options
 
 
+def expand_paths_if_needed(paths, mode, num, fs, name_function):
+    """Expand paths if they have a ``*`` in them.
+
+    :param paths: list of paths
+    mode : str
+        Mode in which to open files.
+    num : int
+        If opening in writing mode, number of files we expect to create.
+    fs : filesystem object
+    name_function : callable
+        If opening in writing mode, this callable is used to generate path
+        names. Names are generated for each partition by
+        ``urlpath.replace('*', name_function(partition_index))``.
+    :return: list of paths
+    """
+    expanded_paths = []
+    paths = list(paths)
+    if 'w' in mode and sum([1 for p in paths if '*' in p]) > 1:
+        raise ValueError("When writing data, only one filename mask can be specified.")
+    for curr_path in paths:
+        if '*' in curr_path:
+            if 'w' in mode:
+                # expand using name_function
+                expanded_paths.extend(_expand_paths(curr_path, name_function, num))
+            else:
+                # expand using glob
+                expanded_paths.extend(fs.glob(curr_path))
+        else:
+            expanded_paths.append(curr_path)
+    # if we generated more paths that asked for, trim the list
+    if 'w' in mode and len(expanded_paths) > num:
+        expanded_paths = expanded_paths[:num]
+    return expanded_paths
+
+
 def get_fs_token_paths(urlpath, mode='rb', num=1, name_function=None,
                        storage_options=None):
     """Filesystem, deterministic token, and paths from a urlpath and options.
@@ -294,9 +329,8 @@ def get_fs_token_paths(urlpath, mode='rb', num=1, name_function=None,
             raise ValueError("When specifying a list of paths, all paths must "
                              "share the same protocol and options")
         update_storage_options(options, storage_options)
-        paths = list(paths)
-
         fs, fs_token = get_fs(protocol, options)
+        paths = expand_paths_if_needed(paths, mode, num, fs, name_function)
 
     elif isinstance(urlpath, (str, unicode)) or hasattr(urlpath, 'name'):
         urlpath, protocol, options = infer_options(urlpath)
diff --git a/dask/dataframe/tseries/resample.py b/dask/dataframe/tseries/resample.py
index 3f5b8f762..de3e15824 100644
--- a/dask/dataframe/tseries/resample.py
+++ b/dask/dataframe/tseries/resample.py
@@ -38,7 +38,8 @@ def _resample_series(series, start, end, reindex_closed, rule,
                      resample_kwargs, how, fill_value, how_args, how_kwargs):
     out = getattr(series.resample(rule, **resample_kwargs), how)(*how_args, **how_kwargs)
     return out.reindex(pd.date_range(start, end, freq=rule,
-                                     closed=reindex_closed),
+                                     closed=reindex_closed,
+                                     name=out.index.name),
                        fill_value=fill_value)
 
 


Test patch:
diff --git a/dask/bytes/tests/test_local.py b/dask/bytes/tests/test_local.py
index dbe58ae0d..8f723a624 100644
--- a/dask/bytes/tests/test_local.py
+++ b/dask/bytes/tests/test_local.py
@@ -29,6 +29,12 @@ files = {'.test.accounts.1.json': (b'{"amount": 100, "name": "Alice"}\n'
                                    b'{"amount": 800, "name": "Dennis"}\n')}
 
 
+csv_files = {'.test.fakedata.1.csv': (b'a,b\n'
+                                      b'1,2\n'),
+             '.test.fakedata.2.csv': (b'a,b\n'
+                                      b'3,4\n')}
+
+
 try:
     # used only in test_with_urls - may be more generally useful
     import pathlib
@@ -86,6 +92,27 @@ def test_urlpath_inference_errors():
                             'should/not/be.csv' 'allowed.csv'})
 
 
+def test_urlpath_expand_read():
+    """Make sure * is expanded in file paths when reading."""
+    # when reading, globs should be expanded to read files by mask
+    with filetexts(csv_files, mode='b'):
+        _, _, paths = get_fs_token_paths('.*.csv')
+        assert len(paths) == 2
+        _, _, paths = get_fs_token_paths(['.*.csv'])
+        assert len(paths) == 2
+
+
+def test_urlpath_expand_write():
+    """Make sure * is expanded in file paths when writing."""
+    _, _, paths = get_fs_token_paths('prefix-*.csv', mode='wb', num=2)
+    assert paths == ['prefix-0.csv', 'prefix-1.csv']
+    _, _, paths = get_fs_token_paths(['prefix-*.csv'], mode='wb', num=2)
+    assert paths == ['prefix-0.csv', 'prefix-1.csv']
+    # we can read with multiple masks, but not write
+    with pytest.raises(ValueError):
+        _, _, paths = get_fs_token_paths(['prefix1-*.csv', 'prefix2-*.csv'], mode='wb', num=2)
+
+
 def test_read_bytes():
     with filetexts(files, mode='b'):
         sample, values = read_bytes('.test.accounts.*')
diff --git a/dask/dataframe/tseries/tests/test_resample.py b/dask/dataframe/tseries/tests/test_resample.py
index 7c3b4f7bf..2b20f3e30 100644
--- a/dask/dataframe/tseries/tests/test_resample.py
+++ b/dask/dataframe/tseries/tests/test_resample.py
@@ -77,3 +77,19 @@ def test_unknown_divisions_error():
         assert False
     except ValueError as e:
         assert 'divisions' in str(e)
+
+
+def test_resample_index_name():
+    import numpy as np
+    from datetime import datetime, timedelta
+
+    date_today = datetime.now()
+    days = pd.date_range(date_today, date_today + timedelta(20), freq='D')
+    data = np.random.randint(1, high=100, size=len(days))
+
+    df = pd.DataFrame({'date': days, 'values': data})
+    df = df.set_index('date')
+
+    ddf = dd.from_pandas(df, npartitions=4)
+
+    assert ddf.resample('D').mean().head().index.name == "date"