Repository: pydata/xarray

Issue:
failing upstream-dev polyfit warning test
In `xarray/tests/test_dataset.py::TestDataset::test_polyfit_warnings`, the second call to `polyfit`:
```python
ds.var1.polyfit("dim2", 10, full=True)
```

emits a warning with the current `upstream-dev` `pandas`:
```
xarray/tests/test_dataset.py::TestDataset::test_polyfit_warnings
  .../xarray/core/alignment.py:307: FutureWarning: Index.__or__ operating as a set operation is deprecated, in the future this will be a logical operation matching Series.__or__.  Use index.union(other) instead
    index = joiner(matching_indexes)
```

since the test makes sure only the first call emits a warning (a `np.RankWarning`), the test fails.

Discussion:

Patch:
diff --git a/doc/whats-new.rst b/doc/whats-new.rst
index 60364a87..dd3120f7 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
@@ -97,6 +107,9 @@ Internal Changes
   By `Mathias Hauser <https://github.com/mathause>`_.
 - Ensure tests are not skipped in the `py38-all-but-dask` test environment
   (:issue:`4509`). By `Mathias Hauser <https://github.com/mathause>`_.
+- Replace the internal use of ``pd.Index.__or__`` and ``pd.Index.__and__`` with ``pd.Index.union``
+  and ``pd.Index.intersection`` as they will stop working as set operations in the future
+  (:issue:`4565`). By `Mathias Hauser <https://github.com/mathause>`_.
 
 .. _whats-new.0.16.1:
 
diff --git a/xarray/core/alignment.py b/xarray/core/alignment.py
index b42a5b2c..21bda8ef 100644
--- a/xarray/core/alignment.py
+++ b/xarray/core/alignment.py
@@ -32,9 +32,9 @@ if TYPE_CHECKING:
 
 def _get_joiner(join):
     if join == "outer":
-        return functools.partial(functools.reduce, operator.or_)
+        return functools.partial(functools.reduce, pd.Index.union)
     elif join == "inner":
-        return functools.partial(functools.reduce, operator.and_)
+        return functools.partial(functools.reduce, pd.Index.intersection)
     elif join == "left":
         return operator.itemgetter(0)
     elif join == "right":
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)),