Fix hybrid scan parquet reader incorrectly parse bloom filter

cpp/benchmarks/io/parquet/experimental/hybrid_scan/hybrid_scan_composer.cpp

@@ -1,6 +1,6 @@
 
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -12,8 +12,6 @@
 #include <cudf/io/text/byte_range_info.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 
-#include <rmm/mr/aligned_resource_adaptor.hpp>
-
 #include <unordered_set>
 #include <vector>
 
@@ -104,12 +102,10 @@ std::vector<cudf::size_type> apply_row_group_filters(
 
   if (filters.contains(hybrid_scan_filter_type::ROW_GROUPS_WITH_BLOOM_FILTERS) and
       bloom_filter_byte_ranges.size()) {
-    // Fetch 32-byte aligned bloom filter data buffers from the input file buffer
-    auto constexpr bloom_filter_alignment = rmm::CUDA_ALLOCATION_ALIGNMENT;
-    auto aligned_mr = rmm::mr::aligned_resource_adaptor(mr, bloom_filter_alignment);
+    // Fetch the header-stripped, 32-byte-aligned bloom filter bitsets from the input file
     auto [bloom_filter_buffers, bloom_filter_data, bloom_read_tasks] =
-      cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        datasource, bloom_filter_byte_ranges, stream, aligned_mr);
+      cudf::io::parquet::fetch_bloom_filters_to_device_async(
+        datasource, bloom_filter_byte_ranges, stream, mr);
     bloom_read_tasks.get();
 
     bloom_filtered_row_groups = reader.filter_row_groups_with_bloom_filters(

cpp/examples/hybrid_scan_io/hybrid_scan_composer.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -16,8 +16,6 @@
 #include <cudf/io/types.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 
-#include <rmm/mr/aligned_resource_adaptor.hpp>
-
 #include <nvtx3/nvtx3.hpp>
 
 #include <unordered_set>
@@ -175,14 +173,13 @@ std::vector<cudf::size_type> apply_row_group_filters(
   bloom_filtered_row_group_indices.reserve(current_row_group_indices.size());
   if (filters.contains(hybrid_scan_filter_type::ROW_GROUPS_WITH_BLOOM_FILTERS) and
       bloom_filter_byte_ranges.size()) {
-    // Fetch 32-byte aligned bloom filter data buffers from the input file buffer
-    auto constexpr bloom_filter_alignment = rmm::CUDA_ALLOCATION_ALIGNMENT;
-    auto aligned_mr = rmm::mr::aligned_resource_adaptor(temp_mr, bloom_filter_alignment);
     if (verbose) { std::cout << "READER: Filter row groups with bloom filters...\n"; }
     timer.reset();
     nvtxRangePush("fetch_bloom_filter_byte_ranges");
+    // Fetch the header-stripped, 32-byte-aligned bloom filter bitsets from the input file
     auto [bloom_filter_buffers, bloom_filter_data, bloom_read_tasks] =
-      fetch_byte_ranges_async(datasource, bloom_filter_byte_ranges, stream, aligned_mr);
+      cudf::io::parquet::fetch_bloom_filters_to_device_async(
+        datasource, bloom_filter_byte_ranges, stream, temp_mr);
     bloom_read_tasks.get();
     nvtxRangePop();
 

cpp/include/cudf/io/parquet_io_utils.hpp

@@ -151,6 +151,28 @@ fetch_byte_ranges_to_device_async(
   rmm::cuda_stream_view stream,
   rmm::device_async_resource_ref mr);
 
+/**
+ * @brief Fetches Parquet bloom filter bitsets from a datasource into device buffers
+ *
+ * @ingroup io_utils
+ *
+ * @param datasource Input datasource
+ * @param bloom_filter_byte_ranges Byte ranges of complete bloom filters to fetch, must span a
+ * complete bloom filter
+ * @param stream CUDA stream
+ * @param mr Device memory resource
+ *
+ * @return A tuple containing the device buffers, the device spans of the bitset data (one per input
+ * range; empty for ranges without a bloom filter), and a future to wait on the read tasks
+ */
+std::tuple<std::vector<rmm::device_buffer>,
+           std::vector<cudf::device_span<uint8_t const>>,
+           std::future<void>>
+fetch_bloom_filters_to_device_async(cudf::io::datasource& datasource,
+                                    cudf::host_span<byte_range_info const> bloom_filter_byte_ranges,
+                                    rmm::cuda_stream_view stream,
+                                    rmm::device_async_resource_ref mr);
+
 /** @} */  // end of group
 }  // namespace io::parquet
 }  // namespace CUDF_EXPORT cudf

cpp/src/io/parquet/bloom_filter_reader.cu

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -32,6 +32,7 @@
 #include <future>
 #include <numeric>
 #include <optional>
+#include <utility>
 
 namespace cudf::io::parquet::detail {
 namespace {
@@ -304,7 +305,6 @@ void read_bloom_filter_data(host_span<std::unique_ptr<datasource> const> sources
                                    cuco::extent<std::size_t>,
                                    cuco::thread_scope_thread,
                                    policy_type>::filter_block_type);
-  auto constexpr words_per_block = policy_type::words_per_block;
 
   // Read tasks for bloom filter data
   std::vector<std::future<std::size_t>> read_tasks;
@@ -322,39 +322,28 @@ void read_bloom_filter_data(host_span<std::unique_ptr<datasource> const> sources
       // Read bloom filter iff present
       auto const bloom_filter_offset = bloom_filter_offsets[chunk].value();
 
-      // If Bloom filter size (header + bitset) is available, just read the entire thing.
-      // Else just read 256 bytes which will contain the entire header and may contain the
-      // entire bitset as well.
-      auto constexpr bloom_filter_size_guess = 256;
+      // If the bloom filter size (header + bitset) is available, read the entire thing. Else read
+      // the max header size, which contains the entire header and may contain the entire bitset.
       auto const initial_read_size =
-        static_cast<std::size_t>(bloom_filter_sizes[chunk].value_or(bloom_filter_size_guess));
+        static_cast<std::size_t>(bloom_filter_sizes[chunk].value_or(bloom_filter_header_max_size));
 
       // Read an initial buffer from source
       auto& source = sources[chunk_source_map[chunk]];
       auto buffer  = source->host_read(bloom_filter_offset, initial_read_size);
 
-      // Deserialize the Bloom filter header from the buffer.
-      BloomFilterHeader header;
-      CompactProtocolReader cp{buffer->data(), buffer->size()};
-      cp.read(&header);
-
-      // Check if the bloom filter header is valid.
-      auto const is_header_valid =
-        (header.num_bytes % words_per_block) == 0 and
-        header.compression.compression == BloomFilterCompression::UNCOMPRESSED and
-        header.algorithm.algorithm == BloomFilterAlgorithm::SPLIT_BLOCK and
-        header.hash.hash == BloomFilterHash::XXHASH;
+      // Deserialize and validate the bloom filter header from the buffer.
+      auto const header_info = parse_bloom_filter_header({buffer->data(), buffer->size()});
 
       // Do not read if the bloom filter is invalid
-      if (not is_header_valid) {
+      if (not header_info.has_value()) {
         bloom_filter_data[chunk] = {};
         CUDF_LOG_WARN("Encountered an invalid bloom filter header. Skipping");
         return;
       }
 
-      // Bloom filter header size
-      auto const bloom_filter_header_size = static_cast<int64_t>(cp.bytecount());
-      auto const bitset_size              = static_cast<std::size_t>(header.num_bytes);
+      // Bloom filter header and bitset sizes
+      auto const bloom_filter_header_size = header_info->first;
+      auto const bitset_size              = header_info->second;
 
       // Check if we already read in the filter bitset in the initial read.
       if (initial_read_size >= bloom_filter_header_size + bitset_size) {
@@ -405,6 +394,29 @@ void read_bloom_filter_data(host_span<std::unique_ptr<datasource> const> sources
 
 }  // namespace
 
+std::optional<std::pair<int64_t, std::size_t>> parse_bloom_filter_header(
+  host_span<uint8_t const> bytes)
+{
+  using policy_type              = arrow_filter_policy<cuda::std::byte>;
+  auto constexpr words_per_block = policy_type::words_per_block;
+
+  // Deserialize the bloom filter header from the front of the buffer
+  BloomFilterHeader header;
+  CompactProtocolReader cp{bytes.data(), bytes.size()};
+  cp.read(&header);
+
+  // Check if the bloom filter header is valid
+  auto const is_header_valid =
+    (header.num_bytes % words_per_block) == 0 and
+    header.compression.compression == BloomFilterCompression::UNCOMPRESSED and
+    header.algorithm.algorithm == BloomFilterAlgorithm::SPLIT_BLOCK and
+    header.hash.hash == BloomFilterHash::XXHASH;
+  if (not is_header_valid) { return std::nullopt; }
+
+  return std::pair{static_cast<int64_t>(cp.bytecount()),
+                   static_cast<std::size_t>(header.num_bytes)};
+}
+
 std::size_t aggregate_reader_metadata::get_bloom_filter_alignment() const
 {
   // Required alignment:

cpp/src/io/parquet/io_utils/parquet_io_utils.cpp

@@ -5,6 +5,7 @@
 
 #include "io/comp/common.hpp"
 #include "io/parquet/parquet_common.hpp"
+#include "io/parquet/reader_impl_helpers.hpp"
 
 #include <cudf/detail/nvtx/ranges.hpp>
 #include <cudf/detail/utilities/cuda_memcpy.hpp>
@@ -25,15 +26,19 @@
 
 #include <algorithm>
 #include <array>
+#include <cstdint>
 #include <cstring>
 #include <format>
 #include <functional>
+#include <iterator>
 #include <mutex>
 #include <numeric>
+#include <optional>
 #include <stdexcept>
 #include <string>
 #include <tuple>
 #include <type_traits>
+#include <utility>
 #include <vector>
 
 /**
@@ -375,6 +380,69 @@ fetch_byte_ranges_to_device_async_impl(
           std::async(std::launch::deferred, sync_function, std::move(device_read_tasks))};
 }
 
+std::tuple<std::vector<rmm::device_buffer>,
+           std::vector<device_spans_per_source_type>,
+           std::future<void>>
+fetch_bloom_filters_to_device_async_impl(
+  cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources,
+  cudf::host_span<cudf::host_span<cudf::io::text::byte_range_info const> const>
+    bloom_filter_byte_ranges_per_source,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  auto const num_sources = datasources.size();
+  CUDF_EXPECTS(num_sources == bloom_filter_byte_ranges_per_source.size(),
+               "Encountered mismatch in number of datasources and bloom filter byte range spans");
+
+  // Read + parse a single bloom filter header to host and return its bitset-only byte range
+  auto const fetch_bitset_range =
+    [](cudf::io::datasource& datasource,
+       cudf::io::text::byte_range_info const& bloom_range) -> cudf::io::text::byte_range_info {
+    // placeholder for a chunk whose column has no bloom filter written
+    if (bloom_range.is_empty()) { return {0, 0}; }
+    auto const header_read_size =
+      std::min<int64_t>(bloom_range.size(), detail::bloom_filter_header_max_size);
+    auto const header      = datasource.host_read(static_cast<std::size_t>(bloom_range.offset()),
+                                             static_cast<std::size_t>(header_read_size));
+    auto const header_info = detail::parse_bloom_filter_header({header->data(), header->size()});
+    CUDF_EXPECTS(header_info.has_value(), "Encountered an invalid bloom filter header");
+    auto const [header_size, bitset_size] = header_info.value();
+    return {bloom_range.offset() + header_size, static_cast<int64_t>(bitset_size)};
+  };
+
+  // Parse each source's bloom filter headers with one task per source
+  auto const bitset_byte_ranges_per_source =
+    dispatch_fetch_tasks(num_sources, [&](std::size_t source_idx) {
+      auto const& bloom_ranges = bloom_filter_byte_ranges_per_source[source_idx];
+      auto& datasource         = datasources[source_idx].get();
+      std::vector<cudf::io::text::byte_range_info> bitset_ranges;
+      bitset_ranges.reserve(bloom_ranges.size());
+      std::transform(
+        bloom_ranges.begin(),
+        bloom_ranges.end(),
+        std::back_inserter(bitset_ranges),
+        [&](auto const& bloom_range) { return fetch_bitset_range(datasource, bloom_range); });
+      return bitset_ranges;
+    });
+
+  // Fetch only the header-free bitsets to device
+  auto result =
+    fetch_byte_ranges_to_device_async(datasources, bitset_byte_ranges_per_source, stream, mr);
+
+  auto const& bitset_spans_per_source = std::get<1>(result);
+  CUDF_EXPECTS(std::all_of(bitset_spans_per_source.begin(),
+                           bitset_spans_per_source.end(),
+                           [](auto const& spans) {
+                             return std::all_of(spans.begin(), spans.end(), [](auto const& span) {
+                               return span.empty() or
+                                      (reinterpret_cast<std::uintptr_t>(span.data()) % 32) == 0;
+                             });
+                           }),
+               "Bloom filter bitset is not 32-byte aligned");
+
+  return result;
+}
+
 }  // namespace
 
 [[nodiscard]] std::size_t metadata_size_hint()
@@ -471,4 +539,29 @@ fetch_byte_ranges_to_device_async(
     mr);
 }
 
+std::tuple<std::vector<rmm::device_buffer>,
+           std::vector<cudf::device_span<uint8_t const>>,
+           std::future<void>>
+fetch_bloom_filters_to_device_async(
+  cudf::io::datasource& datasource,
+  cudf::host_span<cudf::io::text::byte_range_info const> bloom_filter_byte_ranges,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  CUDF_FUNC_RANGE();
+
+  // Wrap the inputs into arrays and delegate to the multi-source implementation
+  std::array<std::reference_wrapper<cudf::io::datasource>, 1> datasources{std::ref(datasource)};
+  std::array<cudf::host_span<cudf::io::text::byte_range_info const>, 1>
+    bloom_filter_byte_ranges_per_source{bloom_filter_byte_ranges};
+
+  auto [buffers, fetched_byte_ranges, fut] = fetch_bloom_filters_to_device_async_impl(
+    {datasources.data(), datasources.size()},
+    {bloom_filter_byte_ranges_per_source.data(), bloom_filter_byte_ranges_per_source.size()},
+    stream,
+    mr);
+
+  return {std::move(buffers), std::move(fetched_byte_ranges.front()), std::move(fut)};
+}
+
 }  // namespace cudf::io::parquet

cpp/src/io/parquet/reader_impl_helpers.hpp

@@ -15,10 +15,12 @@
 
 #include <cstddef>
 #include <functional>
+#include <optional>
 #include <string>
 #include <string_view>
 #include <tuple>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
 namespace cudf::io::parquet::detail {
@@ -142,6 +144,22 @@ struct surviving_row_group_metrics {
   std::optional<size_type> after_bloom_filter;  // number of surviving row groups after bloom filter
 };
 
+/**
+ * @brief Upper bound on the size in bytes of a Parquet `BloomFilterHeader`
+ */
+inline constexpr int64_t bloom_filter_header_max_size = 256;
+
+/**
+ * @brief Parses and validates a Parquet `BloomFilterHeader` from the front of `bytes`
+ *
+ * @param bytes Host bytes starting at the beginning of a bloom filter (header followed by bitset)
+ *
+ * @return A pair of the bloom filter header size and the bitset size in bytes, or `std::nullopt`
+ * if the header is missing or unsupported
+ */
+[[nodiscard]] std::optional<std::pair<int64_t, std::size_t>> parse_bloom_filter_header(
+  host_span<uint8_t const> bytes);
+
 class aggregate_reader_metadata {
  protected:
   std::vector<metadata> per_file_metadata;

cpp/tests/io/experimental/hybrid_scan_composer.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -104,13 +104,10 @@ auto apply_hybrid_scan_filters(cudf::io::datasource& datasource,
   std::vector<cudf::size_type> bloom_filtered_row_group_indices;
   bloom_filtered_row_group_indices.reserve(current_row_group_indices.size());
   if (bloom_filter_byte_ranges.size()) {
-    // Fetch 32 byte aligned bloom filter data buffers from the input file buffer
-    auto aligned_mr = rmm::mr::aligned_resource_adaptor(cudf::get_current_device_resource_ref(),
-                                                        bloom_filter_alignment);
-
+    // Fetch the header-stripped, 32-byte-aligned bloom filter bitsets from the input file
     auto [bloom_filter_buffers, bloom_filter_data, bloom_read_tasks] =
-      cudf::io::parquet::fetch_byte_ranges_to_device_async(
-        datasource, bloom_filter_byte_ranges, stream, aligned_mr);
+      cudf::io::parquet::fetch_bloom_filters_to_device_async(
+        datasource, bloom_filter_byte_ranges, stream, mr);
     bloom_read_tasks.get();
 
     // Filter row groups with bloom filters