Allow tuples as sharedict keys

dask/array/utils.py

@@ -44,7 +44,7 @@ def _check_dsk(dsk):
     if not isinstance(dsk, ShareDict):
         return
 
-    assert all(isinstance(k, str) for k in dsk.dicts)
+    assert all(isinstance(k, (tuple, str)) for k in dsk.dicts)
     freqs = frequencies(concat(dsk.dicts.values()))
     non_one = {k: v for k, v in freqs.items() if v != 1}
     assert not non_one, non_one

dask/delayed.py

@@ -439,7 +439,7 @@ def call_function(func, func_token, args, kwargs, pure=None, nout=None):
     for arg, d in args_dasks:
         if isinstance(d, sharedict.ShareDict):
             dsk.update_with_key(d)
-        elif isinstance(arg, str):
+        elif isinstance(arg, (str, tuple)):
             dsk.update_with_key(d, key=arg)
         else:
             dsk.update(d)

dask/tests/test_delayed.py

@@ -9,6 +9,7 @@
 from dask import set_options, compute
 from dask.compatibility import PY2, PY3
 from dask.delayed import delayed, to_task_dask, Delayed
+from dask.utils_test import inc
 
 
 def test_to_task_dask():
@@ -442,7 +443,8 @@ def myfunc(x):
 
 
 def test_finalize_name():
-    import dask.array as da
+    da = pytest.importorskip('dask.array')
+
     x = da.ones(10, chunks=5)
     v = delayed([x])
     assert set(x.dask).issubset(v.dask)
@@ -453,3 +455,13 @@ def key(s):
         return s.split('-')[0]
 
     assert all(key(k).isalpha() for k in v.dask)
+
+
+def test_keys_from_array():
+    da = pytest.importorskip('dask.array')
+    from dask.array.utils import _check_dsk
+
+    X = da.ones((10, 10), chunks=5).to_delayed().flatten()
+    xs = [delayed(inc)(x) for x in X]
+
+    _check_dsk(xs[0].dask)

docs/source/changelog.rst

@@ -23,7 +23,7 @@ Bag
 Core
 ++++
 
--
+-  Allow tuples as sharedict keys
 
 0.15.4 / 2017-10-06
 -------------------