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

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

@@ -18,6 +18,8 @@
 #include <cuda/stream>
 
 #include <memory>
+#include <span>
+#include <utility>
 #include <vector>
 
 /**
@@ -157,20 +159,64 @@ class hybrid_scan_multifile {
     cuda::stream_ref 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 Span of vectors of 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 vectors containing bloom filter byte ranges and corresponding source indices
    */
-  [[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(std::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
+   *
+   * @param bloom_filter_data Device spans of header-stripped bloom filter bitsets of column
+   *                           chunks with an equality predicate, ordered to match the bloom filter
+   *                           byte ranges returned by `bloom_filters_byte_ranges`
+   * @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 Vectors of filtered per-source row group indices, one per source
+   */
+  [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_bloom_filters(
+    std::span<cudf::device_span<uint8_t const> const> bloom_filter_data,
+    std::span<std::vector<size_type> const> row_group_indices,
+    parquet_reader_options const& options,
+    cuda::stream_ref stream) const;
+
+  /**
+   * @brief Get byte ranges of column chunk dictionary pages for row group pruning
+   *
+   * @param row_group_indices Span of vectors of input row group indices, one per source
+   * @param options Parquet reader options
+   * @return Pair of flattened byte ranges to column chunk dictionary pages subject to the filter
+   *         predicate and their corresponding source indices
+   */
+  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+  dictionary_pages_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 dictionary pages
+   *
+   * @param dictionary_page_data Device spans of dictionary page data of column chunks with an
+   * (in)equality predicate, in the same order as the byte ranges returned by
+   * `dictionary_pages_byte_ranges` including empty spans against empty byte ranges
+   * @param row_group_indices Span of vectors of input row group indices, one per source
+   * @param options Parquet reader options
+   * @param stream CUDA stream used for device memory operations and kernel launches
+   * @return Vector of vectors of filtered row group indices, one per source
+   */
+  [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_dictionary_pages(
+    cudf::host_span<cudf::device_span<uint8_t const> const> dictionary_page_data,
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    parquet_reader_options const& options,
+    cuda::stream_ref stream) const;
 
   /**
    * @brief Builds a boolean survival column of size equal to the total number of rows in the row
@@ -455,35 +501,6 @@ class hybrid_scan_multifile {
    */
   [[nodiscard]] bool has_next_table_chunk() const;
 
-  /**
-   * @brief Get byte ranges of column chunk dictionary pages for row group pruning
-   *
-   * @param row_group_indices Span of vectors of input row group indices, one per source
-   * @param options Parquet reader options
-   * @return Pair of flattened byte ranges to column chunk dictionary pages subject to the filter
-   *         predicate and their corresponding source indices
-   */
-  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
-  dictionary_pages_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 dictionary pages
-   *
-   * @param dictionary_page_data Device spans of dictionary page data of column chunks with an
-   * (in)equality predicate, in the same order as the byte ranges returned by
-   * `dictionary_pages_byte_ranges` including empty spans against empty byte ranges
-   * @param row_group_indices Span of vectors of input row group indices, one per source
-   * @param options Parquet reader options
-   * @param stream CUDA stream used for device memory operations and kernel launches
-   * @return Vector of vectors of filtered row group indices, one per source
-   */
-  [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_dictionary_pages(
-    cudf::host_span<cudf::device_span<uint8_t const> const> dictionary_page_data,
-    cudf::host_span<std::vector<size_type> const> row_group_indices,
-    parquet_reader_options const& options,
-    cuda::stream_ref stream) const;
-
  private:
   std::unique_ptr<detail::hybrid_scan_reader_impl> _impl;
 };

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

@@ -439,7 +439,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::bloom_filters_byte_ranges(
   std::span<std::vector<cudf::size_type> const> row_group_indices,
   std::span<data_type const> output_dtypes,
   std::span<cudf::size_type const> output_column_schemas,
@@ -463,7 +464,7 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_bloom_filter_bytes(
                   std::back_inserter(bloom_filter_col_schemas),
                   [](auto& bloom_filter_literals) { return not bloom_filter_literals.empty(); });
 
-  // No equality literals found, return empty vector
+  // No equality literals found, return empty pair
   if (bloom_filter_col_schemas.empty()) { return {}; }
 
   // Compute total number of input row groups
@@ -476,36 +477,42 @@ 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> bloom_filter_source_map;
+  bloom_filter_source_map.reserve(num_chunks);
+
   // Flag to check if we have at least one valid bloom filter offset
   auto have_bloom_filters = false;
 
   // For all sources
-  std::for_each(cuda::counting_iterator<std::size_t>{0},
-                cuda::counting_iterator{row_group_indices.size()},
-                [&](auto const src_index) {
-                  // Get all row group indices in the data source
-                  auto const& rg_indices = row_group_indices[src_index];
-                  // For all row groups
-                  std::for_each(rg_indices.cbegin(), rg_indices.cend(), [&](auto const rg_index) {
-                    // For all column chunks
-                    std::for_each(
-                      bloom_filter_col_schemas.begin(),
-                      bloom_filter_col_schemas.end(),
-                      [&](auto const schema_idx) {
-                        auto& col_meta = get_column_metadata(rg_index, src_index, schema_idx);
-                        // 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));
-
-                        // Set `have_bloom_filters` if `bloom_filter_offset` is valid
-                        if (col_meta.bloom_filter_offset.has_value()) { have_bloom_filters = true; }
-                      });
-                  });
-                });
+  std::for_each(
+    cuda::counting_iterator<std::size_t>{0},
+    cuda::counting_iterator{row_group_indices.size()},
+    [&](auto const src_index) {
+      // Get all row group indices in the data source
+      auto const& rg_indices = row_group_indices[src_index];
+      // For all row groups
+      std::for_each(rg_indices.cbegin(), rg_indices.cend(), [&](auto const rg_index) {
+        // For all column chunks
+        std::for_each(
+          bloom_filter_col_schemas.begin(),
+          bloom_filter_col_schemas.end(),
+          [&](auto const schema_idx) {
+            auto& col_meta = get_column_metadata(rg_index, src_index, schema_idx);
+            // 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));
+            bloom_filter_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; }
+          });
+      });
+    });
 
   if (not have_bloom_filters) { return {}; }
 
-  return bloom_filter_bytes;
+  return {std::move(bloom_filter_bytes), std::move(bloom_filter_source_map)};
 }
 
 std::pair<std::vector<byte_range_info>, std::vector<cudf::size_type>>
@@ -699,6 +706,13 @@ aggregate_reader_metadata::filter_row_groups_with_bloom_filters(
   // Compute total number of input row groups
   auto const total_row_groups = compute_total_row_groups(row_group_indices);
 
+  // Ensure there is one bloom filter data span per eligible column in each row group
+  CUDF_EXPECTS(bloom_filter_data.size() ==
+                 static_cast<std::size_t>(total_row_groups) * bloom_filter_col_schemas.size(),
+               "Bloom filter data size must match the number of row groups times the number of "
+               "columns with bloom filters and an equality predicate",
+               std::invalid_argument);
+
   // Transform bloom filter data to cuda::std::byte type for apply_bloom_filters
   std::vector<cudf::device_span<cuda::std::byte const>> transformed_bloom_filter_data;
   transformed_bloom_filter_data.reserve(bloom_filter_data.size());

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

@@ -226,13 +226,14 @@ 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 vectors containing bloom filter byte ranges and corresponding source indices
    */
-  [[nodiscard]] std::vector<cudf::io::text::byte_range_info> get_bloom_filter_bytes(
-    std::span<std::vector<size_type> const> row_group_indices,
-    std::span<data_type const> output_dtypes,
-    std::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>>
+  bloom_filters_byte_ranges(std::span<std::vector<size_type> const> row_group_indices,
+                            std::span<data_type const> output_dtypes,
+                            std::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

@@ -284,11 +284,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);
 
+  // Single source: keep only the bloom filter byte ranges, not the source map
   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());
+    _extended_metadata
+      ->bloom_filters_byte_ranges(row_group_indices,
+                                  output_dtypes,
+                                  _output_column_schemas,
+                                  expr_conv.get_converted_expr().value())
+      .first;
   auto const dictionary_page_bytes =
     _extended_metadata
       ->dictionary_pages_byte_ranges(row_group_indices,
@@ -314,6 +317,19 @@ hybrid_scan_reader_impl::dictionary_pages_byte_ranges(
                                                           expr_conv.get_converted_expr().value());
 }
 
+std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+hybrid_scan_reader_impl::bloom_filters_byte_ranges(
+  std::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->bloom_filters_byte_ranges(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(
   std::span<cudf::device_span<uint8_t const> const> dictionary_page_data,

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

@@ -108,12 +108,19 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
     cuda::stream_ref stream);
 
   /**
-   * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::secondary_filters_byte_ranges
+   * @copydoc cudf::io::parquet::experimental::hybrid_scan_reader::secondary_filters_byte_ranges
    */
   [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<byte_range_info>>
   secondary_filters_byte_ranges(std::span<std::vector<size_type> const> row_group_indices,
                                 parquet_reader_options const& options);
 
+  /**
+   * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::bloom_filters_byte_ranges
+   */
+  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+  bloom_filters_byte_ranges(std::span<std::vector<size_type> const> row_group_indices,
+                            parquet_reader_options const& options);
+
   /**
    * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::dictionary_pages_byte_ranges
    */

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

@@ -77,13 +77,24 @@ 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(
-  cudf::host_span<std::vector<size_type> const> row_group_indices,
+std::pair<std::vector<text::byte_range_info>, std::vector<size_type>>
+hybrid_scan_multifile::bloom_filters_byte_ranges(
+  std::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(
+  std::span<cudf::device_span<uint8_t const> const> bloom_filter_data,
+  std::span<std::vector<size_type> const> row_group_indices,
+  parquet_reader_options const& options,
+  cuda::stream_ref stream) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->filter_row_groups_with_bloom_filters(
+    bloom_filter_data, row_group_indices, options, stream);
 }
 
 std::unique_ptr<cudf::column> hybrid_scan_multifile::build_all_true_row_mask(