Repository: dask/dask

Issue:
`dask.visualize` does not work with `filename=None`
According to the documentation for [`dask.visualize`](https://docs.dask.org/en/latest/api.html#dask.visualize),

> **If filename is None, no file will be written**, and we communicate with dot using only pipes.

Unfortunately, it seems this actually throws an error!

```
/python3.8/site-packages/dask/dot.py in <genexpr>(.0)
    274 def graphviz_to_file(g, filename, format):
    275     fmts = [".png", ".pdf", ".dot", ".svg", ".jpeg", ".jpg"]
--> 276     if format is None and any(filename.lower().endswith(fmt) for fmt in fmts):
    277         filename, format = os.path.splitext(filename)
    278         format = format[1:].lower()

AttributeError: 'NoneType' object has no attribute 'lower'
```

Based on the documentation, it sounds like `filename=None` is meant to be supported, but [based on the tests for `dask.visualize`](https://github.com/dask/dask/blob/main/dask/tests/test_base.py#L936-L955), this case was never actually tested for and this error slipped under the radar.

Discussion:
freyam: @martindurant @GenevieveBuckley 

The issue arised when the `format` and the `filename` arguments both were `None`. So, 

https://github.com/dask/dask/blob/1a6a85258bccb14fe37bb54c211beeeffd46cd55/dask/dot.py#L276-L278

would result in an error as it's trying to run `.lower()` on a `NoneType`.

To fix this, I added an extra condition before it reaches the line of error to make sure `format` is not `None` by making it equal to the default ("png").

![image](https://user-images.githubusercontent.com/62539811/120347731-97944580-c30d-11eb-96e7-9f513e46075d.png)


freyam: What kind of tests can I add for this? Not very sure since there is no file being generated!
https://github.com/dask/dask/blob/1a6a85258bccb14fe37bb54c211beeeffd46cd55/dask/tests/test_base.py#L936-L955
IIn this case, what should I check for?

EDIT: surprisingly, this fails many test cases. Will have a look and commit the fixes.
GenevieveBuckley: This is nice work @freyam, thank you for making this!
freyam: I have pushed a separate function for the `no_filename` test and I have kept the monkeypatch mock behavior.
Is this fine? @martindurant @GenevieveBuckley 
or should I just include a one-line test and commend it accordingly?


```py
# To check if visualize function works when filename is None
# Test will fail if the function raises an error
visualize(x, filename=None)
```

freyam: Oops! I have been working on the `main` branch of my forked repository. So, while pulling changes, it got considered as a commit towards this PR.

I am not sure how I overlooked this, I will make sure to use a new branch from next time!

For now, you can consider the second last commit. That passes all the tests.

https://github.com/dask/dask/pull/7740/commits/9bdcbaa8e1323ea0b83bee5740a94628ceff1624

GenevieveBuckley: > LGTM, but someone else should pull the trigger if they are also happy.

ping @dask/maintenance 

mrocklin: A few requests:

1.  In the future, can I ask for a more informative PR title?  This is for a few reasons:
    - Very few people know all of the issue numbers, and so saying "Fixed #XXXX" doesn't help reviewers know what's going on
    - These titles become the title of the commit
2. Can I ask for an informative comment when opening the PR.  This might briefly describe the problem, and then point to the original issue for greater details.  It should then also describe the solution taken so that reviewers can get a sense for what you're trying to do without looking through your code
3. I would prefer not to use monkeypatch if we can avoid it.  At a quick glance at the test I don't know what's going on. I'm sure that it's correct, but it does require a bit more from reviewers to understand.  It's also not typically used in this codebase.
martindurant: OK, from two counts against, lets remove monkeypatch after all
freyam: @mrocklin Yes, I will definitely take care of that. Thank you so much for pointing this out! :100: 
freyam: I just made a simple one-liner test in the already existing `test_visualize()`.

Also, once this PR is merged, I will edit the topmost comment with relevant information about the issue and how I solved it.
martindurant: (I think Matt is asking that you should ensure your title is descriptive before merger)
freyam: @martindurant Oh yes! Changed :)
martindurant: I took the liberty of simplifying it a little
GenevieveBuckley: > Also, once this PR is merged, I will edit the topmost comment with relevant information about the issue and how I solved it.

You don't need to wait until after it's merged, you can go ahead and edit the top description now.
GenevieveBuckley: The pytest output is too verbose, according to GitHub actions which has truncated it. So I *think*, but do not know for sure that failing tests are mostly in dask bag and unrelated to your PR. All the tests pass locally for me, so I'm not entirely sure why our CI is unhappy.

You can re-trigger the CI by adding an [empty git commit](https://coderwall.com/p/vkdekq/git-commit-allow-empty): git commit --allow-empty -m "[EMPTY] Re-trigger CI"``
jrbourbeau: Just opened https://github.com/dask/dask/pull/7771 to run CI against the current `main` branch. That should help us determine if there are CI failures unrelated to the changes here 
freyam: @GenevieveBuckley  Yes, I have already edited the topmost comment to follow the format -> The Issue, The Fix, and The Demo.

Also, I have just submitted an empty commit. Let's see how it performs.
freyam: Looks like @jrbourbeau's suspicion was right. I will wait till #7771 is fixed before committing. 
freyam: Fingers Crosses! :crossed_fingers: 
freyam: > :red_circle: **Error: The process '/usr/share/miniconda3/condabin/mamba' failed with exit code 1**

GenevieveBuckley: Hm, I'm not sure what might have caused that mamba error. You can try the empty commit trick to check if it's just a momentary bit of weirdness (if you see the same thing twice, that might indicate a problem)
freyam: YEAH! Wow, what a roller-coaster ride this issue was! I'm glad it all worked out in the end: :rocket: Learned a lot from this issue. I will try not to repeat any of the mistakes I made this time :heavy_check_mark: 

Patch:
diff --git a/dask/dot.py b/dask/dot.py
index 14a9fbf04..54c638e10 100644
--- a/dask/dot.py
+++ b/dask/dot.py
@@ -273,7 +273,12 @@ def dot_graph(dsk, filename="mydask", format=None, **kwargs):
 
 def graphviz_to_file(g, filename, format):
     fmts = [".png", ".pdf", ".dot", ".svg", ".jpeg", ".jpg"]
-    if format is None and any(filename.lower().endswith(fmt) for fmt in fmts):
+
+    if (
+        format is None
+        and filename is not None
+        and any(filename.lower().endswith(fmt) for fmt in fmts)
+    ):
         filename, format = os.path.splitext(filename)
         format = format[1:].lower()
 
@@ -292,7 +297,7 @@ def graphviz_to_file(g, filename, format):
 
     display_cls = _get_display_cls(format)
 
-    if not filename:
+    if filename is None:
         return display_cls(data=data)
 
     full_filename = ".".join([filename, format])
diff --git a/dask/optimization.py b/dask/optimization.py
index 18640cbf0..803ad11e0 100644
--- a/dask/optimization.py
+++ b/dask/optimization.py
@@ -1,5 +1,6 @@
 import math
 import numbers
+import uuid
 from enum import Enum
 
 from . import config, core, utils
@@ -940,10 +941,12 @@ class SubgraphCallable:
 
     __slots__ = ("dsk", "outkey", "inkeys", "name")
 
-    def __init__(self, dsk, outkey, inkeys, name="subgraph_callable"):
+    def __init__(self, dsk, outkey, inkeys, name=None):
         self.dsk = dsk
         self.outkey = outkey
         self.inkeys = inkeys
+        if name is None:
+            name = f"subgraph_callable-{uuid.uuid4()}"
         self.name = name
 
     def __repr__(self):
@@ -969,4 +972,4 @@ class SubgraphCallable:
         return (SubgraphCallable, (self.dsk, self.outkey, self.inkeys, self.name))
 
     def __hash__(self):
-        return hash(tuple((self.outkey, tuple(self.inkeys), self.name)))
+        return hash(tuple((self.outkey, frozenset(self.inkeys), self.name)))


Test patch:
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index 7e735af25..d5f9a1a6d 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -954,6 +954,10 @@ def test_visualize():
         visualize(x, filename=os.path.join(d, "mydask.png"))
         assert os.path.exists(os.path.join(d, "mydask.png"))
 
+        # To see if visualize() works when the filename parameter is set to None
+        # If the function raises an error, the test will fail
+        x.visualize(filename=None)
+
 
 @pytest.mark.skipif("not da")
 @pytest.mark.skipif(
diff --git a/dask/tests/test_distributed.py b/dask/tests/test_distributed.py
index 019c46680..9925b2054 100644
--- a/dask/tests/test_distributed.py
+++ b/dask/tests/test_distributed.py
@@ -404,6 +404,28 @@ async def test_blockwise_numpy_kwargs(c, s, a, b):
     assert res == 1000
 
 
+def test_blockwise_different_optimization(c):
+    # Regression test for incorrect results due to SubgraphCallable.__eq__
+    # not correctly handling subgraphs with the same outputs and arity but
+    # different internals (GH-7632). The bug is triggered by distributed
+    # because it uses a function cache.
+    da = pytest.importorskip("dask.array")
+    np = pytest.importorskip("numpy")
+
+    u = da.from_array(np.arange(3))
+    v = da.from_array(np.array([10 + 2j, 7 - 3j, 8 + 1j]))
+    cv = v.conj()
+    x = u * cv
+    (cv,) = dask.optimize(cv)
+    y = u * cv
+    expected = np.array([0 + 0j, 7 + 3j, 16 - 2j])
+    with dask.config.set({"optimization.fuse.active": False}):
+        x_value = x.compute()
+        y_value = y.compute()
+    np.testing.assert_equal(x_value, expected)
+    np.testing.assert_equal(y_value, expected)
+
+
 @gen_cluster(client=True)
 async def test_combo_of_layer_types(c, s, a, b):
     """Check pack/unpack of a HLG that has every type of Layers!"""
diff --git a/dask/tests/test_optimization.py b/dask/tests/test_optimization.py
index add1fed55..b8c276a98 100644
--- a/dask/tests/test_optimization.py
+++ b/dask/tests/test_optimization.py
@@ -5,6 +5,7 @@ from functools import partial
 import pytest
 
 import dask
+from dask.base import tokenize
 from dask.core import get_dependencies
 from dask.local import get_sync
 from dask.optimization import (
@@ -20,6 +21,26 @@ from dask.utils import apply, partial_by_order
 from dask.utils_test import add, inc
 
 
+def _subgraph_callables_eq(self, other):
+    return (
+        type(self) is type(other)
+        and self.outkey == other.outkey
+        and set(self.inkeys) == set(other.inkeys)
+        and tokenize(self.dsk) == tokenize(other.dsk)
+    )
+
+
+@pytest.fixture
+def compare_subgraph_callables(monkeypatch):
+    """Ignore name when comparing instances of :class:`SubgraphCallable`.
+
+    They have UUID-generated names which prevents instances generated by the
+    test from comparing equal to reference values. Instead, compare the
+    embedded graph using ``tokenize``.
+    """
+    monkeypatch.setattr(SubgraphCallable, "__eq__", _subgraph_callables_eq)
+
+
 def double(x):
     return x * 2
 
@@ -1134,6 +1155,8 @@ def test_SubgraphCallable():
     assert f(1, 2) == f(1, 2)
 
     f2 = pickle.loads(pickle.dumps(f))
+    assert f2 == f
+    assert hash(f2) == hash(f)
     assert f2(1, 2) == f(1, 2)
 
 
@@ -1156,7 +1179,34 @@ def test_SubgraphCallable_with_numpy():
     assert f1 != f4
 
 
-def test_fuse_subgraphs():
+def test_SubgraphCallable_eq():
+    dsk1 = {"a": 1, "b": 2, "c": (add, "d", "e")}
+    dsk2 = {"a": (inc, 0), "b": (inc, "a"), "c": (add, "d", "e")}
+    f1 = SubgraphCallable(dsk1, "c", ["d", "e"])
+    f2 = SubgraphCallable(dsk2, "c", ["d", "e"])
+    # Different graphs must compare unequal (when no name given)
+    assert f1 != f2
+
+    # Different inputs must compare unequal
+    f3 = SubgraphCallable(dsk2, "c", ["d", "f"], name=f1.name)
+    assert f3 != f1
+
+    # Different outputs must compare unequal
+    f4 = SubgraphCallable(dsk2, "a", ["d", "e"], name=f1.name)
+    assert f4 != f1
+
+    # Reordering the inputs must not prevent equality
+    f5 = SubgraphCallable(dsk1, "c", ["e", "d"], name=f1.name)
+    assert f1 == f5
+    assert hash(f1) == hash(f5)
+
+    # Explicitly named graphs with different names must be unequal
+    unnamed1 = SubgraphCallable(dsk1, "c", ["d", "e"], name="first")
+    unnamed2 = SubgraphCallable(dsk1, "c", ["d", "e"], name="second")
+    assert unnamed1 != unnamed2
+
+
+def test_fuse_subgraphs(compare_subgraph_callables):
     dsk = {
         "x-1": 1,
         "inc-1": (inc, "x-1"),
@@ -1283,7 +1333,7 @@ def test_fuse_subgraphs():
     assert res in sols
 
 
-def test_fuse_subgraphs_linear_chains_of_duplicate_deps():
+def test_fuse_subgraphs_linear_chains_of_duplicate_deps(compare_subgraph_callables):
     dsk = {
         "x-1": 1,
         "add-1": (add, "x-1", "x-1"),