Skip to content
Open
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions cpp/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
9 changes: 7 additions & 2 deletions cpp/benchmarks/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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
)

# ##################################################################################################
Expand Down
76 changes: 76 additions & 0 deletions cpp/benchmarks/rolling/range_rolling_column_bounds_sum.cpp
Original file line number Diff line number Diff line change
@@ -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});
70 changes: 69 additions & 1 deletion cpp/include/cudf/rolling.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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.
Comment thread
mythrocks marked this conversation as resolved.
*
* 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.
*
Expand Down Expand Up @@ -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.
Expand Down
104 changes: 96 additions & 8 deletions cpp/src/rolling/detail/range_rolling.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
Expand Down Expand Up @@ -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
Expand All @@ -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);

Expand All @@ -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
Expand All @@ -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);

Expand All @@ -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
Expand All @@ -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);

Expand All @@ -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
Expand All @@ -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);

Expand Down
Loading
Loading