Repository: pydata/xarray

Issue:
Three test_aggregation[int-method_median] tests failing
Follow-up of #3777.

The three failing tests seems to be failing because `dask_array=None`, which is likely the same kind of issue as #3777: dask-dependant tests ran while dask is not available. The other printed error is strange to me, because `numpy` is at version 1.18.1 on this system.

```
_______________ TestVariable.test_aggregation[int-method_median] _______________

values = array([0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), axis = None, skipna = None
kwargs = {}, func = <function _dask_or_eager_func.<locals>.f at 0x7f3927bdbe50>
msg = 'median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None'

    def f(values, axis=None, skipna=None, **kwargs):
        if kwargs.pop("out", None) is not None:
            raise TypeError(f"`out` is not valid for {name}")
    
        values = asarray(values)
    
        if coerce_strings and values.dtype.kind in "SU":
            values = values.astype(object)
    
        func = None
        if skipna or (skipna is None and values.dtype.kind in "cfO"):
            nanname = "nan" + name
            func = getattr(nanops, nanname)
        else:
            func = _dask_or_eager_func(name, dask_module=dask_module)
    
        try:
>           return func(values, axis=axis, **kwargs)

xarray/core/duck_array_ops.py:307: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (array([0, 0, 0, 0, 0, 0, 0, 0, 0, 1]),), kwargs = {'axis': None}
dispatch_args = (array([0, 0, 0, 0, 0, 0, 0, 0, 0, 1]),)

    def f(*args, **kwargs):
        if list_of_args:
            dispatch_args = args[0]
        else:
            dispatch_args = args[array_args]
>       if any(isinstance(a, dask_array.Array) for a in dispatch_args):

xarray/core/duck_array_ops.py:40: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <tuple_iterator object at 0x7f3927966f70>

>   if any(isinstance(a, dask_array.Array) for a in dispatch_args):
E   AttributeError: 'NoneType' object has no attribute 'Array'

xarray/core/duck_array_ops.py:40: AttributeError

During handling of the above exception, another exception occurred:

self = <xarray.tests.test_units.TestVariable object at 0x7f3927966640>
func = method_median, dtype = <class 'int'>

    @pytest.mark.parametrize(
        "func",
        (
            method("all"),
            method("any"),
            method("argmax"),
            method("argmin"),
            method("argsort"),
            method("cumprod"),
            method("cumsum"),
            method("max"),
            method("mean"),
            method("median"),
            method("min"),
            pytest.param(
                method("prod"),
                marks=pytest.mark.xfail(reason="not implemented by pint"),
            ),
            method("std"),
            method("sum"),
            method("var"),
        ),
        ids=repr,
    )
    def test_aggregation(self, func, dtype):
        array = np.linspace(0, 1, 10).astype(dtype) * (
            unit_registry.m if func.name != "cumprod" else unit_registry.dimensionless
        )
        variable = xr.Variable("x", array)
    
        units = extract_units(func(array))
>       expected = attach_units(func(strip_units(variable)), units)

xarray/tests/test_units.py:1389: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
xarray/tests/test_units.py:374: in __call__
    return func(*all_args, **all_kwargs)
xarray/core/common.py:46: in wrapped_func
    return self.reduce(func, dim, axis, skipna=skipna, **kwargs)
xarray/core/variable.py:1537: in reduce
    data = func(input_data, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

values = array([0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), axis = None, skipna = None
kwargs = {}, func = <function _dask_or_eager_func.<locals>.f at 0x7f3927bdbe50>
msg = 'median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None'

    def f(values, axis=None, skipna=None, **kwargs):
        if kwargs.pop("out", None) is not None:
            raise TypeError(f"`out` is not valid for {name}")
    
        values = asarray(values)
    
        if coerce_strings and values.dtype.kind in "SU":
            values = values.astype(object)
    
        func = None
        if skipna or (skipna is None and values.dtype.kind in "cfO"):
            nanname = "nan" + name
            func = getattr(nanops, nanname)
        else:
            func = _dask_or_eager_func(name, dask_module=dask_module)
    
        try:
            return func(values, axis=axis, **kwargs)
        except AttributeError:
            if isinstance(values, dask_array_type):
                try:  # dask/dask#3133 dask sometimes needs dtype argument
                    # if func does not accept dtype, then raises TypeError
                    return func(values, axis=axis, dtype=values.dtype, **kwargs)
                except (AttributeError, TypeError):
                    msg = "%s is not yet implemented on dask arrays" % name
            else:
                msg = (
                    "%s is not available with skipna=False with the "
                    "installed version of numpy; upgrade to numpy 1.12 "
                    "or newer to use skipna=True or skipna=None" % name
                )
>           raise NotImplementedError(msg)
E           NotImplementedError: median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None

xarray/core/duck_array_ops.py:321: NotImplementedError
______________ TestDataArray.test_aggregation[int-method_median] _______________

values = array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]), axis = None, skipna = None
kwargs = {}, func = <function _dask_or_eager_func.<locals>.f at 0x7f39286dbc10>
msg = 'median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None'

    def f(values, axis=None, skipna=None, **kwargs):
        if kwargs.pop("out", None) is not None:
            raise TypeError(f"`out` is not valid for {name}")
    
        values = asarray(values)
    
        if coerce_strings and values.dtype.kind in "SU":
            values = values.astype(object)
    
        func = None
        if skipna or (skipna is None and values.dtype.kind in "cfO"):
            nanname = "nan" + name
            func = getattr(nanops, nanname)
        else:
            func = _dask_or_eager_func(name, dask_module=dask_module)
    
        try:
>           return func(values, axis=axis, **kwargs)

xarray/core/duck_array_ops.py:307: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]),), kwargs = {'axis': None}
dispatch_args = (array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]),)

    def f(*args, **kwargs):
        if list_of_args:
            dispatch_args = args[0]
        else:
            dispatch_args = args[array_args]
>       if any(isinstance(a, dask_array.Array) for a in dispatch_args):

xarray/core/duck_array_ops.py:40: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <tuple_iterator object at 0x7f39288ad880>

>   if any(isinstance(a, dask_array.Array) for a in dispatch_args):
E   AttributeError: 'NoneType' object has no attribute 'Array'

xarray/core/duck_array_ops.py:40: AttributeError

During handling of the above exception, another exception occurred:

self = <xarray.tests.test_units.TestDataArray object at 0x7f39288ad190>
func = method_median, dtype = <class 'int'>

    @pytest.mark.parametrize(
        "func",
        (
            pytest.param(
                function("all"),
                marks=pytest.mark.xfail(reason="not implemented by pint yet"),
            ),
            pytest.param(
                function("any"),
                marks=pytest.mark.xfail(reason="not implemented by pint yet"),
            ),
            function("argmax"),
            function("argmin"),
            function("max"),
            function("mean"),
            pytest.param(
                function("median"),
                marks=pytest.mark.xfail(reason="not implemented by xarray"),
            ),
            function("min"),
            pytest.param(
                function("prod"),
                marks=pytest.mark.xfail(reason="not implemented by pint yet"),
            ),
            function("sum"),
            function("std"),
            function("var"),
            function("cumsum"),
            pytest.param(
                function("cumprod"),
                marks=pytest.mark.xfail(reason="not implemented by pint yet"),
            ),
            pytest.param(
                method("all"),
                marks=pytest.mark.xfail(reason="not implemented by pint yet"),
            ),
            pytest.param(
                method("any"),
                marks=pytest.mark.xfail(reason="not implemented by pint yet"),
            ),
            method("argmax"),
            method("argmin"),
            method("max"),
            method("mean"),
            method("median"),
            method("min"),
            pytest.param(
                method("prod"),
                marks=pytest.mark.xfail(
                    reason="comparison of quantity with ndarrays in nanops not implemented"
                ),
            ),
            method("sum"),
            method("std"),
            method("var"),
            method("cumsum"),
            pytest.param(
                method("cumprod"),
                marks=pytest.mark.xfail(reason="pint does not implement cumprod yet"),
            ),
        ),
        ids=repr,
    )
    def test_aggregation(self, func, dtype):
        array = np.arange(10).astype(dtype) * (
            unit_registry.m if func.name != "cumprod" else unit_registry.dimensionless
        )
        data_array = xr.DataArray(data=array, dims="x")
    
        # units differ based on the applied function, so we need to
        # first compute the units
        units = extract_units(func(array))
>       expected = attach_units(func(strip_units(data_array)), units)

xarray/tests/test_units.py:2226: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
xarray/tests/test_units.py:374: in __call__
    return func(*all_args, **all_kwargs)
xarray/core/common.py:46: in wrapped_func
    return self.reduce(func, dim, axis, skipna=skipna, **kwargs)
xarray/core/dataarray.py:2235: in reduce
    var = self.variable.reduce(func, dim, axis, keep_attrs, keepdims, **kwargs)
xarray/core/variable.py:1537: in reduce
    data = func(input_data, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

values = array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]), axis = None, skipna = None
kwargs = {}, func = <function _dask_or_eager_func.<locals>.f at 0x7f39286dbc10>
msg = 'median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None'

    def f(values, axis=None, skipna=None, **kwargs):
        if kwargs.pop("out", None) is not None:
            raise TypeError(f"`out` is not valid for {name}")
    
        values = asarray(values)
    
        if coerce_strings and values.dtype.kind in "SU":
            values = values.astype(object)
    
        func = None
        if skipna or (skipna is None and values.dtype.kind in "cfO"):
            nanname = "nan" + name
            func = getattr(nanops, nanname)
        else:
            func = _dask_or_eager_func(name, dask_module=dask_module)
    
        try:
            return func(values, axis=axis, **kwargs)
        except AttributeError:
            if isinstance(values, dask_array_type):
                try:  # dask/dask#3133 dask sometimes needs dtype argument
                    # if func does not accept dtype, then raises TypeError
                    return func(values, axis=axis, dtype=values.dtype, **kwargs)
                except (AttributeError, TypeError):
                    msg = "%s is not yet implemented on dask arrays" % name
            else:
                msg = (
                    "%s is not available with skipna=False with the "
                    "installed version of numpy; upgrade to numpy 1.12 "
                    "or newer to use skipna=True or skipna=None" % name
                )
>           raise NotImplementedError(msg)
E           NotImplementedError: median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None

xarray/core/duck_array_ops.py:321: NotImplementedError
_______________ TestDataset.test_aggregation[int-method_median] ________________

values = <Quantity([0 0 0 0 0 0 0 0 0 1], 'pascal')>, axis = 0, skipna = None
kwargs = {}, func = <function _dask_or_eager_func.<locals>.f at 0x7f392619b820>
msg = 'median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None'

    def f(values, axis=None, skipna=None, **kwargs):
        if kwargs.pop("out", None) is not None:
            raise TypeError(f"`out` is not valid for {name}")
    
        values = asarray(values)
    
        if coerce_strings and values.dtype.kind in "SU":
            values = values.astype(object)
    
        func = None
        if skipna or (skipna is None and values.dtype.kind in "cfO"):
            nanname = "nan" + name
            func = getattr(nanops, nanname)
        else:
            func = _dask_or_eager_func(name, dask_module=dask_module)
    
        try:
>           return func(values, axis=axis, **kwargs)

xarray/core/duck_array_ops.py:307: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = (<Quantity([0 0 0 0 0 0 0 0 0 1], 'pascal')>,), kwargs = {'axis': 0}
dispatch_args = (<Quantity([0 0 0 0 0 0 0 0 0 1], 'pascal')>,)

    def f(*args, **kwargs):
        if list_of_args:
            dispatch_args = args[0]
        else:
            dispatch_args = args[array_args]
>       if any(isinstance(a, dask_array.Array) for a in dispatch_args):

xarray/core/duck_array_ops.py:40: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

.0 = <tuple_iterator object at 0x7f39269995e0>

>   if any(isinstance(a, dask_array.Array) for a in dispatch_args):
E   AttributeError: 'NoneType' object has no attribute 'Array'

xarray/core/duck_array_ops.py:40: AttributeError

During handling of the above exception, another exception occurred:

self = <xarray.tests.test_units.TestDataset object at 0x7f3927adf880>
func = method_median, dtype = <class 'int'>

    @pytest.mark.parametrize(
        "func",
        (
            pytest.param(
                function("all"),
                marks=pytest.mark.xfail(reason="not implemented by pint"),
            ),
            pytest.param(
                function("any"),
                marks=pytest.mark.xfail(reason="not implemented by pint"),
            ),
            function("argmax"),
            function("argmin"),
            function("max"),
            function("min"),
            function("mean"),
            pytest.param(
                function("median"),
                marks=pytest.mark.xfail(
                    reason="np.median does not work with dataset yet"
                ),
            ),
            function("sum"),
            pytest.param(
                function("prod"),
                marks=pytest.mark.xfail(reason="not implemented by pint"),
            ),
            function("std"),
            function("var"),
            function("cumsum"),
            pytest.param(
                function("cumprod"),
                marks=pytest.mark.xfail(reason="fails within xarray"),
            ),
            pytest.param(
                method("all"), marks=pytest.mark.xfail(reason="not implemented by pint")
            ),
            pytest.param(
                method("any"), marks=pytest.mark.xfail(reason="not implemented by pint")
            ),
            method("argmax"),
            method("argmin"),
            method("max"),
            method("min"),
            method("mean"),
            method("median"),
            method("sum"),
            pytest.param(
                method("prod"),
                marks=pytest.mark.xfail(reason="not implemented by pint"),
            ),
            method("std"),
            method("var"),
            method("cumsum"),
            pytest.param(
                method("cumprod"), marks=pytest.mark.xfail(reason="fails within xarray")
            ),
        ),
        ids=repr,
    )
    def test_aggregation(self, func, dtype):
        unit_a = (
            unit_registry.Pa if func.name != "cumprod" else unit_registry.dimensionless
        )
        unit_b = (
            unit_registry.kg / unit_registry.m ** 3
            if func.name != "cumprod"
            else unit_registry.dimensionless
        )
        a = xr.DataArray(data=np.linspace(0, 1, 10).astype(dtype) * unit_a, dims="x")
        b = xr.DataArray(data=np.linspace(-1, 0, 10).astype(dtype) * unit_b, dims="x")
        x = xr.DataArray(data=np.arange(10).astype(dtype) * unit_registry.m, dims="x")
        y = xr.DataArray(
            data=np.arange(10, 20).astype(dtype) * unit_registry.s, dims="x"
        )
    
        ds = xr.Dataset(data_vars={"a": a, "b": b}, coords={"x": x, "y": y})
    
>       actual = func(ds)

xarray/tests/test_units.py:3733: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
xarray/tests/test_units.py:374: in __call__
    return func(*all_args, **all_kwargs)
xarray/core/common.py:83: in wrapped_func
    return self.reduce(
xarray/core/dataset.py:4230: in reduce
    variables[name] = var.reduce(
xarray/core/variable.py:1535: in reduce
    data = func(input_data, axis=axis, **kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

values = <Quantity([0 0 0 0 0 0 0 0 0 1], 'pascal')>, axis = 0, skipna = None
kwargs = {}, func = <function _dask_or_eager_func.<locals>.f at 0x7f392619b820>
msg = 'median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None'

    def f(values, axis=None, skipna=None, **kwargs):
        if kwargs.pop("out", None) is not None:
            raise TypeError(f"`out` is not valid for {name}")
    
        values = asarray(values)
    
        if coerce_strings and values.dtype.kind in "SU":
            values = values.astype(object)
    
        func = None
        if skipna or (skipna is None and values.dtype.kind in "cfO"):
            nanname = "nan" + name
            func = getattr(nanops, nanname)
        else:
            func = _dask_or_eager_func(name, dask_module=dask_module)
    
        try:
            return func(values, axis=axis, **kwargs)
        except AttributeError:
            if isinstance(values, dask_array_type):
                try:  # dask/dask#3133 dask sometimes needs dtype argument
                    # if func does not accept dtype, then raises TypeError
                    return func(values, axis=axis, dtype=values.dtype, **kwargs)
                except (AttributeError, TypeError):
                    msg = "%s is not yet implemented on dask arrays" % name
            else:
                msg = (
                    "%s is not available with skipna=False with the "
                    "installed version of numpy; upgrade to numpy 1.12 "
                    "or newer to use skipna=True or skipna=None" % name
                )
>           raise NotImplementedError(msg)
E           NotImplementedError: median is not available with skipna=False with the installed version of numpy; upgrade to numpy 1.12 or newer to use skipna=True or skipna=None

xarray/core/duck_array_ops.py:321: NotImplementedError
```

However I’m not much knowledgeable on all this, so I’ll defer to you for finding the root cause.

Discussion:
keewis: given that we seem to have quite a few bugs in the dask compatibility code (which only gets called if `dask` is not installed): would it make sense to add a new CI that works like `py38` without `dask`?
keewis: the failing tests should be unrelated
max-sixty: LGTM!

Patch:
diff --git a/.github/ISSUE_TEMPLATE/bug_report.md b/.github/ISSUE_TEMPLATE/bug_report.md
index f24884c6..31fef19b 100644
--- a/.github/ISSUE_TEMPLATE/bug_report.md
+++ b/.github/ISSUE_TEMPLATE/bug_report.md
@@ -24,6 +24,6 @@ assignees: ''
 
 #### Output of ``xr.show_versions()``
 <details>
-# Paste the output here xr.show_versions() here
+<!-- Paste the output here xr.show_versions() here -->
 
 </details>
diff --git a/ci/azure/install.yml b/ci/azure/install.yml
index e11a8b54..958e3c18 100644
--- a/ci/azure/install.yml
+++ b/ci/azure/install.yml
@@ -29,6 +29,7 @@ steps:
         git+https://github.com/zarr-developers/zarr \
         git+https://github.com/Unidata/cftime \
         git+https://github.com/mapbox/rasterio \
+        git+https://github.com/hgrecco/pint \
         git+https://github.com/pydata/bottleneck
   condition: eq(variables['UPSTREAM_DEV'], 'true')
   displayName: Install upstream dev dependencies
diff --git a/doc/whats-new.rst b/doc/whats-new.rst
index e3e4eca7..c5c46022 100644
--- a/doc/whats-new.rst
+++ b/doc/whats-new.rst
@@ -24,9 +24,14 @@ Breaking changes
 
 New Features
 ~~~~~~~~~~~~
+- implement pint support. (:issue:`3594`, :pull:`3706`)
+  By `Justus Magin <https://github.com/keewis>`_.
 
 Bug fixes
 ~~~~~~~~~
+- Use ``dask_array_type`` instead of ``dask_array.Array`` for type
+  checking. (:issue:`3779`, :pull:`3787`)
+  By `Justus Magin <https://github.com/keewis>`_.
 
 Documentation
 ~~~~~~~~~~~~~
diff --git a/xarray/core/dask_array_compat.py b/xarray/core/dask_array_compat.py
index de55de89..05f750a1 100644
--- a/xarray/core/dask_array_compat.py
+++ b/xarray/core/dask_array_compat.py
@@ -3,6 +3,8 @@ from typing import Iterable
 
 import numpy as np
 
+from .pycompat import dask_array_type
+
 try:
     import dask.array as da
     from dask import __version__ as dask_version
@@ -36,7 +38,7 @@ else:
         """
         # If using x._meta, x must be a Dask Array, some libraries (e.g. zarr)
         # implement a _meta attribute that are incompatible with Dask Array._meta
-        if hasattr(x, "_meta") and isinstance(x, da.Array):
+        if hasattr(x, "_meta") and isinstance(x, dask_array_type):
             x = x._meta
 
         if dtype is None and x is None:
diff --git a/xarray/core/duck_array_ops.py b/xarray/core/duck_array_ops.py
index c2fe604a..bc2db93a 100644
--- a/xarray/core/duck_array_ops.py
+++ b/xarray/core/duck_array_ops.py
@@ -37,7 +37,7 @@ def _dask_or_eager_func(
                 dispatch_args = args[0]
             else:
                 dispatch_args = args[array_args]
-            if any(isinstance(a, dask_array.Array) for a in dispatch_args):
+            if any(isinstance(a, dask_array_type) for a in dispatch_args):
                 try:
                     wrapped = getattr(dask_module, name)
                 except AttributeError as e:
@@ -121,6 +121,7 @@ _where = _dask_or_eager_func("where", array_args=slice(3))
 isin = _dask_or_eager_func("isin", array_args=slice(2))
 take = _dask_or_eager_func("take")
 broadcast_to = _dask_or_eager_func("broadcast_to")
+pad = _dask_or_eager_func("pad")
 
 _concatenate = _dask_or_eager_func("concatenate", list_of_args=True)
 _stack = _dask_or_eager_func("stack", list_of_args=True)
@@ -189,8 +190,8 @@ def lazy_array_equiv(arr1, arr2):
         return False
     if (
         dask_array
-        and isinstance(arr1, dask_array.Array)
-        and isinstance(arr2, dask_array.Array)
+        and isinstance(arr1, dask_array_type)
+        and isinstance(arr2, dask_array_type)
     ):
         # GH3068
         if arr1.name == arr2.name:
@@ -261,7 +262,10 @@ def where_method(data, cond, other=dtypes.NA):
 
 
 def fillna(data, other):
-    return where(isnull(data), other, data)
+    # we need to pass data first so pint has a chance of returning the
+    # correct unit
+    # TODO: revert after https://github.com/hgrecco/pint/issues/1019 is fixed
+    return where(notnull(data), data, other)
 
 
 def concatenate(arrays, axis=0):
diff --git a/xarray/core/formatting.py b/xarray/core/formatting.py
index 520fa9b9..89246ff2 100644
--- a/xarray/core/formatting.py
+++ b/xarray/core/formatting.py
@@ -500,6 +500,13 @@ def diff_dim_summary(a, b):
 
 
 def _diff_mapping_repr(a_mapping, b_mapping, compat, title, summarizer, col_width=None):
+    def is_array_like(value):
+        return (
+            hasattr(value, "ndim")
+            and hasattr(value, "shape")
+            and hasattr(value, "dtype")
+        )
+
     def extra_items_repr(extra_keys, mapping, ab_side):
         extra_repr = [summarizer(k, mapping[k], col_width) for k in extra_keys]
         if extra_repr:
@@ -522,7 +529,11 @@ def _diff_mapping_repr(a_mapping, b_mapping, compat, title, summarizer, col_widt
             is_variable = True
         except AttributeError:
             # compare attribute value
-            compatible = a_mapping[k] == b_mapping[k]
+            if is_array_like(a_mapping[k]) or is_array_like(b_mapping[k]):
+                compatible = array_equiv(a_mapping[k], b_mapping[k])
+            else:
+                compatible = a_mapping[k] == b_mapping[k]
+
             is_variable = False
 
         if not compatible:
diff --git a/xarray/core/variable.py b/xarray/core/variable.py
index 74d5d57e..058b7bf5 100644
--- a/xarray/core/variable.py
+++ b/xarray/core/variable.py
@@ -742,7 +742,10 @@ class Variable(
 
             data = as_indexable(self._data)[actual_indexer]
             mask = indexing.create_mask(indexer, self.shape, data)
-            data = duck_array_ops.where(mask, fill_value, data)
+            # we need to invert the mask in order to pass data first. This helps
+            # pint to choose the correct unit
+            # TODO: revert after https://github.com/hgrecco/pint/issues/1019 is fixed
+            data = duck_array_ops.where(np.logical_not(mask), data, fill_value)
         else:
             # array cannot be indexed along dimensions of size 0, so just
             # build the mask directly instead.
@@ -1099,24 +1102,16 @@ class Variable(
         else:
             dtype = self.dtype
 
-        shape = list(self.shape)
-        shape[axis] = min(abs(count), shape[axis])
+        width = min(abs(count), self.shape[axis])
+        dim_pad = (width, 0) if count >= 0 else (0, width)
+        pads = [(0, 0) if d != dim else dim_pad for d in self.dims]
 
-        if isinstance(trimmed_data, dask_array_type):
-            chunks = list(trimmed_data.chunks)
-            chunks[axis] = (shape[axis],)
-            full = functools.partial(da.full, chunks=chunks)
-        else:
-            full = np.full
-
-        filler = full(shape, fill_value, dtype=dtype)
-
-        if count > 0:
-            arrays = [filler, trimmed_data]
-        else:
-            arrays = [trimmed_data, filler]
-
-        data = duck_array_ops.concatenate(arrays, axis)
+        data = duck_array_ops.pad(
+            trimmed_data.astype(dtype),
+            pads,
+            mode="constant",
+            constant_values=fill_value,
+        )
 
         if isinstance(data, dask_array_type):
             # chunked data should come out with the same chunks; this makes


Test patch:
diff --git a/xarray/tests/test_formatting.py b/xarray/tests/test_formatting.py
index 9a1f0bbd..61ecf46b 100644
--- a/xarray/tests/test_formatting.py
+++ b/xarray/tests/test_formatting.py
@@ -3,6 +3,7 @@ from textwrap import dedent
 
 import numpy as np
 import pandas as pd
+import pytest
 
 import xarray as xr
 from xarray.core import formatting
@@ -275,6 +276,44 @@ class TestFormatting:
         except AssertionError:
             assert actual == expected.replace(", dtype=int64", "")
 
+    @pytest.mark.filterwarnings("error")
+    def test_diff_attrs_repr_with_array(self):
+        attrs_a = {"attr": np.array([0, 1])}
+
+        attrs_b = {"attr": 1}
+        expected = dedent(
+            """\
+            Differing attributes:
+            L   attr: [0 1]
+            R   attr: 1
+            """
+        ).strip()
+        actual = formatting.diff_attrs_repr(attrs_a, attrs_b, "equals")
+        assert expected == actual
+
+        attrs_b = {"attr": np.array([-3, 5])}
+        expected = dedent(
+            """\
+            Differing attributes:
+            L   attr: [0 1]
+            R   attr: [-3  5]
+            """
+        ).strip()
+        actual = formatting.diff_attrs_repr(attrs_a, attrs_b, "equals")
+        assert expected == actual
+
+        # should not raise a warning
+        attrs_b = {"attr": np.array([0, 1, 2])}
+        expected = dedent(
+            """\
+            Differing attributes:
+            L   attr: [0 1]
+            R   attr: [0 1 2]
+            """
+        ).strip()
+        actual = formatting.diff_attrs_repr(attrs_a, attrs_b, "equals")
+        assert expected == actual
+
     def test_diff_dataset_repr(self):
         ds_a = xr.Dataset(
             data_vars={
diff --git a/xarray/tests/test_units.py b/xarray/tests/test_units.py
index d98e5e23..75e743c3 100644
--- a/xarray/tests/test_units.py
+++ b/xarray/tests/test_units.py
@@ -1,4 +1,5 @@
 import operator
+from distutils.version import LooseVersion
 
 import numpy as np
 import pandas as pd
@@ -19,6 +20,7 @@ DimensionalityError = pint.errors.DimensionalityError
 unit_registry = pint.UnitRegistry(force_ndarray=True)
 Quantity = unit_registry.Quantity
 
+
 pytestmark = [
     pytest.mark.skipif(
         not IS_NEP18_ACTIVE, reason="NUMPY_EXPERIMENTAL_ARRAY_FUNCTION is not enabled"
@@ -1536,27 +1538,17 @@ class TestVariable(VariableSubclassobjects):
     @pytest.mark.parametrize(
         "unit,error",
         (
-            pytest.param(
-                1,
-                DimensionalityError,
-                id="no_unit",
-                marks=pytest.mark.xfail(reason="uses 0 as a replacement"),
-            ),
+            pytest.param(1, DimensionalityError, id="no_unit"),
             pytest.param(
                 unit_registry.dimensionless, DimensionalityError, id="dimensionless"
             ),
             pytest.param(unit_registry.s, DimensionalityError, id="incompatible_unit"),
-            pytest.param(
-                unit_registry.cm,
-                None,
-                id="compatible_unit",
-                marks=pytest.mark.xfail(reason="converts to fill value's unit"),
-            ),
+            pytest.param(unit_registry.cm, None, id="compatible_unit"),
             pytest.param(unit_registry.m, None, id="identical_unit"),
         ),
     )
     def test_missing_value_fillna(self, unit, error):
-        value = 0
+        value = 10
         array = (
             np.array(
                 [
@@ -1595,13 +1587,7 @@ class TestVariable(VariableSubclassobjects):
             pytest.param(1, id="no_unit"),
             pytest.param(unit_registry.dimensionless, id="dimensionless"),
             pytest.param(unit_registry.s, id="incompatible_unit"),
-            pytest.param(
-                unit_registry.cm,
-                id="compatible_unit",
-                marks=pytest.mark.xfail(
-                    reason="checking for identical units does not work properly, yet"
-                ),
-            ),
+            pytest.param(unit_registry.cm, id="compatible_unit",),
             pytest.param(unit_registry.m, id="identical_unit"),
         ),
     )
@@ -1612,7 +1598,17 @@ class TestVariable(VariableSubclassobjects):
             pytest.param(True, id="with_conversion"),
         ),
     )
-    @pytest.mark.parametrize("func", (method("equals"), method("identical")), ids=repr)
+    @pytest.mark.parametrize(
+        "func",
+        (
+            method("equals"),
+            pytest.param(
+                method("identical"),
+                marks=pytest.mark.skip(reason="behaviour of identical is unclear"),
+            ),
+        ),
+        ids=repr,
+    )
     def test_comparisons(self, func, unit, convert_data, dtype):
         array = np.linspace(0, 1, 9).astype(dtype)
         quantity1 = array * unit_registry.m
@@ -1762,14 +1758,7 @@ class TestVariable(VariableSubclassobjects):
                 unit_registry.dimensionless, DimensionalityError, id="dimensionless"
             ),
             pytest.param(unit_registry.s, DimensionalityError, id="incompatible_unit"),
-            pytest.param(
-                unit_registry.cm,
-                None,
-                id="compatible_unit",
-                marks=pytest.mark.xfail(
-                    reason="getitem_with_mask converts to the unit of other"
-                ),
-            ),
+            pytest.param(unit_registry.cm, None, id="compatible_unit"),
             pytest.param(unit_registry.m, None, id="identical_unit"),
         ),
     )
@@ -1853,12 +1842,7 @@ class TestVariable(VariableSubclassobjects):
             ),
             method("reduce", np.std, "x"),
             method("round", 2),
-            pytest.param(
-                method("shift", {"x": -2}),
-                marks=pytest.mark.xfail(
-                    reason="trying to concatenate ndarray to quantity"
-                ),
-            ),
+            method("shift", {"x": -2}),
             method("transpose", "y", "x"),
         ),
         ids=repr,
@@ -1933,7 +1917,6 @@ class TestVariable(VariableSubclassobjects):
         assert_units_equal(expected, actual)
         xr.testing.assert_identical(expected, actual)
 
-    @pytest.mark.xfail(reason="ignores units")
     @pytest.mark.parametrize(
         "unit,error",
         (
@@ -1948,25 +1931,28 @@ class TestVariable(VariableSubclassobjects):
     )
     def test_concat(self, unit, error, dtype):
         array1 = (
-            np.linspace(0, 5, 3 * 10).reshape(3, 10).astype(dtype) * unit_registry.m
+            np.linspace(0, 5, 9 * 10).reshape(3, 6, 5).astype(dtype) * unit_registry.m
         )
-        array2 = np.linspace(5, 10, 10 * 2).reshape(10, 2).astype(dtype) * unit
+        array2 = np.linspace(5, 10, 10 * 3).reshape(3, 2, 5).astype(dtype) * unit
 
-        variable = xr.Variable(("x", "y"), array1)
-        other = xr.Variable(("y", "z"), array2)
+        variable = xr.Variable(("x", "y", "z"), array1)
+        other = xr.Variable(("x", "y", "z"), array2)
 
         if error is not None:
             with pytest.raises(error):
-                variable.concat(other)
+                xr.Variable.concat([variable, other], dim="y")
 
             return
 
         units = extract_units(variable)
         expected = attach_units(
-            strip_units(variable).concat(strip_units(convert_units(other, units))),
+            xr.Variable.concat(
+                [strip_units(variable), strip_units(convert_units(other, units))],
+                dim="y",
+            ),
             units,
         )
-        actual = variable.concat(other)
+        actual = xr.Variable.concat([variable, other], dim="y")
 
         assert_units_equal(expected, actual)
         xr.testing.assert_identical(expected, actual)
@@ -2036,6 +2022,43 @@ class TestVariable(VariableSubclassobjects):
 
         assert expected == actual
 
+    def test_pad(self, dtype):
+        data = np.arange(4 * 3 * 2).reshape(4, 3, 2).astype(dtype) * unit_registry.m
+        v = xr.Variable(["x", "y", "z"], data)
+
+        xr_args = [{"x": (2, 1)}, {"y": (0, 3)}, {"x": (3, 1), "z": (2, 0)}]
+        np_args = [
+            ((2, 1), (0, 0), (0, 0)),
+            ((0, 0), (0, 3), (0, 0)),
+            ((3, 1), (0, 0), (2, 0)),
+        ]
+        for xr_arg, np_arg in zip(xr_args, np_args):
+            actual = v.pad_with_fill_value(**xr_arg)
+            expected = xr.Variable(
+                v.dims,
+                np.pad(
+                    v.data.astype(float),
+                    np_arg,
+                    mode="constant",
+                    constant_values=np.nan,
+                ),
+            )
+            xr.testing.assert_identical(expected, actual)
+            assert_units_equal(expected, actual)
+            assert isinstance(actual._data, type(v._data))
+
+        # for the boolean array, we pad False
+        data = np.full_like(data, False, dtype=bool).reshape(4, 3, 2)
+        v = xr.Variable(["x", "y", "z"], data)
+        for xr_arg, np_arg in zip(xr_args, np_args):
+            actual = v.pad_with_fill_value(fill_value=data.flat[0], **xr_arg)
+            expected = xr.Variable(
+                v.dims,
+                np.pad(v.data, np_arg, mode="constant", constant_values=v.data.flat[0]),
+            )
+            xr.testing.assert_identical(actual, expected)
+            assert_units_equal(expected, actual)
+
     @pytest.mark.parametrize(
         "unit,error",
         (
@@ -2044,7 +2067,8 @@ class TestVariable(VariableSubclassobjects):
                 DimensionalityError,
                 id="no_unit",
                 marks=pytest.mark.xfail(
-                    reason="is not treated the same as dimensionless"
+                    LooseVersion(pint.__version__) < LooseVersion("0.10.2"),
+                    reason="bug in pint's implementation of np.pad",
                 ),
             ),
             pytest.param(