Remove multi-file secondary-filter API and add bloom-filter pruning to hybrid scan

cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp

@@ -17,6 +17,7 @@
 #include <rmm/resource_ref.hpp>
 
 #include <memory>
+#include <utility>
 #include <vector>
 
 namespace cudf::io::parquet::experimental::detail {
@@ -152,20 +153,37 @@ class hybrid_scan_multifile {
     rmm::cuda_stream_view stream) const;
 
   /**
-   * @brief Get byte ranges of bloom filters and dictionary pages (secondary filters) for row group
-   *        pruning
+   * @brief Get byte ranges of bloom filters for row group pruning
    *
    * @note Device buffers for bloom filter byte ranges must be allocated using a 32 byte
    *       aligned memory resource
    *
    * @param row_group_indices Input row group indices, one per source
    * @param options Parquet reader options
-   * @return Pair of vectors of byte ranges of column chunk with bloom filters and dictionary
-   *         pages subject to filter predicate
+   * @return A pair of (1) a flat vector of byte ranges of column chunk bloom filters subject to the
+   *         filter predicate and (2) a parallel source-index map identifying the source each byte
+   *         range must be fetched from
    */
-  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<byte_range_info>>
-  secondary_filters_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
-                                parquet_reader_options const& options) const;
+  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+  bloom_filters_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
+                            parquet_reader_options const& options) const;
+
+  /**
+   * @brief Filter the row groups using column chunk bloom filters
+   *
+   * @note The `bloom_filter_data` device spans must point to 32-byte aligned addresses
+   *
+   * @param bloom_filter_data Device spans of bloom filters, one inner vector per source
+   * @param row_group_indices Input row group indices
+   * @param options Parquet reader options
+   * @param stream CUDA stream used for device memory operations and kernel launches
+   * @return Filtered per-source row group indices (one inner vector per source)
+   */
+  [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_bloom_filters(
+    cudf::host_span<std::vector<cudf::device_span<uint8_t const>> const> bloom_filter_data,
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    parquet_reader_options const& options,
+    rmm::cuda_stream_view stream) const;
 
  private:
   std::unique_ptr<detail::hybrid_scan_reader_impl> _impl;

cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp

@@ -377,7 +377,8 @@ std::vector<std::vector<cudf::size_type>> aggregate_reader_metadata::filter_row_
   return stats_filtered_row_group_indices.value_or(all_row_group_indices(row_group_indices));
 }
 
-std::vector<byte_range_info> aggregate_reader_metadata::get_bloom_filter_bytes(
+std::pair<std::vector<byte_range_info>, std::vector<cudf::size_type>>
+aggregate_reader_metadata::get_bloom_filter_bytes(
   cudf::host_span<std::vector<cudf::size_type> const> row_group_indices,
   host_span<data_type const> output_dtypes,
   host_span<cudf::size_type const> output_column_schemas,
@@ -411,6 +412,10 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_bloom_filter_bytes(
   std::vector<byte_range_info> bloom_filter_bytes;
   bloom_filter_bytes.reserve(num_chunks);
 
+  // Parallel map identifying the source each emitted byte range must be fetched from
+  std::vector<cudf::size_type> chunk_source_map;
+  chunk_source_map.reserve(num_chunks);
+
   // Flag to check if we have at least one valid bloom filter offset
   auto have_bloom_filters = false;
 
@@ -431,6 +436,7 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_bloom_filter_bytes(
                         // Get bloom filter offsets and sizes
                         bloom_filter_bytes.emplace_back(col_meta.bloom_filter_offset.value_or(0),
                                                         col_meta.bloom_filter_length.value_or(0));
+                        chunk_source_map.emplace_back(static_cast<cudf::size_type>(src_index));
 
                         // Set `have_bloom_filters` if `bloom_filter_offset` is valid
                         if (col_meta.bloom_filter_offset.has_value()) { have_bloom_filters = true; }
@@ -440,7 +446,7 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_bloom_filter_bytes(
 
   if (not have_bloom_filters) { return {}; }
 
-  return bloom_filter_bytes;
+  return {std::move(bloom_filter_bytes), std::move(chunk_source_map)};
 }
 
 std::vector<byte_range_info> aggregate_reader_metadata::get_dictionary_page_bytes(

cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp

@@ -222,13 +222,16 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
    * @param output_column_schemas schema indices of output columns
    * @param filter AST expression to filter row groups based on bloom filters
    *
-   * @return Byte ranges of bloom filters, one per column chunk with equality predicate
+   * @return A pair of (1) byte ranges of bloom filters, one per column chunk with equality
+   *         predicate, and (2) a parallel source-index map identifying the source each byte range
+   *         must be fetched from
    */
-  [[nodiscard]] std::vector<cudf::io::text::byte_range_info> get_bloom_filter_bytes(
-    cudf::host_span<std::vector<size_type> const> row_group_indices,
-    cudf::host_span<data_type const> output_dtypes,
-    cudf::host_span<cudf::size_type const> output_column_schemas,
-    std::reference_wrapper<ast::expression const> filter);
+  [[nodiscard]] std::pair<std::vector<cudf::io::text::byte_range_info>,
+                          std::vector<cudf::size_type>>
+  get_bloom_filter_bytes(cudf::host_span<std::vector<size_type> const> row_group_indices,
+                         cudf::host_span<data_type const> output_dtypes,
+                         cudf::host_span<cudf::size_type const> output_column_schemas,
+                         std::reference_wrapper<ast::expression const> filter);
 
   /**
    * @brief Get the dictionary page byte ranges, one per column chunk with (in)equality predicate

cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp

@@ -283,11 +283,14 @@ hybrid_scan_reader_impl::secondary_filters_byte_ranges(
   CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered");
   auto [expr_conv, output_dtypes] = prepare_filter_and_output_types(options);
 
-  auto const bloom_filter_bytes =
-    _extended_metadata->get_bloom_filter_bytes(row_group_indices,
-                                               output_dtypes,
-                                               _output_column_schemas,
-                                               expr_conv.get_converted_expr().value());
+  // Single-file combined view: take only the bloom byte ranges (`.first`). The per-range source
+  // map (`.second`) is unused here because the single-file reader has exactly one source.
+  auto const bloom_filter_bytes = _extended_metadata
+                                    ->get_bloom_filter_bytes(row_group_indices,
+                                                             output_dtypes,
+                                                             _output_column_schemas,
+                                                             expr_conv.get_converted_expr().value())
+                                    .first;
   auto const dictionary_page_bytes =
     _extended_metadata->get_dictionary_page_bytes(row_group_indices,
                                                   output_dtypes,
@@ -297,6 +300,20 @@ hybrid_scan_reader_impl::secondary_filters_byte_ranges(
   return {bloom_filter_bytes, dictionary_page_bytes};
 }
 
+std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+hybrid_scan_reader_impl::bloom_filters_byte_ranges(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  parquet_reader_options const& options)
+{
+  CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered");
+  auto [expr_conv, output_dtypes] = prepare_filter_and_output_types(options);
+
+  return _extended_metadata->get_bloom_filter_bytes(row_group_indices,
+                                                    output_dtypes,
+                                                    _output_column_schemas,
+                                                    expr_conv.get_converted_expr().value());
+}
+
 std::vector<std::vector<size_type>>
 hybrid_scan_reader_impl::filter_row_groups_with_dictionary_pages(
   cudf::host_span<cudf::device_span<uint8_t const> const> dictionary_page_data,

cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp

@@ -112,6 +112,13 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
   secondary_filters_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
                                 parquet_reader_options const& options);
 
+  /**
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::bloom_filters_byte_ranges
+   */
+  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+  bloom_filters_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
+                            parquet_reader_options const& options);
+
   /**
    * @copydoc cudf::io::experimental::hybrid_scan::filter_row_groups_with_dictionary_pages
    */

cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp

@@ -8,6 +8,8 @@
 #include <cudf/detail/nvtx/ranges.hpp>
 #include <cudf/io/experimental/hybrid_scan_multifile.hpp>
 
+#include <numeric>
+
 namespace cudf::io::parquet::experimental {
 
 hybrid_scan_multifile::hybrid_scan_multifile(
@@ -74,13 +76,40 @@ std::vector<std::vector<size_type>> hybrid_scan_multifile::filter_row_groups_wit
   return _impl->filter_row_groups_with_stats(row_group_indices, options, stream);
 }
 
-std::pair<std::vector<text::byte_range_info>, std::vector<text::byte_range_info>>
-hybrid_scan_multifile::secondary_filters_byte_ranges(
+std::pair<std::vector<text::byte_range_info>, std::vector<size_type>>
+hybrid_scan_multifile::bloom_filters_byte_ranges(
   cudf::host_span<std::vector<size_type> const> row_group_indices,
   parquet_reader_options const& options) const
 {
   CUDF_FUNC_RANGE();
-  return _impl->secondary_filters_byte_ranges(row_group_indices, options);
+  return _impl->bloom_filters_byte_ranges(row_group_indices, options);
+}
+
+std::vector<std::vector<size_type>> hybrid_scan_multifile::filter_row_groups_with_bloom_filters(
+  cudf::host_span<std::vector<cudf::device_span<uint8_t const>> const> bloom_filter_data,
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream) const
+{
+  CUDF_FUNC_RANGE();
+
+  // Concatenate the per-source spans in source order
+  auto const num_chunks =
+    std::accumulate(bloom_filter_data.begin(),
+                    bloom_filter_data.end(),
+                    std::size_t{0},
+                    [](auto sum, auto const& source_data) { return sum + source_data.size(); });
+
+  std::vector<cudf::device_span<uint8_t const>> flattened_bloom_filter_data;
+  flattened_bloom_filter_data.reserve(num_chunks);
+  for (auto const& source_bloom_filter_data : bloom_filter_data) {
+    flattened_bloom_filter_data.insert(flattened_bloom_filter_data.end(),
+                                       source_bloom_filter_data.begin(),
+                                       source_bloom_filter_data.end());
+  }
+
+  return _impl->filter_row_groups_with_bloom_filters(
+    flattened_bloom_filter_data, row_group_indices, options, stream);
 }
 
 }  // namespace cudf::io::parquet::experimental