From 90c1ea130d403275e16ea765c3aaa67c13d08a48 Mon Sep 17 00:00:00 2001
From: Vyas Ramasubramani <vyasr@nvidia.com>
Date: Tue, 25 Aug 2026 19:38:50 -0700
Subject: [PATCH 1/5] Add compressed percentile oracle coverage

---
 .../quantiles/percentile_approx_test.cpp      | 155 ++++++++++++++++++
 1 file changed, 155 insertions(+)

diff --git a/cpp/tests/quantiles/percentile_approx_test.cpp b/cpp/tests/quantiles/percentile_approx_test.cpp
index 6e1fb1f899f0..d4b3a8510f6f 100644
--- a/cpp/tests/quantiles/percentile_approx_test.cpp
+++ b/cpp/tests/quantiles/percentile_approx_test.cpp
@@ -4,6 +4,7 @@
  */
 
 #include <cudf_test/base_fixture.hpp>
+#include <cudf_test/column_utilities.hpp>
 #include <cudf_test/column_wrapper.hpp>
 #include <cudf_test/iterator_utilities.hpp>
 #include <cudf_test/tdigest_utilities.hpp>
@@ -12,6 +13,7 @@
 
 #include <cudf/detail/tdigest/tdigest.hpp>
 #include <cudf/groupby.hpp>
+#include <cudf/lists/lists_column_view.hpp>
 #include <cudf/quantiles.hpp>
 #include <cudf/reduction.hpp>
 #include <cudf/tdigest/tdigest_column_view.hpp>
@@ -21,6 +23,11 @@
 
 #include <cuda/iterator>
 
+#include <algorithm>
+#include <cmath>
+#include <cstddef>
+#include <vector>
+
 namespace {
 struct percentile_approx_dispatch {
   template <typename T, typename Func>
@@ -220,6 +227,70 @@ std::pair<rmm::device_buffer, cudf::size_type> make_null_mask(cudf::column_view
   return cudf::test::detail::make_null_mask(itr, itr + col.size());
 }
 
+std::vector<double> make_compressed_test_values(cudf::size_type group, cudf::size_type row_count)
+{
+  std::vector<double> values(row_count);
+  for (cudf::size_type i = 0; i < row_count; ++i) {
+    auto const permutation =
+      static_cast<double>((i * 37 + group * 101) % row_count) - (row_count / 2.0);
+    auto const trend = static_cast<double>(i) * (0.015 + 0.003 * group);
+    auto const wave  = std::sin(static_cast<double>(i) * 0.037 + group) * (9.0 + group);
+    values[i]        = permutation * (1.5 + 0.2 * group) + trend + wave + group * 500.0;
+  }
+  return values;
+}
+
+double exact_host_quantile(std::vector<double> values, double percentile)
+{
+  std::sort(values.begin(), values.end());
+  auto const rank       = percentile * static_cast<double>(values.size() - 1);
+  auto const lower_rank = static_cast<std::size_t>(std::floor(rank));
+  auto const upper_rank = static_cast<std::size_t>(std::ceil(rank));
+  if (lower_rank == upper_rank) { return values[lower_rank]; }
+
+  auto const weight = rank - static_cast<double>(lower_rank);
+  return values[lower_rank] + (values[upper_rank] - values[lower_rank]) * weight;
+}
+
+std::vector<std::vector<double>> lists_column_to_host(cudf::column_view const& column)
+{
+  auto const lists   = cudf::lists_column_view{column};
+  auto const offsets = cudf::test::to_host<cudf::size_type>(lists.offsets()).first;
+  auto const child   = cudf::test::to_host<double>(lists.child()).first;
+
+  std::vector<std::vector<double>> result(lists.size());
+  for (cudf::size_type row = 0; row < lists.size(); ++row) {
+    result[row].assign(child.begin() + offsets[row], child.begin() + offsets[row + 1]);
+  }
+  return result;
+}
+
+void expect_approx_percentiles_near_exact(std::vector<double> const& values,
+                                          std::vector<double> const& percentiles,
+                                          std::vector<double> const& actual)
+{
+  ASSERT_EQ(actual.size(), percentiles.size());
+
+  auto const [minimum, maximum] = std::minmax_element(values.begin(), values.end());
+  auto const value_tolerance    = (*maximum - *minimum) * 0.08;
+  // The low centroid count intentionally compresses thousands of rows. This bound is loose enough
+  // for both centroid clustering implementations but still catches gross percentile indexing,
+  // interpolation, or weighting regressions against an independent host quantile oracle.
+  for (std::size_t idx = 0; idx < percentiles.size(); ++idx) {
+    auto const exact = exact_host_quantile(values, percentiles[idx]);
+    EXPECT_NEAR(actual[idx], exact, value_tolerance)
+      << "percentile=" << percentiles[idx] << " exact=" << exact << " actual=" << actual[idx];
+  }
+}
+
+struct scoped_cpu_clustering_setting {
+  bool const previous = cudf::tdigest::detail::is_cpu_cluster_computation_disabled;
+  ~scoped_cpu_clustering_setting()
+  {
+    cudf::tdigest::detail::is_cpu_cluster_computation_disabled = previous;
+  }
+};
+
 void simple_with_nulls_test(cudf::data_type input_type, std::vector<std::pair<int, int>> params)
 {
   auto values = cudf::test::generate_standardized_percentile_distribution(input_type);
@@ -490,6 +561,90 @@ TEST_F(PercentileApproxTest, GroupByWithNullsGold)
   CUDF_TEST_EXPECT_COLUMNS_EQUAL(*result, expected);
 }
 
+namespace {
+
+TEST_F(PercentileApproxTest, CompressedTdigestsAgainstHostQuantiles)
+{
+  auto const max_centroids  = 20;
+  auto const rows_per_group = 4096;
+  auto const percentiles    = std::vector<double>{0.01, 0.10, 0.25, 0.50, 0.75, 0.90, 0.99};
+  auto const percentiles_column =
+    cudf::test::fixed_width_column_wrapper<double>(percentiles.begin(), percentiles.end());
+  auto const group_values = std::vector<std::vector<double>>{
+    make_compressed_test_values(0, rows_per_group),
+    make_compressed_test_values(1, rows_per_group),
+  };
+
+  auto const restore_cpu_clustering_setting = scoped_cpu_clustering_setting{};
+  // Small group counts use host-side cluster compression by default. Exercise both cluster
+  // compression paths because the compressed centroid layout is what percentile_approx consumes.
+  for (auto const cpu_clustering_disabled : {true, false}) {
+    SCOPED_TRACE(::testing::Message() << "cpu_clustering_disabled=" << cpu_clustering_disabled);
+    cudf::tdigest::detail::is_cpu_cluster_computation_disabled = cpu_clustering_disabled;
+
+    for (auto const& values : group_values) {
+      auto const values_column =
+        cudf::test::fixed_width_column_wrapper<double>(values.begin(), values.end());
+      auto const tdigest =
+        cudf::reduce(values_column,
+                     *cudf::make_tdigest_aggregation<cudf::reduce_aggregation>(max_centroids),
+                     cudf::data_type{cudf::type_id::STRUCT});
+      auto const tdigest_col = cudf::make_column_from_scalar(*tdigest, 1);
+      auto const result      = cudf::percentile_approx(tdigest_col->view(), percentiles_column);
+
+      auto const actual = lists_column_to_host(result->view());
+      ASSERT_EQ(actual.size(), 1);
+      expect_approx_percentiles_near_exact(values, percentiles, actual.front());
+    }
+
+    std::vector<int32_t> keys;
+    std::vector<double> values;
+    keys.reserve(rows_per_group * group_values.size());
+    values.reserve(rows_per_group * group_values.size());
+    for (cudf::size_type row = 0; row < rows_per_group; ++row) {
+      for (std::size_t group = 0; group < group_values.size(); ++group) {
+        keys.push_back(static_cast<int32_t>(group));
+        values.push_back(group_values[group][row]);
+      }
+    }
+
+    auto const keys_column =
+      cudf::test::fixed_width_column_wrapper<int32_t>(keys.begin(), keys.end());
+    auto const values_column =
+      cudf::test::fixed_width_column_wrapper<double>(values.begin(), values.end());
+
+    cudf::groupby::groupby gb(cudf::table_view{{keys_column}});
+    std::vector<cudf::groupby::aggregation_request> requests;
+    std::vector<std::unique_ptr<cudf::groupby_aggregation>> aggregations;
+    aggregations.push_back(
+      cudf::make_tdigest_aggregation<cudf::groupby_aggregation>(max_centroids));
+    requests.push_back({values_column, std::move(aggregations)});
+    auto const groupby_result = gb.aggregate(requests);
+
+    cudf::tdigest::tdigest_column_view tdv(*groupby_result.second[0].results[0]);
+    auto const result = cudf::percentile_approx(tdv, percentiles_column);
+
+    auto const actual = lists_column_to_host(result->view());
+    auto const result_keys =
+      cudf::test::to_host<int32_t>(groupby_result.first->get_column(0)).first;
+    ASSERT_EQ(actual.size(), result_keys.size());
+    ASSERT_EQ(actual.size(), group_values.size());
+    std::vector<bool> saw_group(group_values.size(), false);
+    for (std::size_t row = 0; row < actual.size(); ++row) {
+      auto const key = result_keys[row];
+      ASSERT_GE(key, 0);
+      ASSERT_LT(key, static_cast<int32_t>(group_values.size()));
+      ASSERT_FALSE(saw_group[key]);
+      saw_group[key] = true;
+      expect_approx_percentiles_near_exact(group_values[key], percentiles, actual[row]);
+    }
+    EXPECT_TRUE(std::all_of(
+      saw_group.cbegin(), saw_group.cend(), [](bool const was_seen) { return was_seen; }));
+  }
+}
+
+}  // namespace
+
 TEST_F(PercentileApproxTest, ReductionWithLowRowCount)
 {
   // Test that the tdigest reduction with a low row count still produces the correct results.

From 30e1d50a61171d79b9b8e75c3e7f27bdecb46915 Mon Sep 17 00:00:00 2001
From: Vyas Ramasubramani <vyasr@nvidia.com>
Date: Thu, 27 Aug 2026 14:55:21 -0700
Subject: [PATCH 2/5] Validate percentile approximation by rank

---
 .../quantiles/percentile_approx_test.cpp      | 66 ++++++++++++++-----
 1 file changed, 50 insertions(+), 16 deletions(-)

diff --git a/cpp/tests/quantiles/percentile_approx_test.cpp b/cpp/tests/quantiles/percentile_approx_test.cpp
index d4b3a8510f6f..db50d6c304e2 100644
--- a/cpp/tests/quantiles/percentile_approx_test.cpp
+++ b/cpp/tests/quantiles/percentile_approx_test.cpp
@@ -26,6 +26,8 @@
 #include <algorithm>
 #include <cmath>
 #include <cstddef>
+#include <iterator>
+#include <utility>
 #include <vector>
 
 namespace {
@@ -240,16 +242,40 @@ std::vector<double> make_compressed_test_values(cudf::size_type group, cudf::siz
   return values;
 }
 
-double exact_host_quantile(std::vector<double> values, double percentile)
+double exact_host_quantile(std::vector<double> const& sorted_values, double percentile)
 {
-  std::sort(values.begin(), values.end());
-  auto const rank       = percentile * static_cast<double>(values.size() - 1);
+  auto const rank       = percentile * static_cast<double>(sorted_values.size() - 1);
   auto const lower_rank = static_cast<std::size_t>(std::floor(rank));
   auto const upper_rank = static_cast<std::size_t>(std::ceil(rank));
-  if (lower_rank == upper_rank) { return values[lower_rank]; }
+  if (lower_rank == upper_rank) { return sorted_values[lower_rank]; }
 
   auto const weight = rank - static_cast<double>(lower_rank);
-  return values[lower_rank] + (values[upper_rank] - values[lower_rank]) * weight;
+  return sorted_values[lower_rank] +
+         (sorted_values[upper_rank] - sorted_values[lower_rank]) * weight;
+}
+
+std::pair<double, double> empirical_rank_interval(std::vector<double> const& sorted_values,
+                                                  double value)
+{
+  auto const denominator = static_cast<double>(sorted_values.size() - 1);
+  auto const first       = std::lower_bound(sorted_values.begin(), sorted_values.end(), value);
+
+  if (first != sorted_values.end() && *first == value) {
+    auto const last = std::upper_bound(first, sorted_values.end(), value);
+    return {static_cast<double>(std::distance(sorted_values.begin(), first)) / denominator,
+            static_cast<double>(std::distance(sorted_values.begin(), last) - 1) / denominator};
+  }
+
+  if (first == sorted_values.begin()) { return {0.0, 0.0}; }
+  if (first == sorted_values.end()) { return {1.0, 1.0}; }
+
+  auto const upper_index = static_cast<std::size_t>(std::distance(sorted_values.begin(), first));
+  auto const lower_index = upper_index - 1;
+  auto const lower_value = sorted_values[lower_index];
+  auto const upper_value = sorted_values[upper_index];
+  auto const fraction    = (value - lower_value) / (upper_value - lower_value);
+  auto const rank        = (static_cast<double>(lower_index) + fraction) / denominator;
+  return {rank, rank};
 }
 
 std::vector<std::vector<double>> lists_column_to_host(cudf::column_view const& column)
@@ -267,19 +293,24 @@ std::vector<std::vector<double>> lists_column_to_host(cudf::column_view const& c
 
 void expect_approx_percentiles_near_exact(std::vector<double> const& values,
                                           std::vector<double> const& percentiles,
-                                          std::vector<double> const& actual)
+                                          std::vector<double> const& actual,
+                                          double max_rank_error)
 {
   ASSERT_EQ(actual.size(), percentiles.size());
 
-  auto const [minimum, maximum] = std::minmax_element(values.begin(), values.end());
-  auto const value_tolerance    = (*maximum - *minimum) * 0.08;
-  // The low centroid count intentionally compresses thousands of rows. This bound is loose enough
-  // for both centroid clustering implementations but still catches gross percentile indexing,
-  // interpolation, or weighting regressions against an independent host quantile oracle.
+  auto sorted_values = values;
+  std::sort(sorted_values.begin(), sorted_values.end());
   for (std::size_t idx = 0; idx < percentiles.size(); ++idx) {
-    auto const exact = exact_host_quantile(values, percentiles[idx]);
-    EXPECT_NEAR(actual[idx], exact, value_tolerance)
-      << "percentile=" << percentiles[idx] << " exact=" << exact << " actual=" << actual[idx];
+    ASSERT_TRUE(std::isfinite(actual[idx]));
+
+    auto const [rank_low, rank_high] = empirical_rank_interval(sorted_values, actual[idx]);
+    auto const nearest_rank          = std::clamp(percentiles[idx], rank_low, rank_high);
+    auto const rank_error            = std::abs(percentiles[idx] - nearest_rank);
+    auto const exact                 = exact_host_quantile(sorted_values, percentiles[idx]);
+
+    EXPECT_LE(rank_error, max_rank_error)
+      << "percentile=" << percentiles[idx] << " actual_rank=[" << rank_low << ", " << rank_high
+      << "] exact_value=" << exact << " actual_value=" << actual[idx];
   }
 }
 
@@ -567,6 +598,8 @@ TEST_F(PercentileApproxTest, CompressedTdigestsAgainstHostQuantiles)
 {
   auto const max_centroids  = 20;
   auto const rows_per_group = 4096;
+  // Regression budget validated for max_centroids=20 with both clustering paths.
+  auto const max_rank_error = 0.025;
   auto const percentiles    = std::vector<double>{0.01, 0.10, 0.25, 0.50, 0.75, 0.90, 0.99};
   auto const percentiles_column =
     cudf::test::fixed_width_column_wrapper<double>(percentiles.begin(), percentiles.end());
@@ -594,7 +627,7 @@ TEST_F(PercentileApproxTest, CompressedTdigestsAgainstHostQuantiles)
 
       auto const actual = lists_column_to_host(result->view());
       ASSERT_EQ(actual.size(), 1);
-      expect_approx_percentiles_near_exact(values, percentiles, actual.front());
+      expect_approx_percentiles_near_exact(values, percentiles, actual.front(), max_rank_error);
     }
 
     std::vector<int32_t> keys;
@@ -636,7 +669,8 @@ TEST_F(PercentileApproxTest, CompressedTdigestsAgainstHostQuantiles)
       ASSERT_LT(key, static_cast<int32_t>(group_values.size()));
       ASSERT_FALSE(saw_group[key]);
       saw_group[key] = true;
-      expect_approx_percentiles_near_exact(group_values[key], percentiles, actual[row]);
+      expect_approx_percentiles_near_exact(
+        group_values[key], percentiles, actual[row], max_rank_error);
     }
     EXPECT_TRUE(std::all_of(
       saw_group.cbegin(), saw_group.cend(), [](bool const was_seen) { return was_seen; }));

From b6326de17bb5063e3508dd7e294a584c96849fbe Mon Sep 17 00:00:00 2001
From: Vyas Ramasubramani <vyasr@nvidia.com>
Date: Fri, 28 Aug 2026 17:16:38 -0700
Subject: [PATCH 3/5] Assert percentile approximation bounds

---
 cpp/tests/quantiles/percentile_approx_test.cpp | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/cpp/tests/quantiles/percentile_approx_test.cpp b/cpp/tests/quantiles/percentile_approx_test.cpp
index 9854cea544f0..9d51b0ed7b29 100644
--- a/cpp/tests/quantiles/percentile_approx_test.cpp
+++ b/cpp/tests/quantiles/percentile_approx_test.cpp
@@ -303,6 +303,8 @@ void expect_approx_percentiles_near_exact(std::vector<double> const& values,
   std::sort(sorted_values.begin(), sorted_values.end());
   for (std::size_t idx = 0; idx < percentiles.size(); ++idx) {
     ASSERT_TRUE(std::isfinite(actual[idx]));
+    ASSERT_GE(actual[idx], sorted_values.front());
+    ASSERT_LE(actual[idx], sorted_values.back());
 
     auto const [rank_low, rank_high] = empirical_rank_interval(sorted_values, actual[idx]);
     auto const nearest_rank          = std::clamp(percentiles[idx], rank_low, rank_high);

From 592a42c73348df17a56adfd09f82434b41368541 Mon Sep 17 00:00:00 2001
From: Vyas Ramasubramani <vyasr@nvidia.com>
Date: Fri, 28 Aug 2026 19:18:41 -0700
Subject: [PATCH 4/5] Fix libcudf streaming build with rapidsmpf 26.10

---
 .../benchmarks/streaming/ndsh/concatenate.cpp |  3 ++-
 .../benchmarks/streaming/ndsh/join.cpp        | 10 ++++++---
 .../streaming/ndsh/parquet_writer.cpp         |  7 ++++--
 .../src/approx_distinct_count.cpp             |  5 ++++-
 cpp/libcudf_streaming/src/bloom_filter.cpp    | 11 +++++++---
 .../src/channel_metadata.cpp                  |  4 +++-
 cpp/libcudf_streaming/src/partition_utils.cpp | 22 +++++++++----------
 7 files changed, 40 insertions(+), 22 deletions(-)

diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp
index 16382695a981..d1f362ad2223 100644
--- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp
+++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp
@@ -55,7 +55,8 @@ streaming::Actor concatenate(std::shared_ptr<streaming::Context> ctx,
     views.reserve(messages.size());
     for (auto&& msg : messages) {
       auto chunk = co_await msg.release<cudf_streaming::table_chunk>().make_available(ctx);
-      rapidsmpf::cuda_stream_join(concat_stream, chunk.stream(), &event);
+      rapidsmpf::cuda_stream_join(
+        std::ranges::single_view{concat_stream}, std::ranges::single_view{chunk.stream()}, &event);
       views.push_back(chunk.table_view());
       chunks.push_back(std::move(chunk));
     }
diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp
index 6cb43c146a2b..f4eb68aed043 100644
--- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp
+++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp
@@ -36,6 +36,7 @@
 
 #include <algorithm>
 #include <memory>
+#include <ranges>
 #include <vector>
 
 namespace rapidsmpf::ndsh {
@@ -63,7 +64,8 @@ coro::task<streaming::Message> broadcast(std::shared_ptr<streaming::Context> ctx
       auto msg = co_await ch_in->receive();
       if (msg.empty()) { break; }
       auto chunk = co_await msg.release<cudf_streaming::table_chunk>().make_available(ctx);
-      rapidsmpf::cuda_stream_join(gather_stream, chunk.stream(), &event);
+      rapidsmpf::cuda_stream_join(
+        std::ranges::single_view{gather_stream}, std::ranges::single_view{chunk.stream()}, &event);
       views.push_back(chunk.table_view());
       chunks.push_back(std::move(chunk));
     }
@@ -177,7 +179,8 @@ streaming::Message semi_join_chunk(std::shared_ptr<streaming::Context> ctx,
 
   auto result_table = std::make_unique<cudf::table>(std::move(result_columns));
   // Deallocation of the join indices will happen on chunk_stream, so add stream dep
-  rapidsmpf::cuda_stream_join(left_chunk.stream(), chunk_stream);
+  rapidsmpf::cuda_stream_join(std::ranges::single_view{left_chunk.stream()},
+                              std::ranges::single_view{chunk_stream});
 
   return to_message(
     sequence, std::make_unique<cudf_streaming::table_chunk>(std::move(result_table), chunk_stream));
@@ -243,7 +246,8 @@ streaming::Message inner_join_chunk(std::shared_ptr<streaming::Context> ctx,
                     std::back_inserter(result_columns));
   // Deallocation of the join indices will happen on build_stream, so add stream dep
   // This also ensure deallocation of the hash_join object waits for completion.
-  rapidsmpf::cuda_stream_join(build_stream, chunk_stream, tmp_event);
+  rapidsmpf::cuda_stream_join(
+    std::ranges::single_view{build_stream}, std::ranges::single_view{chunk_stream}, tmp_event);
   return to_message(sequence,
                     std::make_unique<cudf_streaming::table_chunk>(
                       std::make_unique<cudf::table>(std::move(result_columns)), chunk_stream));
diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp
index c18fdef44932..90ef3dc32472 100644
--- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp
+++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp
@@ -16,6 +16,7 @@
 
 #include <cstddef>
 #include <memory>
+#include <ranges>
 #include <vector>
 
 namespace rapidsmpf::ndsh {
@@ -51,9 +52,11 @@ rapidsmpf::streaming::Actor write_parquet(std::shared_ptr<rapidsmpf::streaming::
     table = chunk.table_view();
     RAPIDSMPF_EXPECTS(static_cast<std::size_t>(table.num_columns()) == column_names.size(),
                       "Mismatching number of column names and chunk columns");
-    rapidsmpf::cuda_stream_join(write_stream, chunk.stream(), &event);
+    rapidsmpf::cuda_stream_join(
+      std::ranges::single_view{write_stream}, std::ranges::single_view{chunk.stream()}, &event);
     writer.write(table);
-    rapidsmpf::cuda_stream_join(chunk.stream(), write_stream, &event);
+    rapidsmpf::cuda_stream_join(
+      std::ranges::single_view{chunk.stream()}, std::ranges::single_view{write_stream}, &event);
   }
   writer.close();
 }
diff --git a/cpp/libcudf_streaming/src/approx_distinct_count.cpp b/cpp/libcudf_streaming/src/approx_distinct_count.cpp
index 2cff62acd9e8..c2de26e7b913 100644
--- a/cpp/libcudf_streaming/src/approx_distinct_count.cpp
+++ b/cpp/libcudf_streaming/src/approx_distinct_count.cpp
@@ -26,6 +26,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <memory>
+#include <ranges>
 #include <utility>
 #include <vector>
 
@@ -115,7 +116,9 @@ rapidsmpf::streaming::Actor cardinality_estimator::estimate(
     auto const table =
       column_indices.empty() ? chunk.table_view() : chunk.table_view().select(column_indices);
     sketch.add(table, chunk.stream());
-    rapidsmpf::cuda_stream_join(sketch_stream, chunk.stream(), &add_event);
+    rapidsmpf::cuda_stream_join(std::ranges::single_view{sketch_stream},
+                                std::ranges::single_view{chunk.stream()},
+                                &add_event);
     reservation.clear();
     if (ch_sampled != nullptr) {
       co_await ch_sampled->send(
diff --git a/cpp/libcudf_streaming/src/bloom_filter.cpp b/cpp/libcudf_streaming/src/bloom_filter.cpp
index 6383304cece7..2e4f7edf3571 100644
--- a/cpp/libcudf_streaming/src/bloom_filter.cpp
+++ b/cpp/libcudf_streaming/src/bloom_filter.cpp
@@ -19,6 +19,8 @@
 #include <rapidsmpf/streaming/core/lineariser.hpp>
 #include <rapidsmpf/streaming/core/message.hpp>
 
+#include <ranges>
+
 namespace cudf_streaming {
 
 std::size_t bloom_filter::aligned_size(std::size_t size) noexcept
@@ -78,7 +80,8 @@ rapidsmpf::streaming::Actor bloom_filter::build(
     // kernels doing that concurrently because the updates are atomic.
     build_event.stream_wait(chunk.stream());
     filter.add(chunk.table_view(), chunk.stream(), mr);
-    rapidsmpf::cuda_stream_join(filter_stream, chunk.stream(), &event);
+    rapidsmpf::cuda_stream_join(
+      std::ranges::single_view{filter_stream}, std::ranges::single_view{chunk.stream()}, &event);
   }
   if (comm_->nranks() > 1) {
     auto reducer = rapidsmpf::streaming::AllReduce(
@@ -130,7 +133,8 @@ rapidsmpf::streaming::Actor bloom_filter::apply(
       ctx_,
       -rapidsmpf::safe_cast<std::int64_t>(chunk.data_alloc_size(rapidsmpf::MemoryType::DEVICE)));
     auto chunk_stream = chunk.stream();
-    rapidsmpf::cuda_stream_join(chunk_stream, stream, &event);
+    rapidsmpf::cuda_stream_join(
+      std::ranges::single_view{chunk_stream}, std::ranges::single_view{stream}, &event);
     // Reservation for the mask construction and guess at output size.
     auto res = co_await ctx_->memory(rapidsmpf::MemoryType::DEVICE)
                  ->reserve_or_wait(rapidsmpf::safe_cast<std::size_t>(chunk.table_view().num_rows())
@@ -142,7 +146,8 @@ rapidsmpf::streaming::Actor bloom_filter::apply(
                                    0);
     auto mask =
       filter.contains(chunk.table_view().select(keys), chunk_stream, ctx_->br()->device_mr());
-    rapidsmpf::cuda_stream_join(stream, chunk_stream, &event);
+    rapidsmpf::cuda_stream_join(
+      std::ranges::single_view{stream}, std::ranges::single_view{chunk_stream}, &event);
     RAPIDSMPF_EXPECTS(mask.size() == static_cast<std::size_t>(chunk.table_view().num_rows()),
                       "Invalid mask size");
     auto mask_view = cudf::column_view{cudf::data_type{cudf::type_id::BOOL8},
diff --git a/cpp/libcudf_streaming/src/channel_metadata.cpp b/cpp/libcudf_streaming/src/channel_metadata.cpp
index fe31d81f5e94..406b927da34a 100644
--- a/cpp/libcudf_streaming/src/channel_metadata.cpp
+++ b/cpp/libcudf_streaming/src/channel_metadata.cpp
@@ -17,6 +17,7 @@
 
 #include <algorithm>
 #include <memory>
+#include <ranges>
 #include <stdexcept>
 #include <utility>
 
@@ -69,7 +70,8 @@ bool ordering::boundaries_aligned_with(ordering const& other, rapidsmpf::BufferR
   auto const lhs    = boundaries->table_view();
   auto const rhs    = other.boundaries->table_view();
   auto const stream = boundaries->stream();
-  rapidsmpf::cuda_stream_join(stream, other.boundaries->stream());
+  rapidsmpf::cuda_stream_join(std::ranges::single_view{stream},
+                              std::ranges::single_view{other.boundaries->stream()});
   for (cudf::size_type i = 0; i < lhs.num_columns(); ++i) {
     auto eq      = cudf::binary_operation(lhs.column(i),
                                      rhs.column(i),
diff --git a/cpp/libcudf_streaming/src/partition_utils.cpp b/cpp/libcudf_streaming/src/partition_utils.cpp
index 0e2528baa538..51b3427c554b 100644
--- a/cpp/libcudf_streaming/src/partition_utils.cpp
+++ b/cpp/libcudf_streaming/src/partition_utils.cpp
@@ -199,10 +199,9 @@ std::unordered_map<rapidsmpf::shuffler::PartID, rapidsmpf::PackedData> partition
     return split_and_pack(table, splits, stream, reservation, reorder_bytes);
   }
 
-  auto res                       = reservation.split(reorder_bytes);
   auto [reordered, split_points] = cudf::hash_partition(
     table, columns_to_hash, num_partitions, hash_function, seed, stream, br->device_mr());
-  res.clear();  // The reorder has landed, hand its bytes back.
+  br->release(reservation, reorder_bytes);  // The reorder has landed, hand its bytes back.
   std::vector<cudf::size_type> splits(split_points.begin() + 1, split_points.end() - 1);
   // Reordering does not change the packed size.
   return split_and_pack(reordered->view(), splits, stream, reservation, reorder_bytes);
@@ -246,9 +245,9 @@ std::unordered_map<rapidsmpf::shuffler::PartID, rapidsmpf::PackedData> split_and
   // at least the size of the table.
   auto const split_bytes =
     packed_and_total_size(table, stream, br->device_mr(), packed_bytes).first;
-  auto res    = reservation.split(split_bytes);
+  check_reservation(reservation, split_bytes);
   auto packed = cudf::contiguous_split(table, splits, stream, br->device_mr());
-  res.clear();  // The split has landed, hand its bytes back.
+  br->release(reservation, split_bytes);  // The split has landed, hand its bytes back.
   ret.reserve(packed.size());
   for (rapidsmpf::shuffler::PartID i = 0; rapidsmpf::safe_cast<std::size_t>(i) < packed.size();
        i++) {
@@ -308,18 +307,17 @@ std::unique_ptr<cudf::table> unpack_and_concat(std::vector<rapidsmpf::PackedData
 
   // Covers the unspill and the concatenation. `move_to_device_buffer()` consumes
   // `non_device_size` of it as it moves each partition, leaving `total_size` accounted
-  // for until the concatenation has allocated it. Splitting once means an undersized
-  // reservation throws before it is mutated, so the caller can reserve more and retry.
-  auto res = reservation.split(total_size + non_device_size);
+  // for until the concatenation has allocated it.
+  check_reservation(reservation, total_size + non_device_size);
 
   for (auto& packed_data : partitions) {
     if (!packed_data.empty()) {
       if (packed_data.data->size > 0) {  // No need to sync empty buffers.
         packed_data_streams.push_back(packed_data.data->stream());
       }
-      unpacked.push_back(cudf::unpack(
-        references.emplace_back(std::move(packed_data.metadata),
-                                br->move_to_device_buffer(std::move(packed_data.data), res))));
+      unpacked.push_back(cudf::unpack(references.emplace_back(
+        std::move(packed_data.metadata),
+        br->move_to_device_buffer(std::move(packed_data.data), reservation))));
     }
   }
 
@@ -332,7 +330,9 @@ std::unique_ptr<cudf::table> unpack_and_concat(std::vector<rapidsmpf::PackedData
     packed_columns.gpu_data->set_stream(stream);
   }
 
-  return cudf::concatenate(unpacked, stream, br->device_mr());
+  auto result = cudf::concatenate(unpacked, stream, br->device_mr());
+  br->release(reservation, total_size);
+  return result;
 }
 
 }  // namespace cudf_streaming

From 63ba596b18172604b4689b3305bad5ba5476c85c Mon Sep 17 00:00:00 2001
From: Vyas Ramasubramani <vyasr@nvidia.com>
Date: Fri, 28 Aug 2026 19:29:20 -0700
Subject: [PATCH 5/5] Revert "Fix libcudf streaming build with rapidsmpf 26.10"

This reverts commit 592a42c73348df17a56adfd09f82434b41368541.
---
 .../benchmarks/streaming/ndsh/concatenate.cpp |  3 +--
 .../benchmarks/streaming/ndsh/join.cpp        | 10 +++------
 .../streaming/ndsh/parquet_writer.cpp         |  7 ++----
 .../src/approx_distinct_count.cpp             |  5 +----
 cpp/libcudf_streaming/src/bloom_filter.cpp    | 11 +++-------
 .../src/channel_metadata.cpp                  |  4 +---
 cpp/libcudf_streaming/src/partition_utils.cpp | 22 +++++++++----------
 7 files changed, 22 insertions(+), 40 deletions(-)

diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp
index d1f362ad2223..16382695a981 100644
--- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp
+++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/concatenate.cpp
@@ -55,8 +55,7 @@ streaming::Actor concatenate(std::shared_ptr<streaming::Context> ctx,
     views.reserve(messages.size());
     for (auto&& msg : messages) {
       auto chunk = co_await msg.release<cudf_streaming::table_chunk>().make_available(ctx);
-      rapidsmpf::cuda_stream_join(
-        std::ranges::single_view{concat_stream}, std::ranges::single_view{chunk.stream()}, &event);
+      rapidsmpf::cuda_stream_join(concat_stream, chunk.stream(), &event);
       views.push_back(chunk.table_view());
       chunks.push_back(std::move(chunk));
     }
diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp
index f4eb68aed043..6cb43c146a2b 100644
--- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp
+++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/join.cpp
@@ -36,7 +36,6 @@
 
 #include <algorithm>
 #include <memory>
-#include <ranges>
 #include <vector>
 
 namespace rapidsmpf::ndsh {
@@ -64,8 +63,7 @@ coro::task<streaming::Message> broadcast(std::shared_ptr<streaming::Context> ctx
       auto msg = co_await ch_in->receive();
       if (msg.empty()) { break; }
       auto chunk = co_await msg.release<cudf_streaming::table_chunk>().make_available(ctx);
-      rapidsmpf::cuda_stream_join(
-        std::ranges::single_view{gather_stream}, std::ranges::single_view{chunk.stream()}, &event);
+      rapidsmpf::cuda_stream_join(gather_stream, chunk.stream(), &event);
       views.push_back(chunk.table_view());
       chunks.push_back(std::move(chunk));
     }
@@ -179,8 +177,7 @@ streaming::Message semi_join_chunk(std::shared_ptr<streaming::Context> ctx,
 
   auto result_table = std::make_unique<cudf::table>(std::move(result_columns));
   // Deallocation of the join indices will happen on chunk_stream, so add stream dep
-  rapidsmpf::cuda_stream_join(std::ranges::single_view{left_chunk.stream()},
-                              std::ranges::single_view{chunk_stream});
+  rapidsmpf::cuda_stream_join(left_chunk.stream(), chunk_stream);
 
   return to_message(
     sequence, std::make_unique<cudf_streaming::table_chunk>(std::move(result_table), chunk_stream));
@@ -246,8 +243,7 @@ streaming::Message inner_join_chunk(std::shared_ptr<streaming::Context> ctx,
                     std::back_inserter(result_columns));
   // Deallocation of the join indices will happen on build_stream, so add stream dep
   // This also ensure deallocation of the hash_join object waits for completion.
-  rapidsmpf::cuda_stream_join(
-    std::ranges::single_view{build_stream}, std::ranges::single_view{chunk_stream}, tmp_event);
+  rapidsmpf::cuda_stream_join(build_stream, chunk_stream, tmp_event);
   return to_message(sequence,
                     std::make_unique<cudf_streaming::table_chunk>(
                       std::make_unique<cudf::table>(std::move(result_columns)), chunk_stream));
diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp
index 90ef3dc32472..c18fdef44932 100644
--- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp
+++ b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/parquet_writer.cpp
@@ -16,7 +16,6 @@
 
 #include <cstddef>
 #include <memory>
-#include <ranges>
 #include <vector>
 
 namespace rapidsmpf::ndsh {
@@ -52,11 +51,9 @@ rapidsmpf::streaming::Actor write_parquet(std::shared_ptr<rapidsmpf::streaming::
     table = chunk.table_view();
     RAPIDSMPF_EXPECTS(static_cast<std::size_t>(table.num_columns()) == column_names.size(),
                       "Mismatching number of column names and chunk columns");
-    rapidsmpf::cuda_stream_join(
-      std::ranges::single_view{write_stream}, std::ranges::single_view{chunk.stream()}, &event);
+    rapidsmpf::cuda_stream_join(write_stream, chunk.stream(), &event);
     writer.write(table);
-    rapidsmpf::cuda_stream_join(
-      std::ranges::single_view{chunk.stream()}, std::ranges::single_view{write_stream}, &event);
+    rapidsmpf::cuda_stream_join(chunk.stream(), write_stream, &event);
   }
   writer.close();
 }
diff --git a/cpp/libcudf_streaming/src/approx_distinct_count.cpp b/cpp/libcudf_streaming/src/approx_distinct_count.cpp
index c2de26e7b913..2cff62acd9e8 100644
--- a/cpp/libcudf_streaming/src/approx_distinct_count.cpp
+++ b/cpp/libcudf_streaming/src/approx_distinct_count.cpp
@@ -26,7 +26,6 @@
 #include <cstddef>
 #include <cstdint>
 #include <memory>
-#include <ranges>
 #include <utility>
 #include <vector>
 
@@ -116,9 +115,7 @@ rapidsmpf::streaming::Actor cardinality_estimator::estimate(
     auto const table =
       column_indices.empty() ? chunk.table_view() : chunk.table_view().select(column_indices);
     sketch.add(table, chunk.stream());
-    rapidsmpf::cuda_stream_join(std::ranges::single_view{sketch_stream},
-                                std::ranges::single_view{chunk.stream()},
-                                &add_event);
+    rapidsmpf::cuda_stream_join(sketch_stream, chunk.stream(), &add_event);
     reservation.clear();
     if (ch_sampled != nullptr) {
       co_await ch_sampled->send(
diff --git a/cpp/libcudf_streaming/src/bloom_filter.cpp b/cpp/libcudf_streaming/src/bloom_filter.cpp
index 2e4f7edf3571..6383304cece7 100644
--- a/cpp/libcudf_streaming/src/bloom_filter.cpp
+++ b/cpp/libcudf_streaming/src/bloom_filter.cpp
@@ -19,8 +19,6 @@
 #include <rapidsmpf/streaming/core/lineariser.hpp>
 #include <rapidsmpf/streaming/core/message.hpp>
 
-#include <ranges>
-
 namespace cudf_streaming {
 
 std::size_t bloom_filter::aligned_size(std::size_t size) noexcept
@@ -80,8 +78,7 @@ rapidsmpf::streaming::Actor bloom_filter::build(
     // kernels doing that concurrently because the updates are atomic.
     build_event.stream_wait(chunk.stream());
     filter.add(chunk.table_view(), chunk.stream(), mr);
-    rapidsmpf::cuda_stream_join(
-      std::ranges::single_view{filter_stream}, std::ranges::single_view{chunk.stream()}, &event);
+    rapidsmpf::cuda_stream_join(filter_stream, chunk.stream(), &event);
   }
   if (comm_->nranks() > 1) {
     auto reducer = rapidsmpf::streaming::AllReduce(
@@ -133,8 +130,7 @@ rapidsmpf::streaming::Actor bloom_filter::apply(
       ctx_,
       -rapidsmpf::safe_cast<std::int64_t>(chunk.data_alloc_size(rapidsmpf::MemoryType::DEVICE)));
     auto chunk_stream = chunk.stream();
-    rapidsmpf::cuda_stream_join(
-      std::ranges::single_view{chunk_stream}, std::ranges::single_view{stream}, &event);
+    rapidsmpf::cuda_stream_join(chunk_stream, stream, &event);
     // Reservation for the mask construction and guess at output size.
     auto res = co_await ctx_->memory(rapidsmpf::MemoryType::DEVICE)
                  ->reserve_or_wait(rapidsmpf::safe_cast<std::size_t>(chunk.table_view().num_rows())
@@ -146,8 +142,7 @@ rapidsmpf::streaming::Actor bloom_filter::apply(
                                    0);
     auto mask =
       filter.contains(chunk.table_view().select(keys), chunk_stream, ctx_->br()->device_mr());
-    rapidsmpf::cuda_stream_join(
-      std::ranges::single_view{stream}, std::ranges::single_view{chunk_stream}, &event);
+    rapidsmpf::cuda_stream_join(stream, chunk_stream, &event);
     RAPIDSMPF_EXPECTS(mask.size() == static_cast<std::size_t>(chunk.table_view().num_rows()),
                       "Invalid mask size");
     auto mask_view = cudf::column_view{cudf::data_type{cudf::type_id::BOOL8},
diff --git a/cpp/libcudf_streaming/src/channel_metadata.cpp b/cpp/libcudf_streaming/src/channel_metadata.cpp
index 406b927da34a..fe31d81f5e94 100644
--- a/cpp/libcudf_streaming/src/channel_metadata.cpp
+++ b/cpp/libcudf_streaming/src/channel_metadata.cpp
@@ -17,7 +17,6 @@
 
 #include <algorithm>
 #include <memory>
-#include <ranges>
 #include <stdexcept>
 #include <utility>
 
@@ -70,8 +69,7 @@ bool ordering::boundaries_aligned_with(ordering const& other, rapidsmpf::BufferR
   auto const lhs    = boundaries->table_view();
   auto const rhs    = other.boundaries->table_view();
   auto const stream = boundaries->stream();
-  rapidsmpf::cuda_stream_join(std::ranges::single_view{stream},
-                              std::ranges::single_view{other.boundaries->stream()});
+  rapidsmpf::cuda_stream_join(stream, other.boundaries->stream());
   for (cudf::size_type i = 0; i < lhs.num_columns(); ++i) {
     auto eq      = cudf::binary_operation(lhs.column(i),
                                      rhs.column(i),
diff --git a/cpp/libcudf_streaming/src/partition_utils.cpp b/cpp/libcudf_streaming/src/partition_utils.cpp
index 51b3427c554b..0e2528baa538 100644
--- a/cpp/libcudf_streaming/src/partition_utils.cpp
+++ b/cpp/libcudf_streaming/src/partition_utils.cpp
@@ -199,9 +199,10 @@ std::unordered_map<rapidsmpf::shuffler::PartID, rapidsmpf::PackedData> partition
     return split_and_pack(table, splits, stream, reservation, reorder_bytes);
   }
 
+  auto res                       = reservation.split(reorder_bytes);
   auto [reordered, split_points] = cudf::hash_partition(
     table, columns_to_hash, num_partitions, hash_function, seed, stream, br->device_mr());
-  br->release(reservation, reorder_bytes);  // The reorder has landed, hand its bytes back.
+  res.clear();  // The reorder has landed, hand its bytes back.
   std::vector<cudf::size_type> splits(split_points.begin() + 1, split_points.end() - 1);
   // Reordering does not change the packed size.
   return split_and_pack(reordered->view(), splits, stream, reservation, reorder_bytes);
@@ -245,9 +246,9 @@ std::unordered_map<rapidsmpf::shuffler::PartID, rapidsmpf::PackedData> split_and
   // at least the size of the table.
   auto const split_bytes =
     packed_and_total_size(table, stream, br->device_mr(), packed_bytes).first;
-  check_reservation(reservation, split_bytes);
+  auto res    = reservation.split(split_bytes);
   auto packed = cudf::contiguous_split(table, splits, stream, br->device_mr());
-  br->release(reservation, split_bytes);  // The split has landed, hand its bytes back.
+  res.clear();  // The split has landed, hand its bytes back.
   ret.reserve(packed.size());
   for (rapidsmpf::shuffler::PartID i = 0; rapidsmpf::safe_cast<std::size_t>(i) < packed.size();
        i++) {
@@ -307,17 +308,18 @@ std::unique_ptr<cudf::table> unpack_and_concat(std::vector<rapidsmpf::PackedData
 
   // Covers the unspill and the concatenation. `move_to_device_buffer()` consumes
   // `non_device_size` of it as it moves each partition, leaving `total_size` accounted
-  // for until the concatenation has allocated it.
-  check_reservation(reservation, total_size + non_device_size);
+  // for until the concatenation has allocated it. Splitting once means an undersized
+  // reservation throws before it is mutated, so the caller can reserve more and retry.
+  auto res = reservation.split(total_size + non_device_size);
 
   for (auto& packed_data : partitions) {
     if (!packed_data.empty()) {
       if (packed_data.data->size > 0) {  // No need to sync empty buffers.
         packed_data_streams.push_back(packed_data.data->stream());
       }
-      unpacked.push_back(cudf::unpack(references.emplace_back(
-        std::move(packed_data.metadata),
-        br->move_to_device_buffer(std::move(packed_data.data), reservation))));
+      unpacked.push_back(cudf::unpack(
+        references.emplace_back(std::move(packed_data.metadata),
+                                br->move_to_device_buffer(std::move(packed_data.data), res))));
     }
   }
 
@@ -330,9 +332,7 @@ std::unique_ptr<cudf::table> unpack_and_concat(std::vector<rapidsmpf::PackedData
     packed_columns.gpu_data->set_stream(stream);
   }
 
-  auto result = cudf::concatenate(unpacked, stream, br->device_mr());
-  br->release(reservation, total_size);
-  return result;
+  return cudf::concatenate(unpacked, stream, br->device_mr());
 }
 
 }  // namespace cudf_streaming