SNOW-1757443: Implement GroupBy rolling (#3686)

Signed-off-by: Labanya Mukhopadhyay <[email protected]>

Author: sfc-gh-lmukhopadhyay

CHANGELOG.md

@@ -84,6 +84,7 @@
 - Added support for the `dtypes` parameter of `pd.get_dummies`
 - Added support for `nunique` in `df.pivot_table`, `df.agg` and other places where aggregate functions can be used.
 - Added support for `DataFrame.interpolate` and `Series.interpolate` with the "linear", "ffill"/"pad", and "backfill"/bfill" methods. These use the SQL `INTERPOLATE_LINEAR`, `INTERPOLATE_FFILL`, and `INTERPOLATE_BFILL` functions (PuPr).
+- Added support for `Dataframe.groupby.rolling()`.
 
 #### Improvements
 

docs/source/modin/supported/groupby_supported.rst

@@ -153,7 +153,10 @@ Computations/descriptive stats
 |                             |                                 | will be lost. ``rule`` frequencies 's', 'min',     |
 |                             |                                 | 'h', and 'D' are supported.                        |
 +-----------------------------+---------------------------------+----------------------------------------------------+
-| ``rolling``                 | N                               |                                                    |
+| ``rolling``                 | P                               | Implemented for DataframeGroupby objects. ``N`` for|
+|                             |                                 | ``on``, non-integer ``window``, ``axis = 1``,      |
+|                             |                                 | ``method`` != ``single``, ``min_periods = 0``, or  |
+|                             |                                 | ``closed``  != ``None``.                           |
 +-----------------------------+---------------------------------+----------------------------------------------------+
 | ``sample``                  | N                               |                                                    |
 +-----------------------------+---------------------------------+----------------------------------------------------+

src/snowflake/snowpark/modin/plugin/compiler/snowflake_query_compiler.py

@@ -2424,7 +2424,76 @@ def expanding():
         pass
 
     def rolling():
-        pass
+        """
+        Return a rolling grouper, providing rolling functionality per group.
+
+        This implementation supports both fixed window-based and time-based rolling operations
+        with groupby functionality.
+
+        Parameters
+        ----------
+        window : int, timedelta, str, offset, or BaseIndexer subclass
+            Size of the moving window.
+            If an integer, the fixed number of observations used for each window.
+            If a timedelta, str, or offset, the time period of each window. Each window will be a variable sized based on the observations included in the time-period. This is only valid for datetimelike indexes. To learn more about the offsets & frequency strings, please see `this link <https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#offset-aliases>`.
+            If a BaseIndexer subclass, the window boundaries based on the defined `get_window_bounds` method. Additional rolling keyword arguments, namely `min_periods`, `center`, `closed` and `step` will be passed to `get_window_bounds`.
+
+        min_periods : int, default None
+            Minimum number of observations in window required to have a value; otherwise, result is `np.nan`.
+            For a window that is specified by an offset, `min_periods` will default to 1.
+            For a window that is specified by an integer, `min_periods` will default to the size of the window.
+
+        center : bool, default False
+            If False, set the window labels as the right edge of the window index.
+            If True, set the window labels as the center of the window index.
+
+        win_type : str, default None
+            If `None`, all points are evenly weighted.
+            If a string, it must be a valid scipy.signal window function.
+            Certain Scipy window types require additional parameters to be passed in the aggregation function. The additional parameters must match the keywords specified in the Scipy window type method signature.
+
+        on : str, optional
+            For a DataFrame, a column label or Index level on which to calculate the rolling window, rather than the DataFrame's index.
+            Provided integer column is ignored and excluded from result since an integer index is not used to calculate the rolling window.
+
+        axis : int or str, default 0
+            If `0` or `'index'`, roll across the rows.
+            If `1` or `'columns'`, roll across the columns.
+            For Series this parameter is unused and defaults to 0.
+
+        closed : str, default None
+            If `'right'`, the first point in the window is excluded from calculations.
+            If `'left'`, the last point in the window is excluded from calculations.
+            If `'both'`, no points in the window are excluded from calculations.
+            If `'neither'`, the first and last points in the window are excluded from calculations.
+            Default `None` (`'right'`).
+
+        method : str {'single', 'table'}, default 'single'
+            Execute the rolling operation per single column or row (`'single'`) or over the entire object (`'table'`).
+            This argument is only implemented when specifying `engine='numba'` in the method call.
+
+        Returns
+        -------
+        Rolling
+            A rolling grouper, providing rolling functionality per group.
+
+        Examples
+        --------
+        >>> df = pd.DataFrame({'A': [1, 1, 2, 2], 'B': [1, 2, 3, 4], 'C': [0.362, 0.227, 1.267, -0.562]})
+        >>> df
+           A  B      C
+        0  1  1  0.362
+        1  1  2  0.227
+        2  2  3  1.267
+        3  2  4 -0.562
+        >>> df.groupby('A').rolling(2).sum() # doctest: +NORMALIZE_WHITESPACE
+               B      C
+        A
+        1 0  NaN    NaN
+          1  3.0  0.589
+        2 2  NaN    NaN
+          3  7.0  0.705
+        """
 
     def hist():
         pass

src/snowflake/snowpark/modin/plugin/docstrings/groupby.py

@@ -847,9 +847,35 @@ def resample(
 
 
 @register_df_groupby_override("rolling")
-def rolling(self, *args, **kwargs):
-    # TODO: SNOW-1063349: Modin upgrade - modin.pandas.groupby.DataFrameGroupBy functions
-    ErrorMessage.method_not_implemented_error(name="rolling", class_="GroupBy")
+def rolling(
+    self,
+    window,
+    min_periods: Union[int, None] = None,
+    center: bool = False,
+    win_type: Union[str, None] = None,
+    on: Union[str, None] = None,
+    axis: Union[int, str] = 0,
+    closed: Union[str, None] = None,
+    method: str = "single",
+    **kwargs,
+):
+    from snowflake.snowpark.modin.plugin.extensions.rolling_groupby_overrides import (
+        RollingGroupby,
+    )
+
+    return RollingGroupby(
+        dataframe=self._df,
+        by=self._by,
+        window=window,
+        min_periods=min_periods,
+        center=center,
+        win_type=win_type,
+        on=on,
+        axis=axis,
+        closed=closed,
+        method=method,
+        dropna=self._kwargs.get("dropna", True),
+    )
 
 
 @register_df_groupby_override("sample")