Rework approach to cudf-streaming bloom filter sizing

cpp/include/cudf/reduction/bloom_filter.cuh

@@ -0,0 +1,27 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cuco/bloom_filter_policies.cuh>
+
+#include <cstdint>
+
+namespace cudf {
+
+/**
+ * @brief Policy describing the Apache Arrow Block-Split Bloom Filter layout.
+ *
+ * Uses cuco's `parametric_filter_policy` with the Apache Arrow layout: 256-bit blocks (8 x
+ * `uint32_t`), 8 fingerprint bits per key, fully horizontal add (Theta=8), and fully vertical
+ * contains (Phi=8). This layout is bit-compatible with Apache Arrow.
+ *
+ * @tparam Hash The hash function used to generate a hash for each key.
+ */
+template <typename Hash>
+using arrow_filter_policy =
+  cuco::parametric_filter_policy<Hash, std::uint32_t, 8, 8, 8, 1, 1, 8, false, false>;
+
+}  // namespace cudf

cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp

@@ -357,8 +357,8 @@ int main(int argc, char** argv)
   int device;
   RAPIDSMPF_CUDA_TRY(cudaGetDevice(&device));
   RAPIDSMPF_CUDA_TRY(cudaDeviceGetAttribute(&l2size, cudaDevAttrL2CacheSize, device));
-  auto const num_filter_blocks =
-    cudf_streaming::bloom_filter::fitting_num_blocks(static_cast<std::size_t>(l2size));
+  auto const filter_size =
+    cudf_streaming::bloom_filter::aligned_size(static_cast<std::size_t>(l2size) * 2 / 3);
 
   for (int i = 0; i < arguments.num_iterations; i++) {
     int op_id{0};
@@ -406,7 +406,7 @@ int main(int argc, char** argv)
       actors.push_back(fanout_bounded(
         ctx, comm, customer_x_orders, bloom_filter_input, {0}, customer_x_orders_input));
       auto bloom_filter =
-        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, num_filter_blocks);
+        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, filter_size);
       actors.push_back(bloom_filter.build(
         bloom_filter_input, bloom_filter_output, static_cast<rapidsmpf::OpID>(10 * i + op_id++)));
       // Out: l_orderkey, l_extendedprice, l_discount

cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp

@@ -256,8 +256,8 @@ int main(int argc, char** argv)
   int device;
   RAPIDSMPF_CUDA_TRY(cudaGetDevice(&device));
   RAPIDSMPF_CUDA_TRY(cudaDeviceGetAttribute(&l2size, cudaDevAttrL2CacheSize, device));
-  auto const num_filter_blocks =
-    cudf_streaming::bloom_filter::fitting_num_blocks(static_cast<std::size_t>(l2size));
+  auto const filter_size =
+    cudf_streaming::bloom_filter::aligned_size(static_cast<std::size_t>(l2size) * 2 / 3);
 
   for (int i = 0; i < arguments.num_iterations; i++) {
     rapidsmpf::OpID op_id{0};
@@ -302,7 +302,7 @@ int main(int argc, char** argv)
       // Build bloom filter from filtered orders' o_orderkey
       auto bloom_filter_output = ctx->create_channel();
       auto bloom_filter =
-        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, num_filter_blocks);
+        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, filter_size);
       actors.push_back(bloom_filter.build(
         bloom_filter_input, bloom_filter_output, static_cast<rapidsmpf::OpID>(10 * i + op_id++)));
 

cpp/libcudf_streaming/benchmarks/streaming/ndsh/q21.cpp

@@ -499,8 +499,8 @@ int main(int argc, char** argv)
   int device;
   RAPIDSMPF_CUDA_TRY(cudaGetDevice(&device));
   RAPIDSMPF_CUDA_TRY(cudaDeviceGetAttribute(&l2size, cudaDevAttrL2CacheSize, device));
-  auto const num_filter_blocks =
-    cudf_streaming::bloom_filter::fitting_num_blocks(static_cast<std::size_t>(l2size));
+  auto const filter_size =
+    cudf_streaming::bloom_filter::aligned_size(static_cast<std::size_t>(l2size) * 2 / 3);
   for (int i = 0; i < arguments.num_iterations; i++) {
     int op_id{0};
     std::vector<rapidsmpf::streaming::Actor> actors;
@@ -670,7 +670,7 @@ int main(int argc, char** argv)
                                             rapidsmpf::streaming::actor::FanoutPolicy::UNBOUNDED));
       auto bloom_output = ctx->create_channel();
       auto bloom_filter =
-        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, num_filter_blocks);
+        cudf_streaming::bloom_filter(ctx, comm, cudf::DEFAULT_HASH_SEED, filter_size);
       // Select the relevant key column(s) and build filter.
       actors.push_back(populate_bloom_filter(ctx,
                                              comm,

cpp/libcudf_streaming/include/cudf_streaming/bloom_filter.hpp

@@ -43,18 +43,25 @@ struct bloom_filter {
    * @param ctx Streaming context.
    * @param comm Communicator for the collective operation.
    * @param seed Hash seed used when hashing values into the filter.
-   * @param num_filter_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be positive and satisfy
+   * `aligned_size(filter_size) == filter_size`, and must not exceed the maximum size supported by
+   * the filter policy.
+   *
+   * @throws std::logic_error If `filter_size` is zero, incorrectly aligned, or exceeds the policy
+   * maximum.
    */
   explicit bloom_filter(std::shared_ptr<rapidsmpf::streaming::Context> ctx,
                         std::shared_ptr<rapidsmpf::Communicator> comm,
                         std::uint64_t seed,
-                        std::size_t num_filter_blocks) noexcept
-    : ctx_{std::move(ctx)},
-      comm_{std::move(comm)},
-      seed_{seed},
-      num_filter_blocks_{num_filter_blocks}
-  {
-  }
+                        std::size_t filter_size);
+
+  /**
+   * @brief Find the largest valid filter size no greater than a byte count.
+   *
+   * @param size Byte count to align.
+   * @return Largest valid filter size less than or equal to `size`.
+   */
+  [[nodiscard]] static std::size_t aligned_size(std::size_t size) noexcept;
 
   /**
    * @brief Gets the communicator associated with this bloom_filter.
@@ -100,22 +107,10 @@ struct bloom_filter {
     std::shared_ptr<rapidsmpf::streaming::Channel> ch_out,
     std::vector<cudf::size_type> keys);
 
-  /**
-   * @brief Compute number of filter blocks that fit in the given L2 cache size.
-   *
-   * @param l2size L2 cache size in bytes.
-   * @return Number of filter blocks that fit.
-   */
-  [[nodiscard]] static std::size_t fitting_num_blocks(std::size_t l2size) noexcept
-  {
-    using StorageType = std::uint32_t;
-    return (l2size * 2) / (3 * sizeof(StorageType));
-  }
-
  private:
   std::shared_ptr<rapidsmpf::streaming::Context> ctx_{};
   std::shared_ptr<rapidsmpf::Communicator> comm_{};
   std::uint64_t seed_{};
-  std::size_t num_filter_blocks_{};
+  std::size_t filter_size_{};
 };
 }  // namespace cudf_streaming

cpp/libcudf_streaming/include/cudf_streaming/detail/device_bloom_filter.hpp

@@ -28,40 +28,58 @@ struct device_bloom_filter {
   /**
    * @brief Create a filter.
    *
-   * @param num_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be a positive multiple of the filter
+   * block size and no greater than the maximum supported by the filter policy.
    * @param seed Seed used for hashing each value.
    * @param storage Storage to view as a bloom filter, must be appropriately
    * initialized.
    */
-  device_bloom_filter(std::size_t num_blocks, std::uint64_t seed, void* storage);
+  device_bloom_filter(std::size_t filter_size, std::uint64_t seed, void* storage);
 
   /**
    * @brief Create a read-only filter.
    *
-   * @param num_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be a positive multiple of the filter
+   * block size and no greater than the maximum supported by the filter policy.
    * @param seed Seed used for hashing each value.
    * @param storage View of storage, must be appropriately initialized.
    *
    * @return A const-qualified bloom filter viewing the underlying storage.
    */
-  static device_bloom_filter const view(std::size_t num_blocks,
+  static device_bloom_filter const view(std::size_t filter_size,
                                         std::uint64_t seed,
                                         void const* storage);
 
   /**
    * @brief Create uninitialized storage for a filter.
    *
-   * @param num_blocks Number of blocks in the filter.
+   * @param filter_size Filter storage size in bytes. Must be a positive multiple of the filter
+   * block size and no greater than the maximum supported by the filter policy.
    * @param stream CUDA stream for device operations.
    * @param mr Memory resource for allocations.
    *
    * @return Unique pointer to a device buffer containing storage for the requested
-   * number of filter blocks.
+   * filter size.
    */
-  static std::unique_ptr<rmm::device_buffer> storage(std::size_t num_blocks,
+  static std::unique_ptr<rmm::device_buffer> storage(std::size_t filter_size,
                                                      rmm::cuda_stream_view stream,
                                                      rmm::device_async_resource_ref mr);
 
+  /**
+   * @brief Find the largest valid filter size no greater than a byte count.
+   *
+   * @param size Byte count to align.
+   * @return Largest valid filter size less than or equal to `size`.
+   */
+  [[nodiscard]] static std::size_t aligned_size(std::size_t size) noexcept;
+
+  /**
+   *  Return the largest storage size supported by the filter policy.
+   *
+   *  Maximum valid filter size in bytes.
+   */
+  [[nodiscard]] static std::size_t max_size() noexcept;
+
   /**
    * @brief Add values to the filter.
    *
@@ -111,14 +129,6 @@ struct device_bloom_filter {
    */
   [[nodiscard]] std::size_t size() const noexcept;
 
-  /**
-   * @brief @return Number of blocks to use if the filter should fit in a given L2 cache
-   * size.
-   *
-   * @param l2size Size of the L2 cache in bytes.
-   */
-  [[nodiscard]] static std::size_t fitting_num_blocks(std::size_t l2size) noexcept;
-
  private:
   std::size_t num_blocks_;  ///< Number of blocks used in the filter.
   std::uint64_t seed_;      ///< Seed used when hashing values.

cpp/libcudf_streaming/src/bloom_filter.cpp

@@ -21,6 +21,24 @@
 
 namespace cudf_streaming {
 
+std::size_t bloom_filter::aligned_size(std::size_t size) noexcept
+{
+  return detail::device_bloom_filter::aligned_size(size);
+}
+
+bloom_filter::bloom_filter(std::shared_ptr<rapidsmpf::streaming::Context> ctx,
+                           std::shared_ptr<rapidsmpf::Communicator> comm,
+                           std::uint64_t seed,
+                           std::size_t filter_size)
+  : ctx_{std::move(ctx)}, comm_{std::move(comm)}, seed_{seed}, filter_size_{filter_size}
+{
+  RAPIDSMPF_EXPECTS(filter_size_ > 0, "Bloom filter storage size must be positive");
+  RAPIDSMPF_EXPECTS(filter_size_ == aligned_size(filter_size_),
+                    "Bloom filter storage size must be a multiple of the filter block size");
+  RAPIDSMPF_EXPECTS(filter_size_ <= detail::device_bloom_filter::max_size(),
+                    "Bloom filter storage exceeds the maximum size supported by its policy");
+}
+
 rapidsmpf::streaming::Actor bloom_filter::build(
   std::shared_ptr<rapidsmpf::streaming::Channel> ch_in,
   std::shared_ptr<rapidsmpf::streaming::Channel> ch_out,
@@ -35,10 +53,9 @@ rapidsmpf::streaming::Actor bloom_filter::build(
   auto filter_stream = br->stream_pool()->get_stream();
   rapidsmpf::CudaEvent event;
   auto storage =
-    cudf_streaming::detail::device_bloom_filter::storage(num_filter_blocks_, filter_stream, mr);
+    cudf_streaming::detail::device_bloom_filter::storage(filter_size_, filter_stream, mr);
   RAPIDSMPF_CUDA_TRY(cudaMemsetAsync(storage->data(), 0, storage->size(), filter_stream));
-  auto filter =
-    cudf_streaming::detail::device_bloom_filter(num_filter_blocks_, seed_, storage->data());
+  auto filter = cudf_streaming::detail::device_bloom_filter(filter_size_, seed_, storage->data());
   rapidsmpf::CudaEvent build_event;
   build_event.record(filter_stream);
   while (!ch_out->is_shutdown()) {
@@ -69,15 +86,15 @@ rapidsmpf::streaming::Actor bloom_filter::build(
       comm_,
       br->move(std::move(storage), filter_stream),
       br->move(
-        cudf_streaming::detail::device_bloom_filter::storage(num_filter_blocks_, filter_stream, mr),
+        cudf_streaming::detail::device_bloom_filter::storage(filter_size_, filter_stream, mr),
         filter_stream),
       tag,
-      [num_blocks = num_filter_blocks_, seed = seed_](rapidsmpf::Buffer const* left,
-                                                      rapidsmpf::Buffer* right) {
+      [filter_size = filter_size_, seed = seed_](rapidsmpf::Buffer const* left,
+                                                 rapidsmpf::Buffer* right) {
         right->write_access([&](std::byte* out_bytes, rmm::cuda_stream_view stream) {
           auto const in =
-            cudf_streaming::detail::device_bloom_filter::view(num_blocks, seed, left->data());
-          cudf_streaming::detail::device_bloom_filter(num_blocks, seed, out_bytes)
+            cudf_streaming::detail::device_bloom_filter::view(filter_size, seed, left->data());
+          cudf_streaming::detail::device_bloom_filter(filter_size, seed, out_bytes)
             .merge(in, stream);
         });
       });
@@ -102,9 +119,8 @@ rapidsmpf::streaming::Actor bloom_filter::apply(
                     "Bloom filter channel contained more than one message");
   auto stream = storage.stream();
   rapidsmpf::CudaEvent event;
-  auto filter =
-    cudf_streaming::detail::device_bloom_filter(num_filter_blocks_, seed_, storage.data());
-  auto meta = co_await ch_in->receive_metadata();
+  auto filter = cudf_streaming::detail::device_bloom_filter(filter_size_, seed_, storage.data());
+  auto meta   = co_await ch_in->receive_metadata();
   if (!meta.empty()) { co_await ch_out->send_metadata(std::move(meta)); }
   while (!ch_out->is_shutdown()) {
     auto msg = co_await ch_in->receive();