Repository: dask/dask
Pull request #3238 (2018-03-02 05:44:22)

Fix handling of non-iterable non-string keys for task graphs
The first issue on lines 306-317 in subs is fixing the assumption that if type_arg is type(key) then arg is an ndarray, when it can actually be any non-core python type at this point in the branching.  It adds a fallback to check simple equality in case the arg is not sized or is not iterable.

The second issue is where '->'.join(cycle) will fail (in trying to create an error message) if the elements of cycle cannot be added to strings.  This is resolved by calling str(x) on each before joining.

- [Y] Tests added / passed - added test_core.py:test_subs_unexpected_hashable_key, passes all existing tests
- [N] Passes `flake8 dask` - Passes for my code, errors elsewhere
- [?] Fully documented, including `docs/source/changelog.rst` for all changes
      and one of the `docs/source/*-api.rst` files for new API - Not sure what changes need to be made

mrocklin: In principle this seems safe to me.  cc'ing @eriknw though for his thoughts on `subs`.

The test failures are odd.  I'm restrating travis-ci in case it's an issue on their end.  

I recommend adding an entry to docs/source/changelog.rst under the most recent `Core` subheading.  Hopefully by following the pattern in that document things should be clear.  If not then please don't hesitate to ask.
dancollins34: @mrocklin It looks like travis is still failing on some sort of network error. Any ideas? It doesn't look related to something I did?

Ill put an entry in the changelog right now
mrocklin: Yeah, I'm not sure what's up with the Python 2 failure.  Hopefully by pushing the new commit things will be fine.  The Python 3 failures seem genuine though.
dancollins34: In what way? I looked through the 3.5 one, and it looked like an IO failure in bag.  Am I missing something?  I didn't have this issue on my local machine, but ill build with 3.5 and try

mrocklin: Ah, perhaps I was looking at 2.7.  Anyway:

```python
=================================== FAILURES ===================================
______________________ test_subs_unexpected_hashable_key _______________________
[gw2] linux2 -- Python 2.7.14 /home/travis/miniconda/envs/test-environment/bin/python
    def test_subs_unexpected_hashable_key():
        class UnexpectedButHashable:
            def __init__(self):
                self.name = "a"
    
            def __hash__(self):
                return hash(self.name)
    
            def __eq__(self, other):
                return isinstance(other, UnexpectedButHashable)
    
>       assert subs((id, UnexpectedButHashable()), UnexpectedButHashable(), 1) == (id, 1)
dask/tests/test_core.py:235: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
task = (<built-in function id>, <dask.tests.test_core.UnexpectedButHashable instance at 0x7f268a87db00>)
key = <dask.tests.test_core.UnexpectedButHashable instance at 0x7f26767f7e60>
val = 1
    def subs(task, key, val):
        """ Perform a substitution on a task
    
        Examples
        --------
    
        >>> subs((inc, 'x'), 'x', 1)  # doctest: +SKIP
        (inc, 1)
        """
        type_task = type(task)
        if not (type_task is tuple and task and callable(task[0])):  # istask(task):
            try:
                if type_task is type(key) and task == key:
                    return val
            except Exception:
                pass
            if type_task is list:
                return [subs(x, key, val) for x in task]
            return task
        newargs = []
        for arg in task[1:]:
            type_arg = type(arg)
            if type_arg is tuple and arg and callable(arg[0]):  # istask(task):
                arg = subs(arg, key, val)
            elif type_arg is list:
                arg = [subs(x, key, val) for x in arg]
            elif type_arg is type(key):
                try:
                    # Can't do a simple equality check, since this may trigger
                    # a FutureWarning from NumPy about array equality
                    # https://github.com/dask/dask/pull/2457
>                   if len(arg) == len(key) and all(type(aa) == type(bb) and aa == bb
                                                    for aa, bb in zip(arg, key)):
E                                                   AttributeError: UnexpectedButHashable instance has no attribute '__len__'
dask/core.py:311: AttributeError
```
mrocklin: @martindurant any reason this might pop up?

```python
___________________________________ test_bag ___________________________________
[gw1] linux -- Python 3.5.4 /home/travis/miniconda/envs/test-environment/bin/python
    @pytest.mark.network
    def test_bag():
        # This test pulls from different hosts
        db = pytest.importorskip('dask.bag')
        urls = ['https://raw.githubusercontent.com/weierophinney/pastebin/'
                'master/public/js-src/dojox/data/tests/stores/patterns.csv',
                'https://en.wikipedia.org']
        b = db.read_text(urls)
        assert b.npartitions == 2
>       b.compute()
dask/bytes/tests/test_http.py:135: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dask/base.py:155: in compute
    (result,) = compute(self, traverse=False, **kwargs)
dask/base.py:404: in compute
    results = get(dsk, keys, **kwargs)
dask/multiprocessing.py:177: in get
    raise_exception=reraise, **kwargs)
dask/local.py:521: in get_async
    raise_exception(exc, tb)
dask/compatibility.py:66: in reraise
    raise exc.with_traceback(tb)
dask/local.py:290: in execute_task
    result = _execute_task(task, data)
dask/local.py:271: in _execute_task
    return func(*args2)
dask/bag/core.py:1502: in reify
    seq = list(seq)
dask/bag/text.py:88: in file_to_blocks
    for line in f:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
>   raise ValueError('File size is unknown, must read all data')
E   ValueError: File size is unknown, must read all data
```
dancollins34: @mrocklin It looks from the stacktrace like len() throws a type error in python 2 but an attribute error in python 3?  I'd think since it was just a fallback we could check for Exception, but I don't know what your thoughts on that are.  Alternatively we could check for TypeError and AttributeError
mrocklin: I would check for (TypeError, AttributeError)

On Sat, Mar 3, 2018 at 8:10 PM, dancollins34 <notifications@github.com>
wrote:

> @mrocklin <https://github.com/mrocklin> It looks from the stacktrace like
> len() throws a type error in python 2 but an attribute error in python 3?
> I'd think since it was just a fallback we could check for Exception, but I
> don't know what your thoughts on that are. Alternatively we could check for
> TypeError and AttributeError
>
> —
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/pull/3238#issuecomment-370193349>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/AASszPcvPYZ2bEExImfAIt-lTf3iCchvks5taz77gaJpZM4SZbkX>
> .
>

dancollins34: @mrocklin Okay, Ill change it to check for (TypeError, AttributeError), once I finish running the tests in a python 3.5 environment locally to check if the bag test was likely a travis issue.
dancollins34: Actually looks like it was a 3.4 -> 3.5 change, just pushed the two error check, hopefully that passes.
martindurant: > @martindurant any reason this might pop up?

The CSV is fixed in size, but I can suppose the wiki front page could sometimes become bigger than the requested block-size (5MB)? `db.read_text(urls)` could set a `blocksize` larger than anything we could reasonably expect to get around that, there is no option to always read the whole file in one go.
dancollins34: @mrocklin fixed
mrocklin: Excellent.  Merging.  Thank you for the work on this @dancollins34 !

Patch:
diff --git a/dask/core.py b/dask/core.py
index e5e7d41fa..5cc3b8281 100644
--- a/dask/core.py
+++ b/dask/core.py
@@ -304,12 +304,18 @@ def subs(task, key, val):
         elif type_arg is list:
             arg = [subs(x, key, val) for x in arg]
         elif type_arg is type(key):
-            # Can't do a simple equality check, since this may trigger
-            # a FutureWarning from NumPy about array equality
-            # https://github.com/dask/dask/pull/2457
-            if len(arg) == len(key) and all(type(aa) == type(bb) and aa == bb
-                                            for aa, bb in zip(arg, key)):
-                arg = val
+            try:
+                # Can't do a simple equality check, since this may trigger
+                # a FutureWarning from NumPy about array equality
+                # https://github.com/dask/dask/pull/2457
+                if len(arg) == len(key) and all(type(aa) == type(bb) and aa == bb
+                                                for aa, bb in zip(arg, key)):
+                    arg = val
+
+            except (TypeError, AttributeError):
+                # Handle keys which are not sized (len() fails), but are hashable
+                if arg == key:
+                    arg = val
         newargs.append(arg)
     return task[:1] + tuple(newargs)
 
@@ -365,7 +371,7 @@ def _toposort(dsk, keys=None, returncycle=False, dependencies=None):
                         if returncycle:
                             return cycle
                         else:
-                            cycle = '->'.join(cycle)
+                            cycle = '->'.join(str(x) for x in cycle)
                             raise RuntimeError('Cycle detected in Dask: %s' % cycle)
                     next_nodes.append(nxt)
 
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 3a151bddf..a1d79f716 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -42,6 +42,8 @@ Bag
 Core
 ++++
 
+- Fixed bug when using unexpected but hashable types for keys (:pr:`3238`) `Daniel Collins`_
+
 
 0.17.1 / 2018-02-22
 -------------------
@@ -1025,3 +1027,4 @@ Other
 .. _`Max Epstein`: https://github.com/MaxPowerWasTaken
 .. _`Simon Perkins`: https://github.com/sjperkins
 .. _`Richard Postelnik`: https://github.com/postelrich
+.. _`Daniel Collins`: https://github.com/dancollins34


Test patch:
diff --git a/dask/tests/test_core.py b/dask/tests/test_core.py
index d53291917..b8999f3f9 100644
--- a/dask/tests/test_core.py
+++ b/dask/tests/test_core.py
@@ -221,6 +221,20 @@ def test_subs_with_surprisingly_friendly_eq():
         assert subs(df, 'x', 1) is df
 
 
+def test_subs_unexpected_hashable_key():
+    class UnexpectedButHashable:
+        def __init__(self):
+            self.name = "a"
+
+        def __hash__(self):
+            return hash(self.name)
+
+        def __eq__(self, other):
+            return isinstance(other, UnexpectedButHashable)
+
+    assert subs((id, UnexpectedButHashable()), UnexpectedButHashable(), 1) == (id, 1)
+
+
 def test_quote():
     literals = [[1, 2, 3], (add, 1, 2),
                 [1, [2, 3]], (add, 1, (add, 2, 3))]