Merge pull request #8 from jcrist/delayed-csv-names

Fix tokenize in dask.delayed

dask/delayed.py

@@ -106,7 +106,7 @@ def tokenize(*args, **kwargs):
         unique identifier is always used.
     """
     if kwargs.pop('pure', False):
-        return base.tokenize(*args)
+        return base.tokenize(*args, **kwargs)
     else:
         return str(uuid.uuid4())
 
@@ -218,7 +218,8 @@ def delayed(obj, name=None, pure=False):
     if not dasks:
         return DelayedLeaf(obj, pure=pure, name=name)
     else:
-        name = name or '%s-%s' % (type(obj).__name__, tokenize(task))
+        if not name:
+            name = '%s-%s' % (type(obj).__name__, tokenize(task, pure=pure))
         dasks.append({name: task})
         return Delayed(name, dasks)
 

dask/tests/test_delayed.py

@@ -144,6 +144,14 @@ def test_literates():
     assert delayed(lit)['a'].compute() == 1
 
 
+def test_literates_keys():
+    a = delayed(1)
+    b = a + 1
+    lit = (a, b, 3)
+    assert delayed(lit).key != delayed(lit).key
+    assert delayed(lit, pure=True).key == delayed(lit, pure=True).key
+
+
 def test_lists_are_concrete():
     a = delayed(1)
     b = delayed(2)
@@ -184,8 +192,13 @@ def mysum(a, b, c=(), **kwargs):
     ten = dmysum(1, 2, c=[delayed(3), 0], four=dmysum(2, 2))
     assert ten.compute() == 10
     dmysum = delayed(mysum, pure=True)
-    ten = dmysum(1, 2, c=[delayed(3), 0], four=dmysum(2, 2))
+    c = [delayed(3), 0]
+    ten = dmysum(1, 2, c=c, four=dmysum(2, 2))
     assert ten.compute() == 10
+    assert dmysum(1, 2, c=c, four=dmysum(2, 2)).key == ten.key
+    assert dmysum(1, 2, c=c, four=dmysum(2, 3)).key != ten.key
+    assert dmysum(1, 2, c=c, four=4).key != ten.key
+    assert dmysum(1, 2, c=c, four=4).key != dmysum(2, 2, c=c, four=4).key
 
 
 def test_array_delayed():