Implement data page pruning using Parquet page index stats (#18873)

Contributes to #17896. Part of #18011. Implements feature request in #9269

This PR implements discarding of Parquet data pages using the page level (min/max) statistics contained in the page index section of a parquet file, in the experimental Parquet reader for optimizing hybrid scan queries.

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

Approvers:
  - Vukasin Milovanovic (https://github.com/vuule)
  - Kyle Edwards (https://github.com/KyleFromNVIDIA)
  - Shruti Shivakumar (https://github.com/shrshi)

URL: #18873

Author: mhaseeb123

cpp/CMakeLists.txt

@@ -534,6 +534,7 @@ add_library(
   src/io/parquet/experimental/hybrid_scan.cpp
   src/io/parquet/experimental/hybrid_scan_helpers.cpp
   src/io/parquet/experimental/hybrid_scan_impl.cpp
+  src/io/parquet/experimental/page_index_filter.cu
   src/io/parquet/page_data.cu
   src/io/parquet/chunk_dict.cu
   src/io/parquet/page_enc.cu

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

@@ -203,6 +203,49 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
     host_span<cudf::size_type const> output_column_schemas,
     std::reference_wrapper<ast::expression const> filter,
     rmm::cuda_stream_view stream) const;
+
+  /**
+   * @brief Filter data pages using statistics page-level statistics based on predicate filter
+   *
+   * @param row_group_indices Input row groups indices
+   * @param output_dtypes Datatypes of output columns
+   * @param output_column_schemas schema indices of output columns
+   * @param filter AST expression to filter data pages based on `PageIndex` statistics
+   * @param stream CUDA stream used for device memory operations and kernel launches
+   * @param mr Device memory resource used to allocate the returned column's device memory
+   *
+   * @return A boolean column representing a mask of rows surviving the predicate filter at
+   *         page-level
+   */
+  [[nodiscard]] std::unique_ptr<cudf::column> filter_data_pages_with_stats(
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    cudf::host_span<cudf::data_type const> output_dtypes,
+    cudf::host_span<cudf::size_type const> output_column_schemas,
+    std::reference_wrapper<ast::expression const> filter,
+    rmm::cuda_stream_view stream,
+    rmm::device_async_resource_ref mr) const;
+
+  /**
+   * @brief Computes which data pages need decoding to construct input columns based on the row mask
+   *
+   * Compute a vector of boolean vectors indicating which data pages need to be decoded to
+   * construct each input column based on the row mask, one vector per column
+   *
+   * @param row_mask Boolean column indicating which rows need to be read after page-pruning
+   * @param row_group_indices Input row groups indices
+   * @param output_dtypes Datatypes of output columns
+   * @param output_column_schemas schema indices of output columns
+   * @param stream CUDA stream used for device memory operations and kernel launches
+   *
+   * @return A vector of boolean vectors indicating which data pages need to be decoded to produce
+   *         the output table based on the input row mask, one per input column
+   */
+  [[nodiscard]] std::vector<thrust::host_vector<bool>> compute_data_page_mask(
+    cudf::column_view row_mask,
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    cudf::host_span<cudf::data_type const> output_dtypes,
+    cudf::host_span<cudf::size_type const> output_column_schemas,
+    rmm::cuda_stream_view stream) const;
 };
 
 /**

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

@@ -262,7 +262,29 @@ hybrid_scan_reader_impl::filter_data_pages_with_stats(
   rmm::cuda_stream_view stream,
   rmm::device_async_resource_ref mr)
 {
-  return {};
+  CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered");
+  CUDF_EXPECTS(options.get_filter().has_value(), "Encountered empty converted filter expression");
+
+  select_columns(read_mode::FILTER_COLUMNS, options);
+
+  table_metadata metadata;
+  populate_metadata(metadata);
+  auto expr_conv = named_to_reference_converter(options.get_filter(), metadata);
+  CUDF_EXPECTS(expr_conv.get_converted_expr().has_value(),
+               "Columns names in filter expression must be convertible to index references");
+  auto output_dtypes = get_output_types(_output_buffers_template);
+
+  auto row_mask = _metadata->filter_data_pages_with_stats(row_group_indices,
+                                                          output_dtypes,
+                                                          _output_column_schemas,
+                                                          expr_conv.get_converted_expr().value(),
+                                                          stream,
+                                                          mr);
+
+  auto data_page_mask = _metadata->compute_data_page_mask(
+    row_mask->view(), row_group_indices, output_dtypes, _output_column_schemas, stream);
+
+  return {std::move(row_mask), std::move(data_page_mask)};
 }
 
 std::pair<std::vector<byte_range_info>, std::vector<cudf::size_type>>