Repository: modin-project/modin

Issue:
Implement `groupby.rolling` API
Currently it's defaulting to pandas: https://github.com/modin-project/modin/blob/2a0566efc70661428583fbd799861d9355c158f4/modin/pandas/groupby.py#L1144-L1145


Discussion:

Patch:
diff --git a/modin/core/dataframe/algebra/default2pandas/groupby.py b/modin/core/dataframe/algebra/default2pandas/groupby.py
--- a/modin/core/dataframe/algebra/default2pandas/groupby.py
+++ b/modin/core/dataframe/algebra/default2pandas/groupby.py
@@ -601,10 +601,13 @@ def register(cls, func, **kwargs):
     #   2. `.apply(func)` applies func to a DataFrames, holding a whole group (group-wise).
     #   3. `.transform(func)` is the same as `.apply()` but also broadcast the `func`
     #      result to the group's original shape.
+    #   4. 'direct' mode means that the passed `func` has to be applied directly
+    #      to the `pandas.DataFrameGroupBy` object.
     _aggregation_methods_dict = {
         "axis_wise": pandas.core.groupby.DataFrameGroupBy.aggregate,
         "group_wise": pandas.core.groupby.DataFrameGroupBy.apply,
         "transform": pandas.core.groupby.DataFrameGroupBy.transform,
+        "direct": lambda grp, func, *args, **kwargs: func(grp, *args, **kwargs),
     }
 
     @classmethod
@@ -637,4 +640,5 @@ class SeriesGroupByDefault(GroupByDefault):
         "axis_wise": pandas.core.groupby.SeriesGroupBy.aggregate,
         "group_wise": pandas.core.groupby.SeriesGroupBy.apply,
         "transform": pandas.core.groupby.SeriesGroupBy.transform,
+        "direct": lambda grp, func, *args, **kwargs: func(grp, *args, **kwargs),
     }
diff --git a/modin/core/dataframe/algebra/default2pandas/rolling.py b/modin/core/dataframe/algebra/default2pandas/rolling.py
--- a/modin/core/dataframe/algebra/default2pandas/rolling.py
+++ b/modin/core/dataframe/algebra/default2pandas/rolling.py
@@ -37,9 +37,9 @@ def _build_rolling(cls, func):
             Function that takes pandas DataFrame and applies `func` on a rolling window.
         """
 
-        def fn(df, rolling_args, *args, **kwargs):
+        def fn(df, rolling_kwargs, *args, **kwargs):
             """Create rolling window for the passed frame and execute specified `func` on it."""
-            roller = df.rolling(*rolling_args)
+            roller = df.rolling(**rolling_kwargs)
 
             if type(func) == property:
                 return func.fget(roller)
diff --git a/modin/core/storage_formats/base/doc_utils.py b/modin/core/storage_formats/base/doc_utils.py
--- a/modin/core/storage_formats/base/doc_utils.py
+++ b/modin/core/storage_formats/base/doc_utils.py
@@ -598,11 +598,11 @@ def doc_window_method(
     if action is None:
         action = f"compute {result}"
     if win_type == "rolling window":
-        window_args_name = "rolling_args"
+        window_args_name = "rolling_kwargs"
     elif win_type == "expanding window":
         window_args_name = "expanding_args"
     else:
-        window_args_name = "window_args"
+        window_args_name = "window_kwargs"
 
     # We need that `params` value ended with new line to have
     # an empty line between "parameters" and "return" sections
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -3021,6 +3021,68 @@ def groupby_size(
             result.columns = result.columns[:-1].append(pandas.Index(["size"]))
         return result
 
+    @doc_utils.add_refer_to("GroupBy.rolling")
+    def groupby_rolling(
+        self,
+        by,
+        agg_func,
+        axis,
+        groupby_kwargs,
+        rolling_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        """
+        Group QueryCompiler data and apply passed aggregation function to a rolling window in each group.
+
+        Parameters
+        ----------
+        by : BaseQueryCompiler, column or index label, Grouper or list of such
+            Object that determine groups.
+        agg_func : str, dict or callable(Series | DataFrame) -> scalar | Series | DataFrame
+            Function to apply to the GroupBy object.
+        axis : {0, 1}
+            Axis to group and apply aggregation function along.
+            0 is for index, when 1 is for columns.
+        groupby_kwargs : dict
+            GroupBy parameters as expected by ``modin.pandas.DataFrame.groupby`` signature.
+        rolling_kwargs : dict
+            Parameters to build a rolling window as expected by ``modin.pandas.window.RollingGroupby`` signature.
+        agg_args : list-like
+            Positional arguments to pass to the `agg_func`.
+        agg_kwargs : dict
+            Key arguments to pass to the `agg_func`.
+        drop : bool, default: False
+            If `by` is a QueryCompiler indicates whether or not by-data came
+            from the `self`.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            QueryCompiler containing the result of groupby aggregation.
+        """
+        if isinstance(agg_func, str):
+            str_func = agg_func
+
+            def agg_func(window, *args, **kwargs):
+                return getattr(window, str_func)(*args, **kwargs)
+
+        else:
+            assert callable(agg_func)
+        return self.groupby_agg(
+            by=by,
+            agg_func=lambda grp, *args, **kwargs: agg_func(
+                grp.rolling(**rolling_kwargs), *args, **kwargs
+            ),
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            how="direct",
+            drop=drop,
+        )
+
     @doc_utils.add_refer_to("GroupBy.aggregate")
     def groupby_agg(
         self,
@@ -5760,9 +5822,9 @@ def str_casefold(self):
         **kwargs : dict""",
         build_rules="udf_aggregation",
     )
-    def rolling_aggregate(self, fold_axis, rolling_args, func, *args, **kwargs):
+    def rolling_aggregate(self, fold_axis, rolling_kwargs, func, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.aggregate)(
-            self, rolling_args, func, *args, **kwargs
+            self, rolling_kwargs, func, *args, **kwargs
         )
 
     # FIXME: at the query compiler method `rolling_apply` is an alias for `rolling_aggregate`,
@@ -5787,7 +5849,7 @@ def rolling_aggregate(self, fold_axis, rolling_args, func, *args, **kwargs):
     def rolling_apply(
         self,
         fold_axis,
-        rolling_args,
+        rolling_kwargs,
         func,
         raw=False,
         engine=None,
@@ -5796,7 +5858,7 @@ def rolling_apply(
         kwargs=None,
     ):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.apply)(
-            self, rolling_args, func, raw, engine, engine_kwargs, args, kwargs
+            self, rolling_kwargs, func, raw, engine, engine_kwargs, args, kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5810,18 +5872,18 @@ def rolling_apply(
         **kwargs : dict""",
     )
     def rolling_corr(
-        self, fold_axis, rolling_args, other=None, pairwise=None, *args, **kwargs
+        self, fold_axis, rolling_kwargs, other=None, pairwise=None, *args, **kwargs
     ):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.corr)(
-            self, rolling_args, other, pairwise, *args, **kwargs
+            self, rolling_kwargs, other, pairwise, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
         window_cls_name="Rolling", result="number of non-NA values", refer_to="count"
     )
-    def rolling_count(self, fold_axis, rolling_args):
+    def rolling_count(self, fold_axis, rolling_kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.count)(
-            self, rolling_args
+            self, rolling_kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5835,10 +5897,10 @@ def rolling_count(self, fold_axis, rolling_args):
         **kwargs : dict""",
     )
     def rolling_cov(
-        self, fold_axis, rolling_args, other=None, pairwise=None, ddof=1, **kwargs
+        self, fold_axis, rolling_kwargs, other=None, pairwise=None, ddof=1, **kwargs
     ):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.cov)(
-            self, rolling_args, other, pairwise, ddof, **kwargs
+            self, rolling_kwargs, other, pairwise, ddof, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5847,9 +5909,9 @@ def rolling_cov(
         refer_to="kurt",
         params="**kwargs : dict",
     )
-    def rolling_kurt(self, fold_axis, rolling_args, **kwargs):
+    def rolling_kurt(self, fold_axis, rolling_kwargs, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.kurt)(
-            self, rolling_args, **kwargs
+            self, rolling_kwargs, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5860,9 +5922,9 @@ def rolling_kurt(self, fold_axis, rolling_args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def rolling_max(self, fold_axis, rolling_args, *args, **kwargs):
+    def rolling_max(self, fold_axis, rolling_kwargs, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.max)(
-            self, rolling_args, *args, **kwargs
+            self, rolling_kwargs, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5873,9 +5935,9 @@ def rolling_max(self, fold_axis, rolling_args, *args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def rolling_mean(self, fold_axis, rolling_args, *args, **kwargs):
+    def rolling_mean(self, fold_axis, rolling_kwargs, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.mean)(
-            self, rolling_args, *args, **kwargs
+            self, rolling_kwargs, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5884,9 +5946,9 @@ def rolling_mean(self, fold_axis, rolling_args, *args, **kwargs):
         refer_to="median",
         params="**kwargs : dict",
     )
-    def rolling_median(self, fold_axis, rolling_args, **kwargs):
+    def rolling_median(self, fold_axis, rolling_kwargs, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.median)(
-            self, rolling_args, **kwargs
+            self, rolling_kwargs, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5897,9 +5959,9 @@ def rolling_median(self, fold_axis, rolling_args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def rolling_min(self, fold_axis, rolling_args, *args, **kwargs):
+    def rolling_min(self, fold_axis, rolling_kwargs, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.min)(
-            self, rolling_args, *args, **kwargs
+            self, rolling_kwargs, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5912,10 +5974,10 @@ def rolling_min(self, fold_axis, rolling_args, *args, **kwargs):
         **kwargs : dict""",
     )
     def rolling_quantile(
-        self, fold_axis, rolling_args, quantile, interpolation="linear", **kwargs
+        self, fold_axis, rolling_kwargs, quantile, interpolation="linear", **kwargs
     ):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.quantile)(
-            self, rolling_args, quantile, interpolation, **kwargs
+            self, rolling_kwargs, quantile, interpolation, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5924,9 +5986,9 @@ def rolling_quantile(
         refer_to="skew",
         params="**kwargs : dict",
     )
-    def rolling_skew(self, fold_axis, rolling_args, **kwargs):
+    def rolling_skew(self, fold_axis, rolling_kwargs, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.skew)(
-            self, rolling_args, **kwargs
+            self, rolling_kwargs, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5938,9 +6000,9 @@ def rolling_skew(self, fold_axis, rolling_args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def rolling_std(self, fold_axis, rolling_args, ddof=1, *args, **kwargs):
+    def rolling_std(self, fold_axis, rolling_kwargs, ddof=1, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.std)(
-            self, rolling_args, ddof, *args, **kwargs
+            self, rolling_kwargs, ddof, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5951,9 +6013,9 @@ def rolling_std(self, fold_axis, rolling_args, ddof=1, *args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def rolling_sum(self, fold_axis, rolling_args, *args, **kwargs):
+    def rolling_sum(self, fold_axis, rolling_kwargs, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.sum)(
-            self, rolling_args, *args, **kwargs
+            self, rolling_kwargs, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5964,9 +6026,9 @@ def rolling_sum(self, fold_axis, rolling_args, *args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def rolling_sem(self, fold_axis, rolling_args, *args, **kwargs):
+    def rolling_sem(self, fold_axis, rolling_kwargs, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.sem)(
-            self, rolling_args, *args, **kwargs
+            self, rolling_kwargs, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5978,9 +6040,9 @@ def rolling_sem(self, fold_axis, rolling_args, *args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def rolling_var(self, fold_axis, rolling_args, ddof=1, *args, **kwargs):
+    def rolling_var(self, fold_axis, rolling_kwargs, ddof=1, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.var)(
-            self, rolling_args, ddof, *args, **kwargs
+            self, rolling_kwargs, ddof, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -5998,7 +6060,7 @@ def rolling_var(self, fold_axis, rolling_args, ddof=1, *args, **kwargs):
     def rolling_rank(
         self,
         fold_axis,
-        rolling_args,
+        rolling_kwargs,
         method="average",
         ascending=True,
         pct=False,
@@ -6008,7 +6070,7 @@ def rolling_rank(
     ):
         return RollingDefault.register(pandas.core.window.rolling.Rolling.rank)(
             self,
-            rolling_args,
+            rolling_kwargs,
             method=method,
             ascending=ascending,
             pct=pct,
@@ -6371,9 +6433,9 @@ def expanding_rank(
         *args : iterable
         **kwargs : dict""",
     )
-    def window_mean(self, fold_axis, window_args, *args, **kwargs):
+    def window_mean(self, fold_axis, window_kwargs, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.Window.mean)(
-            self, window_args, *args, **kwargs
+            self, window_kwargs, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -6386,9 +6448,9 @@ def window_mean(self, fold_axis, window_args, *args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def window_std(self, fold_axis, window_args, ddof=1, *args, **kwargs):
+    def window_std(self, fold_axis, window_kwargs, ddof=1, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.Window.std)(
-            self, window_args, ddof, *args, **kwargs
+            self, window_kwargs, ddof, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -6400,9 +6462,9 @@ def window_std(self, fold_axis, window_args, ddof=1, *args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def window_sum(self, fold_axis, window_args, *args, **kwargs):
+    def window_sum(self, fold_axis, window_kwargs, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.Window.sum)(
-            self, window_args, *args, **kwargs
+            self, window_kwargs, *args, **kwargs
         )
 
     @doc_utils.doc_window_method(
@@ -6415,9 +6477,9 @@ def window_sum(self, fold_axis, window_args, *args, **kwargs):
         *args : iterable
         **kwargs : dict""",
     )
-    def window_var(self, fold_axis, window_args, ddof=1, *args, **kwargs):
+    def window_var(self, fold_axis, window_kwargs, ddof=1, *args, **kwargs):
         return RollingDefault.register(pandas.core.window.Window.var)(
-            self, window_args, ddof, *args, **kwargs
+            self, window_kwargs, ddof, *args, **kwargs
         )
 
     # End of Window methods
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -21,7 +21,6 @@
 import re
 import numpy as np
 import pandas
-import functools
 from pandas.api.types import is_scalar
 from pandas.core.common import is_bool_indexer
 from pandas.core.indexing import check_bool_indexer
@@ -1378,81 +1377,83 @@ def expanding_corr(
     )
 
     window_mean = Fold.register(
-        lambda df, rolling_args, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).mean(*args, **kwargs)
+        lambda df, rolling_kwargs, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).mean(*args, **kwargs)
         )
     )
     window_sum = Fold.register(
-        lambda df, rolling_args, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).sum(*args, **kwargs)
+        lambda df, rolling_kwargs, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).sum(*args, **kwargs)
         )
     )
     window_var = Fold.register(
-        lambda df, rolling_args, ddof, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).var(ddof=ddof, *args, **kwargs)
+        lambda df, rolling_kwargs, ddof, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).var(ddof=ddof, *args, **kwargs)
         )
     )
     window_std = Fold.register(
-        lambda df, rolling_args, ddof, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).std(ddof=ddof, *args, **kwargs)
+        lambda df, rolling_kwargs, ddof, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).std(ddof=ddof, *args, **kwargs)
         )
     )
     rolling_count = Fold.register(
-        lambda df, rolling_args: pandas.DataFrame(df.rolling(*rolling_args).count())
+        lambda df, rolling_kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).count()
+        )
     )
     rolling_sum = Fold.register(
-        lambda df, rolling_args, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).sum(*args, **kwargs)
+        lambda df, rolling_kwargs, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).sum(*args, **kwargs)
         )
     )
     rolling_sem = Fold.register(
-        lambda df, rolling_args, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).sem(*args, **kwargs)
+        lambda df, rolling_kwargs, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).sem(*args, **kwargs)
         )
     )
     rolling_mean = Fold.register(
-        lambda df, rolling_args, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).mean(*args, **kwargs)
+        lambda df, rolling_kwargs, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).mean(*args, **kwargs)
         )
     )
     rolling_median = Fold.register(
-        lambda df, rolling_args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).median(**kwargs)
+        lambda df, rolling_kwargs, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).median(**kwargs)
         )
     )
     rolling_var = Fold.register(
-        lambda df, rolling_args, ddof, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).var(ddof=ddof, *args, **kwargs)
+        lambda df, rolling_kwargs, ddof, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).var(ddof=ddof, *args, **kwargs)
         )
     )
     rolling_std = Fold.register(
-        lambda df, rolling_args, ddof, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).std(ddof=ddof, *args, **kwargs)
+        lambda df, rolling_kwargs, ddof, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).std(ddof=ddof, *args, **kwargs)
         )
     )
     rolling_min = Fold.register(
-        lambda df, rolling_args, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).min(*args, **kwargs)
+        lambda df, rolling_kwargs, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).min(*args, **kwargs)
         )
     )
     rolling_max = Fold.register(
-        lambda df, rolling_args, *args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).max(*args, **kwargs)
+        lambda df, rolling_kwargs, *args, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).max(*args, **kwargs)
         )
     )
     rolling_skew = Fold.register(
-        lambda df, rolling_args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).skew(**kwargs)
+        lambda df, rolling_kwargs, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).skew(**kwargs)
         )
     )
     rolling_kurt = Fold.register(
-        lambda df, rolling_args, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).kurt(**kwargs)
+        lambda df, rolling_kwargs, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).kurt(**kwargs)
         )
     )
     rolling_apply = Fold.register(
-        lambda df, rolling_args, func, raw, engine, engine_kwargs, args, kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).apply(
+        lambda df, rolling_kwargs, func, raw, engine, engine_kwargs, args, kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).apply(
                 func=func,
                 raw=raw,
                 engine=engine,
@@ -1463,15 +1464,15 @@ def expanding_corr(
         )
     )
     rolling_quantile = Fold.register(
-        lambda df, rolling_args, quantile, interpolation, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).quantile(
+        lambda df, rolling_kwargs, quantile, interpolation, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).quantile(
                 quantile=quantile, interpolation=interpolation, **kwargs
             )
         )
     )
     rolling_rank = Fold.register(
-        lambda df, rolling_args, method, ascending, pct, numeric_only, **kwargs: pandas.DataFrame(
-            df.rolling(*rolling_args).rank(
+        lambda df, rolling_kwargs, method, ascending, pct, numeric_only, **kwargs: pandas.DataFrame(
+            df.rolling(**rolling_kwargs).rank(
                 method=method,
                 ascending=ascending,
                 pct=pct,
@@ -1481,43 +1482,43 @@ def expanding_corr(
         )
     )
 
-    def rolling_corr(self, axis, rolling_args, other, pairwise, *args, **kwargs):
+    def rolling_corr(self, axis, rolling_kwargs, other, pairwise, *args, **kwargs):
         if len(self.columns) > 1:
             return self.default_to_pandas(
-                lambda df: pandas.DataFrame.rolling(df, *rolling_args).corr(
+                lambda df: pandas.DataFrame.rolling(df, **rolling_kwargs).corr(
                     other=other, pairwise=pairwise, *args, **kwargs
                 )
             )
         else:
             return Fold.register(
                 lambda df: pandas.DataFrame(
-                    df.rolling(*rolling_args).corr(
+                    df.rolling(**rolling_kwargs).corr(
                         other=other, pairwise=pairwise, *args, **kwargs
                     )
                 )
             )(self, axis)
 
-    def rolling_cov(self, axis, rolling_args, other, pairwise, ddof, **kwargs):
+    def rolling_cov(self, axis, rolling_kwargs, other, pairwise, ddof, **kwargs):
         if len(self.columns) > 1:
             return self.default_to_pandas(
-                lambda df: pandas.DataFrame.rolling(df, *rolling_args).cov(
+                lambda df: pandas.DataFrame.rolling(df, **rolling_kwargs).cov(
                     other=other, pairwise=pairwise, ddof=ddof, **kwargs
                 )
             )
         else:
             return Fold.register(
                 lambda df: pandas.DataFrame(
-                    df.rolling(*rolling_args).cov(
+                    df.rolling(**rolling_kwargs).cov(
                         other=other, pairwise=pairwise, ddof=ddof, **kwargs
                     )
                 )
             )(self, axis)
 
-    def rolling_aggregate(self, axis, rolling_args, func, *args, **kwargs):
+    def rolling_aggregate(self, axis, rolling_kwargs, func, *args, **kwargs):
         new_modin_frame = self._modin_frame.apply_full_axis(
             axis,
             lambda df: pandas.DataFrame(
-                df.rolling(*rolling_args).aggregate(func=func, *args, **kwargs)
+                df.rolling(**rolling_kwargs).aggregate(func=func, *args, **kwargs)
             ),
             new_index=self.index,
         )
@@ -3439,9 +3440,11 @@ def _groupby_shuffle(
         else:
             obj = self
 
-        agg_func = functools.partial(
-            GroupByDefault.get_aggregation_method(how), func=agg_func
-        )
+        agg_method = GroupByDefault.get_aggregation_method(how)
+        original_agg_func = agg_func
+
+        def agg_func(grp, *args, **kwargs):
+            return agg_method(grp, original_agg_func, *args, **kwargs)
 
         result = obj._modin_frame.groupby(
             axis=axis,
@@ -3504,6 +3507,51 @@ def groupby_cov(
             drop=drop,
         )
 
+    def groupby_rolling(
+        self,
+        by,
+        agg_func,
+        axis,
+        groupby_kwargs,
+        rolling_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        # 'corr' and 'cov' require knowledge about the whole row axis (all columns have
+        # to be available in the same partitions), this requirement is not being satisfied
+        # in the current groupby implementation
+        unsupported_groupby = (
+            agg_func in ("corr", "cov") or rolling_kwargs.get("on") is not None
+        )
+
+        if isinstance(agg_func, str):
+            str_func = agg_func
+
+            def agg_func(window, *args, **kwargs):
+                return getattr(window, str_func)(*args, **kwargs)
+
+        else:
+            assert callable(agg_func)
+
+        if unsupported_groupby:
+            obj = super(PandasQueryCompiler, self)
+        else:
+            obj = self
+
+        return obj.groupby_agg(
+            by=by,
+            agg_func=lambda grp, *args, **kwargs: agg_func(
+                grp.rolling(**rolling_kwargs), *args, **kwargs
+            ),
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            how="direct",
+            drop=drop,
+        )
+
     def groupby_agg(
         self,
         by,
@@ -3559,12 +3607,12 @@ def groupby_agg(
                 how == "axis_wise"
             ), f"Only 'axis_wise' aggregation is supported with dictionary functions, got: {how}"
         else:
-            agg_func = functools.partial(
-                (
-                    SeriesGroupByDefault if series_groupby else GroupByDefault
-                ).get_aggregation_method(how),
-                func=agg_func,
-            )
+            agg_method = (
+                SeriesGroupByDefault if series_groupby else GroupByDefault
+            ).get_aggregation_method(how)
+
+            def agg_func(grp, *args, **kwargs):
+                return agg_method(grp, original_agg_func, *args, **kwargs)
 
         # since we're going to modify `groupby_kwargs` dict in a `groupby_agg_builder`,
         # we want to copy it to not propagate these changes into source dict, in case
diff --git a/modin/pandas/groupby.py b/modin/pandas/groupby.py
--- a/modin/pandas/groupby.py
+++ b/modin/pandas/groupby.py
@@ -39,6 +39,7 @@
 from modin.core.dataframe.algebra.default2pandas.groupby import GroupBy
 from modin.config import IsExperimental
 from .series import Series
+from .window import RollingGroupby
 from .utils import is_label
 
 
@@ -135,7 +136,20 @@ def __init__(
         }
         self._kwargs.update(kwargs)
 
-    def __override(self, **kwargs):
+    def _override(self, **kwargs):
+        """
+        Override groupby parameters.
+
+        Parameters
+        ----------
+        **kwargs : dict
+            Parameters to override.
+
+        Returns
+        -------
+        DataFrameGroupBy
+            A groupby object with new parameters.
+        """
         new_kw = dict(
             df=self._df,
             by=self._by,
@@ -836,7 +850,7 @@ def do_relabel(obj_to_relabel):
             # for list-list aggregation pandas always puts
             # groups as index in the result, ignoring as_index,
             # so we have to reset it to default value
-            res = self.__override(as_index=True)._wrap_aggregation(
+            res = self._override(as_index=True)._wrap_aggregation(
                 qc_method=type(self._query_compiler).groupby_agg,
                 numeric_only=False,
                 agg_func=func,
@@ -916,7 +930,7 @@ def var(self, ddof=1, engine=None, engine_kwargs=None, numeric_only=False):
         )
 
     def get_group(self, name, obj=None):
-        work_object = self.__override(
+        work_object = self._override(
             df=obj if obj is not None else self._df, as_index=True
         )
 
@@ -1067,7 +1081,7 @@ def median(self, numeric_only=False):
 
     def head(self, n=5):
         # groupby().head()/.tail() ignore as_index, so override it to True
-        work_object = self.__override(as_index=True)
+        work_object = self._override(as_index=True)
 
         return work_object._check_index(
             work_object._wrap_aggregation(
@@ -1184,7 +1198,7 @@ def cumcount(self, ascending=True):
 
     def tail(self, n=5):
         # groupby().head()/.tail() ignore as_index, so override it to True
-        work_object = self.__override(as_index=True)
+        work_object = self._override(as_index=True)
         return work_object._check_index(
             work_object._wrap_aggregation(
                 type(work_object._query_compiler).groupby_tail,
@@ -1199,7 +1213,7 @@ def expanding(self, *args, **kwargs):
         return self._default_to_pandas(lambda df: df.expanding(*args, **kwargs))
 
     def rolling(self, *args, **kwargs):
-        return self._default_to_pandas(lambda df: df.rolling(*args, **kwargs))
+        return RollingGroupby(self, *args, **kwargs)
 
     def hist(
         self,
diff --git a/modin/pandas/window.py b/modin/pandas/window.py
--- a/modin/pandas/window.py
+++ b/modin/pandas/window.py
@@ -20,6 +20,7 @@
 from modin.logging import ClassLogger
 from modin.utils import _inherit_docstrings
 from modin.pandas.utils import cast_function_modin2pandas
+from modin.error_message import ErrorMessage
 
 
 @_inherit_docstrings(pandas.core.window.rolling.Window)
@@ -39,44 +40,44 @@ def __init__(
     ):
         self._dataframe = dataframe
         self._query_compiler = dataframe._query_compiler
-        self.window_args = [
-            window,
-            min_periods,
-            center,
-            win_type,
-            on,
-            axis,
-            closed,
-            step,
-            method,
-        ]
+        self.window_kwargs = {
+            "window": window,
+            "min_periods": min_periods,
+            "center": center,
+            "win_type": win_type,
+            "on": on,
+            "axis": axis,
+            "closed": closed,
+            "step": step,
+            "method": method,
+        }
         self.axis = axis
 
     def mean(self, *args, **kwargs):
         return self._dataframe.__constructor__(
             query_compiler=self._query_compiler.window_mean(
-                self.axis, self.window_args, *args, **kwargs
+                self.axis, self.window_kwargs, *args, **kwargs
             )
         )
 
     def sum(self, *args, **kwargs):
         return self._dataframe.__constructor__(
             query_compiler=self._query_compiler.window_sum(
-                self.axis, self.window_args, *args, **kwargs
+                self.axis, self.window_kwargs, *args, **kwargs
             )
         )
 
     def var(self, ddof=1, *args, **kwargs):
         return self._dataframe.__constructor__(
             query_compiler=self._query_compiler.window_var(
-                self.axis, self.window_args, ddof, *args, **kwargs
+                self.axis, self.window_kwargs, ddof, *args, **kwargs
             )
         )
 
     def std(self, ddof=1, *args, **kwargs):
         return self._dataframe.__constructor__(
             query_compiler=self._query_compiler.window_std(
-                self.axis, self.window_args, ddof, *args, **kwargs
+                self.axis, self.window_kwargs, ddof, *args, **kwargs
             )
         )
 
@@ -103,81 +104,87 @@ def __init__(
             raise NotImplementedError("step parameter is not implemented yet.")
         self._dataframe = dataframe
         self._query_compiler = dataframe._query_compiler
-        self.rolling_args = [
-            window,
-            min_periods,
-            center,
-            win_type,
-            on,
-            axis,
-            closed,
-            step,
-            method,
-        ]
+        self.rolling_kwargs = {
+            "window": window,
+            "min_periods": min_periods,
+            "center": center,
+            "win_type": win_type,
+            "on": on,
+            "axis": axis,
+            "closed": closed,
+            "step": step,
+            "method": method,
+        }
         self.axis = axis
 
+    def _call_qc_method(self, method_name, *args, **kwargs):
+        """
+        Call a query compiler method for the specified rolling aggregation.
+
+        Parameters
+        ----------
+        method_name : str
+            Name of the aggregation.
+        *args : tuple
+            Positional arguments to pass to the query compiler method.
+        **kwargs : dict
+            Keyword arguments to pass to the query compiler method.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            QueryCompiler holding the result of the aggregation.
+        """
+        qc_method = getattr(self._query_compiler, f"rolling_{method_name}")
+        return qc_method(self.axis, self.rolling_kwargs, *args, **kwargs)
+
+    def _aggregate(self, method_name, *args, **kwargs):
+        """
+        Run the specified rolling aggregation.
+
+        Parameters
+        ----------
+        method_name : str
+            Name of the aggregation.
+        *args : tuple
+            Positional arguments to pass to the aggregation.
+        **kwargs : dict
+            Keyword arguments to pass to the aggregation.
+
+        Returns
+        -------
+        DataFrame or Series
+            Result of the aggregation.
+        """
+        qc_result = self._call_qc_method(method_name, *args, **kwargs)
+        return self._dataframe.__constructor__(query_compiler=qc_result)
+
     def count(self):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_count(
-                self.axis, self.rolling_args
-            )
-        )
+        return self._aggregate("count")
 
     def sem(self, *args, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_sem(
-                self.axis, self.rolling_args, *args, **kwargs
-            )
-        )
+        return self._aggregate("sem", *args, **kwargs)
 
     def sum(self, *args, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_sum(
-                self.axis, self.rolling_args, *args, **kwargs
-            )
-        )
+        return self._aggregate("sum", *args, **kwargs)
 
     def mean(self, *args, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_mean(
-                self.axis, self.rolling_args, *args, **kwargs
-            )
-        )
+        return self._aggregate("mean", *args, **kwargs)
 
     def median(self, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_median(
-                self.axis, self.rolling_args, **kwargs
-            )
-        )
+        return self._aggregate("median", **kwargs)
 
     def var(self, ddof=1, *args, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_var(
-                self.axis, self.rolling_args, ddof, *args, **kwargs
-            )
-        )
+        return self._aggregate("var", ddof, *args, **kwargs)
 
     def std(self, ddof=1, *args, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_std(
-                self.axis, self.rolling_args, ddof, *args, **kwargs
-            )
-        )
+        return self._aggregate("std", ddof, *args, **kwargs)
 
     def min(self, *args, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_min(
-                self.axis, self.rolling_args, *args, **kwargs
-            )
-        )
+        return self._aggregate("min", *args, **kwargs)
 
     def max(self, *args, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_max(
-                self.axis, self.rolling_args, *args, **kwargs
-            )
-        )
+        return self._aggregate("max", *args, **kwargs)
 
     def corr(self, other=None, pairwise=None, *args, **kwargs):
         from .dataframe import DataFrame
@@ -188,11 +195,7 @@ def corr(self, other=None, pairwise=None, *args, **kwargs):
         elif isinstance(other, Series):
             other = other._query_compiler.to_pandas().squeeze()
 
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_corr(
-                self.axis, self.rolling_args, other, pairwise, *args, **kwargs
-            )
-        )
+        return self._aggregate("corr", other, pairwise, *args, **kwargs)
 
     def cov(self, other=None, pairwise=None, ddof: Optional[int] = 1, **kwargs):
         from .dataframe import DataFrame
@@ -203,25 +206,13 @@ def cov(self, other=None, pairwise=None, ddof: Optional[int] = 1, **kwargs):
         elif isinstance(other, Series):
             other = other._query_compiler.to_pandas().squeeze()
 
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_cov(
-                self.axis, self.rolling_args, other, pairwise, ddof, **kwargs
-            )
-        )
+        return self._aggregate("cov", other, pairwise, ddof, **kwargs)
 
     def skew(self, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_skew(
-                self.axis, self.rolling_args, **kwargs
-            )
-        )
+        return self._aggregate("skew", **kwargs)
 
     def kurt(self, **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_kurt(
-                self.axis, self.rolling_args, **kwargs
-            )
-        )
+        return self._aggregate("kurt", **kwargs)
 
     def apply(
         self,
@@ -233,18 +224,7 @@ def apply(
         kwargs=None,
     ):
         func = cast_function_modin2pandas(func)
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_apply(
-                self.axis,
-                self.rolling_args,
-                func,
-                raw,
-                engine,
-                engine_kwargs,
-                args,
-                kwargs,
-            )
-        )
+        return self._aggregate("apply", func, raw, engine, engine_kwargs, args, kwargs)
 
     def aggregate(
         self,
@@ -255,9 +235,8 @@ def aggregate(
         from .dataframe import DataFrame
 
         dataframe = DataFrame(
-            query_compiler=self._query_compiler.rolling_aggregate(
-                self.axis,
-                self.rolling_args,
+            query_compiler=self._call_qc_method(
+                "aggregate",
                 func,
                 *args,
                 **kwargs,
@@ -265,7 +244,7 @@ def aggregate(
         )
         if isinstance(self._dataframe, DataFrame):
             return dataframe
-        elif is_list_like(func):
+        elif is_list_like(func) and dataframe.columns.nlevels > 1:
             dataframe.columns = dataframe.columns.droplevel()
             return dataframe
         else:
@@ -274,26 +253,86 @@ def aggregate(
     agg = aggregate
 
     def quantile(self, quantile, interpolation="linear", **kwargs):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_quantile(
-                self.axis, self.rolling_args, quantile, interpolation, **kwargs
-            )
-        )
+        return self._aggregate("quantile", quantile, interpolation, **kwargs)
 
     def rank(
         self, method="average", ascending=True, pct=False, numeric_only=False, **kwargs
     ):
-        return self._dataframe.__constructor__(
-            query_compiler=self._query_compiler.rolling_rank(
-                self.axis,
-                self.rolling_args,
-                method,
-                ascending,
-                pct,
-                numeric_only,
-                **kwargs,
-            )
+        return self._aggregate("rank", method, ascending, pct, numeric_only, **kwargs)
+
+
+@_inherit_docstrings(Rolling)
+class RollingGroupby(Rolling):
+    def __init__(self, groupby_obj, *args, **kwargs):
+        self._as_index = groupby_obj._kwargs.get("as_index", True)
+        self._groupby_obj = (
+            groupby_obj if self._as_index else groupby_obj._override(as_index=True)
         )
+        super().__init__(self._groupby_obj._df, *args, **kwargs)
+
+    def sem(self, *args, **kwargs):
+        ErrorMessage.missmatch_with_pandas(
+            operation="RollingGroupby.sem() when 'as_index=False'",
+            message=(
+                "The group columns won't be involved in the aggregation.\n"
+                + "See this gh-issue for more information: https://github.com/modin-project/modin/issues/6291"
+            ),
+        )
+        return super().sem(*args, **kwargs)
+
+    def corr(self, other=None, pairwise=None, *args, **kwargs):
+        # pandas behavior is that it always assumes that 'as_index=True' for the '.corr()' method
+        return super().corr(
+            *args, as_index=True, other=other, pairwise=pairwise, **kwargs
+        )
+
+    def cov(self, other=None, pairwise=None, ddof: Optional[int] = 1, **kwargs):
+        # pandas behavior is that it always assumes that 'as_index=True' for the '.cov()' method
+        return super().cov(as_index=True, other=other, pairwise=pairwise, **kwargs)
+
+    def _aggregate(self, method_name, *args, as_index=None, **kwargs):
+        """
+        Run the specified rolling aggregation.
+
+        Parameters
+        ----------
+        method_name : str
+            Name of the aggregation.
+        *args : tuple
+            Positional arguments to pass to the aggregation.
+        as_index : bool, optional
+            Whether the result should have the group labels as index levels or as columns.
+            If not specified the parameter value will be taken from groupby kwargs.
+        **kwargs : dict
+            Keyword arguments to pass to the aggregation.
+
+        Returns
+        -------
+        DataFrame or Series
+            Result of the aggregation.
+        """
+        res = self._groupby_obj._wrap_aggregation(
+            qc_method=type(self._query_compiler).groupby_rolling,
+            numeric_only=False,
+            agg_args=args,
+            agg_kwargs=kwargs,
+            agg_func=method_name,
+            rolling_kwargs=self.rolling_kwargs,
+        )
+
+        if as_index is None:
+            as_index = self._as_index
+
+        if not as_index:
+            res = res.reset_index(
+                level=[i for i in range(len(self._groupby_obj._internal_by))],
+                drop=False,
+            )
+
+        return res
+
+    def _call_qc_method(self, method_name, *args, **kwargs):
+        return self._aggregate(method_name, *args, **kwargs)._query_compiler
 
 
 @_inherit_docstrings(


Test patch:
diff --git a/modin/pandas/test/test_groupby.py b/modin/pandas/test/test_groupby.py
--- a/modin/pandas/test/test_groupby.py
+++ b/modin/pandas/test/test_groupby.py
@@ -2651,3 +2651,109 @@ def test_groupby_pct_change_diff_6194():
         df._to_pandas(),
         lambda df: df.groupby(by="by").diff(),
     )
+
+
+def eval_rolling(md_window, pd_window):
+    eval_general(md_window, pd_window, lambda window: window.count())
+    eval_general(md_window, pd_window, lambda window: window.sum())
+    eval_general(md_window, pd_window, lambda window: window.mean())
+    eval_general(md_window, pd_window, lambda window: window.median())
+    eval_general(md_window, pd_window, lambda window: window.var())
+    eval_general(md_window, pd_window, lambda window: window.std())
+    eval_general(md_window, pd_window, lambda window: window.min())
+    eval_general(md_window, pd_window, lambda window: window.max())
+    eval_general(md_window, pd_window, lambda window: window.corr())
+    eval_general(md_window, pd_window, lambda window: window.cov())
+    eval_general(md_window, pd_window, lambda window: window.skew())
+    eval_general(md_window, pd_window, lambda window: window.kurt())
+    eval_general(
+        md_window, pd_window, lambda window: window.apply(lambda df: (df + 10).sum())
+    )
+    eval_general(md_window, pd_window, lambda window: window.agg("sum"))
+    eval_general(md_window, pd_window, lambda window: window.quantile(0.2))
+    eval_general(md_window, pd_window, lambda window: window.rank())
+
+    if not md_window._as_index:
+        # There's a mismatch in group columns when 'as_index=False'
+        # see: https://github.com/modin-project/modin/issues/6291
+        by_cols = list(md_window._groupby_obj._internal_by)
+        eval_general(
+            md_window,
+            pd_window,
+            lambda window: window.sem().drop(columns=by_cols, errors="ignore"),
+        )
+    else:
+        eval_general(
+            md_window,
+            pd_window,
+            lambda window: window.sem(),
+        )
+
+
+@pytest.mark.parametrize("center", [True, False])
+@pytest.mark.parametrize("closed", ["right", "left", "both", "neither"])
+@pytest.mark.parametrize("as_index", [True, False])
+def test_rolling_int_window(center, closed, as_index):
+    col_part1 = pd.DataFrame(
+        {
+            "by": np.tile(np.arange(15), 10),
+            "col1": np.arange(150),
+            "col2": np.arange(10, 160),
+        }
+    )
+    col_part2 = pd.DataFrame({"col3": np.arange(20, 170)})
+
+    md_df = pd.concat([col_part1, col_part2], axis=1)
+    pd_df = md_df._to_pandas()
+
+    if StorageFormat.get() == "Pandas":
+        assert md_df._query_compiler._modin_frame._partitions.shape[1] == 2
+
+    md_window = md_df.groupby("by", as_index=as_index).rolling(
+        3, center=center, closed=closed
+    )
+    pd_window = pd_df.groupby("by", as_index=as_index).rolling(
+        3, center=center, closed=closed
+    )
+    eval_rolling(md_window, pd_window)
+
+
+@pytest.mark.parametrize("center", [True, False])
+@pytest.mark.parametrize("closed", ["right", "left", "both", "neither"])
+@pytest.mark.parametrize("as_index", [True, False])
+@pytest.mark.parametrize("on", [None, "col4"])
+def test_rolling_timedelta_window(center, closed, as_index, on):
+    col_part1 = pd.DataFrame(
+        {
+            "by": np.tile(np.arange(15), 10),
+            "col1": np.arange(150),
+            "col2": np.arange(10, 160),
+        }
+    )
+    col_part2 = pd.DataFrame({"col3": np.arange(20, 170)})
+
+    if on is not None:
+        col_part2[on] = pandas.DatetimeIndex(
+            [
+                datetime.date(2020, 1, 1) + datetime.timedelta(hours=12) * i
+                for i in range(150)
+            ]
+        )
+
+    md_df = pd.concat([col_part1, col_part2], axis=1)
+    md_df.index = pandas.DatetimeIndex(
+        [datetime.date(2020, 1, 1) + datetime.timedelta(days=1) * i for i in range(150)]
+    )
+
+    pd_df = md_df._to_pandas()
+
+    if StorageFormat.get() == "Pandas":
+        assert md_df._query_compiler._modin_frame._partitions.shape[1] == 2
+
+    md_window = md_df.groupby("by", as_index=as_index).rolling(
+        datetime.timedelta(days=3), center=center, closed=closed, on=on
+    )
+    pd_window = pd_df.groupby("by", as_index=as_index).rolling(
+        datetime.timedelta(days=3), center=center, closed=closed, on=on
+    )
+    eval_rolling(md_window, pd_window)