Add streaming_groupby for stateful streaming aggregation

ci/validate_wheel.sh

@@ -21,11 +21,11 @@ PYDISTCHECK_ARGS=(
 if [[ "${package_dir}" == "python/libcudf" ]]; then
     if [[ "${RAPIDS_CUDA_MAJOR}" == "12" ]]; then
         PYDISTCHECK_ARGS+=(
-            --max-allowed-size-compressed '675M'
+            --max-allowed-size-compressed '700M'
         )
     else
         PYDISTCHECK_ARGS+=(
-            --max-allowed-size-compressed '325M'
+            --max-allowed-size-compressed '350M'
         )
     fi
 elif [[ "${package_dir}" != "python/cudf" ]] && \

cpp/CMakeLists.txt

@@ -496,6 +496,7 @@ add_library(
   src/filling/repeat.cu
   src/filling/sequence.cu
   src/groupby/common/m2_var_std.cu
+  src/groupby/common/utils.cpp
   src/groupby/groupby.cu
   src/groupby/hash/compute_global_memory_aggs.cu
   src/groupby/hash/compute_global_memory_aggs_null.cu
@@ -539,6 +540,16 @@ add_library(
   src/groupby/sort/host_udf_aggregation.cpp
   src/groupby/sort/scan.cpp
   src/groupby/sort/sort_helper.cu
+  src/groupby/streaming_groupby.cpp
+  src/groupby/streaming_groupby/aggregate.cu
+  src/groupby/streaming_groupby/impl.cu
+  src/groupby/streaming_groupby/insert.cu
+  src/groupby/streaming_groupby/insert_first.cu
+  src/groupby/streaming_groupby/insert_first_nested.cu
+  src/groupby/streaming_groupby/insert_nested.cu
+  src/groupby/streaming_groupby/insert_subsequent.cu
+  src/groupby/streaming_groupby/insert_subsequent_nested.cu
+  src/groupby/streaming_groupby/merge.cu
   src/hash/md5_hash.cu
   src/hash/murmurhash3_x86_32.cu
   src/hash/murmurhash3_x64_128.cu

cpp/benchmarks/groupby/group_max.cpp

@@ -84,11 +84,85 @@ void bench_groupby_max(nvbench::state& state, nvbench::type_list<Type>)
 template <typename Type>
 void bench_groupby_max_cardinality(nvbench::state& state, nvbench::type_list<Type>)
 {
-  auto constexpr num_rows         = 20'000'000;
-  auto constexpr null_probability = 0.;
-  auto const cardinality          = static_cast<cudf::size_type>(state.get_int64("cardinality"));
+  auto const num_rows         = static_cast<cudf::size_type>(state.get_int64("num_rows"));
+  auto const cardinality      = static_cast<cudf::size_type>(state.get_int64("cardinality"));
+  auto const num_aggregations = state.get_int64("num_aggregations");
+  auto const is_streaming     = state.get_string("api") == "streaming";
+
+  // TODO: streaming groupby reuses the cudf hash element_aggregator, which has
+  // no decimal128 MIN/MAX/SUM specialization (no native 128-bit atomics).  The
+  // stateless `normal` path falls back to sort-based aggregation, but streaming
+  // has no fallback and rejects the request.  Re-enable once streaming has a
+  // non-atomic aggregator path or 128-bit atomics gain hardware support.
+  if (is_streaming && std::is_same_v<Type, numeric::decimal128>) {
+    state.skip("streaming groupby does not support decimal128 MAX/MIN/SUM");
+    return;
+  }
 
-  groupby_max_helper<Type>(state, num_rows, cardinality, null_probability);
+  auto const keys = [&] {
+    data_profile const profile =
+      data_profile_builder()
+        .cardinality(cardinality)
+        .no_validity()
+        .distribution(cudf::type_to_id<int32_t>(), distribution_id::UNIFORM, 0, num_rows);
+    return create_random_column(cudf::type_to_id<int32_t>(), row_count{num_rows}, profile);
+  }();
+
+  auto const make_values = [&]() {
+    auto builder = data_profile_builder().cardinality(0).no_validity().distribution(
+      cudf::type_to_id<Type>(), distribution_id::UNIFORM, 0, num_rows);
+    return create_random_column(
+      cudf::type_to_id<Type>(), row_count{num_rows}, data_profile{builder});
+  };
+
+  std::vector<std::unique_ptr<cudf::column>> val_cols;
+  for (int64_t i = 0; i < num_aggregations; i++) {
+    val_cols.emplace_back(make_values());
+  }
+
+  auto keys_view = keys->view();
+
+  auto const mem_stats_logger = cudf::memory_stats_logger();
+  state.set_cuda_stream(nvbench::make_cuda_stream_view(cudf::get_default_stream().value()));
+
+  if (is_streaming) {
+    std::vector<cudf::column_view> all_columns = {keys_view, keys_view, keys_view};
+    for (auto const& vc : val_cols) {
+      all_columns.push_back(vc->view());
+    }
+    auto const full_table = cudf::table_view(all_columns);
+
+    std::vector<cudf::size_type> key_indices = {0, 1, 2};
+    std::vector<cudf::groupby::streaming_aggregation_request> requests;
+    for (int64_t i = 0; i < num_aggregations; i++) {
+      cudf::groupby::streaming_aggregation_request req;
+      req.column_index = static_cast<cudf::size_type>(3 + i);
+      req.aggregation  = cudf::make_max_aggregation<cudf::groupby_aggregation>();
+      requests.push_back(std::move(req));
+    }
+    state.exec(nvbench::exec_tag::sync, [&](nvbench::launch& launch) {
+      auto sgb = cudf::groupby::streaming_groupby(key_indices, requests, num_rows);
+      sgb.aggregate(full_table);
+      auto const result = sgb.finalize();
+    });
+  } else {
+    std::vector<cudf::groupby::aggregation_request> requests;
+    for (int64_t i = 0; i < num_aggregations; i++) {
+      requests.emplace_back();
+      requests[i].values = val_cols[i]->view();
+      requests[i].aggregations.push_back(cudf::make_max_aggregation<cudf::groupby_aggregation>());
+    }
+    state.exec(nvbench::exec_tag::sync, [&](nvbench::launch& launch) {
+      auto gb_obj = cudf::groupby::groupby(cudf::table_view({keys_view, keys_view, keys_view}));
+      auto const result = gb_obj.aggregate(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 * num_aggregations) / 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_TYPES(bench_groupby_max,
@@ -102,5 +176,7 @@ NVBENCH_BENCH_TYPES(bench_groupby_max,
 NVBENCH_BENCH_TYPES(bench_groupby_max_cardinality,
                     NVBENCH_TYPE_AXES(nvbench::type_list<int32_t, numeric::decimal128>))
   .set_name("groupby_max_cardinality")
+  .add_int64_axis("num_rows", {20'000'000})
   .add_int64_axis("num_aggregations", {1, 2, 3, 4, 5, 6, 7, 8})
-  .add_int64_axis("cardinality", {20, 50, 100, 1'000, 10'000, 100'000, 1'000'000});
+  .add_int64_axis("cardinality", {20, 50, 100, 1'000, 10'000, 100'000, 1'000'000})
+  .add_string_axis("api", {"normal", "streaming"});

cpp/include/cudf/detail/aggregation/aggregation.hpp

@@ -14,6 +14,8 @@
 #include <cudf/utilities/span.hpp>
 #include <cudf/utilities/traits.hpp>
 
+#include <cuda/std/type_traits>
+
 #include <functional>
 #include <numeric>
 #include <utility>
@@ -1341,9 +1343,9 @@ data_type target_type(data_type source_type, aggregation::Kind k);
  * @tparam k The aggregation to perform
  */
 template <typename Source, aggregation::Kind k>
-constexpr inline bool is_valid_aggregation()
+CUDF_HOST_DEVICE constexpr inline bool is_valid_aggregation()
 {
-  return (not std::is_void_v<target_type_t<Source, k>>);
+  return (not cuda::std::is_void_v<target_type_t<Source, k>>);
 }
 
 /**

cpp/include/cudf/groupby.hpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2024, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -137,7 +137,7 @@ class groupby {
    * result is the same order as was specified in the request.
    *
    * The returned `table` contains the group labels for each group, i.e., the
-   * unique rows from `keys`. Element `i` across all aggregation results
+   * distinct rows from `keys`. Element `i` across all aggregation results
    * belongs to the group at row `i` in the group labels table.
    *
    * The order of the rows in the group labels is arbitrary. Furthermore,
@@ -169,7 +169,7 @@ class groupby {
    * perform
    * @param stream CUDA stream used for device memory operations and kernel launches.
    * @param mr Device memory resource used to allocate the returned table and columns' device memory
-   * @return Pair containing the table with each group's unique key and
+   * @return Pair containing the table with each group's distinct key and
    * a vector of aggregation_results for each request in the same order as
    * specified in `requests`.
    */
@@ -407,6 +407,189 @@ class groupby {
     rmm::cuda_stream_view stream,
     rmm::device_async_resource_ref mr);
 };
+
+/**
+ * @brief Request for a single streaming groupby aggregation on a column.
+ *
+ * Analogous to `aggregation_request` but identifies the value column by index rather than
+ * by `column_view`, since data arrives in batches after construction, and carries exactly
+ * one aggregation per request.
+ *
+ * `column_index` refers to the position of the value column in the `table_view` passed
+ * to `streaming_groupby::aggregate()`.  Multiple aggregations on the same column are
+ * expressed as separate requests (e.g., `[{col, sum}, {col, mean}]`).  Internal
+ * deduplication ensures redundant computations are shared automatically.
+ */
+struct streaming_aggregation_request {
+  size_type column_index;                            ///< Index of the value column
+  std::unique_ptr<groupby_aggregation> aggregation;  ///< Desired aggregation
+};
+
+/**
+ * @brief Stateful streaming groupby that accumulates partial aggregates across batches.
+ *
+ * `streaming_groupby` and the stateless `groupby` serve different use cases.  Use
+ * the stateless `groupby` for single-shot aggregation when all input fits in memory
+ * at once.  Use `streaming_groupby` when input arrives over multiple batches and
+ * memory efficiency matters: peak memory does not scale with the number of *input*
+ * rows, because only the distinct keys seen so far and one aggregation slot per
+ * group are kept across batches.  Arbitrarily long high-duplicate streams therefore
+ * accumulate without running out of memory.
+ *
+ * If memory is not a concern, concatenating all batches and calling the stateless
+ * `groupby` once is also a valid choice.  Reach for `streaming_groupby` when
+ * (a) the cumulative input does not fit in memory, or (b) partial-state aggregation
+ * across distributed workers (`merge()`) is part of the workload.
+ *
+ * Per-batch cost is O(batch_size): each batch does direct hash table insertion
+ * and in-place aggregation updates against the persistent state. Partial states
+ * can be combined via `merge()`, and final results are produced via `finalize()`.
+ *
+ * The `max_distinct_keys` parameter sets the upper bound on the number of distinct key
+ * combinations across the lifetime of this object.  The persistent state is sized to
+ * `max_distinct_keys` (constant for the lifetime of the object); the stored distinct
+ * keys grow with the number of distinct keys actually seen, so the incremental key
+ * storage is O(`distinct_keys()` × key_size) and does not scale with cumulative input
+ * rows.
+ *
+ * Cumulative input rows are not bounded — only cumulative distinct keys.  A single
+ * batch may also not exceed `max_distinct_keys` rows; this is an implementation
+ * limit because each in-flight batch row is encoded as `max_distinct_keys + row_idx`
+ * inside the hash set, which must fit in `cudf::size_type`.
+ *
+ * All column types (including variable-width types such as strings, lists, and structs)
+ * are supported for key columns.  Only hash-based aggregation kinds are supported; use
+ * `is_streaming_groupby_supported()` to query a specific (value type, aggregation kind)
+ * combination.
+ *
+ * Supported aggregation kinds:
+ *   SUM, SUM_OF_SQUARES, PRODUCT, MIN, MAX, COUNT_VALID, COUNT_ALL,
+ *   MEAN, M2, VARIANCE, STD
+ *
+ * @throws std::invalid_argument for unsupported aggregation kinds
+ * @throws std::invalid_argument if a single batch exceeds `max_distinct_keys` rows
+ * @throws cudf::logic_error if cumulative distinct keys exceed `max_distinct_keys`
+ */
+class streaming_groupby {
+ public:
+  streaming_groupby() = delete;
+  ~streaming_groupby();
+  streaming_groupby(streaming_groupby const&)            = delete;
+  streaming_groupby& operator=(streaming_groupby const&) = delete;
+
+  /** @brief Move constructor. */
+  streaming_groupby(streaming_groupby&&) noexcept;
+
+  /**
+   * @brief Move assignment operator.
+   * @return Reference to this object.
+   */
+  streaming_groupby& operator=(streaming_groupby&&) noexcept;
+
+  /**
+   * @brief Construct a streaming groupby object with a persistent hash table.
+   *
+   * @param key_indices Indices of columns in the data table that serve as groupby keys
+   * @param requests The aggregations to perform and which columns to aggregate
+   * @param max_distinct_keys Upper bound on distinct key combinations. The hash set,
+   *        companion vectors, and aggregation results table are all sized to this
+   *        capacity. Cumulative input rows are not bounded.
+   * @param null_handling Indicates whether rows in keys that contain NULL values should be included
+   *
+   * @throws std::invalid_argument if `max_distinct_keys <= 0`
+   * @throws std::invalid_argument if any requested aggregation kind is unsupported
+   */
+  explicit streaming_groupby(host_span<size_type const> key_indices,
+                             host_span<streaming_aggregation_request const> requests,
+                             size_type max_distinct_keys,
+                             null_policy null_handling = null_policy::EXCLUDE);
+
+  /**
+   * @brief Feed a batch of data into the streaming aggregation.
+   *
+   * Batch keys are inserted into the persistent hash set and aggregation results
+   * are updated atomically. The input `data` table is not referenced after this
+   * call returns.
+   *
+   * @param data Table containing both key and value columns
+   * @param stream CUDA stream used for device memory operations and kernel launches
+   *
+   * @throws std::invalid_argument if `data.num_rows()` exceeds `max_distinct_keys`
+   * @throws cudf::logic_error if cumulative distinct keys exceed `max_distinct_keys`
+   */
+  void aggregate(table_view const& data, rmm::cuda_stream_view stream = cudf::get_default_stream());
+
+  /**
+   * @brief Merge another streaming_groupby's accumulated partial state into this one.
+   *
+   * Extracts the other object's accumulated intermediate state and merges it into this
+   * object's persistent hash table. The other object is not modified.
+   * Both objects must have been constructed with compatible aggregation requests,
+   * and this object must have had at least one `aggregate()` call.
+   *
+   * @param other The streaming_groupby whose partial state to merge
+   * @param stream CUDA stream used for device memory operations and kernel launches
+   *
+   * @throws std::invalid_argument if the other object has more distinct keys than
+   * `max_distinct_keys`
+   * @throws cudf::logic_error if this object has not been initialized via `aggregate()`
+   * @throws cudf::logic_error if distinct keys exceed `max_distinct_keys` after merge
+   */
+  void merge(streaming_groupby const& other,
+             rmm::cuda_stream_view stream = cudf::get_default_stream());
+
+  /**
+   * @brief Finalize the accumulated partial aggregates into final results.
+   *
+   * For most aggregation kinds the partial state is the final result. For kinds like
+   * MEAN, VARIANCE, or STD, a finalization step converts the internal partial representation
+   * (e.g., sum+count) into the user-facing result.
+   *
+   * This does not modify the internal state; `aggregate()` may be called again afterward.
+   *
+   * @param stream CUDA stream used for device memory operations and kernel launches
+   * @param mr Device memory resource used to allocate the returned table and columns
+   * @return Pair of distinct keys table and a vector of aggregation_results (one per request)
+   *
+   * @throws cudf::logic_error if no data has been accumulated
+   */
+  [[nodiscard]] std::pair<std::unique_ptr<table>, std::vector<aggregation_result>> finalize(
+    rmm::cuda_stream_view stream      = cudf::get_default_stream(),
+    rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref()) const;
+
+  /**
+   * @brief Returns the number of distinct keys accumulated so far.
+   *
+   * Returns 0 before any successful `aggregate()` or `merge()` call.
+   *
+   * @return The current count of distinct keys in the persistent hash table
+   */
+  [[nodiscard]] size_type distinct_keys() const noexcept;
+
+ private:
+  struct impl;
+  std::unique_ptr<impl> _impl;
+
+  void do_aggregate(table_view const& data, rmm::cuda_stream_view stream);
+  void do_merge(streaming_groupby const& other, rmm::cuda_stream_view stream);
+  [[nodiscard]] std::pair<std::unique_ptr<table>, std::vector<aggregation_result>> do_finalize(
+    rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) const;
+};
+
+/**
+ * @brief Returns true if `streaming_groupby` supports the given value type and
+ * aggregation kind combination.
+ *
+ * Use this to query support without constructing a `streaming_groupby`.  A `true`
+ * return implies that an `aggregate()` call with a value column of `values_type` and
+ * an aggregation of `kind` will not be rejected on type/kind grounds.
+ *
+ * @param values_type Type of the value column the aggregation would run on
+ * @param kind Aggregation kind
+ * @return True if the combination is supported, false otherwise
+ */
+[[nodiscard]] bool is_streaming_groupby_supported(data_type values_type, aggregation::Kind kind);
+
 /** @} */
 }  // namespace groupby
 }  // namespace CUDF_EXPORT cudf