Repository: pydata/xarray

Issue:
distributed.Client.compute fails on DataArray
As of
- dask 2.1.0
- distributed 2.1.0
- xarray 0.12.1 or git head (didn't try older versions):

```python
>>> import xarray
>>> import distributed
>>> client = distributed.Client(set_as_default=False)
>>> ds = xarray.Dataset({'d': ('x', [1, 2])}).chunk(1)
>>> client.compute(ds).result()
<xarray.Dataset>
Dimensions:  (x: 2)
Dimensions without coordinates: x
Data variables:
    d        (x) int64 1 2

>>> client.compute(ds.d).result()
distributed.worker - WARNING -  Compute Failed
Function:  _dask_finalize
args:      ([[array([1]), array([2])]], <function Dataset._dask_postcompute at 0x316a1db70>, ([(True, <this-array>, (<function Variable._dask_finalize at 0x3168f7f28>, (<function finalize at 0x1166bb8c8>, (), ('x',), OrderedDict(), None)))], set(), {'x': 2}, None, None, None, None), 'd')
kwargs:    {}
Exception: KeyError(<this-array>)

---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
<ipython-input-8-2dbfe1b2ff17> in <module>
----> 1 client.compute(ds.d).result()

/anaconda3/lib/python3.7/site-packages/distributed/client.py in result(self, timeout)
    226         result = self.client.sync(self._result, callback_timeout=timeout, raiseit=False)
    227         if self.status == "error":
--> 228             six.reraise(*result)
    229         elif self.status == "cancelled":
    230             raise result

/anaconda3/lib/python3.7/site-packages/six.py in reraise(tp, value, tb)
    690                 value = tp()
    691             if value.__traceback__ is not tb:
--> 692                 raise value.with_traceback(tb)
    693             raise value
    694         finally:

~/PycharmProjects/xarray/xarray/core/dataarray.py in _dask_finalize()
    706     def _dask_finalize(results, func, args, name):
    707         ds = func(results, *args)
--> 708         variable = ds._variables.pop(_THIS_ARRAY)
    709         coords = ds._variables
    710         return DataArray(variable, coords, name=name, fastpath=True)
```

Discussion:
crusaderky: Ready for review and merge.
The CI failures are caused by #3174

Patch:
diff --git a/doc/whats-new.rst b/doc/whats-new.rst
index afb8a9b2..3ad2f268 100644
--- a/doc/whats-new.rst
+++ b/doc/whats-new.rst
@@ -55,8 +55,11 @@ Bug fixes
   due to a ``datetime`` issue in NumPy (:issue:`2334`).
   By `Graham Inggs <https://github.com/ginggs>`_.
 - Fixed bug in ``combine_by_coords()`` causing a `ValueError` if the input had
-  an unused dimension with coordinates which were not monotonic (:issue`3150`).
+  an unused dimension with coordinates which were not monotonic (:issue:`3150`).
   By `Tom Nicholas <http://github.com/TomNicholas>`_.
+- Fixed crash when applying ``distributed.Client.compute()`` to a DataArray
+  (:issue:`3171`). By `Guido Imperiale <https://github.com/crusaderky>`_.
+
 
 .. _whats-new.0.12.3:
 
diff --git a/xarray/core/dataarray.py b/xarray/core/dataarray.py
index f993888c..0e286133 100644
--- a/xarray/core/dataarray.py
+++ b/xarray/core/dataarray.py
@@ -300,7 +300,7 @@ class DataArray(AbstractArray, DataWithCoords):
         self,
         variable: Variable = None,
         coords=None,
-        name: Union[Hashable, None, ReprObject] = __default,
+        name: Optional[Hashable] = __default,
     ) -> 'DataArray':
         if variable is None:
             variable = self.variable
@@ -313,7 +313,7 @@ class DataArray(AbstractArray, DataWithCoords):
     def _replace_maybe_drop_dims(
             self,
             variable: Variable,
-            name: Union[str, None, utils.ReprObject] = __default
+            name: Optional[Hashable] = __default
     ) -> 'DataArray':
         if variable.dims == self.dims and variable.shape == self.shape:
             coords = self._coords.copy()
@@ -356,7 +356,7 @@ class DataArray(AbstractArray, DataWithCoords):
     def _from_temp_dataset(
         self,
         dataset: Dataset,
-        name: Union[Hashable, ReprObject] = __default
+        name: Hashable = __default
     ) -> 'DataArray':
         variable = dataset._variables.pop(_THIS_ARRAY)
         coords = dataset._variables
diff --git a/xarray/core/utils.py b/xarray/core/utils.py
index 8ce30a3c..08c911f5 100644
--- a/xarray/core/utils.py
+++ b/xarray/core/utils.py
@@ -462,12 +462,22 @@ class NDArrayMixin(NdimSizeLenMixin):
 class ReprObject:
     """Object that prints as the given value, for use with sentinel values.
     """
+    __slots__ = ('_value', )
+
     def __init__(self, value: str):
         self._value = value
 
     def __repr__(self) -> str:
         return self._value
 
+    def __eq__(self, other) -> bool:
+        if isinstance(other, ReprObject):
+            return self._value == other._value
+        return False
+
+    def __hash__(self) -> int:
+        return hash((ReprObject, self._value))
+
 
 @contextlib.contextmanager
 def close_on_error(f):


Test patch:
diff --git a/xarray/tests/test_utils.py b/xarray/tests/test_utils.py
index bcd960e4..ce4c5cc8 100644
--- a/xarray/tests/test_utils.py
+++ b/xarray/tests/test_utils.py
@@ -1,5 +1,6 @@
 from collections import OrderedDict
 from datetime import datetime
+from typing import Hashable
 
 import numpy as np
 import pandas as pd
@@ -179,6 +180,21 @@ class TestDictionaries:
 def test_repr_object():
     obj = utils.ReprObject('foo')
     assert repr(obj) == 'foo'
+    assert isinstance(obj, Hashable)
+    assert not isinstance(obj, str)
+
+
+def test_repr_object_magic_methods():
+    o1 = utils.ReprObject('foo')
+    o2 = utils.ReprObject('foo')
+    o3 = utils.ReprObject('bar')
+    o4 = 'foo'
+    assert o1 == o2
+    assert o1 != o3
+    assert o1 != o4
+    assert hash(o1) == hash(o2)
+    assert hash(o1) != hash(o3)
+    assert hash(o1) != hash(o4)
 
 
 def test_is_remote_uri():