Implement row group pruning with bloom filters in experimental PQ reader (#18545)

Contributes to #17896. Part of #18011.

This PR implements row group pruning with bloom filters in the experimental Parquet reader optimized for hybrid scan queries. Dictionary based row group pruning is still WIP in a separate branch and so this PR has empty definitions where needed.

Note: Unfortunately, we can't add any tests for this feature as we don't yet have capability of writing parquet files with bloom filters. However, the code that filters row groups with bloom filters is identical to already tested code at:  https://github.com/rapidsai/cudf/blob/branch-25.06/cpp/src/io/parquet/predicate_pushdown.cpp#L198-L240

Authors:
  - Muhammad Haseeb (https://github.com/mhaseeb123)

Approvers:
  - Vukasin Milovanovic (https://github.com/vuule)
  - Nghia Truong (https://github.com/ttnghia)
  - Bradley Dice (https://github.com/bdice)

URL: #18545

Author: mhaseeb123

cpp/CMakeLists.txt

@@ -546,6 +546,7 @@ add_library(
   src/io/parquet/compact_protocol_reader.cpp
   src/io/parquet/compact_protocol_writer.cpp
   src/io/parquet/decode_preprocess.cu
+  src/io/parquet/experimental/dictionary_page_filter.cu
   src/io/parquet/experimental/hybrid_scan.cpp
   src/io/parquet/experimental/hybrid_scan_helpers.cpp
   src/io/parquet/experimental/hybrid_scan_impl.cpp

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

@@ -339,12 +339,16 @@ class hybrid_scan_reader {
     rmm::cuda_stream_view stream) const;
 
   /**
-   * @brief Get byte ranges of bloom filters and dictionary pages (secondary filters) for
-   * further row group pruning
+   * @brief Get byte ranges of bloom filters and dictionary pages (secondary 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 groups indices
    * @param options Parquet reader options
-   * @return Pair of vectors of byte ranges to per-column-chunk bloom filters and dictionary pages
+   * @return Pair of vectors of byte ranges of column chunk with bloom filters and dictionary
+   *         pages subject to filter predicate
    */
   [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<byte_range_info>>
   secondary_filters_byte_ranges(cudf::host_span<size_type const> row_group_indices,
@@ -353,7 +357,8 @@ class hybrid_scan_reader {
   /**
    * @brief Filter the row groups using column chunk dictionary pages
    *
-   * @param dictionary_page_data Device buffers containing per-column-chunk dictionary page data
+   * @param dictionary_page_data Device buffers containing dictionary page data of column chunks
+   *                             with (in)equality predicate
    * @param row_group_indices Input row groups indices
    * @param options Parquet reader options
    * @param stream CUDA stream used for device memory operations and kernel launches
@@ -368,7 +373,11 @@ class hybrid_scan_reader {
   /**
    * @brief Filter the row groups using column chunk bloom filters
    *
-   * @param bloom_filter_data Device buffers containing per-column-chunk bloom filter data
+   * @note The `bloom_filter_data` device buffers must be allocated using a 32
+   *       byte aligned memory resource
+   *
+   * @param bloom_filter_data Device buffers containing bloom filter data of column chunks with
+   *                          an equality predicate
    * @param row_group_indices Input row groups indices
    * @param options Parquet reader options
    * @param stream CUDA stream used for device memory operations and kernel launches

cpp/src/io/parquet/bloom_filter_reader.cu

@@ -513,15 +513,15 @@ std::optional<std::vector<std::vector<size_type>>> aggregate_reader_metadata::ap
   host_span<std::vector<ast::literal*> const> literals,
   size_type total_row_groups,
   host_span<data_type const> output_dtypes,
-  host_span<int const> equality_col_schemas,
+  host_span<int const> bloom_filter_col_schemas,
   std::reference_wrapper<ast::expression const> filter,
   rmm::cuda_stream_view stream) const
 {
   // Number of input table columns
   auto const num_input_columns = static_cast<cudf::size_type>(output_dtypes.size());
 
   // Get parquet types for the predicate columns
-  auto const parquet_types = get_parquet_types(input_row_group_indices, equality_col_schemas);
+  auto const parquet_types = get_parquet_types(input_row_group_indices, bloom_filter_col_schemas);
 
   // Create spans from bloom filter bitset buffers to use in cuco::bloom_filter_ref.
   std::vector<cudf::device_span<cuda::std::byte>> h_bloom_filter_spans;
@@ -542,7 +542,7 @@ std::optional<std::vector<std::vector<size_type>>> aggregate_reader_metadata::ap
   bloom_filter_caster const bloom_filter_col{bloom_filter_spans,
                                              parquet_types,
                                              static_cast<size_t>(total_row_groups),
-                                             equality_col_schemas.size()};
+                                             bloom_filter_col_schemas.size()};
 
   // Converts bloom filter membership for equality predicate columns to a table
   // containing a column for each `col[i] == literal` predicate to be evaluated.
@@ -584,8 +584,6 @@ std::optional<std::vector<std::vector<size_type>>> aggregate_reader_metadata::ap
                                             stream);
 }
 
-equality_literals_collector::equality_literals_collector() = default;
-
 equality_literals_collector::equality_literals_collector(ast::expression const& expr,
                                                          cudf::size_type num_input_columns)
   : _num_input_columns{num_input_columns}
@@ -604,7 +602,7 @@ std::reference_wrapper<ast::expression const> equality_literals_collector::visit
   ast::column_reference const& expr)
 {
   CUDF_EXPECTS(expr.get_table_source() == ast::table_reference::LEFT,
-               "BloomfilterAST supports only left table");
+               "DictionaryAST and BloomfilterAST support only left table");
   CUDF_EXPECTS(expr.get_column_index() < _num_input_columns,
                "Column index cannot be more than number of columns in the table");
   return expr;
@@ -613,7 +611,7 @@ std::reference_wrapper<ast::expression const> equality_literals_collector::visit
 std::reference_wrapper<ast::expression const> equality_literals_collector::visit(
   ast::column_name_reference const& expr)
 {
-  CUDF_FAIL("Column name reference is not supported in BloomfilterAST");
+  CUDF_FAIL("Column name reference is not supported in DictionaryAST and BloomfilterAST");
 }
 
 std::reference_wrapper<ast::expression const> equality_literals_collector::visit(

cpp/src/io/parquet/experimental/dictionary_page_filter.cu

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "hybrid_scan_helpers.hpp"
+
+#include <cudf/ast/detail/expression_transformer.hpp>
+#include <cudf/ast/detail/operators.hpp>
+#include <cudf/ast/expressions.hpp>
+#include <cudf/detail/cuco_helpers.hpp>
+#include <cudf/detail/transform.hpp>
+#include <cudf/hashing/detail/xxhash_64.cuh>
+#include <cudf/io/parquet_schema.hpp>
+#include <cudf/logger.hpp>
+#include <cudf/utilities/span.hpp>
+#include <cudf/utilities/traits.hpp>
+#include <cudf/utilities/type_checks.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/device_buffer.hpp>
+#include <rmm/exec_policy.hpp>
+
+#include <thrust/iterator/counting_iterator.h>
+#include <thrust/tabulate.h>
+
+#include <future>
+#include <numeric>
+#include <optional>
+
+namespace cudf::io::parquet::experimental::detail {
+
+using parquet::detail::chunk_page_info;
+using parquet::detail::ColumnChunkDesc;
+using parquet::detail::decode_error;
+using parquet::detail::PageInfo;
+
+dictionary_literals_collector::dictionary_literals_collector(ast::expression const& expr,
+                                                             cudf::size_type num_input_columns)
+{
+  _num_input_columns = num_input_columns;
+  _literals.resize(num_input_columns);
+  expr.accept(*this);
+}
+
+std::reference_wrapper<ast::expression const> dictionary_literals_collector::visit(
+  ast::operation const& expr)
+{
+  using cudf::ast::ast_operator;
+  auto const operands = expr.get_operands();
+  auto const op       = expr.get_operator();
+
+  if (auto* v = dynamic_cast<ast::column_reference const*>(&operands[0].get())) {
+    // First operand should be column reference, second should be literal.
+    CUDF_EXPECTS(cudf::ast::detail::ast_operator_arity(op) == 2,
+                 "Only binary operations are supported on column reference");
+    auto const literal_ptr = dynamic_cast<ast::literal const*>(&operands[1].get());
+    CUDF_EXPECTS(literal_ptr != nullptr,
+                 "Second operand of binary operation with column reference must be a literal");
+    v->accept(*this);
+
+    // Push to the corresponding column's literals and operators list iff EQUAL or NOT_EQUAL
+    // operator is seen
+    if (op == ast_operator::EQUAL or op == ast::ast_operator::NOT_EQUAL) {
+      auto const col_idx = v->get_column_index();
+      _literals[col_idx].emplace_back(const_cast<ast::literal*>(literal_ptr));
+    }
+  } else {
+    // Visit the operands and ignore any output as we only want to collect literals
+    std::ignore = visit_operands(operands);
+  }
+
+  return expr;
+}
+
+}  // namespace cudf::io::parquet::experimental::detail