Repository: pydata/xarray

Issue:
Option to skip tests in `weighted()`
When working with large dask-array weights, [this check](https://github.com/pydata/xarray/blob/adc55ac4d2883e0c6647f3983c3322ca2c690514/xarray/core/weighted.py#L103-L107) triggers computation of the array. This affects [xgcms ability to layer operations lazily](https://github.com/xgcm/xgcm/blob/f0a0afd666184dd7d15bcff6f22ac50716b6c78a/xgcm/grid.py#L1737)

Would you be open to implement an option to skip this test, maybe with a warning displayed? Happy to submit a PR.



Discussion:
pep8speaks: Hello @jbusecke! Thanks for opening this PR. We checked the lines you've touched for [PEP 8](https://www.python.org/dev/peps/pep-0008) issues, and found:

* In the file [`xarray/core/weighted.py`](https://github.com/pydata/xarray/blob/c3b47db42c40dfe7620c6d3ebcbde5565f9052f8/xarray/core/weighted.py):

> [Line 102:1](https://github.com/pydata/xarray/blob/c3b47db42c40dfe7620c6d3ebcbde5565f9052f8/xarray/core/weighted.py#L102): [W293](https://duckduckgo.com/?q=pep8%20W293) blank line contains whitespace
> [Line 111:1](https://github.com/pydata/xarray/blob/c3b47db42c40dfe7620c6d3ebcbde5565f9052f8/xarray/core/weighted.py#L111): [W293](https://duckduckgo.com/?q=pep8%20W293) blank line contains whitespace


jbusecke: The Ci environments without dask are failing. Should I add some pytest skip logic, or what is the best way to handle this?
max-sixty: Yes, `@requires_dask` on the test
keewis: since the test you added requires `dask` it definitely should get a `@requires_dask`.

That won't fix all the failing tests, though: `map_blocks` is only valid if we operate on `dask` arrays. I guess you either have to apply the function directly on other arrays, or we have to make `DataArray.map_blocks` on non-`dask` arrays call to `DataArray.map`.
mathause: I think you need to do someting along the lines of:

```python
if dask_duck_array(weights):
    import dask.array as dsa
    dsa.map_blocks(_weight_check, weights.data, dtype=weights.dtype)
else:
    _weight_check()
```

jbusecke: I did have to fiddle with this a bit. I did change `.isnull()` to `np.isnan()` and replace the data of `weights`,  otherwise the error would not be raised for the dask array at all. 
This does not look terribly elegant to me, but passes tests locally. Waiting for the CI to see if the others pass aswell. Happy to make further changes, and thanks for all the input already.
jbusecke: Do you think this works or are further changes needed? Many thanks for the guidance so far!
jbusecke: I am getting some failures for `py38-upstream` (TestDataset.test_polyfit_warnings) that seem unrelated?
jbusecke: Similarly on `MacOSX py38` this fails: TestDask.test_save_mfdataset_compute_false_roundtrip
keewis: the `TestDask.test_save_mfdataset_compute_false_roundtrip` failure should be #4539, and the failing `upstream-dev` CI shouldn't be related, either (my guess is that a dependency – most likely `numpy` – added a new warning). I'll investigate and open a new issue for that.
jbusecke: Ok I think this should be good to go. I have implemented all the requested changes. The remaining failures are related to other problems upstream (I think). Anything else I should add here?
jbusecke: I am not understanding why that `MinimumVersionPolicy` test is failing (or what it does haha). Is this something upstream, or should I fix?
keewis: this CI is sometimes flaky, it's usually enough to rerun it. I'll do that for you once the other CI finished.
jbusecke: Seems like all the other test are passing (minus the two upstream problems discussed before).

Patch:
diff --git a/doc/whats-new.rst b/doc/whats-new.rst
index 60364a87..ce132d77 100644
--- a/doc/whats-new.rst
+++ b/doc/whats-new.rst
@@ -23,6 +23,11 @@ v0.16.2 (unreleased)
 Breaking changes
 ~~~~~~~~~~~~~~~~
 
+- :py:attr:`DataArray.rolling` and :py:attr:`Dataset.rolling` no longer support passing ``keep_attrs``
+  via its constructor. Pass ``keep_attrs`` via the applied function, i.e. use
+  ``ds.rolling(...).mean(keep_attrs=False)`` instead of ``ds.rolling(..., keep_attrs=False).mean()``
+  Rolling operations now keep their attributes per default (:pull:`4510`).
+  By `Mathias Hauser <https://github.com/mathause>`_.
 
 New Features
 ~~~~~~~~~~~~
@@ -38,6 +43,8 @@ New Features
   By `Stephan Hoyer <https://github.com/shoyer>`_.
 - Added typehints in :py:func:`align` to reflect that the same type received in ``objects`` arg will be returned (:pull:`4522`).
   By `Michal Baumgartner <https://github.com/m1so>`_.
+- :py:meth:`Dataset.weighted` and :py:meth:`DataArray.weighted` are now executing value checks lazily if weights are provided as dask arrays (:issue:`4541`, :pull:`4559`).
+  By `Julius Busecke <https://github.com/jbusecke>`_.
 
 Bug fixes
 ~~~~~~~~~
@@ -64,6 +71,9 @@ Bug fixes
   By `Mathias Hauser <https://github.com/mathause>`_.
 - :py:func:`combine_by_coords` now raises an informative error when passing coordinates
   with differing calendars (:issue:`4495`). By `Mathias Hauser <https://github.com/mathause>`_.
+- :py:attr:`DataArray.rolling` and :py:attr:`Dataset.rolling` now also keep the attributes and names of of (wrapped)
+  ``DataArray`` objects, previously only the global attributes were retained (:issue:`4497`, :pull:`4510`).
+  By `Mathias Hauser <https://github.com/mathause>`_.
 - Improve performance where reading small slices from huge dimensions was slower than necessary (:pull:`4560`). By `Dion Häfner <https://github.com/dionhaefner>`_.
 
 Documentation
diff --git a/xarray/core/common.py b/xarray/core/common.py
index eda31a16..7078a4c1 100644
--- a/xarray/core/common.py
+++ b/xarray/core/common.py
@@ -811,10 +811,6 @@ class DataWithCoords(SupportsArithmetic, AttrAccessMixin):
             setting min_periods equal to the size of the window.
         center : bool or mapping, default: False
             Set the labels at the center of the window.
-        keep_attrs : bool, optional
-            If True, the object's attributes (`attrs`) will be copied from
-            the original object to the new one.  If False (default), the new
-            object will be returned without attributes.
         **window_kwargs : optional
             The keyword arguments form of ``dim``.
             One of dim or window_kwargs must be provided.
@@ -863,8 +859,6 @@ class DataWithCoords(SupportsArithmetic, AttrAccessMixin):
         core.rolling.DataArrayRolling
         core.rolling.DatasetRolling
         """
-        if keep_attrs is None:
-            keep_attrs = _get_keep_attrs(default=False)
 
         dim = either_dict_or_kwargs(dim, window_kwargs, "rolling")
         return self._rolling_cls(
diff --git a/xarray/core/rolling.py b/xarray/core/rolling.py
index 0bffc215..38cb11b5 100644
--- a/xarray/core/rolling.py
+++ b/xarray/core/rolling.py
@@ -22,6 +22,10 @@ Reduce this object's data windows by applying `{name}` along its dimension.
 
 Parameters
 ----------
+keep_attrs : bool, default: None
+    If True, the attributes (``attrs``) will be copied from the original
+    object to the new one. If False, the new object will be returned
+    without attributes. If None uses the global default.
 **kwargs : dict
     Additional keyword arguments passed on to `{name}`.
 
@@ -56,17 +60,13 @@ class Rolling:
             Object to window.
         windows : mapping of hashable to int
             A mapping from the name of the dimension to create the rolling
-            exponential window along (e.g. `time`) to the size of the moving window.
+            window along (e.g. `time`) to the size of the moving window.
         min_periods : int, default: None
             Minimum number of observations in window required to have a value
             (otherwise result is NA). The default, None, is equivalent to
             setting min_periods equal to the size of the window.
         center : bool, default: False
             Set the labels at the center of the window.
-        keep_attrs : bool, optional
-            If True, the object's attributes (`attrs`) will be copied from
-            the original object to the new one.  If False (default), the new
-            object will be returned without attributes.
 
         Returns
         -------
@@ -88,8 +88,13 @@ class Rolling:
 
         self.min_periods = np.prod(self.window) if min_periods is None else min_periods
 
-        if keep_attrs is None:
-            keep_attrs = _get_keep_attrs(default=False)
+        if keep_attrs is not None:
+            warnings.warn(
+                "Passing ``keep_attrs`` to ``rolling`` is deprecated and will raise an"
+                " error in xarray 0.18. Please pass ``keep_attrs`` directly to the"
+                " applied function. Note that keep_attrs is now True per default.",
+                FutureWarning,
+            )
         self.keep_attrs = keep_attrs
 
     def __repr__(self):
@@ -110,9 +115,12 @@ class Rolling:
         array_agg_func = getattr(duck_array_ops, name)
         bottleneck_move_func = getattr(bottleneck, "move_" + name, None)
 
-        def method(self, **kwargs):
+        def method(self, keep_attrs=None, **kwargs):
+
+            keep_attrs = self._get_keep_attrs(keep_attrs)
+
             return self._numpy_or_bottleneck_reduce(
-                array_agg_func, bottleneck_move_func, **kwargs
+                array_agg_func, bottleneck_move_func, keep_attrs=keep_attrs, **kwargs
             )
 
         method.__name__ = name
@@ -130,8 +138,9 @@ class Rolling:
     var = _reduce_method("var")
     median = _reduce_method("median")
 
-    def count(self):
-        rolling_count = self._counts()
+    def count(self, keep_attrs=None):
+        keep_attrs = self._get_keep_attrs(keep_attrs)
+        rolling_count = self._counts(keep_attrs=keep_attrs)
         enough_periods = rolling_count >= self.min_periods
         return rolling_count.where(enough_periods)
 
@@ -157,6 +166,19 @@ class Rolling:
                 "Mapping argument is necessary for {}d-rolling.".format(len(self.dim))
             )
 
+    def _get_keep_attrs(self, keep_attrs):
+
+        if keep_attrs is None:
+            # TODO: uncomment the next line and remove the others after the deprecation
+            # keep_attrs = _get_keep_attrs(default=True)
+
+            if self.keep_attrs is None:
+                keep_attrs = _get_keep_attrs(default=True)
+            else:
+                keep_attrs = self.keep_attrs
+
+        return keep_attrs
+
 
 class DataArrayRolling(Rolling):
     __slots__ = ("window_labels",)
@@ -180,10 +202,6 @@ class DataArrayRolling(Rolling):
             setting min_periods equal to the size of the window.
         center : bool, default: False
             Set the labels at the center of the window.
-        keep_attrs : bool, optional
-            If True, the object's attributes (`attrs`) will be copied from
-            the original object to the new one.  If False (default), the new
-            object will be returned without attributes.
 
         Returns
         -------
@@ -196,8 +214,6 @@ class DataArrayRolling(Rolling):
         Dataset.rolling
         Dataset.groupby
         """
-        if keep_attrs is None:
-            keep_attrs = _get_keep_attrs(default=False)
         super().__init__(
             obj, windows, min_periods=min_periods, center=center, keep_attrs=keep_attrs
         )
@@ -220,7 +236,12 @@ class DataArrayRolling(Rolling):
             yield (label, window)
 
     def construct(
-        self, window_dim=None, stride=1, fill_value=dtypes.NA, **window_dim_kwargs
+        self,
+        window_dim=None,
+        stride=1,
+        fill_value=dtypes.NA,
+        keep_attrs=None,
+        **window_dim_kwargs,
     ):
         """
         Convert this rolling object to xr.DataArray,
@@ -230,11 +251,14 @@ class DataArrayRolling(Rolling):
         ----------
         window_dim : str or mapping, optional
             A mapping from dimension name to the new window dimension names.
-            Just a string can be used for 1d-rolling.
-        stride : int or mapping of int, optional
+        stride : int or mapping of int, default: 1
             Size of stride for the rolling window.
         fill_value : default: dtypes.NA
             Filling value to match the dimension size.
+        keep_attrs : bool, default: None
+            If True, the attributes (``attrs``) will be copied from the original
+            object to the new one. If False, the new object will be returned
+            without attributes. If None uses the global default.
         **window_dim_kwargs : {dim: new_name, ...}, optional
             The keyword arguments form of ``window_dim``.
 
@@ -279,6 +303,8 @@ class DataArrayRolling(Rolling):
 
         from .dataarray import DataArray
 
+        keep_attrs = self._get_keep_attrs(keep_attrs)
+
         if window_dim is None:
             if len(window_dim_kwargs) == 0:
                 raise ValueError(
@@ -294,14 +320,21 @@ class DataArrayRolling(Rolling):
         window = self.obj.variable.rolling_window(
             self.dim, self.window, window_dim, self.center, fill_value=fill_value
         )
+
+        attrs = self.obj.attrs if keep_attrs else {}
+
         result = DataArray(
-            window, dims=self.obj.dims + tuple(window_dim), coords=self.obj.coords
+            window,
+            dims=self.obj.dims + tuple(window_dim),
+            coords=self.obj.coords,
+            attrs=attrs,
+            name=self.obj.name,
         )
         return result.isel(
             **{d: slice(None, None, s) for d, s in zip(self.dim, stride)}
         )
 
-    def reduce(self, func, **kwargs):
+    def reduce(self, func, keep_attrs=None, **kwargs):
         """Reduce the items in this group by applying `func` along some
         dimension(s).
 
@@ -311,6 +344,10 @@ class DataArrayRolling(Rolling):
             Function which can be called in the form
             `func(x, **kwargs)` to return the result of collapsing an
             np.ndarray over an the rolling dimension.
+        keep_attrs : bool, default: None
+            If True, the attributes (``attrs``) will be copied from the original
+            object to the new one. If False, the new object will be returned
+            without attributes. If None uses the global default.
         **kwargs : dict
             Additional keyword arguments passed on to `func`.
 
@@ -349,19 +386,24 @@ class DataArrayRolling(Rolling):
                [ 4.,  9., 15., 18.]])
         Dimensions without coordinates: a, b
         """
+
+        keep_attrs = self._get_keep_attrs(keep_attrs)
+
         rolling_dim = {
             d: utils.get_temp_dimname(self.obj.dims, f"_rolling_dim_{d}")
             for d in self.dim
         }
-        windows = self.construct(rolling_dim)
-        result = windows.reduce(func, dim=list(rolling_dim.values()), **kwargs)
+        windows = self.construct(rolling_dim, keep_attrs=keep_attrs)
+        result = windows.reduce(
+            func, dim=list(rolling_dim.values()), keep_attrs=keep_attrs, **kwargs
+        )
 
         # Find valid windows based on count.
-        counts = self._counts()
+        counts = self._counts(keep_attrs=False)
         return result.where(counts >= self.min_periods)
 
-    def _counts(self):
-        """ Number of non-nan entries in each rolling window. """
+    def _counts(self, keep_attrs):
+        """Number of non-nan entries in each rolling window."""
 
         rolling_dim = {
             d: utils.get_temp_dimname(self.obj.dims, f"_rolling_dim_{d}")
@@ -372,17 +414,17 @@ class DataArrayRolling(Rolling):
         # The use of skipna==False is also faster since it does not need to
         # copy the strided array.
         counts = (
-            self.obj.notnull()
+            self.obj.notnull(keep_attrs=keep_attrs)
             .rolling(
                 center={d: self.center[i] for i, d in enumerate(self.dim)},
                 **{d: w for d, w in zip(self.dim, self.window)},
             )
-            .construct(rolling_dim, fill_value=False)
-            .sum(dim=list(rolling_dim.values()), skipna=False)
+            .construct(rolling_dim, fill_value=False, keep_attrs=keep_attrs)
+            .sum(dim=list(rolling_dim.values()), skipna=False, keep_attrs=keep_attrs)
         )
         return counts
 
-    def _bottleneck_reduce(self, func, **kwargs):
+    def _bottleneck_reduce(self, func, keep_attrs, **kwargs):
         from .dataarray import DataArray
 
         # bottleneck doesn't allow min_count to be 0, although it should
@@ -398,8 +440,8 @@ class DataArrayRolling(Rolling):
         padded = self.obj.variable
         if self.center[0]:
             if is_duck_dask_array(padded.data):
-                # Workaround to make the padded chunk size is larger than
-                # self.window-1
+                # workaround to make the padded chunk size larger than
+                # self.window - 1
                 shift = -(self.window[0] + 1) // 2
                 offset = (self.window[0] - 1) // 2
                 valid = (slice(None),) * axis + (
@@ -422,16 +464,19 @@ class DataArrayRolling(Rolling):
 
         if self.center[0]:
             values = values[valid]
-        result = DataArray(values, self.obj.coords)
 
-        return result
+        attrs = self.obj.attrs if keep_attrs else {}
+
+        return DataArray(values, self.obj.coords, attrs=attrs, name=self.obj.name)
 
     def _numpy_or_bottleneck_reduce(
-        self, array_agg_func, bottleneck_move_func, **kwargs
+        self, array_agg_func, bottleneck_move_func, keep_attrs, **kwargs
     ):
         if "dim" in kwargs:
             warnings.warn(
-                f"Reductions will be applied along the rolling dimension '{self.dim}'. Passing the 'dim' kwarg to reduction operations has no effect and will raise an error in xarray 0.16.0.",
+                f"Reductions are applied along the rolling dimension(s) "
+                f"'{self.dim}'. Passing the 'dim' kwarg to reduction "
+                f"operations has no effect.",
                 DeprecationWarning,
                 stacklevel=3,
             )
@@ -445,9 +490,11 @@ class DataArrayRolling(Rolling):
             # TODO: renable bottleneck with dask after the issues
             # underlying https://github.com/pydata/xarray/issues/2940 are
             # fixed.
-            return self._bottleneck_reduce(bottleneck_move_func, **kwargs)
+            return self._bottleneck_reduce(
+                bottleneck_move_func, keep_attrs=keep_attrs, **kwargs
+            )
         else:
-            return self.reduce(array_agg_func, **kwargs)
+            return self.reduce(array_agg_func, keep_attrs=keep_attrs, **kwargs)
 
 
 class DatasetRolling(Rolling):
@@ -472,10 +519,6 @@ class DatasetRolling(Rolling):
             setting min_periods equal to the size of the window.
         center : bool or mapping of hashable to bool, default: False
             Set the labels at the center of the window.
-        keep_attrs : bool, optional
-            If True, the object's attributes (`attrs`) will be copied from
-            the original object to the new one.  If False (default), the new
-            object will be returned without attributes.
 
         Returns
         -------
@@ -494,7 +537,7 @@ class DatasetRolling(Rolling):
         # Keep each Rolling object as a dictionary
         self.rollings = {}
         for key, da in self.obj.data_vars.items():
-            # keeps rollings only for the dataset depending on slf.dim
+            # keeps rollings only for the dataset depending on self.dim
             dims, center = [], {}
             for i, d in enumerate(self.dim):
                 if d in da.dims:
@@ -503,23 +546,27 @@ class DatasetRolling(Rolling):
 
             if len(dims) > 0:
                 w = {d: windows[d] for d in dims}
-                self.rollings[key] = DataArrayRolling(
-                    da, w, min_periods, center, keep_attrs
-                )
+                self.rollings[key] = DataArrayRolling(da, w, min_periods, center)
 
-    def _dataset_implementation(self, func, **kwargs):
+    def _dataset_implementation(self, func, keep_attrs, **kwargs):
         from .dataset import Dataset
 
+        keep_attrs = self._get_keep_attrs(keep_attrs)
+
         reduced = {}
         for key, da in self.obj.data_vars.items():
             if any(d in da.dims for d in self.dim):
-                reduced[key] = func(self.rollings[key], **kwargs)
+                reduced[key] = func(self.rollings[key], keep_attrs=keep_attrs, **kwargs)
             else:
-                reduced[key] = self.obj[key]
-        attrs = self.obj.attrs if self.keep_attrs else {}
+                reduced[key] = self.obj[key].copy()
+                # we need to delete the attrs of the copied DataArray
+                if not keep_attrs:
+                    reduced[key].attrs = {}
+
+        attrs = self.obj.attrs if keep_attrs else {}
         return Dataset(reduced, coords=self.obj.coords, attrs=attrs)
 
-    def reduce(self, func, **kwargs):
+    def reduce(self, func, keep_attrs=None, **kwargs):
         """Reduce the items in this group by applying `func` along some
         dimension(s).
 
@@ -529,6 +576,10 @@ class DatasetRolling(Rolling):
             Function which can be called in the form
             `func(x, **kwargs)` to return the result of collapsing an
             np.ndarray over an the rolling dimension.
+        keep_attrs : bool, default: None
+            If True, the attributes (``attrs``) will be copied from the original
+            object to the new one. If False, the new object will be returned
+            without attributes. If None uses the global default.
         **kwargs : dict
             Additional keyword arguments passed on to `func`.
 
@@ -538,14 +589,18 @@ class DatasetRolling(Rolling):
             Array with summarized data.
         """
         return self._dataset_implementation(
-            functools.partial(DataArrayRolling.reduce, func=func), **kwargs
+            functools.partial(DataArrayRolling.reduce, func=func),
+            keep_attrs=keep_attrs,
+            **kwargs,
         )
 
-    def _counts(self):
-        return self._dataset_implementation(DataArrayRolling._counts)
+    def _counts(self, keep_attrs):
+        return self._dataset_implementation(
+            DataArrayRolling._counts, keep_attrs=keep_attrs
+        )
 
     def _numpy_or_bottleneck_reduce(
-        self, array_agg_func, bottleneck_move_func, **kwargs
+        self, array_agg_func, bottleneck_move_func, keep_attrs, **kwargs
     ):
         return self._dataset_implementation(
             functools.partial(
@@ -553,6 +608,7 @@ class DatasetRolling(Rolling):
                 array_agg_func=array_agg_func,
                 bottleneck_move_func=bottleneck_move_func,
             ),
+            keep_attrs=keep_attrs,
             **kwargs,
         )
 
@@ -587,6 +643,8 @@ class DatasetRolling(Rolling):
 
         from .dataset import Dataset
 
+        keep_attrs = self._get_keep_attrs(keep_attrs)
+
         if window_dim is None:
             if len(window_dim_kwargs) == 0:
                 raise ValueError(
@@ -599,22 +657,30 @@ class DatasetRolling(Rolling):
         )
         stride = self._mapping_to_list(stride, default=1)
 
-        if keep_attrs is None:
-            keep_attrs = _get_keep_attrs(default=True)
-
         dataset = {}
         for key, da in self.obj.data_vars.items():
-            # keeps rollings only for the dataset depending on slf.dim
+            # keeps rollings only for the dataset depending on self.dim
             dims = [d for d in self.dim if d in da.dims]
             if len(dims) > 0:
                 wi = {d: window_dim[i] for i, d in enumerate(self.dim) if d in da.dims}
                 st = {d: stride[i] for i, d in enumerate(self.dim) if d in da.dims}
+
                 dataset[key] = self.rollings[key].construct(
-                    window_dim=wi, fill_value=fill_value, stride=st
+                    window_dim=wi,
+                    fill_value=fill_value,
+                    stride=st,
+                    keep_attrs=keep_attrs,
                 )
             else:
-                dataset[key] = da
-        return Dataset(dataset, coords=self.obj.coords).isel(
+                dataset[key] = da.copy()
+
+            # as the DataArrays can be copied we need to delete the attrs
+            if not keep_attrs:
+                dataset[key].attrs = {}
+
+        attrs = self.obj.attrs if keep_attrs else {}
+
+        return Dataset(dataset, coords=self.obj.coords, attrs=attrs).isel(
             **{d: slice(None, None, s) for d, s in zip(self.dim, stride)}
         )
 
diff --git a/xarray/core/weighted.py b/xarray/core/weighted.py
index 96b4c79f..ab4a0958 100644
--- a/xarray/core/weighted.py
+++ b/xarray/core/weighted.py
@@ -1,7 +1,9 @@
 from typing import TYPE_CHECKING, Hashable, Iterable, Optional, Union, overload
 
+from . import duck_array_ops
 from .computation import dot
 from .options import _get_keep_attrs
+from .pycompat import is_duck_dask_array
 
 if TYPE_CHECKING:
     from .dataarray import DataArray, Dataset
@@ -100,12 +102,24 @@ class Weighted:
         if not isinstance(weights, DataArray):
             raise ValueError("`weights` must be a DataArray")
 
-        if weights.isnull().any():
-            raise ValueError(
-                "`weights` cannot contain missing values. "
-                "Missing values can be replaced by `weights.fillna(0)`."
+        def _weight_check(w):
+            # Ref https://github.com/pydata/xarray/pull/4559/files#r515968670
+            if duck_array_ops.isnull(w).any():
+                raise ValueError(
+                    "`weights` cannot contain missing values. "
+                    "Missing values can be replaced by `weights.fillna(0)`."
+                )
+            return w
+
+        if is_duck_dask_array(weights.data):
+            # assign to copy - else the check is not triggered
+            weights = weights.copy(
+                data=weights.data.map_blocks(_weight_check, dtype=weights.dtype)
             )
 
+        else:
+            _weight_check(weights.data)
+
         self.obj = obj
         self.weights = weights
 


Test patch:
diff --git a/xarray/tests/test_dataarray.py b/xarray/tests/test_dataarray.py
index 78b12ddd..e944c020 100644
--- a/xarray/tests/test_dataarray.py
+++ b/xarray/tests/test_dataarray.py
@@ -6297,6 +6297,7 @@ def test_rolling_properties(da):
     # catching invalid args
     with pytest.raises(ValueError, match="window must be > 0"):
         da.rolling(time=-2)
+
     with pytest.raises(ValueError, match="min_periods must be greater than zero"):
         da.rolling(time=2, min_periods=0)
 
@@ -6317,7 +6318,7 @@ def test_rolling_wrapped_bottleneck(da, name, center, min_periods):
     )
     assert_array_equal(actual.values, expected)
 
-    with pytest.warns(DeprecationWarning, match="Reductions will be applied"):
+    with pytest.warns(DeprecationWarning, match="Reductions are applied"):
         getattr(rolling_obj, name)(dim="time")
 
     # Test center
@@ -6336,7 +6337,7 @@ def test_rolling_wrapped_dask(da_dask, name, center, min_periods, window):
     rolling_obj = da_dask.rolling(time=window, min_periods=min_periods, center=center)
     actual = getattr(rolling_obj, name)().load()
     if name != "count":
-        with pytest.warns(DeprecationWarning, match="Reductions will be applied"):
+        with pytest.warns(DeprecationWarning, match="Reductions are applied"):
             getattr(rolling_obj, name)(dim="time")
     # numpy version
     rolling_obj = da_dask.load().rolling(
@@ -6540,6 +6541,92 @@ def test_ndrolling_construct(center, fill_value):
     assert_allclose(actual, expected)
 
 
+@pytest.mark.parametrize(
+    "funcname, argument",
+    [
+        ("reduce", (np.mean,)),
+        ("mean", ()),
+        ("construct", ("window_dim",)),
+        ("count", ()),
+    ],
+)
+def test_rolling_keep_attrs(funcname, argument):
+
+    attrs_da = {"da_attr": "test"}
+
+    data = np.linspace(10, 15, 100)
+    coords = np.linspace(1, 10, 100)
+
+    da = DataArray(
+        data, dims=("coord"), coords={"coord": coords}, attrs=attrs_da, name="name"
+    )
+
+    # attrs are now kept per default
+    func = getattr(da.rolling(dim={"coord": 5}), funcname)
+    result = func(*argument)
+    assert result.attrs == attrs_da
+    assert result.name == "name"
+
+    # discard attrs
+    func = getattr(da.rolling(dim={"coord": 5}), funcname)
+    result = func(*argument, keep_attrs=False)
+    assert result.attrs == {}
+    assert result.name == "name"
+
+    # test discard attrs using global option
+    func = getattr(da.rolling(dim={"coord": 5}), funcname)
+    with set_options(keep_attrs=False):
+        result = func(*argument)
+    assert result.attrs == {}
+    assert result.name == "name"
+
+    # keyword takes precedence over global option
+    func = getattr(da.rolling(dim={"coord": 5}), funcname)
+    with set_options(keep_attrs=False):
+        result = func(*argument, keep_attrs=True)
+    assert result.attrs == attrs_da
+    assert result.name == "name"
+
+    func = getattr(da.rolling(dim={"coord": 5}), funcname)
+    with set_options(keep_attrs=True):
+        result = func(*argument, keep_attrs=False)
+    assert result.attrs == {}
+    assert result.name == "name"
+
+
+def test_rolling_keep_attrs_deprecated():
+
+    attrs_da = {"da_attr": "test"}
+
+    data = np.linspace(10, 15, 100)
+    coords = np.linspace(1, 10, 100)
+
+    da = DataArray(
+        data,
+        dims=("coord"),
+        coords={"coord": coords},
+        attrs=attrs_da,
+    )
+
+    # deprecated option
+    with pytest.warns(
+        FutureWarning, match="Passing ``keep_attrs`` to ``rolling`` is deprecated"
+    ):
+        result = da.rolling(dim={"coord": 5}, keep_attrs=False).construct("window_dim")
+
+    assert result.attrs == {}
+
+    # the keep_attrs in the reduction function takes precedence
+    with pytest.warns(
+        FutureWarning, match="Passing ``keep_attrs`` to ``rolling`` is deprecated"
+    ):
+        result = da.rolling(dim={"coord": 5}, keep_attrs=True).construct(
+            "window_dim", keep_attrs=False
+        )
+
+    assert result.attrs == {}
+
+
 def test_raise_no_warning_for_nan_in_binary_ops():
     with pytest.warns(None) as record:
         xr.DataArray([1, 2, np.NaN]) > 0
diff --git a/xarray/tests/test_dataset.py b/xarray/tests/test_dataset.py
index 62cf3b8c..61e80557 100644
--- a/xarray/tests/test_dataset.py
+++ b/xarray/tests/test_dataset.py
@@ -5989,33 +5989,115 @@ def test_coarsen_keep_attrs():
     xr.testing.assert_identical(ds, ds2)
 
 
-def test_rolling_keep_attrs():
-    _attrs = {"units": "test", "long_name": "testing"}
+@pytest.mark.parametrize(
+    "funcname, argument",
+    [
+        ("reduce", (np.mean,)),
+        ("mean", ()),
+        ("construct", ("window_dim",)),
+        ("count", ()),
+    ],
+)
+def test_rolling_keep_attrs(funcname, argument):
+    global_attrs = {"units": "test", "long_name": "testing"}
+    da_attrs = {"da_attr": "test"}
+    da_not_rolled_attrs = {"da_not_rolled_attr": "test"}
 
-    var1 = np.linspace(10, 15, 100)
-    var2 = np.linspace(5, 10, 100)
+    data = np.linspace(10, 15, 100)
     coords = np.linspace(1, 10, 100)
 
     ds = Dataset(
-        data_vars={"var1": ("coord", var1), "var2": ("coord", var2)},
+        data_vars={"da": ("coord", data), "da_not_rolled": ("no_coord", data)},
         coords={"coord": coords},
-        attrs=_attrs,
+        attrs=global_attrs,
     )
+    ds.da.attrs = da_attrs
+    ds.da_not_rolled.attrs = da_not_rolled_attrs
+
+    # attrs are now kept per default
+    func = getattr(ds.rolling(dim={"coord": 5}), funcname)
+    result = func(*argument)
+    assert result.attrs == global_attrs
+    assert result.da.attrs == da_attrs
+    assert result.da_not_rolled.attrs == da_not_rolled_attrs
+    assert result.da.name == "da"
+    assert result.da_not_rolled.name == "da_not_rolled"
+
+    # discard attrs
+    func = getattr(ds.rolling(dim={"coord": 5}), funcname)
+    result = func(*argument, keep_attrs=False)
+    assert result.attrs == {}
+    assert result.da.attrs == {}
+    assert result.da_not_rolled.attrs == {}
+    assert result.da.name == "da"
+    assert result.da_not_rolled.name == "da_not_rolled"
+
+    # test discard attrs using global option
+    func = getattr(ds.rolling(dim={"coord": 5}), funcname)
+    with set_options(keep_attrs=False):
+        result = func(*argument)
+
+    assert result.attrs == {}
+    assert result.da.attrs == {}
+    assert result.da_not_rolled.attrs == {}
+    assert result.da.name == "da"
+    assert result.da_not_rolled.name == "da_not_rolled"
+
+    # keyword takes precedence over global option
+    func = getattr(ds.rolling(dim={"coord": 5}), funcname)
+    with set_options(keep_attrs=False):
+        result = func(*argument, keep_attrs=True)
+
+    assert result.attrs == global_attrs
+    assert result.da.attrs == da_attrs
+    assert result.da_not_rolled.attrs == da_not_rolled_attrs
+    assert result.da.name == "da"
+    assert result.da_not_rolled.name == "da_not_rolled"
+
+    func = getattr(ds.rolling(dim={"coord": 5}), funcname)
+    with set_options(keep_attrs=True):
+        result = func(*argument, keep_attrs=False)
 
-    # Test dropped attrs
-    dat = ds.rolling(dim={"coord": 5}, min_periods=None, center=False).mean()
-    assert dat.attrs == {}
+    assert result.attrs == {}
+    assert result.da.attrs == {}
+    assert result.da_not_rolled.attrs == {}
+    assert result.da.name == "da"
+    assert result.da_not_rolled.name == "da_not_rolled"
 
-    # Test kept attrs using dataset keyword
-    dat = ds.rolling(
-        dim={"coord": 5}, min_periods=None, center=False, keep_attrs=True
-    ).mean()
-    assert dat.attrs == _attrs
 
-    # Test kept attrs using global option
-    with set_options(keep_attrs=True):
-        dat = ds.rolling(dim={"coord": 5}, min_periods=None, center=False).mean()
-    assert dat.attrs == _attrs
+def test_rolling_keep_attrs_deprecated():
+    global_attrs = {"units": "test", "long_name": "testing"}
+    attrs_da = {"da_attr": "test"}
+
+    data = np.linspace(10, 15, 100)
+    coords = np.linspace(1, 10, 100)
+
+    ds = Dataset(
+        data_vars={"da": ("coord", data)},
+        coords={"coord": coords},
+        attrs=global_attrs,
+    )
+    ds.da.attrs = attrs_da
+
+    # deprecated option
+    with pytest.warns(
+        FutureWarning, match="Passing ``keep_attrs`` to ``rolling`` is deprecated"
+    ):
+        result = ds.rolling(dim={"coord": 5}, keep_attrs=False).construct("window_dim")
+
+    assert result.attrs == {}
+    assert result.da.attrs == {}
+
+    # the keep_attrs in the reduction function takes precedence
+    with pytest.warns(
+        FutureWarning, match="Passing ``keep_attrs`` to ``rolling`` is deprecated"
+    ):
+        result = ds.rolling(dim={"coord": 5}, keep_attrs=True).construct(
+            "window_dim", keep_attrs=False
+        )
+
+    assert result.attrs == {}
+    assert result.da.attrs == {}
 
 
 def test_rolling_properties(ds):
diff --git a/xarray/tests/test_weighted.py b/xarray/tests/test_weighted.py
index 48fad296..2366b982 100644
--- a/xarray/tests/test_weighted.py
+++ b/xarray/tests/test_weighted.py
@@ -5,6 +5,8 @@ import xarray as xr
 from xarray import DataArray
 from xarray.tests import assert_allclose, assert_equal, raises_regex
 
+from . import raise_if_dask_computes, requires_dask
+
 
 @pytest.mark.parametrize("as_dataset", (True, False))
 def test_weighted_non_DataArray_weights(as_dataset):
@@ -29,6 +31,24 @@ def test_weighted_weights_nan_raises(as_dataset, weights):
         data.weighted(DataArray(weights))
 
 
+@requires_dask
+@pytest.mark.parametrize("as_dataset", (True, False))
+@pytest.mark.parametrize("weights", ([np.nan, 2], [np.nan, np.nan]))
+def test_weighted_weights_nan_raises_dask(as_dataset, weights):
+
+    data = DataArray([1, 2]).chunk({"dim_0": -1})
+    if as_dataset:
+        data = data.to_dataset(name="data")
+
+    weights = DataArray(weights).chunk({"dim_0": -1})
+
+    with raise_if_dask_computes():
+        weighted = data.weighted(weights)
+
+    with pytest.raises(ValueError, match="`weights` cannot contain missing values."):
+        weighted.sum().load()
+
+
 @pytest.mark.parametrize(
     ("weights", "expected"),
     (([1, 2], 3), ([2, 0], 2), ([0, 0], np.nan), ([-1, 1], np.nan)),