Repository: dask/dask

Issue:
Add "empty" mode for `da.pad`
NumPy 1.17.0 adds a new mode to `np.pad` called "empty", which pads the provided array with uninitialized arrays (like `np.empty`). Would be good to add this to `da.pad` as well.

Discussion:
quasiben: I see this is  your first time contributing, welcome!

@thomasjpfan this looks really great! There is currently a test failure due to incompatibility with older versions of NumPy.  You can skip the skip the test like we do here:

https://github.com/dask/dask/blob/54deb7fa4cdbbf3e124be6834224272be27ace65/dask/array/tests/test_routines.py#L812-L814
thomasjpfan: Thank you for looking at this so quickly!

Updated PR with a `pytest.mark.skipif` in the `parametrize` to skip if the numpy version is not >1.17
quasiben: Thanks for updating so quickly!  As I'm not very familiar with the array side of dask I'm going to wait until tomorrow to merge giving other reviewers some time to comment

Patch:
diff --git a/continuous_integration/appveyor/environment.yaml b/continuous_integration/appveyor/environment.yaml
index 6b0102296..7c1dd0b63 100644
--- a/continuous_integration/appveyor/environment.yaml
+++ b/continuous_integration/appveyor/environment.yaml
@@ -10,7 +10,6 @@ dependencies:
   - pytest
   - cloudpickle
   - distributed
-  - msgpack-python==0.6.2
   - toolz
   - bcolz
   - bokeh
diff --git a/continuous_integration/travis/travis-36.yaml b/continuous_integration/travis/travis-36.yaml
index e21d2c8d6..45078fffb 100644
--- a/continuous_integration/travis/travis-36.yaml
+++ b/continuous_integration/travis/travis-36.yaml
@@ -34,8 +34,7 @@ dependencies:
   - cloudpickle
   - crick
   - cytoolz
-  - distributed==2.0.1
-  - msgpack-python==0.6.2
+  - distributed
   - graphviz
   - ipython
   - lz4
diff --git a/continuous_integration/travis/travis-37.yaml b/continuous_integration/travis/travis-37.yaml
index 4a134554d..d6bcebef2 100644
--- a/continuous_integration/travis/travis-37.yaml
+++ b/continuous_integration/travis/travis-37.yaml
@@ -32,8 +32,7 @@ dependencies:
   - cloudpickle
   - crick
   - cytoolz
-  - distributed==2.0.1
-  - msgpack-python==0.6.2
+  - distributed
   - graphviz
   - ipython
   - lz4
diff --git a/continuous_integration/travis/travis-38-dev.yaml b/continuous_integration/travis/travis-38-dev.yaml
index 16fab121f..070f33d2e 100644
--- a/continuous_integration/travis/travis-38-dev.yaml
+++ b/continuous_integration/travis/travis-38-dev.yaml
@@ -32,7 +32,7 @@ dependencies:
   - cloudpickle
   - crick
   - cytoolz
-  - distributed==2.0.1
+  - distributed
   - graphviz
   - ipython
   - lz4
diff --git a/continuous_integration/travis/travis-38.yaml b/continuous_integration/travis/travis-38.yaml
index 86f32d5f8..34edf05b2 100644
--- a/continuous_integration/travis/travis-38.yaml
+++ b/continuous_integration/travis/travis-38.yaml
@@ -32,8 +32,7 @@ dependencies:
   - cloudpickle
   - crick
   - cytoolz
-  - distributed==2.0.1
-  - msgpack-python==0.6.2
+  - distributed
   - graphviz
   - ipython
   - lz4
diff --git a/dask/array/creation.py b/dask/array/creation.py
index 963056c7a..b53a16628 100644
--- a/dask/array/creation.py
+++ b/dask/array/creation.py
@@ -945,6 +945,11 @@ def pad_edge(array, pad_width, mode, *args):
                         zip(pad_arrays, end_values, pad_width[d], pad_chunks)
                     )
                 ]
+        elif mode == "empty":
+            pad_arrays = [
+                empty(s, dtype=array.dtype, chunks=c)
+                for s, c in zip(pad_shapes, pad_chunks)
+            ]
 
         result = concatenate([pad_arrays[0], result, pad_arrays[1]], axis=d)
 
@@ -1119,7 +1124,7 @@ def pad(array, pad_width, mode, **kwargs):
         kwargs.setdefault("end_values", 0)
     elif mode in ["reflect", "symmetric"]:
         kwargs.setdefault("reflect_type", "even")
-    elif mode in ["edge", "wrap"]:
+    elif mode in ["edge", "wrap", "empty"]:
         if kwargs:
             raise TypeError("Got unsupported keyword arguments.")
     elif callable(mode):
@@ -1132,7 +1137,7 @@ def pad(array, pad_width, mode, **kwargs):
 
     if mode in ["maximum", "mean", "median", "minimum"]:
         return pad_stats(array, pad_width, mode, *kwargs.values())
-    elif mode in ["constant", "edge", "linear_ramp"]:
+    elif mode in ["constant", "edge", "linear_ramp", "empty"]:
         return pad_edge(array, pad_width, mode, *kwargs.values())
     elif mode in ["reflect", "symmetric", "wrap"]:
         return pad_reuse(array, pad_width, mode, *kwargs.values())
diff --git a/dask/base.py b/dask/base.py
index 1612b8408..e3ac1e9d4 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -683,6 +683,11 @@ def normalize_literal(lit):
     return "literal", normalize_token(lit())
 
 
+@normalize_token.register(range)
+def normalize_range(r):
+    return list(map(normalize_token, [r.start, r.stop, r.step]))
+
+
 @normalize_token.register(object)
 def normalize_object(o):
     method = getattr(o, "__dask_tokenize__", None)
diff --git a/docs/source/dataframe-groupby.rst b/docs/source/dataframe-groupby.rst
index 46422b1c5..3156f3eed 100644
--- a/docs/source/dataframe-groupby.rst
+++ b/docs/source/dataframe-groupby.rst
@@ -207,3 +207,16 @@ Finally, we create and use the aggregation
    a
    a  2
    b  4
+
+Another example of a custom aggregation is the Dask DataFrame version of 
+Pandas' ``groupby('g').agg(list)``:
+
+.. code-block:: python
+
+   >>> import itertools as it
+   >>> collect_list = dd.Aggregation(
+   ...     name="collect_list",
+   ...     chunk=lambda s: s.apply(list),
+   ...     agg=lambda s0: s0.apply(lambda chunks: list(it.chain.from_iterable(chunks))),
+   ... )
+   >>> df.groupby('g').agg(collect_list)


Test patch:
diff --git a/dask/array/tests/test_creation.py b/dask/array/tests/test_creation.py
index 64eda3673..2f73f8405 100644
--- a/dask/array/tests/test_creation.py
+++ b/dask/array/tests/test_creation.py
@@ -640,6 +640,16 @@ skip_stat_length = pytest.mark.xfail(_numpy_117, reason="numpy-14061")
         ((10, 11), (4, 5), 0, "reflect", {}),
         ((10, 11), (4, 5), 0, "symmetric", {}),
         ((10, 11), (4, 5), 0, "wrap", {}),
+        pytest.param(
+            (10, 11),
+            (4, 5),
+            0,
+            "empty",
+            {},
+            marks=pytest.mark.skipif(
+                not _numpy_117, reason="requires NumPy>=1.17 for empty mode support"
+            ),
+        ),
     ],
 )
 def test_pad_0_width(shape, chunks, pad_width, mode, kwargs):
@@ -683,6 +693,16 @@ def test_pad_0_width(shape, chunks, pad_width, mode, kwargs):
         ((10,), (3,), ((2, 3)), "maximum", {"stat_length": (1, 2)}),
         ((10, 11), (4, 5), ((1, 4), (2, 3)), "mean", {"stat_length": ((3, 4), (2, 1))}),
         ((10,), (3,), ((2, 3)), "minimum", {"stat_length": (2, 3)}),
+        pytest.param(
+            (10,),
+            (3,),
+            1,
+            "empty",
+            {},
+            marks=pytest.mark.skipif(
+                not _numpy_117, reason="requires NumPy>=1.17 for empty mode support"
+            ),
+        ),
     ],
 )
 def test_pad(shape, chunks, pad_width, mode, kwargs):
@@ -692,7 +712,11 @@ def test_pad(shape, chunks, pad_width, mode, kwargs):
     np_r = np.pad(np_a, pad_width, mode, **kwargs)
     da_r = da.pad(da_a, pad_width, mode, **kwargs)
 
-    assert_eq(np_r, da_r)
+    if mode == "empty":
+        # empty pads lead to undefined values which may be different
+        assert_eq(np_r[pad_width:-pad_width], da_r[pad_width:-pad_width])
+    else:
+        assert_eq(np_r, da_r)
 
 
 @pytest.mark.parametrize("kwargs", [{}, {"scaler": 2}])
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 663a77846..6f96e529d 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -394,6 +394,13 @@ def test_tokenize_ordered_dict():
     assert tokenize(a) != tokenize(c)
 
 
+def test_tokenize_range():
+    assert tokenize(range(5, 10, 2)) == tokenize(range(5, 10, 2))  # Identical ranges
+    assert tokenize(range(5, 10, 2)) != tokenize(range(1, 10, 2))  # Different start
+    assert tokenize(range(5, 10, 2)) != tokenize(range(5, 15, 2))  # Different stop
+    assert tokenize(range(5, 10, 2)) != tokenize(range(5, 10, 1))  # Different step
+
+
 @pytest.mark.skipif("not np")
 def test_tokenize_object_array_with_nans():
     a = np.array([u"foo", u"Jos\xe9", np.nan], dtype="O")
diff --git a/dask/tests/test_dot.py b/dask/tests/test_dot.py
index 0fb245d40..a069878cc 100644
--- a/dask/tests/test_dot.py
+++ b/dask/tests/test_dot.py
@@ -217,11 +217,10 @@ def test_dot_graph_defaults():
         ),
     ],
 )
-def test_filenames_and_formats(filename, format, target, expected_result_type):
-    result = dot_graph(dsk, filename=filename, format=format)
-    assert os.path.isfile(target)
+def test_filenames_and_formats(tmpdir, filename, format, target, expected_result_type):
+    result = dot_graph(dsk, filename=str(tmpdir.join(filename)), format=format)
+    assert tmpdir.join(target).exists()
     assert isinstance(result, expected_result_type)
-    ensure_not_exists(target)
 
 
 def test_delayed_kwargs_apply():