From ff29c5a95a4fda628e790a182b8d316126e25ec7 Mon Sep 17 00:00:00 2001 From: Lawrence Mitchell Date: Wed, 1 Jul 2026 17:29:21 +0100 Subject: [PATCH 1/5] Rework approach to cudf-streaming bloom filter sizing Rather than asking for a fitting number of blocks for an size, just require the user to provide the size of the filter directly. Since it must be a multiple of the storage type, also offer an aligned_size utility that takes an input and makes that happen. Additionally, share the new cuco parametric filter policy across cudf and cudf-streaming. --- cpp/include/cudf/reduction/bloom_filter.cuh | 27 ++++++ .../benchmarks/streaming/ndsh/q03.cpp | 6 +- .../benchmarks/streaming/ndsh/q04.cpp | 6 +- .../benchmarks/streaming/ndsh/q21.cpp | 6 +- .../include/cudf_streaming/bloom_filter.hpp | 37 ++++---- .../detail/device_bloom_filter.hpp | 40 +++++---- cpp/libcudf_streaming/src/bloom_filter.cpp | 38 +++++--- .../src/detail/device_bloom_filter.cu | 54 ++++++------ cpp/libcudf_streaming/tests/CMakeLists.txt | 6 +- .../tests/streaming/test_bloom_filter.cu | 86 +++++++++++++++++++ .../streaming/test_bloom_filter_config.cpp | 41 +++++++++ cpp/src/io/parquet/bloom_filter_reader.cu | 13 +-- cpp/tests/io/parquet_bloom_filter_test.cu | 17 +--- .../cudf_polars/streaming/actor_graph/join.py | 6 +- .../cudf_streaming/bloom_filter.pxd | 11 +-- .../cudf_streaming/bloom_filter.pyi | 4 +- .../cudf_streaming/bloom_filter.pyx | 21 ++--- .../cudf_streaming/tests/test_bloom_filter.py | 15 +++- 18 files changed, 302 insertions(+), 132 deletions(-) create mode 100644 cpp/include/cudf/reduction/bloom_filter.cuh create mode 100644 cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu create mode 100644 cpp/libcudf_streaming/tests/streaming/test_bloom_filter_config.cpp diff --git a/cpp/include/cudf/reduction/bloom_filter.cuh b/cpp/include/cudf/reduction/bloom_filter.cuh new file mode 100644 index 000000000000..c8c7b3d56df8 --- /dev/null +++ b/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 + +#include + +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 +using arrow_filter_policy = + cuco::parametric_filter_policy; + +} // namespace cudf diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp index af94c8992bf1..900cc69f3856 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q03.cpp +++ b/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(l2size)); + auto const filter_size = + cudf_streaming::bloom_filter::aligned_size(static_cast(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(10 * i + op_id++))); // Out: l_orderkey, l_extendedprice, l_discount diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp index 515fa9bff1dc..86bc8482507e 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q04.cpp +++ b/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(l2size)); + auto const filter_size = + cudf_streaming::bloom_filter::aligned_size(static_cast(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(10 * i + op_id++))); diff --git a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q21.cpp b/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q21.cpp index 635b4b0a7f5a..7eaa08229187 100644 --- a/cpp/libcudf_streaming/benchmarks/streaming/ndsh/q21.cpp +++ b/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(l2size)); + auto const filter_size = + cudf_streaming::bloom_filter::aligned_size(static_cast(l2size) * 2 / 3); for (int i = 0; i < arguments.num_iterations; i++) { int op_id{0}; std::vector 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, diff --git a/cpp/libcudf_streaming/include/cudf_streaming/bloom_filter.hpp b/cpp/libcudf_streaming/include/cudf_streaming/bloom_filter.hpp index 7ea1ec567b35..9e87cc832b2f 100644 --- a/cpp/libcudf_streaming/include/cudf_streaming/bloom_filter.hpp +++ b/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 ctx, std::shared_ptr 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 ch_out, std::vector 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 ctx_{}; std::shared_ptr comm_{}; std::uint64_t seed_{}; - std::size_t num_filter_blocks_{}; + std::size_t filter_size_{}; }; } // namespace cudf_streaming diff --git a/cpp/libcudf_streaming/include/cudf_streaming/detail/device_bloom_filter.hpp b/cpp/libcudf_streaming/include/cudf_streaming/detail/device_bloom_filter.hpp index 3fcc8f5ddaf1..435087c60ffb 100644 --- a/cpp/libcudf_streaming/include/cudf_streaming/detail/device_bloom_filter.hpp +++ b/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 storage(std::size_t num_blocks, + static std::unique_ptr 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. diff --git a/cpp/libcudf_streaming/src/bloom_filter.cpp b/cpp/libcudf_streaming/src/bloom_filter.cpp index 78acf8431f05..c3411c25a6cb 100644 --- a/cpp/libcudf_streaming/src/bloom_filter.cpp +++ b/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 ctx, + std::shared_ptr 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 ch_in, std::shared_ptr 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(); diff --git a/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu b/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu index 0525f139babf..6cc947d5f5d8 100644 --- a/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu +++ b/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu @@ -18,7 +18,8 @@ #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wsign-conversion" #endif -#include +#include + #include #include #include @@ -33,6 +34,7 @@ #include +#include #include #include @@ -53,47 +55,50 @@ namespace cudf_streaming::detail { namespace { using KeyType = std::uint64_t; -using BloomFilterRefType = - cuco::bloom_filter_ref, - cuco::thread_scope_device, - cuco::parametric_filter_policy, - std::uint32_t, - 8, - 8, - 8, - 1, - 1, - 8, - false, - false>>; -using StorageType = BloomFilterRefType::filter_block_type; +using BloomFilterPolicy = cudf::arrow_filter_policy>; +using BloomFilterRefType = cuco::bloom_filter_ref, + cuco::thread_scope_device, + BloomFilterPolicy>; +using StorageType = BloomFilterRefType::filter_block_type; + +std::size_t num_blocks(std::size_t filter_size) +{ + RAPIDSMPF_EXPECTS(filter_size >= sizeof(StorageType), + "Bloom filter storage must contain at least one filter block"); + RAPIDSMPF_EXPECTS(filter_size == device_bloom_filter::aligned_size(filter_size), + "Bloom filter storage size must be a multiple of the filter block size"); + auto const blocks = filter_size / sizeof(StorageType); + RAPIDSMPF_EXPECTS(blocks <= BloomFilterPolicy::max_filter_blocks, + "Bloom filter storage exceeds the maximum size supported by its policy"); + return blocks; +} } // namespace -device_bloom_filter::device_bloom_filter(std::size_t num_blocks, std::uint64_t seed, void* storage) - : num_blocks_{num_blocks}, seed_{seed}, storage_{storage} +device_bloom_filter::device_bloom_filter(std::size_t filter_size, std::uint64_t seed, void* storage) + : num_blocks_{num_blocks(filter_size)}, seed_{seed}, storage_{storage} { RAPIDSMPF_EXPECTS( reinterpret_cast(storage_) % std::alignment_of_v == 0, "Allocation for bloom filter is not aligned."); } -device_bloom_filter const device_bloom_filter::view(std::size_t num_blocks, +device_bloom_filter const device_bloom_filter::view(std::size_t filter_size, std::uint64_t seed, void const* storage) { // const-cast is safe because the returned object is also const and therefore can't // call methods that throw away constness. - return device_bloom_filter(num_blocks, seed, const_cast(storage)); + return device_bloom_filter(filter_size, seed, const_cast(storage)); } -std::unique_ptr device_bloom_filter::storage(std::size_t num_blocks, +std::unique_ptr device_bloom_filter::storage(std::size_t filter_size, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { return std::make_unique( - num_blocks * sizeof(StorageType), std::alignment_of_v, stream, mr); + num_blocks(filter_size) * sizeof(StorageType), std::alignment_of_v, stream, mr); } void device_bloom_filter::add(cudf::table_view const& values_to_hash, @@ -135,11 +140,6 @@ rmm::device_uvector device_bloom_filter::contains(cudf::table_view const& return result; } -std::size_t device_bloom_filter::fitting_num_blocks(std::size_t l2size) noexcept -{ - return (l2size * 2) / (3 * sizeof(StorageType)); -} - void* device_bloom_filter::data() noexcept { return storage_; } void const* device_bloom_filter::data() const noexcept { return storage_; } diff --git a/cpp/libcudf_streaming/tests/CMakeLists.txt b/cpp/libcudf_streaming/tests/CMakeLists.txt index 2fef944da8a3..1aae79f1f4bd 100644 --- a/cpp/libcudf_streaming/tests/CMakeLists.txt +++ b/cpp/libcudf_streaming/tests/CMakeLists.txt @@ -59,7 +59,7 @@ target_compile_options( ) target_link_libraries( libcudf_streaming_test_sources - PRIVATE cudf_streaming rapidsmpf::rapidsmpf cudf::cudftestutil cudf::cudftestutil_impl + PRIVATE cudf_streaming rapidsmpf::rapidsmpf cudf::cudftestutil cudf::cudftestutil_impl cuco::cuco PUBLIC GTest::gmock GTest::gtest ) # cudf::cudftestutil_impl injects cudf test-utility .cu sources (via INTERFACE_SOURCES) that are @@ -70,7 +70,9 @@ target_compile_options( ) target_sources( libcudf_streaming_test_sources - PRIVATE streaming/test_table_chunk.cpp + PRIVATE streaming/test_bloom_filter.cu + streaming/test_bloom_filter_config.cpp + streaming/test_table_chunk.cpp streaming/test_read_parquet.cpp streaming/test_channel_metadata.cpp streaming/test_partition.cpp diff --git a/cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu b/cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu new file mode 100644 index 000000000000..b891a81e7ec1 --- /dev/null +++ b/cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu @@ -0,0 +1,86 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include + +#include +#include +#include + +#include + +#include + +#include +#include +#include +#include + +#include +#include +#include +#include + +namespace { + +using policy_type = cudf::arrow_filter_policy>; + +__global__ void block_index_kernel(std::uint32_t upper_hash, + std::size_t num_blocks, + std::uint32_t* result) +{ + *result = policy_type{}.block_index(upper_hash, cuco::extent{num_blocks}); +} + +TEST(BloomFilterPolicyTest, UsesBlocksBeyondFormerArrowLimit) +{ + constexpr auto arrow_max_blocks = std::size_t{4'194'304}; + constexpr auto num_blocks = arrow_max_blocks + 1; + constexpr auto upper_hash = std::numeric_limits::max(); + auto const stream = cudf::get_default_stream(); + rmm::device_scalar index{0, stream}; + + block_index_kernel<<<1, 1, 0, stream.value()>>>(upper_hash, num_blocks, index.data()); + CUDF_CHECK_CUDA(stream.value()); + + EXPECT_EQ(index.value(stream), arrow_max_blocks); +} + +TEST(DeviceBloomFilterTest, RejectsStorageBeyondPolicyLimit) +{ + using filter_ref_type = cuco::bloom_filter_ref, + cuco::thread_scope_device, + policy_type>; + constexpr auto block_size = sizeof(filter_ref_type::filter_block_type); + constexpr auto too_large = (policy_type::max_filter_blocks + std::size_t{1}) * block_size; + auto const stream = cudf::get_default_stream(); + + EXPECT_THROW(cudf_streaming::detail::device_bloom_filter::storage( + too_large, stream, cudf::get_current_device_resource_ref()), + std::logic_error); +} + +TEST(DeviceBloomFilterTest, RequiresAlignedStorageSize) +{ + constexpr auto unaligned_size = std::size_t{65}; + constexpr auto aligned_size = std::size_t{64}; + auto const stream = cudf::get_default_stream(); + + EXPECT_THROW(cudf_streaming::detail::device_bloom_filter::storage( + unaligned_size, stream, cudf::get_current_device_resource_ref()), + std::logic_error); + + auto storage = cudf_streaming::detail::device_bloom_filter::storage( + aligned_size, stream, cudf::get_current_device_resource_ref()); + + EXPECT_EQ(storage->size(), aligned_size); + + auto const filter = + cudf_streaming::detail::device_bloom_filter{aligned_size, 0, storage->data(), stream}; + EXPECT_EQ(filter.size(), aligned_size); +} + +} // namespace diff --git a/cpp/libcudf_streaming/tests/streaming/test_bloom_filter_config.cpp b/cpp/libcudf_streaming/tests/streaming/test_bloom_filter_config.cpp new file mode 100644 index 000000000000..e50807fbc222 --- /dev/null +++ b/cpp/libcudf_streaming/tests/streaming/test_bloom_filter_config.cpp @@ -0,0 +1,41 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include + +#include + +#include +#include +#include +#include + +namespace { + +TEST(BloomFilterTest, AlignsStorageSize) +{ + EXPECT_EQ(cudf_streaming::bloom_filter::aligned_size(31), 0); + EXPECT_EQ(cudf_streaming::bloom_filter::aligned_size(32), 32); + EXPECT_EQ(cudf_streaming::bloom_filter::aligned_size(65), 64); +} + +TEST(BloomFilterTest, RequiresAlignedStorageSize) +{ + auto make_filter = [](std::size_t filter_size) { + return cudf_streaming::bloom_filter{std::shared_ptr{}, + std::shared_ptr{}, + 0, + filter_size}; + }; + + EXPECT_THROW(make_filter(0), std::logic_error); + EXPECT_THROW(make_filter(65), std::logic_error); + EXPECT_THROW(make_filter(cudf_streaming::bloom_filter::aligned_size( + std::numeric_limits::max())), + std::logic_error); + EXPECT_NO_THROW(make_filter(64)); +} + +} // namespace diff --git a/cpp/src/io/parquet/bloom_filter_reader.cu b/cpp/src/io/parquet/bloom_filter_reader.cu index e0591f2d2115..e303ba0165f9 100644 --- a/cpp/src/io/parquet/bloom_filter_reader.cu +++ b/cpp/src/io/parquet/bloom_filter_reader.cu @@ -16,6 +16,7 @@ #include #include #include +#include #include #include #include @@ -24,7 +25,6 @@ #include #include -#include #include #include #include @@ -49,16 +49,7 @@ namespace { * @tparam Key The type of the values to generate a fingerprint for. */ template -using arrow_filter_policy = cuco::parametric_filter_policy, - std::uint32_t, - 8, - 8, - 8, - 1, - 1, - 8, - false, - false>; +using arrow_filter_policy = cudf::arrow_filter_policy>; /** * @brief Converts bloom filter membership results (for each column chunk) to a device column. diff --git a/cpp/tests/io/parquet_bloom_filter_test.cu b/cpp/tests/io/parquet_bloom_filter_test.cu index 4de68ff3c991..7fbb21714eff 100644 --- a/cpp/tests/io/parquet_bloom_filter_test.cu +++ b/cpp/tests/io/parquet_bloom_filter_test.cu @@ -11,12 +11,12 @@ #include #include #include +#include #include #include #include -#include #include @@ -26,19 +26,8 @@ class ParquetBloomFilterTest : public cudf::test::BaseFixture {}; TEST_F(ParquetBloomFilterTest, TestStrings) { - using key_type = StringType; - // Apache Arrow Block-Split Bloom Filter layout, hashing keys with cudf's `XXHash_64` (matching - // `cudf::io::parquet::detail::arrow_filter_policy`). - using policy_type = cuco::parametric_filter_policy, - std::uint32_t, - 8, - 8, - 8, - 1, - 1, - 8, - false, - false>; + using key_type = StringType; + using policy_type = cudf::arrow_filter_policy>; using word_type = policy_type::word_type; std::size_t constexpr num_filter_blocks = 4; diff --git a/python/cudf_polars/cudf_polars/streaming/actor_graph/join.py b/python/cudf_polars/cudf_polars/streaming/actor_graph/join.py index 46cb66a04f92..a984e69957d1 100644 --- a/python/cudf_polars/cudf_polars/streaming/actor_graph/join.py +++ b/python/cudf_polars/cudf_polars/streaming/actor_graph/join.py @@ -806,9 +806,9 @@ def make_filter_tasks( ch_left = context.create_channel() bloom_apply_output = ch_left - # TODO: configure based on GPU L2 size - nblocks = BloomFilter.fitting_num_blocks(32 * 1024 * 1024) - filter = BloomFilter(context, comm, LIBCUDF_DEFAULT_HASH_SEED, nblocks) + # TODO: Make the filter size configurable. + filter_size = 32 * 1024 * 1024 + filter = BloomFilter(context, comm, LIBCUDF_DEFAULT_HASH_SEED, filter_size) filter_tasks: list[Coroutine[Any, Any, None]] = [] chs_to_shutdown = [ bloom_build_output, diff --git a/python/cudf_streaming/cudf_streaming/bloom_filter.pxd b/python/cudf_streaming/cudf_streaming/bloom_filter.pxd index 4e1c64db46bd..84345a7ac165 100644 --- a/python/cudf_streaming/cudf_streaming/bloom_filter.pxd +++ b/python/cudf_streaming/cudf_streaming/bloom_filter.pxd @@ -5,6 +5,7 @@ from libc.stddef cimport size_t from libc.stdint cimport uint64_t from libcpp.memory cimport shared_ptr, unique_ptr +from rapidsmpf._detail.exception_handling cimport ex_handler from rapidsmpf.communicator.communicator cimport Communicator, cpp_Communicator from rapidsmpf.streaming.core.context cimport cpp_Context @@ -15,15 +16,15 @@ cdef extern from "" nogil: shared_ptr[cpp_Context] ctx, shared_ptr[cpp_Communicator] comm, uint64_t seed, - size_t num_filter_blocks, - ) noexcept + size_t filter_size, + ) except +ex_handler const shared_ptr[cpp_Communicator]& comm() noexcept cdef extern from "" nogil: - size_t cpp_fitting_num_blocks \ - "cudf_streaming::bloom_filter::fitting_num_blocks"( - size_t l2size + size_t cpp_aligned_size \ + "cudf_streaming::bloom_filter::aligned_size"( + size_t size ) noexcept diff --git a/python/cudf_streaming/cudf_streaming/bloom_filter.pyi b/python/cudf_streaming/cudf_streaming/bloom_filter.pyi index ea879d755771..0c8f6ab0ff4c 100644 --- a/python/cudf_streaming/cudf_streaming/bloom_filter.pyi +++ b/python/cudf_streaming/cudf_streaming/bloom_filter.pyi @@ -26,12 +26,12 @@ class BloomFilter: ctx: Context, comm: Communicator, seed: int, - num_filter_blocks: int, + filter_size: int, ) -> None: ... @property def comm(self) -> Communicator: ... @staticmethod - def fitting_num_blocks(l2size: int) -> int: ... + def aligned_size(size: int) -> int: ... async def build( self, ctx: Context, diff --git a/python/cudf_streaming/cudf_streaming/bloom_filter.pyx b/python/cudf_streaming/cudf_streaming/bloom_filter.pyx index 5116a8aa44e2..23b4c6509423 100644 --- a/python/cudf_streaming/cudf_streaming/bloom_filter.pyx +++ b/python/cudf_streaming/cudf_streaming/bloom_filter.pyx @@ -118,8 +118,9 @@ cdef class BloomFilter: The communicator the bloom filter construction is collective over. seed Seed used for hashing values into the bloom filter. - num_filter_blocks - Number of blocks used to size the filter. + filter_size + Filter storage size in bytes. Must be positive and satisfy + ``BloomFilter.aligned_size(filter_size) == filter_size``. """ def __init__( @@ -127,7 +128,7 @@ cdef class BloomFilter: Context ctx not None, Communicator comm not None, uint64_t seed, - size_t num_filter_blocks, + size_t filter_size, ): self._comm = comm with nogil: @@ -135,7 +136,7 @@ cdef class BloomFilter: ctx._handle, comm._handle, seed, - num_filter_blocks, + filter_size, ) def __dealloc__(self): @@ -154,22 +155,22 @@ cdef class BloomFilter: return self._comm @staticmethod - def fitting_num_blocks(size_t l2size): + def aligned_size(size_t size): """ - Return the number of blocks needed to fit within an L2 cache size. + Return the largest valid filter size no greater than a byte count. Parameters ---------- - l2size - Size of the L2 cache in bytes. + size + Byte count to align. Returns ------- - Number of blocks to use in the filter. + Largest valid filter size less than or equal to ``size``. """ cdef size_t ret with nogil: - ret = cpp_fitting_num_blocks(l2size) + ret = cpp_aligned_size(size) return ret async def build( diff --git a/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py b/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py index d00e7926e50e..846fa690c399 100644 --- a/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py +++ b/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py @@ -42,6 +42,17 @@ def make_table( ) +def test_aligned_size() -> None: + assert BloomFilter.aligned_size(31) == 0 + assert BloomFilter.aligned_size(32) == 32 + assert BloomFilter.aligned_size(65) == 64 + + +def test_requires_aligned_size(context: Context, comm: Communicator) -> None: + with pytest.raises(RuntimeError, match="must be a multiple"): + BloomFilter(context, comm, seed=0, filter_size=65) + + @define_actor() async def add_metadata( ctx: Context, ch_in: Channel[TableChunk], ch_out: Channel[TableChunk] @@ -94,13 +105,13 @@ def run_bloom_filter_pipeline( probe_table: TableChunk, *, seed: int = 42, - l2size: int = 1 << 20, + filter_size: int = 1 << 20, ) -> list[Message]: bloom = BloomFilter( context, comm, seed=seed, - num_filter_blocks=BloomFilter.fitting_num_blocks(l2size), + filter_size=filter_size, ) build_msg = Message(0, build_table) From d8b3e844ad08f44423461dcf74da4cbd3c375fd5 Mon Sep 17 00:00:00 2001 From: Lawrence Mitchell Date: Wed, 22 Jul 2026 17:37:36 +0100 Subject: [PATCH 2/5] Fix constructor in tests --- cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu b/cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu index b891a81e7ec1..b8fc5b2e978a 100644 --- a/cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu +++ b/cpp/libcudf_streaming/tests/streaming/test_bloom_filter.cu @@ -78,8 +78,7 @@ TEST(DeviceBloomFilterTest, RequiresAlignedStorageSize) EXPECT_EQ(storage->size(), aligned_size); - auto const filter = - cudf_streaming::detail::device_bloom_filter{aligned_size, 0, storage->data(), stream}; + auto const filter = cudf_streaming::detail::device_bloom_filter{aligned_size, 0, storage->data()}; EXPECT_EQ(filter.size(), aligned_size); } From 160bf30f92823102672c0180c3b2feae21024666 Mon Sep 17 00:00:00 2001 From: Lawrence Mitchell Date: Wed, 22 Jul 2026 18:00:33 +0100 Subject: [PATCH 3/5] Add back in the implementations --- cpp/libcudf_streaming/src/detail/device_bloom_filter.cu | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu b/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu index 6cc947d5f5d8..4377647c8d7e 100644 --- a/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu +++ b/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu @@ -140,6 +140,15 @@ rmm::device_uvector device_bloom_filter::contains(cudf::table_view const& return result; } +std::size_t aligned_size(std::size_t size) noexcept +{ + return rmm::align_down(size, std::alignment_of_v); +} + +std::size_t max_size() noexcept { + return BloomFilterPolicy::max_filter_blocks * sizeof(StorageType) +}; + void* device_bloom_filter::data() noexcept { return storage_; } void const* device_bloom_filter::data() const noexcept { return storage_; } From 04042dd52ea8e8fc04602f36f49dfac33a0fc774 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Wed, 22 Jul 2026 22:15:07 +0000 Subject: [PATCH 4/5] Minor fixes --- cpp/libcudf_streaming/src/detail/device_bloom_filter.cu | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu b/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu index 4377647c8d7e..ad3a2217ba13 100644 --- a/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu +++ b/cpp/libcudf_streaming/src/detail/device_bloom_filter.cu @@ -140,14 +140,15 @@ rmm::device_uvector device_bloom_filter::contains(cudf::table_view const& return result; } -std::size_t aligned_size(std::size_t size) noexcept +std::size_t device_bloom_filter::aligned_size(std::size_t size) noexcept { return rmm::align_down(size, std::alignment_of_v); } -std::size_t max_size() noexcept { - return BloomFilterPolicy::max_filter_blocks * sizeof(StorageType) -}; +std::size_t device_bloom_filter::max_size() noexcept +{ + return BloomFilterPolicy::max_filter_blocks * sizeof(StorageType); +} void* device_bloom_filter::data() noexcept { return storage_; } From 26dd17295a0f68a832dcc4bff9cc9739c567df23 Mon Sep 17 00:00:00 2001 From: Tom Augspurger Date: Wed, 22 Jul 2026 16:38:57 -0700 Subject: [PATCH 5/5] update test usage --- python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py b/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py index 846fa690c399..50089c2f4250 100644 --- a/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py +++ b/python/cudf_streaming/cudf_streaming/tests/test_bloom_filter.py @@ -196,7 +196,7 @@ def test_bloom_filter_build_exception_no_shutdown( context, comm, seed=42, - num_filter_blocks=BloomFilter.fitting_num_blocks(1 << 20), + filter_size=(1 << 20), ) ch_in: Channel[TableChunk] = context.create_channel() ch_out: Channel[BloomFilterChunk] = context.create_channel()