Add column-valued PRECEDING/FOLLOWING bounds to range rolling windows

cpp/CMakeLists.txt

@@ -992,7 +992,9 @@ add_library(
   src/rolling/grouped_rolling.cu
   src/rolling/range_rolling.cu
   src/rolling/range_rolling_bounded_closed.cu
+  src/rolling/range_rolling_bounded_closed_column.cu
   src/rolling/range_rolling_bounded_open.cu
+  src/rolling/range_rolling_bounded_open_column.cu
   src/rolling/range_rolling_current_row.cu
   src/rolling/range_rolling_multi_column.cu
   src/rolling/range_rolling_unbounded.cu

cpp/benchmarks/CMakeLists.txt

@@ -516,8 +516,13 @@ ConfigureNVBench(RESHAPE_NVBENCH reshape/interleave.cpp reshape/table_to_array.c
 # * rolling benchmark
 # ---------------------------------------------------------------------------------
 ConfigureNVBench(
-  ROLLING_NVBENCH rolling/grouped_range_rolling_sum.cu rolling/grouped_rolling_sum.cpp
-  rolling/multi_orderby_range_rolling_sum.cpp rolling/range_rolling_sum.cu rolling/rolling_sum.cpp
+  ROLLING_NVBENCH
+  rolling/grouped_range_rolling_sum.cu
+  rolling/grouped_rolling_sum.cpp
+  rolling/multi_orderby_range_rolling_sum.cpp
+  rolling/range_rolling_column_bounds_sum.cpp
+  rolling/range_rolling_sum.cu
+  rolling/rolling_sum.cpp
 )
 
 # ##################################################################################################

cpp/benchmarks/rolling/range_rolling_column_bounds_sum.cpp

@@ -0,0 +1,76 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <benchmarks/common/generate_input.hpp>
+#include <benchmarks/common/memory_stats.hpp>
+
+#include <cudf/aggregation.hpp>
+#include <cudf/column/column_factories.hpp>
+#include <cudf/filling.hpp>
+#include <cudf/rolling.hpp>
+#include <cudf/scalar/scalar.hpp>
+#include <cudf/table/table_view.hpp>
+#include <cudf/types.hpp>
+#include <cudf/utilities/default_stream.hpp>
+
+#include <nvbench/nvbench.cuh>
+
+#include <cstdint>
+#include <vector>
+
+// Per-row (column-valued) RANGE bounds: `bounded_closed_column`. This mirrors the scalar
+// `range_rolling_sum` benchmark so the two can be compared directly to quantify the overhead of
+// reading a per-row delta instead of broadcasting a single scalar. Constant delta columns keep the
+// window sizes (and therefore the aggregation work) identical to the scalar case, isolating the
+// cost of the per-row read.
+void bench_range_rolling_column_bounds_sum(nvbench::state& state)
+{
+  auto const num_rows        = static_cast<cudf::size_type>(state.get_int64("num_rows"));
+  auto const preceding_range = state.get_int64("preceding_range");
+  auto const following_range = state.get_int64("following_range");
+
+  auto vals = [&] {
+    data_profile const profile = data_profile_builder().cardinality(0).no_validity().distribution(
+      cudf::type_to_id<std::int32_t>(), distribution_id::UNIFORM, 0, 100);
+    return create_random_column(cudf::type_to_id<std::int32_t>(), row_count{num_rows}, profile);
+  }();
+
+  // Equally-spaced ascending integer orderby (1 unit apart), so `preceding_range`/`following_range`
+  // approximately control the number of rows in each window.
+  auto const orderby = cudf::sequence(
+    num_rows, cudf::numeric_scalar<std::int64_t>(0), cudf::numeric_scalar<std::int64_t>(1));
+
+  // Per-row delta columns (matching the orderby type), each filled with a single constant.
+  auto const preceding_col =
+    cudf::make_column_from_scalar(cudf::numeric_scalar<std::int64_t>(preceding_range), num_rows);
+  auto const following_col =
+    cudf::make_column_from_scalar(cudf::numeric_scalar<std::int64_t>(following_range), num_rows);
+
+  std::vector<cudf::rolling_request> requests;
+  requests.push_back({vals->view(), 1, cudf::make_sum_aggregation<cudf::rolling_aggregation>()});
+
+  auto const mem_stats_logger = cudf::memory_stats_logger();
+  state.set_cuda_stream(nvbench::make_cuda_stream_view(cudf::get_default_stream().value()));
+  state.exec(nvbench::exec_tag::sync, [&](nvbench::launch& launch) {
+    auto const result =
+      cudf::grouped_range_rolling_window(cudf::table_view{},
+                                         orderby->view(),
+                                         cudf::order::ASCENDING,
+                                         cudf::null_order::BEFORE,
+                                         cudf::bounded_closed_column{preceding_col->view()},
+                                         cudf::bounded_closed_column{following_col->view()},
+                                         requests);
+  });
+  auto const elapsed_time = state.get_summary("nv/cold/time/gpu/mean").get_float64("value");
+  state.add_element_count(static_cast<double>(num_rows) / elapsed_time / 1'000'000., "Mrows/s");
+  state.add_buffer_size(
+    mem_stats_logger.peak_memory_usage(), "peak_memory_usage", "peak_memory_usage");
+}
+
+NVBENCH_BENCH(bench_range_rolling_column_bounds_sum)
+  .set_name("range_rolling_column_bounds_sum")
+  .add_int64_power_of_two_axis("num_rows", {14, 22, 28})
+  .add_int64_axis("preceding_range", {100})
+  .add_int64_axis("following_range", {100});

cpp/include/cudf/rolling.hpp

@@ -6,6 +6,7 @@
 #pragma once
 
 #include <cudf/aggregation.hpp>
+#include <cudf/column/column_view.hpp>
 #include <cudf/rolling/range_window_bounds.hpp>
 #include <cudf/types.hpp>
 #include <cudf/utilities/default_stream.hpp>
@@ -88,6 +89,65 @@ struct bounded_open {
   [[nodiscard]] cudf::scalar const* delta() const noexcept { return &delta_; }
 };
 
+/**
+ * @brief Strongly typed wrapper for bounded closed rolling windows whose delta varies row-to-row.
+ *
+ * Unlike `bounded_closed`, which applies a single scalar delta to every row, this endpoint reads a
+ * per-row delta from a column: row `i`'s endpoint is computed from `orderby[i]` and `delta[i]`.
+ * This lets engines evaluate windows such as `RANGE BETWEEN <expr> PRECEDING AND ...` where the
+ * bound is a projected column rather than a literal.
+ *
+ * The delta column must have exactly one entry per orderby row, must not contain nulls, and must
+ * have the same type as the orderby column (or, when the orderby column is a TIMESTAMP, the
+ * matching DURATION type). Per-row delta values must be finite, otherwise behaviour is undefined.
+ *
+ * Fixed-point (decimal) orderby columns are not supported, unlike the scalar `bounded_closed` /
+ * `bounded_open` endpoints.
+ *
+ * The endpoints of this window are included.
+ */
+struct bounded_closed_column {
+  cudf::column_view delta_;  ///< Per-row delta column, one entry per orderby row. Must not contain
+                             ///< nulls and must match the orderby column's type.
+
+  /**
+   * @brief Construct a bounded closed rolling window with a per-row delta column.
+   *
+   * @param delta Per-row delta column. Must not contain nulls and must match the orderby type.
+   */
+  bounded_closed_column(cudf::column_view delta) : delta_{delta} {}
+  /**
+   * @brief Return the per-row delta column.
+   * @return the per-row delta column.
+   */
+  [[nodiscard]] cudf::column_view delta() const noexcept { return delta_; }
+};
+
+/**
+ * @brief Strongly typed wrapper for bounded open rolling windows whose delta varies row-to-row.
+ *
+ * The column-valued analogue of `bounded_open`. See `bounded_closed_column` for the per-row delta
+ * column requirements.
+ *
+ * The endpoints of this window are excluded.
+ */
+struct bounded_open_column {
+  cudf::column_view delta_;  ///< Per-row delta column, one entry per orderby row. Must not contain
+                             ///< nulls and must match the orderby column's type.
+
+  /**
+   * @brief Construct a bounded open rolling window with a per-row delta column.
+   *
+   * @param delta Per-row delta column. Must not contain nulls and must match the orderby type.
+   */
+  bounded_open_column(cudf::column_view delta) : delta_{delta} {}
+  /**
+   * @brief Return the per-row delta column.
+   * @return the per-row delta column.
+   */
+  [[nodiscard]] cudf::column_view delta() const noexcept { return delta_; }
+};
+
 /**
  * @brief Strongly typed wrapper for unbounded rolling windows.
  *
@@ -115,8 +175,16 @@ struct current_row {
 
 /**
  * @brief The type of the range-based rolling window endpoint.
+ *
+ * `bounded_closed_column` and `bounded_open_column` carry a per-row delta column (one entry per
+ * orderby row) instead of a single scalar delta, so the window width can vary row-to-row.
  */
-using range_window_type = std::variant<unbounded, current_row, bounded_closed, bounded_open>;
+using range_window_type = std::variant<unbounded,
+                                       current_row,
+                                       bounded_closed,
+                                       bounded_open,
+                                       bounded_closed_column,
+                                       bounded_open_column>;
 
 /**
  * @brief A request for a rolling aggregation on a column.

cpp/src/rolling/detail/range_rolling.hpp

@@ -17,12 +17,50 @@
 
 #include <cuda/stream>
 
+#include <concepts>
 #include <memory>
 #include <optional>
+#include <type_traits>
 #include <utility>
+#include <variant>
 
 namespace cudf::detail {
 
+/**
+ * @brief Normalized delta source for a single range-window endpoint.
+ *
+ * A range-window endpoint carries at most one delta, and each endpoint kind supplies it
+ * differently: `bounded_closed`/`bounded_open` hold a single scalar delta (exposed as a
+ * `cudf::scalar const*`), the column-valued endpoints hold a per-row delta `cudf::column_view`, and
+ * `unbounded`/`current_row` carry no delta at all (`std::monostate`). Normalizing to this variant
+ * lets a single typed value be threaded through the dispatch stack instead of a pair of nullable
+ * pointers, while keeping the endpoints' public accessors unchanged.
+ */
+using range_window_delta = std::variant<std::monostate, cudf::scalar const*, cudf::column_view>;
+
+/**
+ * @brief Normalize a range-window endpoint's delta into a single typed source.
+ *
+ * `unbounded`/`current_row` carry no delta and normalize to `std::monostate`; every other endpoint
+ * forwards its public `delta()` accessor (a `cudf::scalar const*` for the scalar-valued bounded
+ * windows, a `cudf::column_view` for the column-valued ones).
+ *
+ * @tparam Window The endpoint tag type.
+ * @param window The endpoint tag.
+ * @return The endpoint's delta as a `range_window_delta`.
+ */
+template <typename Window>
+[[nodiscard]] range_window_delta normalize_delta(Window const& window)
+{
+  using WindowType = std::remove_cvref_t<Window>;
+  if constexpr (std::same_as<WindowType, cudf::unbounded> ||
+                std::same_as<WindowType, cudf::current_row>) {
+    return std::monostate{};
+  } else {
+    return window.delta();
+  }
+}
+
 /**
  * @brief Constructs preceding and following window-size columns for a single-column RANGE window.
  *
@@ -55,7 +93,7 @@ namespace cudf::detail {
  * @param order Sort order of the order-by column
  * @param grouping Preprocessed grouping information, if any
  * @param nulls_at_start Whether nulls are ordered before non-null values
- * @param row_delta Must be null for an unbounded window
+ * @param delta Must hold `std::monostate` (no delta) for an unbounded window
  * @param stream CUDA stream used for device memory operations and kernel launches
  * @param mr Device memory resource used to allocate the returned column's device memory
  * @return Column containing the window size for each row
@@ -67,7 +105,7 @@ namespace cudf::detail {
   order order,
   std::optional<rolling::preprocessed_group_info> const& grouping,
   bool nulls_at_start,
-  scalar const* row_delta,
+  range_window_delta const& delta,
   cuda::stream_ref stream,
   rmm::device_async_resource_ref mr);
 
@@ -80,7 +118,7 @@ namespace cudf::detail {
  * @param order Sort order of the order-by column
  * @param grouping Preprocessed grouping information, if any
  * @param nulls_at_start Whether nulls are ordered before non-null values
- * @param row_delta Must be null for a current-row window
+ * @param delta Must hold `std::monostate` (no delta) for a current-row window
  * @param stream CUDA stream used for device memory operations and kernel launches
  * @param mr Device memory resource used to allocate the returned column's device memory
  * @return Column containing the window size for each row
@@ -92,7 +130,7 @@ namespace cudf::detail {
   order order,
   std::optional<rolling::preprocessed_group_info> const& grouping,
   bool nulls_at_start,
-  scalar const* row_delta,
+  range_window_delta const& delta,
   cuda::stream_ref stream,
   rmm::device_async_resource_ref mr);
 
@@ -105,7 +143,7 @@ namespace cudf::detail {
  * @param order Sort order of the order-by column
  * @param grouping Preprocessed grouping information, if any
  * @param nulls_at_start Whether nulls are ordered before non-null values
- * @param row_delta Must be non-null and contain the bounded-window delta
+ * @param delta Must hold a non-null `scalar const*` with the bounded-window delta
  * @param stream CUDA stream used for device memory operations and kernel launches
  * @param mr Device memory resource used to allocate the returned column's device memory
  * @return Column containing the window size for each row
@@ -117,7 +155,7 @@ namespace cudf::detail {
   order order,
   std::optional<rolling::preprocessed_group_info> const& grouping,
   bool nulls_at_start,
-  scalar const* row_delta,
+  range_window_delta const& delta,
   cuda::stream_ref stream,
   rmm::device_async_resource_ref mr);
 
@@ -130,7 +168,7 @@ namespace cudf::detail {
  * @param order Sort order of the order-by column
  * @param grouping Preprocessed grouping information, if any
  * @param nulls_at_start Whether nulls are ordered before non-null values
- * @param row_delta Must be non-null and contain the bounded-window delta
+ * @param delta Must hold a non-null `scalar const*` with the bounded-window delta
  * @param stream CUDA stream used for device memory operations and kernel launches
  * @param mr Device memory resource used to allocate the returned column's device memory
  * @return Column containing the window size for each row
@@ -142,7 +180,57 @@ namespace cudf::detail {
   order order,
   std::optional<rolling::preprocessed_group_info> const& grouping,
   bool nulls_at_start,
-  scalar const* row_delta,
+  range_window_delta const& delta,
+  cuda::stream_ref stream,
+  rmm::device_async_resource_ref mr);
+
+/**
+ * @brief Dispatches computation of a bounded-closed RANGE window-size column with a per-row delta.
+ *
+ * @param window Bounded-closed column-valued window tag
+ * @param orderby Sorted order-by column
+ * @param direction Direction of the window
+ * @param order Sort order of the order-by column
+ * @param grouping Preprocessed grouping information, if any
+ * @param nulls_at_start Whether nulls are ordered before non-null values
+ * @param delta Must hold a `column_view` with one delta per orderby row
+ * @param stream CUDA stream used for device memory operations and kernel launches
+ * @param mr Device memory resource used to allocate the returned column's device memory
+ * @return Column containing the window size for each row
+ */
+[[nodiscard]] std::unique_ptr<column> dispatch_range_window(
+  bounded_closed_column window,
+  column_view const& orderby,
+  rolling::direction direction,
+  order order,
+  std::optional<rolling::preprocessed_group_info> const& grouping,
+  bool nulls_at_start,
+  range_window_delta const& delta,
+  cuda::stream_ref stream,
+  rmm::device_async_resource_ref mr);
+
+/**
+ * @brief Dispatches computation of a bounded-open RANGE window-size column with a per-row delta.
+ *
+ * @param window Bounded-open column-valued window tag
+ * @param orderby Sorted order-by column
+ * @param direction Direction of the window
+ * @param order Sort order of the order-by column
+ * @param grouping Preprocessed grouping information, if any
+ * @param nulls_at_start Whether nulls are ordered before non-null values
+ * @param delta Must hold a `column_view` with one delta per orderby row
+ * @param stream CUDA stream used for device memory operations and kernel launches
+ * @param mr Device memory resource used to allocate the returned column's device memory
+ * @return Column containing the window size for each row
+ */
+[[nodiscard]] std::unique_ptr<column> dispatch_range_window(
+  bounded_open_column window,
+  column_view const& orderby,
+  rolling::direction direction,
+  order order,
+  std::optional<rolling::preprocessed_group_info> const& grouping,
+  bool nulls_at_start,
+  range_window_delta const& delta,
   cuda::stream_ref stream,
   rmm::device_async_resource_ref mr);