Remove cudf._lib.rolling in favor of inlining pylibcudf (#17423)

Contributes to #17317

Authors:
  - Matthew Roeschke (https://github.com/mroeschke)

Approvers:
  - Matthew Murray (https://github.com/Matt711)

URL: #17423

python/cudf/cudf/_lib/CMakeLists.txt

@@ -31,7 +31,6 @@ set(cython_sources
     reduce.pyx
     replace.pyx
     reshape.pyx
-    rolling.pyx
     round.pyx
     scalar.pyx
     sort.pyx

python/cudf/cudf/_lib/__init__.py

@@ -19,7 +19,6 @@
     reduce,
     replace,
     reshape,
-    rolling,
     round,
     sort,
     stream_compaction,

python/cudf/cudf/core/window/rolling.py

@@ -8,8 +8,11 @@
 import pandas as pd
 from pandas.api.indexers import BaseIndexer
 
+import pylibcudf as plc
+
 import cudf
 from cudf import _lib as libcudf
+from cudf._lib.aggregation import make_aggregation
 from cudf.api.types import is_integer, is_number
 from cudf.core.buffer import acquire_spill_lock
 from cudf.core.column.column import as_column
@@ -284,16 +287,37 @@ def _apply_agg_column(self, source_column, agg_name):
             )
             window = None
 
-        return libcudf.rolling.rolling(
-            source_column=source_column,
-            pre_column_window=preceding_window,
-            fwd_column_window=following_window,
-            window=window,
-            min_periods=min_periods,
-            center=self.center,
-            op=agg_name,
-            agg_params=self.agg_params,
-        )
+        with acquire_spill_lock():
+            if window is None:
+                if self.center:
+                    # TODO: we can support this even though Pandas currently does not
+                    raise NotImplementedError(
+                        "center is not implemented for offset-based windows"
+                    )
+                pre = preceding_window.to_pylibcudf(mode="read")
+                fwd = following_window.to_pylibcudf(mode="read")
+            else:
+                if self.center:
+                    pre = (window // 2) + 1
+                    fwd = window - (pre)
+                else:
+                    pre = window
+                    fwd = 0
+
+            return libcudf.column.Column.from_pylibcudf(
+                plc.rolling.rolling_window(
+                    source_column.to_pylibcudf(mode="read"),
+                    pre,
+                    fwd,
+                    min_periods,
+                    make_aggregation(
+                        agg_name,
+                        {"dtype": source_column.dtype}
+                        if callable(agg_name)
+                        else self.agg_params,
+                    ).c_obj,
+                )
+            )
 
     def _reduce(
         self,