From 121bf9dd8225c64844a163e2e6991c92479f1990 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 01:15:18 +0000
Subject: [PATCH 01/23] Add hybrid scan multifile reader basics

---
 cpp/CMakeLists.txt                            |   1 +
 .../io/experimental/hybrid_scan_multifile.hpp | 146 +++++++++++++++
 .../io/parquet/experimental/hybrid_scan.cpp   |  15 +-
 .../experimental/hybrid_scan_helpers.cpp      | 166 ++++++++++++------
 .../experimental/hybrid_scan_helpers.hpp      |  39 ++--
 .../parquet/experimental/hybrid_scan_impl.cpp |  25 ++-
 .../parquet/experimental/hybrid_scan_impl.hpp |  35 ++--
 .../experimental/hybrid_scan_multifile.cpp    |  60 +++++++
 8 files changed, 381 insertions(+), 106 deletions(-)
 create mode 100644 cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
 create mode 100644 cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp

diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
index 63c292646773..d3576d2a6b5a 100644
--- a/cpp/CMakeLists.txt
+++ b/cpp/CMakeLists.txt
@@ -627,6 +627,7 @@ add_library(
   src/io/parquet/experimental/hybrid_scan_chunking.cu
   src/io/parquet/experimental/hybrid_scan_helpers.cpp
   src/io/parquet/experimental/hybrid_scan_impl.cpp
+  src/io/parquet/experimental/hybrid_scan_multifile.cpp
   src/io/parquet/experimental/hybrid_scan_preprocess.cu
   src/io/parquet/experimental/page_index_filter.cu
   src/io/parquet/experimental/page_index_filter_utils.cu
diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
new file mode 100644
index 000000000000..13bbf97ee499
--- /dev/null
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -0,0 +1,146 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cudf/io/experimental/hybrid_scan.hpp>
+#include <cudf/io/parquet.hpp>
+#include <cudf/io/parquet_schema.hpp>
+#include <cudf/io/text/byte_range_info.hpp>
+#include <cudf/io/types.hpp>
+#include <cudf/types.hpp>
+#include <cudf/utilities/export.hpp>
+
+#include <rmm/cuda_stream_view.hpp>
+#include <rmm/resource_ref.hpp>
+
+#include <memory>
+#include <vector>
+
+namespace CUDF_EXPORT cudf {
+namespace io::parquet::experimental::detail {
+/**
+ * @brief Internal experimental Parquet reader optimized for highly selective filters, called a
+ *        Hybrid Scan operation.
+ */
+class hybrid_scan_reader_impl;
+}  // namespace io::parquet::experimental::detail
+}  // namespace CUDF_EXPORT cudf
+
+//! Using `byte_range_info` from cudf::io::text
+using cudf::io::text::byte_range_info;
+
+namespace CUDF_EXPORT cudf {
+namespace io::parquet::experimental {
+/**
+ * @addtogroup io_readers
+ * @{
+ * @file
+ */
+
+/**
+ * @brief Multi-file variant of the experimental Hybrid Scan Parquet reader
+ *
+ * Vectorizes `hybrid_scan_reader` APIs to support multiple Parquet sources. Inputs and outputs are
+ * indexed by source order except for the row mask which is a single BOOL8 column spanning all rows
+ * from all sources concatenated in source order, then row-group order within a source.
+ *
+ * @note Detailed usage documentation will be added once all APIs are in place.
+ */
+class hybrid_scan_multifile {
+ public:
+  /**
+   * @brief Constructor for the multi-file experimental Parquet reader
+   *
+   * @param footer_bytes Host span of Parquet file footer byte spans, one per source
+   * @param options Parquet reader options
+   */
+  explicit hybrid_scan_multifile(cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes,
+                                 parquet_reader_options const& options);
+
+  /**
+   * @brief Constructor for the multi-file experimental Parquet reader
+   *
+   * @param parquet_metadata Host span of pre-populated Parquet file metadata, one per source
+   * @param options Parquet reader options
+   */
+  explicit hybrid_scan_multifile(cudf::host_span<FileMetaData const> parquet_metadata,
+                                 parquet_reader_options const& options);
+
+  /**
+   * @brief Destructor for the multi-file experimental Parquet reader
+   */
+  ~hybrid_scan_multifile();
+
+  /**
+   * @brief Get the per-source Parquet file footer metadata
+   *
+   * @return Vector of file metadata, one per source
+   */
+  [[nodiscard]] std::vector<FileMetaData> parquet_metadata() const;
+
+  /**
+   * @brief Get the per-source byte range of the page index in each Parquet file
+   *
+   * The returned vector always has one entry per source. A source's entry is a
+   * default-constructed `byte_range_info{}` if that source has no row groups, no columns, or no
+   * page index offsets. A `CUDF_LOG_WARN` is emitted once if some sources have a page index and
+   * others do not.
+   *
+   * @return Vector of page index byte ranges, one per source
+   */
+  [[nodiscard]] std::vector<byte_range_info> page_index_byte_range() const;
+
+  /**
+   * @brief Setup the per-source page index within each Parquet file metadata
+   *
+   * Materializes `ColumnIndex` and `OffsetIndex` (page index) inside each source's
+   * `FileMetaData`. The input span size must equal the number of sources. A per-source empty
+   * span is skipped with a one-time warning. Sources whose corresponding span is non-empty must
+   * have row groups and valid page index offsets.
+   *
+   * @param page_index_bytes Host span of Parquet page index buffer bytes, one per source
+   */
+  void setup_page_index(
+    cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const;
+
+  /**
+   * @brief Get all available per-source row group indices from the parquet files
+   *
+   * If `options.get_row_groups()` is non-empty, its size must equal the number of sources and it
+   * is returned as-is. Otherwise builds `[0 .. per_source_num_row_groups[i])` for each source.
+   *
+   * @param options Parquet reader options
+   * @return Vector of row group indices, one inner vector per source
+   */
+  [[nodiscard]] std::vector<std::vector<size_type>> all_row_groups(
+    parquet_reader_options const& options) const;
+
+  /**
+   * @brief Get the total number of top-level rows in the per-source row groups
+   *
+   * @param row_group_indices Input per-source row group indices (one inner vector per source)
+   * @return Total number of top-level rows across all sources
+   */
+  [[nodiscard]] size_type total_rows_in_row_groups(
+    cudf::host_span<std::vector<size_type> const> row_group_indices) const;
+
+  /**
+   * @brief Resets the current column selection
+   *
+   * Resets the current column selection state forcing column re-selection in subsequent filter,
+   * byte range, setup chunking and materialization APIs. This is useful if the filter expression
+   * has been cascaded (and-ed) to include new columns.
+   */
+  void reset_column_selection() const;
+
+ private:
+  std::unique_ptr<detail::hybrid_scan_reader_impl> _impl;
+};
+
+/** @} */  // end of group
+
+}  // namespace io::parquet::experimental
+}  // namespace CUDF_EXPORT cudf
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan.cpp b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
index b6243fafe60a..3223ca99c8f8 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
@@ -15,33 +15,36 @@ namespace cudf::io::parquet::experimental {
 
 hybrid_scan_reader::hybrid_scan_reader(cudf::host_span<uint8_t const> footer_bytes,
                                        parquet_reader_options const& options)
-  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(footer_bytes, options)}
 {
+  auto const footers = std::vector<cudf::host_span<uint8_t const>>{footer_bytes};
+  _impl              = std::make_unique<detail::hybrid_scan_reader_impl>(footers, options);
 }
 
 hybrid_scan_reader::hybrid_scan_reader(FileMetaData const& parquet_metadata,
                                        parquet_reader_options const& options)
-  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(parquet_metadata, options)}
 {
+  auto const metadatas = std::vector<FileMetaData>{parquet_metadata};
+  _impl                = std::make_unique<detail::hybrid_scan_reader_impl>(metadatas, options);
 }
 
 hybrid_scan_reader::~hybrid_scan_reader() = default;
 
 [[nodiscard]] text::byte_range_info hybrid_scan_reader::page_index_byte_range() const
 {
-  return _impl->page_index_byte_range();
+  return _impl->page_index_byte_range().front();
 }
 
 [[nodiscard]] FileMetaData hybrid_scan_reader::parquet_metadata() const
 {
-  return _impl->parquet_metadata();
+  return _impl->parquet_metadata().front();
 }
 
 void hybrid_scan_reader::setup_page_index(cudf::host_span<uint8_t const> page_index_bytes) const
 {
   CUDF_FUNC_RANGE();
 
-  return _impl->setup_page_index(page_index_bytes);
+  auto const per_source = std::vector<cudf::host_span<uint8_t const>>{page_index_bytes};
+  return _impl->setup_page_index(per_source);
 }
 
 std::vector<cudf::size_type> hybrid_scan_reader::all_row_groups(
@@ -53,7 +56,7 @@ std::vector<cudf::size_type> hybrid_scan_reader::all_row_groups(
   // If row groups are specified in parquet reader options, return them as is
   if (options.get_row_groups().size() == 1) { return options.get_row_groups().front(); }
 
-  return _impl->all_row_groups(options);
+  return _impl->all_row_groups(options).front();
 }
 
 size_type hybrid_scan_reader::total_rows_in_row_groups(
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
index f6e6985ea4ea..e93054073637 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
@@ -72,23 +72,34 @@ metadata::metadata(cudf::host_span<uint8_t const> footer_bytes)
   sanitize_schema();
 }
 
-aggregate_reader_metadata::aggregate_reader_metadata(FileMetaData const& parquet_metadata,
-                                                     bool use_arrow_schema,
-                                                     bool has_cols_from_mismatched_srcs)
+aggregate_reader_metadata::aggregate_reader_metadata(
+  cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes,
+  bool use_arrow_schema,
+  bool has_cols_from_mismatched_srcs)
   : aggregate_reader_metadata_base(host_span<std::unique_ptr<datasource> const>{}, false, false)
 {
-  // Just copy over the FileMetaData struct to the internal metadata struct
-  per_file_metadata.emplace_back(metadata{parquet_metadata});
+  CUDF_EXPECTS(not footer_bytes.empty(), "At least one source must be provided");
+  per_file_metadata.reserve(footer_bytes.size());
+  std::transform(footer_bytes.begin(),
+                 footer_bytes.end(),
+                 std::back_inserter(per_file_metadata),
+                 [](auto const& fb) { return metadata{fb}; });
   initialize_internals(use_arrow_schema, has_cols_from_mismatched_srcs);
 }
 
-aggregate_reader_metadata::aggregate_reader_metadata(cudf::host_span<uint8_t const> footer_bytes,
-                                                     bool use_arrow_schema,
-                                                     bool has_cols_from_mismatched_srcs)
+aggregate_reader_metadata::aggregate_reader_metadata(
+  cudf::host_span<FileMetaData const> parquet_metadatas,
+  bool use_arrow_schema,
+  bool has_cols_from_mismatched_srcs)
   : aggregate_reader_metadata_base(host_span<std::unique_ptr<datasource> const>{}, false, false)
 {
-  // Re-initialize internal variables here as base class was initialized without a source
-  per_file_metadata.emplace_back(metadata{footer_bytes});
+  CUDF_EXPECTS(not parquet_metadatas.empty(), "At least one source must be provided");
+  per_file_metadata.reserve(parquet_metadatas.size());
+  // Just copy over the FileMetaData structs to the internal metadata structs
+  std::transform(parquet_metadatas.begin(),
+                 parquet_metadatas.end(),
+                 std::back_inserter(per_file_metadata),
+                 [](auto const& parquet_metadata) { return metadata{parquet_metadata}; });
   initialize_internals(use_arrow_schema, has_cols_from_mismatched_srcs);
 }
 
@@ -102,13 +113,15 @@ void aggregate_reader_metadata::initialize_internals(bool use_arrow_schema,
 
   // Force all non-nullable (REQUIRED) columns to be nullable without modifying REPEATED columns to
   // preserve list structures
-  auto& schema = per_file_metadata.front().schema;
-  std::for_each(schema.begin() + 1, schema.end(), [](auto& col) {
-    // TODO: Store information of whichever column schema we modified here and restore it to
-    // `REQUIRED` if we end up not pruning any pages out of it
-    if (col.repetition_type == FieldRepetitionType::REQUIRED) {
-      col.repetition_type = FieldRepetitionType::OPTIONAL;
-    }
+  std::for_each(per_file_metadata.begin(), per_file_metadata.end(), [](auto& pfm) {
+    auto& schema = pfm.schema;
+    std::for_each(schema.begin() + 1, schema.end(), [](auto& col) {
+      // TODO: Store information of whichever column schema we modified here and restore it to
+      // `REQUIRED` if we end up not pruning any pages out of it
+      if (col.repetition_type == FieldRepetitionType::REQUIRED) {
+        col.repetition_type = FieldRepetitionType::OPTIONAL;
+      }
+    });
   });
 
   // Collect and apply arrow:schema from Parquet's key value metadata section
@@ -122,53 +135,90 @@ void aggregate_reader_metadata::initialize_internals(bool use_arrow_schema,
   }
 }
 
-text::byte_range_info aggregate_reader_metadata::page_index_byte_range() const
+std::vector<text::byte_range_info> aggregate_reader_metadata::page_index_byte_range() const
 {
-  auto& schema     = per_file_metadata.front();
-  auto& row_groups = schema.row_groups;
-
-  if (row_groups.size() and row_groups.front().columns.size()) {
-    auto const min_offset = schema.row_groups.front().columns.front().column_index_offset;
-    auto const& last_col  = schema.row_groups.back().columns.back();
-    auto const max_offset = last_col.offset_index_offset + last_col.offset_index_length;
-    return {min_offset, (max_offset - min_offset)};
-  }
+  std::vector<text::byte_range_info> page_index_byte_ranges;
+  page_index_byte_ranges.reserve(per_file_metadata.size());
+  std::transform(per_file_metadata.begin(),
+                 per_file_metadata.end(),
+                 std::back_inserter(page_index_byte_ranges),
+                 [](auto const& pfm) -> text::byte_range_info {
+                   auto const& row_groups = pfm.row_groups;
+                   if (row_groups.empty() or row_groups.front().columns.empty()) { return {}; }
+
+                   auto const min_offset = row_groups.front().columns.front().column_index_offset;
+                   auto const& last_col  = row_groups.back().columns.back();
+                   auto const max_offset =
+                     last_col.offset_index_offset + last_col.offset_index_length;
+
+                   if (max_offset <= min_offset) { return {}; }
+                   return {min_offset, max_offset - min_offset};
+                 });
 
-  return {};
+  return page_index_byte_ranges;
 }
 
-FileMetaData aggregate_reader_metadata::parquet_metadata() const
+std::vector<FileMetaData> aggregate_reader_metadata::parquet_metadata() const
 {
-  return per_file_metadata.front();
+  return {per_file_metadata.begin(), per_file_metadata.end()};
 }
 
-void aggregate_reader_metadata::setup_page_index(cudf::host_span<uint8_t const> page_index_bytes)
+void aggregate_reader_metadata::setup_page_index(
+  cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes)
 {
-  // Return early if empty page index buffer span
-  if (page_index_bytes.empty()) {
-    CUDF_LOG_WARN("Hybrid scan reader encountered empty page index buffer");
-    return;
-  }
+  CUDF_EXPECTS(page_index_bytes.size() == per_file_metadata.size(),
+               "Page index byte span count must equal the number of sources");
+
+  auto iter = cuda::zip_iterator(page_index_bytes.begin(), per_file_metadata.begin());
+  std::for_each(iter, iter + page_index_bytes.size(), [&](auto const& pair) {
+    auto const& pgidx_bytes = cuda::std::get<0>(pair);
+
+    // Return early if empty page index buffer span
+    if (pgidx_bytes.empty()) { return; }
 
-  // Get the file metadata and setup the page index
-  auto& file_metadata    = per_file_metadata.front();
-  auto const& row_groups = file_metadata.row_groups;
+    // Get the file metadata and setup the page index
+    auto& file_metadata    = cuda::std::get<1>(pair);
+    auto const& row_groups = file_metadata.row_groups;
 
-  // Check for empty parquet file
-  CUDF_EXPECTS(not row_groups.empty() and not row_groups.front().columns.empty(),
-               "No column chunks in Parquet schema to read page index for");
+    // Check for empty parquet file
+    CUDF_EXPECTS(not row_groups.empty() and not row_groups.front().columns.empty(),
+                 "No column chunks in Parquet schema to read page index for");
 
-  // Set the first ColumnChunk's offset of ColumnIndex as the adjusted zero offset
-  int64_t const min_offset = row_groups.front().columns.front().column_index_offset;
+    // Set the first ColumnChunk's offset of ColumnIndex as the adjusted zero offset
+    int64_t const min_offset = row_groups.front().columns.front().column_index_offset;
 
-  // Check if the page index buffer is valid
-  {
-    auto const& last_col  = row_groups.back().columns.back();
-    auto const max_offset = last_col.offset_index_offset + last_col.offset_index_length;
-    CUDF_EXPECTS(max_offset > min_offset, "Encountered an invalid page index buffer");
+    // Check if the page index buffer is valid
+    {
+      auto const& last_col  = row_groups.back().columns.back();
+      auto const max_offset = last_col.offset_index_offset + last_col.offset_index_length;
+      CUDF_EXPECTS(max_offset > min_offset, "Encountered an invalid page index buffer");
+    }
+
+    file_metadata.setup_page_index(pgidx_bytes, min_offset);
+  });
+}
+
+std::vector<std::vector<size_type>> aggregate_reader_metadata::all_row_groups(
+  parquet_reader_options const& options) const
+{
+  auto const& opts_row_groups = options.get_row_groups();
+  if (not opts_row_groups.empty()) {
+    CUDF_EXPECTS(opts_row_groups.size() == per_file_metadata.size(),
+                 "Row groups in parquet reader options must have one inner vector per source");
+    return opts_row_groups;
   }
 
-  file_metadata.setup_page_index(page_index_bytes, min_offset);
+  std::vector<std::vector<size_type>> row_groups;
+  row_groups.reserve(per_file_metadata.size());
+  std::transform(per_file_metadata.begin(),
+                 per_file_metadata.end(),
+                 std::back_inserter(row_groups),
+                 [](auto const& pfm) {
+                   std::vector<size_type> indices(pfm.row_groups.size());
+                   std::iota(indices.begin(), indices.end(), size_type{0});
+                   return indices;
+                 });
+  return row_groups;
 }
 
 size_type aggregate_reader_metadata::total_rows_in_row_groups(
@@ -231,8 +281,8 @@ aggregate_reader_metadata::select_payload_columns(
     return filter_columns_set;
   };
 
-  // If payload columns are specified, only select payload columns that do not appear in the filter
-  // expression
+  // If payload columns are specified, only select payload columns that do not appear in the
+  // filter expression
   if (payload_column_names.has_value()) {
     valid_payload_columns = *payload_column_names;
     // Remove filter columns from the provided payload column names
@@ -634,12 +684,12 @@ std::reference_wrapper<ast::expression const> named_to_reference_converter::visi
 {
   // Map the column index to its name
   auto const col_name_iter = _column_indices_to_names.find(expr.get_column_index());
-  CUDF_EXPECTS(
-    col_name_iter != _column_indices_to_names.end(),
-    "Column index in the filter expression not found in the column indices to names map. Note that "
-    "only top-level columns except structs and lists are supported in "
-    "Parquet filter expression",
-    std::invalid_argument);
+  CUDF_EXPECTS(col_name_iter != _column_indices_to_names.end(),
+               "Column index in the filter expression not found in the column indices to names "
+               "map. Note that "
+               "only top-level columns except structs and lists are supported in "
+               "Parquet filter expression",
+               std::invalid_argument);
   auto const col_name = col_name_iter->second;
   auto col_index_it   = _column_name_to_index.find(col_name);
   CUDF_EXPECTS(col_index_it != _column_name_to_index.end(),
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
index df17ae493cd2..9a8879163782 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
@@ -80,22 +80,22 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
   /**
    * @brief Constructor for aggregate_reader_metadata
    *
-   * @param footer_bytes Host span of Parquet file footer buffer bytes
+   * @param footer_bytes Host span of Parquet file footer buffer bytes, one per source
    * @param use_arrow_schema Whether to use Arrow schema
    * @param has_cols_from_mismatched_srcs Whether to have columns from mismatched sources
    */
-  aggregate_reader_metadata(cudf::host_span<uint8_t const> footer_bytes,
+  aggregate_reader_metadata(cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes,
                             bool use_arrow_schema,
                             bool has_cols_from_mismatched_srcs);
 
   /**
    * @brief Constructor for aggregate_reader_metadata
    *
-   * @param parquet_metadata Pre-populated Parquet file metadata
+   * @param parquet_metadatas Host span of pre-populated Parquet file metadata, one per source
    * @param use_arrow_schema Whether to use Arrow schema
    * @param has_cols_from_mismatched_srcs Whether to have columns from mismatched sources
    */
-  aggregate_reader_metadata(FileMetaData const& parquet_metadata,
+  aggregate_reader_metadata(cudf::host_span<FileMetaData const> parquet_metadatas,
                             bool use_arrow_schema,
                             bool has_cols_from_mismatched_srcs);
 
@@ -110,21 +110,38 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
   void initialize_internals(bool use_arrow_schema, bool has_cols_from_mismatched_srcs);
 
   /**
-   * @brief Fetch the byte range of the page index in the Parquet file
+   * @brief Fetch the byte range of the page index in each Parquet file
+   *
+   * @return Vector of byte ranges of the page index, one per source
    */
-  [[nodiscard]] text::byte_range_info page_index_byte_range() const;
+  [[nodiscard]] std::vector<text::byte_range_info> page_index_byte_range() const;
 
   /**
-   * @brief Get the Parquet file metadata
+   * @brief Get the Parquet file metadata for every source
+   *
+   * @return Vector of file metadata, one per source
    */
-  [[nodiscard]] FileMetaData parquet_metadata() const;
+  [[nodiscard]] std::vector<FileMetaData> parquet_metadata() const;
 
   /**
-   * @brief Setup and populate the page index structs in `FileMetaData`
+   * @brief Setup and populate the page index structs in every source's `FileMetaData`
+   *
+   * @param page_index_bytes Host span of Parquet page index buffer bytes, one per source
+   */
+  void setup_page_index(cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes);
+
+  /**
+   * @brief Get all available row group indices, one inner vector per source
+   *
+   * If `options.get_row_groups()` is non-empty, validates that its size equals the number of
+   * sources and returns it as-is. Otherwise returns `[0 .. per_source_num_row_groups[i])` for
+   * each source.
    *
-   * @param page_index_bytes Host span of Parquet page index buffer bytes
+   * @param options Parquet reader options
+   * @return Vector of row group indices, one inner vector per source
    */
-  void setup_page_index(cudf::host_span<uint8_t const> page_index_bytes);
+  [[nodiscard]] std::vector<std::vector<size_type>> all_row_groups(
+    parquet_reader_options const& options) const;
 
   /**
    * @brief Get the total number of top-level rows in the row groups
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
index 7b53b4345911..5ef78f19299e 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
@@ -64,10 +64,10 @@ namespace {
 
 }  // namespace
 
-hybrid_scan_reader_impl::hybrid_scan_reader_impl(cudf::host_span<uint8_t const> footer_bytes,
-                                                 parquet_reader_options const& options)
+hybrid_scan_reader_impl::hybrid_scan_reader_impl(
+  cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes,
+  parquet_reader_options const& options)
 {
-  // Open and parse the source dataset metadata
   _metadata = std::make_unique<aggregate_reader_metadata>(
     footer_bytes,
     options.is_enabled_use_arrow_schema(),
@@ -76,28 +76,28 @@ hybrid_scan_reader_impl::hybrid_scan_reader_impl(cudf::host_span<uint8_t const>
   _extended_metadata = static_cast<aggregate_reader_metadata*>(_metadata.get());
 }
 
-hybrid_scan_reader_impl::hybrid_scan_reader_impl(FileMetaData const& parquet_metadata,
-                                                 parquet_reader_options const& options)
+hybrid_scan_reader_impl::hybrid_scan_reader_impl(
+  cudf::host_span<FileMetaData const> parquet_metadatas, parquet_reader_options const& options)
 {
   _metadata = std::make_unique<aggregate_reader_metadata>(
-    parquet_metadata,
+    parquet_metadatas,
     options.is_enabled_use_arrow_schema(),
     options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas());
   _extended_metadata = static_cast<aggregate_reader_metadata*>(_metadata.get());
 }
 
-FileMetaData hybrid_scan_reader_impl::parquet_metadata() const
+std::vector<FileMetaData> hybrid_scan_reader_impl::parquet_metadata() const
 {
   return _extended_metadata->parquet_metadata();
 }
 
-byte_range_info hybrid_scan_reader_impl::page_index_byte_range() const
+std::vector<byte_range_info> hybrid_scan_reader_impl::page_index_byte_range() const
 {
   return _extended_metadata->page_index_byte_range();
 }
 
 void hybrid_scan_reader_impl::setup_page_index(
-  cudf::host_span<uint8_t const> page_index_bytes) const
+  cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const
 {
   _extended_metadata->setup_page_index(page_index_bytes);
 }
@@ -200,13 +200,10 @@ void hybrid_scan_reader_impl::select_columns(read_columns_mode read_columns_mode
                  [](auto const& buff) { return inline_column_buffer::empty_like(buff); });
 }
 
-std::vector<size_type> hybrid_scan_reader_impl::all_row_groups(
+std::vector<std::vector<size_type>> hybrid_scan_reader_impl::all_row_groups(
   parquet_reader_options const& options) const
 {
-  auto const num_row_groups = _extended_metadata->get_num_row_groups();
-  auto row_groups_indices   = std::vector<size_type>(num_row_groups);
-  std::iota(row_groups_indices.begin(), row_groups_indices.end(), size_type{0});
-  return row_groups_indices;
+  return _extended_metadata->all_row_groups(options);
 }
 
 size_type hybrid_scan_reader_impl::total_rows_in_row_groups(
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
index 3f16699a3d18..0e85f066d347 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
@@ -39,44 +39,45 @@ using text::byte_range_info;
 class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
  public:
   /**
-   * @brief Constructor for the experimental parquet reader implementation to optimally read
-   * Parquet files subject to highly selective filters
+   * @brief Constructor for the experimental parquet reader implementation
    *
-   * @param footer_bytes Host span of parquet file footer bytes
+   * @param footer_bytes Span of parquet file footer byte spans, one per source
    * @param options Parquet reader options
    */
-  explicit hybrid_scan_reader_impl(cudf::host_span<uint8_t const> footer_bytes,
-                                   parquet_reader_options const& options);
+  explicit hybrid_scan_reader_impl(
+    cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes,
+    parquet_reader_options const& options);
 
   /**
-   * @brief Constructor for the experimental parquet reader implementation to optimally read
-   * Parquet files subject to highly selective filters
+   * @brief Constructor for the experimental parquet reader implementation
    *
-   * @param parquet_metadata Pre-populated Parquet file metadata
+   * @param parquet_metadatas Span of pre-populated Parquet file metadata, one per source
    * @param options Parquet reader options
    */
-  explicit hybrid_scan_reader_impl(FileMetaData const& parquet_metadata,
+  explicit hybrid_scan_reader_impl(cudf::host_span<FileMetaData const> parquet_metadatas,
                                    parquet_reader_options const& options);
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::parquet_metadata
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::parquet_metadata
    */
-  [[nodiscard]] FileMetaData parquet_metadata() const;
+  [[nodiscard]] std::vector<FileMetaData> parquet_metadata() const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::page_index_byte_range
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::page_index_byte_range
    */
-  [[nodiscard]] byte_range_info page_index_byte_range() const;
+  [[nodiscard]] std::vector<byte_range_info> page_index_byte_range() const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::setup_page_index
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::setup_page_index
    */
-  void setup_page_index(cudf::host_span<uint8_t const> page_index_bytes) const;
+  void setup_page_index(
+    cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::all_row_groups
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::all_row_groups
    */
-  [[nodiscard]] std::vector<size_type> all_row_groups(parquet_reader_options const& options) const;
+  [[nodiscard]] std::vector<std::vector<size_type>> all_row_groups(
+    parquet_reader_options const& options) const;
 
   /**
    * @copydoc cudf::io::experimental::hybrid_scan::total_rows_in_row_groups
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
new file mode 100644
index 000000000000..32819903f530
--- /dev/null
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
@@ -0,0 +1,60 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "hybrid_scan_impl.hpp"
+
+#include <cudf/detail/nvtx/ranges.hpp>
+#include <cudf/io/experimental/hybrid_scan_multifile.hpp>
+
+namespace cudf::io::parquet::experimental {
+
+hybrid_scan_multifile::hybrid_scan_multifile(
+  cudf::host_span<cudf::host_span<uint8_t const> const> footer_bytes,
+  parquet_reader_options const& options)
+  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(footer_bytes, options)}
+{
+}
+
+hybrid_scan_multifile::hybrid_scan_multifile(cudf::host_span<FileMetaData const> parquet_metadata,
+                                             parquet_reader_options const& options)
+  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(parquet_metadata, options)}
+{
+}
+
+hybrid_scan_multifile::~hybrid_scan_multifile() = default;
+
+std::vector<FileMetaData> hybrid_scan_multifile::parquet_metadata() const
+{
+  return _impl->parquet_metadata();
+}
+
+std::vector<text::byte_range_info> hybrid_scan_multifile::page_index_byte_range() const
+{
+  return _impl->page_index_byte_range();
+}
+
+void hybrid_scan_multifile::setup_page_index(
+  cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const
+{
+  CUDF_FUNC_RANGE();
+  _impl->setup_page_index(page_index_bytes);
+}
+
+std::vector<std::vector<size_type>> hybrid_scan_multifile::all_row_groups(
+  parquet_reader_options const& options) const
+{
+  return _impl->all_row_groups(options);
+}
+
+size_type hybrid_scan_multifile::total_rows_in_row_groups(
+  cudf::host_span<std::vector<size_type> const> row_group_indices) const
+{
+  if (row_group_indices.empty()) { return 0; }
+  return _impl->total_rows_in_row_groups(row_group_indices);
+}
+
+void hybrid_scan_multifile::reset_column_selection() const { _impl->reset_column_selection(); }
+
+}  // namespace cudf::io::parquet::experimental

From b763cdb973c0313beb8e5bb8cdff63bf22de3134 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 01:20:19 +0000
Subject: [PATCH 02/23] Minor

---
 cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index 13bbf97ee499..6b6418879e21 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -47,7 +47,9 @@ namespace io::parquet::experimental {
  * indexed by source order except for the row mask which is a single BOOL8 column spanning all rows
  * from all sources concatenated in source order, then row-group order within a source.
  *
- * @note Detailed usage documentation will be added once all APIs are in place.
+ * @note Detailed usage documentation will be added once all APIs are in place. This reader will
+ * eventually move to `hybrid_scan.hpp` and the existing single-file reader (`hybrid_scan_reader`)
+ * will become its subclass. Only keeping this separate here for now to reduce noise.
  */
 class hybrid_scan_multifile {
  public:

From c9bf41923d20e12e958a7584994279b33acc1fdb Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 01:27:41 +0000
Subject: [PATCH 03/23] Clean up claude's comments

---
 .../cudf/io/experimental/hybrid_scan_multifile.hpp  | 13 -------------
 1 file changed, 13 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index 6b6418879e21..45ea96ab3628 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -86,11 +86,6 @@ class hybrid_scan_multifile {
   /**
    * @brief Get the per-source byte range of the page index in each Parquet file
    *
-   * The returned vector always has one entry per source. A source's entry is a
-   * default-constructed `byte_range_info{}` if that source has no row groups, no columns, or no
-   * page index offsets. A `CUDF_LOG_WARN` is emitted once if some sources have a page index and
-   * others do not.
-   *
    * @return Vector of page index byte ranges, one per source
    */
   [[nodiscard]] std::vector<byte_range_info> page_index_byte_range() const;
@@ -98,11 +93,6 @@ class hybrid_scan_multifile {
   /**
    * @brief Setup the per-source page index within each Parquet file metadata
    *
-   * Materializes `ColumnIndex` and `OffsetIndex` (page index) inside each source's
-   * `FileMetaData`. The input span size must equal the number of sources. A per-source empty
-   * span is skipped with a one-time warning. Sources whose corresponding span is non-empty must
-   * have row groups and valid page index offsets.
-   *
    * @param page_index_bytes Host span of Parquet page index buffer bytes, one per source
    */
   void setup_page_index(
@@ -111,9 +101,6 @@ class hybrid_scan_multifile {
   /**
    * @brief Get all available per-source row group indices from the parquet files
    *
-   * If `options.get_row_groups()` is non-empty, its size must equal the number of sources and it
-   * is returned as-is. Otherwise builds `[0 .. per_source_num_row_groups[i])` for each source.
-   *
    * @param options Parquet reader options
    * @return Vector of row group indices, one inner vector per source
    */

From 795f05849d2c677eb057a0856ac6bf8b379a88ab Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 19:48:29 +0000
Subject: [PATCH 04/23] Add gtests

---
 cpp/tests/CMakeLists.txt                      |   1 +
 .../hybrid_scan_multifile_filters_test.cpp    | 221 ++++++++++++++++++
 2 files changed, 222 insertions(+)
 create mode 100644 cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp

diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt
index ce7bfdbed0f0..e709a52ace20 100644
--- a/cpp/tests/CMakeLists.txt
+++ b/cpp/tests/CMakeLists.txt
@@ -349,6 +349,7 @@ ConfigureTest(
   HYBRID_SCAN_TEST
   io/experimental/hybrid_scan_composer.cpp
   io/experimental/hybrid_scan_filters_test.cpp
+  io/experimental/hybrid_scan_multifile_filters_test.cpp
   io/experimental/hybrid_scan_test.cpp
   io/parquet_common.cpp
   io/parquet_test.cpp
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
new file mode 100644
index 000000000000..0c62cc8b223c
--- /dev/null
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -0,0 +1,221 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "hybrid_scan_common.hpp"
+
+#include <cudf_test/base_fixture.hpp>
+
+#include <cudf/copying.hpp>
+#include <cudf/io/experimental/hybrid_scan_multifile.hpp>
+#include <cudf/io/parquet.hpp>
+#include <cudf/io/parquet_io_utils.hpp>
+#include <cudf/io/text/byte_range_info.hpp>
+#include <cudf/utilities/default_stream.hpp>
+#include <cudf/utilities/memory_resource.hpp>
+#include <cudf/utilities/span.hpp>
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+namespace {
+
+/**
+ * @brief Struct to hold multifile datasources, and footer buffers along with their byte spans
+ */
+struct multifile_inputs {
+  std::vector<std::unique_ptr<cudf::io::datasource>> datasources;
+  std::vector<std::unique_ptr<cudf::io::datasource::buffer>> footer_buffers;
+  std::vector<cudf::host_span<uint8_t const>> footer_byte_spans;
+};
+
+template <typename Buffers>
+multifile_inputs build_multifile_inputs(Buffers const& file_buffers)
+{
+  multifile_inputs out;
+  out.datasources.reserve(file_buffers.size());
+  out.footer_buffers.reserve(file_buffers.size());
+  out.footer_byte_spans.reserve(file_buffers.size());
+  for (auto const& buf : file_buffers) {
+    out.datasources.emplace_back(cudf::io::datasource::create(cudf::host_span<std::byte const>(
+      reinterpret_cast<std::byte const*>(buf.data()), buf.size())));
+    out.footer_buffers.emplace_back(
+      cudf::io::parquet::fetch_footer_to_host(*out.datasources.back()));
+    out.footer_byte_spans.emplace_back(*out.footer_buffers.back());
+  }
+  return out;
+}
+
+/**
+ * @brief Creates a parquet buffer with zero-rows and same schema as table from
+ * `create_parquet_with_stats`
+ */
+template <typename T>
+std::vector<char> create_empty_parquet_with_stats()
+{
+  auto const non_empty = std::get<0>(create_parquet_with_stats<T, 1>());
+  auto const empty     = cudf::empty_like(non_empty->view());
+
+  cudf::io::table_input_metadata output_metadata(empty->view());
+  output_metadata.column_metadata[0].set_name("col0");
+  output_metadata.column_metadata[1].set_name("col1");
+  output_metadata.column_metadata[2].set_name("col2");
+
+  std::vector<char> buffer;
+  auto out_opts =
+    cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&buffer}, empty->view())
+      .metadata(std::move(output_metadata))
+      .stats_level(cudf::io::statistics_freq::STATISTICS_COLUMN)
+      .build();
+  cudf::io::write_parquet(out_opts);
+  return buffer;
+}
+
+}  // namespace
+
+struct HybridScanMultifileFiltersTest : public cudf::test::BaseFixture {};
+
+TEST_F(HybridScanMultifileFiltersTest, Metadata)
+{
+  using T = cudf::timestamp_ms;
+
+  // Create two parquet sources, each with 4 row groups and 5000 rows per row
+  // group
+  auto constexpr rows_per_row_group = page_size_for_ordered_tests;
+  auto constexpr num_sources        = 2;
+
+  // Build sources with different seeds
+  std::vector<std::vector<char>> file_buffers;
+  file_buffers.reserve(num_sources);
+  auto constexpr num_concat = 1;
+  srand(0xbad);
+  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, num_concat>()));
+  srand(0xf00d);
+  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, num_concat>()));
+
+  // Filtering AST - col0 < 100
+  auto literal_value =
+    cudf::timestamp_scalar<T>(T(typename T::duration(100)), true, cudf::get_default_stream());
+  auto literal           = cudf::ast::literal(literal_value);
+  auto col_ref_0         = cudf::ast::column_name_reference("col0");
+  auto filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref_0, literal);
+
+  // Construct reader from footer bytes
+  auto inputs = build_multifile_inputs(file_buffers);
+
+  cudf::io::parquet_reader_options options =
+    cudf::io::parquet_reader_options::builder().filter(filter_expression).build();
+  auto const reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+    cudf::host_span<cudf::host_span<uint8_t const> const>{inputs.footer_byte_spans}, options);
+
+  // Get parquet metadata and check
+  auto parquet_metadata = reader->parquet_metadata();
+  ASSERT_EQ(parquet_metadata.size(), num_sources);
+  for (auto const& meta : parquet_metadata) {
+    ASSERT_FALSE(meta.row_groups.empty());
+    EXPECT_FALSE(meta.row_groups[0].columns[0].offset_index.has_value());
+    EXPECT_FALSE(meta.row_groups[0].columns[0].column_index.has_value());
+  }
+
+  // Setup page index
+  auto const page_index_byte_ranges = reader->page_index_byte_range();
+  ASSERT_EQ(page_index_byte_ranges.size(), num_sources);
+  EXPECT_TRUE(std::all_of(page_index_byte_ranges.begin(),
+                          page_index_byte_ranges.end(),
+                          [](auto const& range) { return not range.is_empty(); }));
+
+  std::vector<std::unique_ptr<cudf::io::datasource::buffer>> page_index_buffers;
+  std::vector<cudf::host_span<uint8_t const>> page_index_byte_spans;
+  page_index_buffers.reserve(num_sources);
+  page_index_byte_spans.reserve(num_sources);
+
+  auto iter = cuda::zip_iterator(page_index_byte_ranges.begin(), inputs.datasources.begin());
+  std::for_each(iter, iter + num_sources, [&](auto const& pair) {
+    auto const& pgidx_byte_range = cuda::std::get<0>(pair);
+    auto const& datasource       = cuda::std::get<1>(pair);
+    page_index_buffers.emplace_back(
+      cudf::io::parquet::fetch_page_index_to_host(*datasource, pgidx_byte_range));
+    page_index_byte_spans.emplace_back(*page_index_buffers.back());
+  });
+
+  reader->setup_page_index(
+    cudf::host_span<cudf::host_span<uint8_t const> const>{page_index_byte_spans});
+
+  // Check if page index is now present in each parquet metadata
+  parquet_metadata = reader->parquet_metadata();
+  for (auto const& meta : parquet_metadata) {
+    EXPECT_TRUE(meta.row_groups[0].columns[0].offset_index.has_value());
+    EXPECT_TRUE(meta.row_groups[0].columns[0].column_index.has_value());
+  }
+
+  // Check all row groups
+  auto input_row_group_indices = reader->all_row_groups(options);
+  ASSERT_EQ(input_row_group_indices.size(), num_sources);
+  EXPECT_TRUE(std::all_of(
+    input_row_group_indices.begin(), input_row_group_indices.end(), [](auto const& rgs) {
+      return rgs == (std::vector<cudf::size_type>{0, 1, 2, 3});
+    }));
+
+  // Set explicit row groups (per-source) via options
+  options.set_row_groups({{0, 1}, {2, 3}});
+  input_row_group_indices = reader->all_row_groups(options);
+
+  // Check if the row groups are set correctly
+  ASSERT_EQ(input_row_group_indices.size(), num_sources);
+  EXPECT_EQ(input_row_group_indices[0], (std::vector<cudf::size_type>{0, 1}));
+  EXPECT_EQ(input_row_group_indices[1], (std::vector<cudf::size_type>{2, 3}));
+  EXPECT_EQ(reader->total_rows_in_row_groups(input_row_group_indices),
+            2 * rows_per_row_group * num_sources);
+
+  // Construct a new reader from a span of existing FileMetaData
+  auto const reader_with_existing_metadata =
+    std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+      cudf::host_span<cudf::io::parquet::FileMetaData const>{parquet_metadata}, options);
+
+  // Check if the new metadata is the same as the existing one
+  auto const new_metadata = reader_with_existing_metadata->parquet_metadata();
+  ASSERT_EQ(new_metadata.size(), num_sources);
+  EXPECT_TRUE(std::all_of(new_metadata.begin(), new_metadata.end(), [&](auto const& meta) {
+    return meta.row_groups.size() == parquet_metadata.front().row_groups.size();
+  }));
+}
+
+TEST_F(HybridScanMultifileFiltersTest, EmptySource)
+{
+  using T = uint32_t;
+
+  srand(0xc0ffee);
+
+  // Create two parquet source. First one with non-zero rows and the second one with zero rows.
+  auto constexpr num_sources = 2;
+  std::vector<std::vector<char>> file_buffers;
+  file_buffers.reserve(num_sources);
+  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1>()));
+  file_buffers.emplace_back(create_empty_parquet_with_stats<T>());
+
+  auto inputs = build_multifile_inputs(file_buffers);
+
+  cudf::io::parquet_reader_options options = cudf::io::parquet_reader_options::builder().build();
+  auto const reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+    inputs.footer_byte_spans, options);
+
+  // Check parquet metadata
+  auto const parquet_metadata = reader->parquet_metadata();
+  ASSERT_EQ(parquet_metadata.size(), num_sources);
+  EXPECT_FALSE(parquet_metadata.front().row_groups.empty());
+  EXPECT_TRUE(parquet_metadata.back().row_groups.empty());
+
+  // Check row group indices
+  auto const all_rgs = reader->all_row_groups(options);
+  ASSERT_EQ(all_rgs.size(), num_sources);
+  EXPECT_EQ(all_rgs.front(), (std::vector<cudf::size_type>{0, 1, 2, 3}));
+  EXPECT_TRUE(all_rgs.back().empty());
+
+  // Check page index byte ranges
+  auto const page_index_byte_ranges = reader->page_index_byte_range();
+  ASSERT_EQ(page_index_byte_ranges.size(), num_sources);
+  EXPECT_FALSE(page_index_byte_ranges.front().is_empty());
+  EXPECT_TRUE(page_index_byte_ranges.back().is_empty());
+}

From f8b90ea171218fba4d46b09ae3332d977950db3f Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 13:53:22 -0700
Subject: [PATCH 05/23] Apply suggestions from code review

Co-authored-by: Yunsong Wang <12716979+PointKernel@users.noreply.github.com>
---
 cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp    | 4 ++--
 cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp       | 4 ++--
 .../io/experimental/hybrid_scan_multifile_filters_test.cpp    | 2 +-
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index 45ea96ab3628..f542173fc23d 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -19,7 +19,7 @@
 #include <memory>
 #include <vector>
 
-namespace CUDF_EXPORT cudf {
+namespace cudf {
 namespace io::parquet::experimental::detail {
 /**
  * @brief Internal experimental Parquet reader optimized for highly selective filters, called a
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
index 9a8879163782..a81aa827aebf 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
@@ -114,14 +114,14 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
    *
    * @return Vector of byte ranges of the page index, one per source
    */
-  [[nodiscard]] std::vector<text::byte_range_info> page_index_byte_range() const;
+  [[nodiscard]] std::vector<text::byte_range_info> page_index_byte_ranges() const;
 
   /**
    * @brief Get the Parquet file metadata for every source
    *
    * @return Vector of file metadata, one per source
    */
-  [[nodiscard]] std::vector<FileMetaData> parquet_metadata() const;
+  [[nodiscard]] std::vector<FileMetaData> parquet_metadatas() const;
 
   /**
    * @brief Setup and populate the page index structs in every source's `FileMetaData`
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index 0c62cc8b223c..bee1f7edfcc0 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 

From 90aef6178439d5225f88e9a9ae702339dd0abeda Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 13:53:53 -0700
Subject: [PATCH 06/23] Update
 cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp

Co-authored-by: Yunsong Wang <12716979+PointKernel@users.noreply.github.com>
---
 cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
index 32819903f530..97592add9ac6 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 

From 8876f5711fd067eda4391285ea7c4c09aea7f380 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 21:43:35 +0000
Subject: [PATCH 07/23] Apply suggestions from @PointKernel (thanks!)

---
 .../cudf/io/experimental/hybrid_scan.hpp       |  6 ++----
 .../io/experimental/hybrid_scan_multifile.hpp  | 18 ++++++++----------
 .../io/parquet/experimental/hybrid_scan.cpp    | 14 +++++++-------
 .../experimental/hybrid_scan_helpers.cpp       | 18 ++++++++----------
 .../experimental/hybrid_scan_helpers.hpp       |  2 +-
 .../parquet/experimental/hybrid_scan_impl.cpp  | 12 ++++++------
 .../parquet/experimental/hybrid_scan_impl.hpp  | 12 ++++++------
 .../experimental/hybrid_scan_multifile.cpp     | 12 ++++++------
 .../hybrid_scan_multifile_filters_test.cpp     | 17 ++++++++---------
 9 files changed, 52 insertions(+), 59 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan.hpp b/cpp/include/cudf/io/experimental/hybrid_scan.hpp
index 0f7ead8a3dfd..709acfee2804 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan.hpp
@@ -19,15 +19,13 @@
 #include <utility>
 #include <vector>
 
-namespace CUDF_EXPORT cudf {
-namespace io::parquet::experimental::detail {
+namespace cudf::io::parquet::experimental::detail {
 /**
  * @brief Internal experimental Parquet reader optimized for highly selective filters, called a
  *        Hybrid Scan operation.
  */
 class hybrid_scan_reader_impl;
-}  // namespace io::parquet::experimental::detail
-}  // namespace CUDF_EXPORT cudf
+}  // namespace cudf::io::parquet::experimental::detail
 
 //! Using `byte_range_info` from cudf::io::text
 using cudf::io::text::byte_range_info;
diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index f542173fc23d..70ff792660bf 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -19,15 +19,13 @@
 #include <memory>
 #include <vector>
 
-namespace cudf {
-namespace io::parquet::experimental::detail {
+namespace cudf::io::parquet::experimental::detail {
 /**
  * @brief Internal experimental Parquet reader optimized for highly selective filters, called a
  *        Hybrid Scan operation.
  */
 class hybrid_scan_reader_impl;
-}  // namespace io::parquet::experimental::detail
-}  // namespace CUDF_EXPORT cudf
+}  // namespace cudf::io::parquet::experimental::detail
 
 //! Using `byte_range_info` from cudf::io::text
 using cudf::io::text::byte_range_info;
@@ -77,25 +75,25 @@ class hybrid_scan_multifile {
   ~hybrid_scan_multifile();
 
   /**
-   * @brief Get the per-source Parquet file footer metadata
+   * @brief Get parquet metadatas for all sources
    *
-   * @return Vector of file metadata, one per source
+   * @return Vector of parquet metadata, one per source
    */
-  [[nodiscard]] std::vector<FileMetaData> parquet_metadata() const;
+  [[nodiscard]] std::vector<FileMetaData> parquet_metadatas() const;
 
   /**
-   * @brief Get the per-source byte range of the page index in each Parquet file
+   * @brief Get byte ranges of the page index for all sources
    *
    * @return Vector of page index byte ranges, one per source
    */
-  [[nodiscard]] std::vector<byte_range_info> page_index_byte_range() const;
+  [[nodiscard]] std::vector<byte_range_info> page_index_byte_ranges() const;
 
   /**
    * @brief Setup the per-source page index within each Parquet file metadata
    *
    * @param page_index_bytes Host span of Parquet page index buffer bytes, one per source
    */
-  void setup_page_index(
+  void setup_page_indexes(
     cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const;
 
   /**
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan.cpp b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
index 3223ca99c8f8..96ec82d30f18 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
@@ -15,28 +15,28 @@ namespace cudf::io::parquet::experimental {
 
 hybrid_scan_reader::hybrid_scan_reader(cudf::host_span<uint8_t const> footer_bytes,
                                        parquet_reader_options const& options)
+  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(
+      std::vector<cudf::host_span<uint8_t const>>{footer_bytes}, options)}
 {
-  auto const footers = std::vector<cudf::host_span<uint8_t const>>{footer_bytes};
-  _impl              = std::make_unique<detail::hybrid_scan_reader_impl>(footers, options);
 }
 
 hybrid_scan_reader::hybrid_scan_reader(FileMetaData const& parquet_metadata,
                                        parquet_reader_options const& options)
+  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(
+      std::vector<FileMetaData>{parquet_metadata}, options)}
 {
-  auto const metadatas = std::vector<FileMetaData>{parquet_metadata};
-  _impl                = std::make_unique<detail::hybrid_scan_reader_impl>(metadatas, options);
 }
 
 hybrid_scan_reader::~hybrid_scan_reader() = default;
 
 [[nodiscard]] text::byte_range_info hybrid_scan_reader::page_index_byte_range() const
 {
-  return _impl->page_index_byte_range().front();
+  return _impl->page_index_byte_ranges().front();
 }
 
 [[nodiscard]] FileMetaData hybrid_scan_reader::parquet_metadata() const
 {
-  return _impl->parquet_metadata().front();
+  return _impl->parquet_metadatas().front();
 }
 
 void hybrid_scan_reader::setup_page_index(cudf::host_span<uint8_t const> page_index_bytes) const
@@ -44,7 +44,7 @@ void hybrid_scan_reader::setup_page_index(cudf::host_span<uint8_t const> page_in
   CUDF_FUNC_RANGE();
 
   auto const per_source = std::vector<cudf::host_span<uint8_t const>>{page_index_bytes};
-  return _impl->setup_page_index(per_source);
+  return _impl->setup_page_indexes(per_source);
 }
 
 std::vector<cudf::size_type> hybrid_scan_reader::all_row_groups(
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
index e93054073637..4cebe6551da6 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
@@ -135,15 +135,15 @@ void aggregate_reader_metadata::initialize_internals(bool use_arrow_schema,
   }
 }
 
-std::vector<text::byte_range_info> aggregate_reader_metadata::page_index_byte_range() const
+std::vector<text::byte_range_info> aggregate_reader_metadata::page_index_byte_ranges() const
 {
   std::vector<text::byte_range_info> page_index_byte_ranges;
   page_index_byte_ranges.reserve(per_file_metadata.size());
   std::transform(per_file_metadata.begin(),
                  per_file_metadata.end(),
                  std::back_inserter(page_index_byte_ranges),
-                 [](auto const& pfm) -> text::byte_range_info {
-                   auto const& row_groups = pfm.row_groups;
+                 [](auto const& file_metadata) -> text::byte_range_info {
+                   auto const& row_groups = file_metadata.row_groups;
                    if (row_groups.empty() or row_groups.front().columns.empty()) { return {}; }
 
                    auto const min_offset = row_groups.front().columns.front().column_index_offset;
@@ -158,12 +158,12 @@ std::vector<text::byte_range_info> aggregate_reader_metadata::page_index_byte_ra
   return page_index_byte_ranges;
 }
 
-std::vector<FileMetaData> aggregate_reader_metadata::parquet_metadata() const
+std::vector<FileMetaData> aggregate_reader_metadata::parquet_metadatas() const
 {
   return {per_file_metadata.begin(), per_file_metadata.end()};
 }
 
-void aggregate_reader_metadata::setup_page_index(
+void aggregate_reader_metadata::setup_page_indexes(
   cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes)
 {
   CUDF_EXPECTS(page_index_bytes.size() == per_file_metadata.size(),
@@ -171,15 +171,13 @@ void aggregate_reader_metadata::setup_page_index(
 
   auto iter = cuda::zip_iterator(page_index_bytes.begin(), per_file_metadata.begin());
   std::for_each(iter, iter + page_index_bytes.size(), [&](auto const& pair) {
-    auto const& pgidx_bytes = cuda::std::get<0>(pair);
+    // Get the page index bytes and file metadata
+    auto const& [pgidx_bytes, file_metadata] = pair;
+    auto const& row_groups                   = file_metadata.row_groups;
 
     // Return early if empty page index buffer span
     if (pgidx_bytes.empty()) { return; }
 
-    // Get the file metadata and setup the page index
-    auto& file_metadata    = cuda::std::get<1>(pair);
-    auto const& row_groups = file_metadata.row_groups;
-
     // Check for empty parquet file
     CUDF_EXPECTS(not row_groups.empty() and not row_groups.front().columns.empty(),
                  "No column chunks in Parquet schema to read page index for");
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
index a81aa827aebf..2bc10699847e 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
@@ -128,7 +128,7 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
    *
    * @param page_index_bytes Host span of Parquet page index buffer bytes, one per source
    */
-  void setup_page_index(cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes);
+  void setup_page_indexes(cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes);
 
   /**
    * @brief Get all available row group indices, one inner vector per source
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
index 5ef78f19299e..eababea05f4b 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
@@ -86,20 +86,20 @@ hybrid_scan_reader_impl::hybrid_scan_reader_impl(
   _extended_metadata = static_cast<aggregate_reader_metadata*>(_metadata.get());
 }
 
-std::vector<FileMetaData> hybrid_scan_reader_impl::parquet_metadata() const
+std::vector<FileMetaData> hybrid_scan_reader_impl::parquet_metadatas() const
 {
-  return _extended_metadata->parquet_metadata();
+  return _extended_metadata->parquet_metadatas();
 }
 
-std::vector<byte_range_info> hybrid_scan_reader_impl::page_index_byte_range() const
+std::vector<byte_range_info> hybrid_scan_reader_impl::page_index_byte_ranges() const
 {
-  return _extended_metadata->page_index_byte_range();
+  return _extended_metadata->page_index_byte_ranges();
 }
 
-void hybrid_scan_reader_impl::setup_page_index(
+void hybrid_scan_reader_impl::setup_page_indexes(
   cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const
 {
-  _extended_metadata->setup_page_index(page_index_bytes);
+  _extended_metadata->setup_page_indexes(page_index_bytes);
 }
 
 void hybrid_scan_reader_impl::select_columns(read_columns_mode read_columns_mode,
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
index 0e85f066d347..21d727f51be3 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
@@ -58,19 +58,19 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
                                    parquet_reader_options const& options);
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan_multifile::parquet_metadata
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::parquet_metadatas
    */
-  [[nodiscard]] std::vector<FileMetaData> parquet_metadata() const;
+  [[nodiscard]] std::vector<FileMetaData> parquet_metadatas() const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan_multifile::page_index_byte_range
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::page_index_byte_ranges
    */
-  [[nodiscard]] std::vector<byte_range_info> page_index_byte_range() const;
+  [[nodiscard]] std::vector<byte_range_info> page_index_byte_ranges() const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan_multifile::setup_page_index
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::setup_page_indexes
    */
-  void setup_page_index(
+  void setup_page_indexes(
     cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const;
 
   /**
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
index 97592add9ac6..31b3cb5a6443 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
@@ -25,21 +25,21 @@ hybrid_scan_multifile::hybrid_scan_multifile(cudf::host_span<FileMetaData const>
 
 hybrid_scan_multifile::~hybrid_scan_multifile() = default;
 
-std::vector<FileMetaData> hybrid_scan_multifile::parquet_metadata() const
+std::vector<FileMetaData> hybrid_scan_multifile::parquet_metadatas() const
 {
-  return _impl->parquet_metadata();
+  return _impl->parquet_metadatas();
 }
 
-std::vector<text::byte_range_info> hybrid_scan_multifile::page_index_byte_range() const
+std::vector<text::byte_range_info> hybrid_scan_multifile::page_index_byte_ranges() const
 {
-  return _impl->page_index_byte_range();
+  return _impl->page_index_byte_ranges();
 }
 
-void hybrid_scan_multifile::setup_page_index(
+void hybrid_scan_multifile::setup_page_indexes(
   cudf::host_span<cudf::host_span<uint8_t const> const> page_index_bytes) const
 {
   CUDF_FUNC_RANGE();
-  _impl->setup_page_index(page_index_bytes);
+  _impl->setup_page_indexes(page_index_bytes);
 }
 
 std::vector<std::vector<size_type>> hybrid_scan_multifile::all_row_groups(
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index bee1f7edfcc0..1a0c135207f4 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -111,7 +111,7 @@ TEST_F(HybridScanMultifileFiltersTest, Metadata)
     cudf::host_span<cudf::host_span<uint8_t const> const>{inputs.footer_byte_spans}, options);
 
   // Get parquet metadata and check
-  auto parquet_metadata = reader->parquet_metadata();
+  auto parquet_metadata = reader->parquet_metadatas();
   ASSERT_EQ(parquet_metadata.size(), num_sources);
   for (auto const& meta : parquet_metadata) {
     ASSERT_FALSE(meta.row_groups.empty());
@@ -120,7 +120,7 @@ TEST_F(HybridScanMultifileFiltersTest, Metadata)
   }
 
   // Setup page index
-  auto const page_index_byte_ranges = reader->page_index_byte_range();
+  auto const page_index_byte_ranges = reader->page_index_byte_ranges();
   ASSERT_EQ(page_index_byte_ranges.size(), num_sources);
   EXPECT_TRUE(std::all_of(page_index_byte_ranges.begin(),
                           page_index_byte_ranges.end(),
@@ -133,18 +133,17 @@ TEST_F(HybridScanMultifileFiltersTest, Metadata)
 
   auto iter = cuda::zip_iterator(page_index_byte_ranges.begin(), inputs.datasources.begin());
   std::for_each(iter, iter + num_sources, [&](auto const& pair) {
-    auto const& pgidx_byte_range = cuda::std::get<0>(pair);
-    auto const& datasource       = cuda::std::get<1>(pair);
+    auto const& [pgidx_byte_range, datasource] = pair;
     page_index_buffers.emplace_back(
       cudf::io::parquet::fetch_page_index_to_host(*datasource, pgidx_byte_range));
     page_index_byte_spans.emplace_back(*page_index_buffers.back());
   });
 
-  reader->setup_page_index(
+  reader->setup_page_indexes(
     cudf::host_span<cudf::host_span<uint8_t const> const>{page_index_byte_spans});
 
   // Check if page index is now present in each parquet metadata
-  parquet_metadata = reader->parquet_metadata();
+  parquet_metadata = reader->parquet_metadatas();
   for (auto const& meta : parquet_metadata) {
     EXPECT_TRUE(meta.row_groups[0].columns[0].offset_index.has_value());
     EXPECT_TRUE(meta.row_groups[0].columns[0].column_index.has_value());
@@ -175,7 +174,7 @@ TEST_F(HybridScanMultifileFiltersTest, Metadata)
       cudf::host_span<cudf::io::parquet::FileMetaData const>{parquet_metadata}, options);
 
   // Check if the new metadata is the same as the existing one
-  auto const new_metadata = reader_with_existing_metadata->parquet_metadata();
+  auto const new_metadata = reader_with_existing_metadata->parquet_metadatas();
   ASSERT_EQ(new_metadata.size(), num_sources);
   EXPECT_TRUE(std::all_of(new_metadata.begin(), new_metadata.end(), [&](auto const& meta) {
     return meta.row_groups.size() == parquet_metadata.front().row_groups.size();
@@ -202,7 +201,7 @@ TEST_F(HybridScanMultifileFiltersTest, EmptySource)
     inputs.footer_byte_spans, options);
 
   // Check parquet metadata
-  auto const parquet_metadata = reader->parquet_metadata();
+  auto const parquet_metadata = reader->parquet_metadatas();
   ASSERT_EQ(parquet_metadata.size(), num_sources);
   EXPECT_FALSE(parquet_metadata.front().row_groups.empty());
   EXPECT_TRUE(parquet_metadata.back().row_groups.empty());
@@ -214,7 +213,7 @@ TEST_F(HybridScanMultifileFiltersTest, EmptySource)
   EXPECT_TRUE(all_rgs.back().empty());
 
   // Check page index byte ranges
-  auto const page_index_byte_ranges = reader->page_index_byte_range();
+  auto const page_index_byte_ranges = reader->page_index_byte_ranges();
   ASSERT_EQ(page_index_byte_ranges.size(), num_sources);
   EXPECT_FALSE(page_index_byte_ranges.front().is_empty());
   EXPECT_TRUE(page_index_byte_ranges.back().is_empty());

From a5752762ca11682482f7830bd9e95833a0c934d4 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 21:47:35 +0000
Subject: [PATCH 08/23] Minor

---
 cpp/src/io/parquet/experimental/hybrid_scan.cpp | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/cpp/src/io/parquet/experimental/hybrid_scan.cpp b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
index 96ec82d30f18..3bc68cb97752 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
@@ -42,9 +42,7 @@ hybrid_scan_reader::~hybrid_scan_reader() = default;
 void hybrid_scan_reader::setup_page_index(cudf::host_span<uint8_t const> page_index_bytes) const
 {
   CUDF_FUNC_RANGE();
-
-  auto const per_source = std::vector<cudf::host_span<uint8_t const>>{page_index_bytes};
-  return _impl->setup_page_indexes(per_source);
+  return _impl->setup_page_indexes(std::vector<cudf::host_span<uint8_t const>>{page_index_bytes});
 }
 
 std::vector<cudf::size_type> hybrid_scan_reader::all_row_groups(

From c7d7cb6da3bf534f9def672c02ce719d51b0631a Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 21:51:32 +0000
Subject: [PATCH 09/23] Minor changes

---
 .../experimental/hybrid_scan_helpers.cpp      | 29 ++++++++++---------
 1 file changed, 16 insertions(+), 13 deletions(-)

diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
index 4cebe6551da6..0d29fdfd358c 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
@@ -202,7 +202,7 @@ std::vector<std::vector<size_type>> aggregate_reader_metadata::all_row_groups(
   auto const& opts_row_groups = options.get_row_groups();
   if (not opts_row_groups.empty()) {
     CUDF_EXPECTS(opts_row_groups.size() == per_file_metadata.size(),
-                 "Row groups in parquet reader options must have one inner vector per source");
+                 "Row groups in parquet reader options must specify one vector per data source");
     return opts_row_groups;
   }
 
@@ -224,18 +224,21 @@ size_type aggregate_reader_metadata::total_rows_in_row_groups(
 {
   std::size_t total_rows = 0;
 
-  std::for_each(cuda::counting_iterator<std::size_t>{0},
-                cuda::counting_iterator{row_group_indices.size()},
-                [&](auto const src_idx) {
-                  auto const& pfm = per_file_metadata[src_idx];
-                  for (auto const row_group_idx : row_group_indices[src_idx]) {
-                    CUDF_EXPECTS(std::cmp_less(row_group_idx, pfm.row_groups.size()),
-                                 "Row group index out of bounds");
-                    total_rows += pfm.row_groups[row_group_idx].num_rows;
-                  }
-                });
-  CUDF_EXPECTS(std::cmp_less_equal(total_rows, std::numeric_limits<size_type>::max()),
-               "Total number of rows exceeds cudf::size_type's limit");
+  std::for_each(
+    cuda::counting_iterator<std::size_t>{0},
+    cuda::counting_iterator{row_group_indices.size()},
+    [&](auto const src_idx) {
+      auto const& pfm = per_file_metadata[src_idx];
+      for (auto const row_group_idx : row_group_indices[src_idx]) {
+        CUDF_EXPECTS(
+          std::cmp_greater_equal(row_group_idx, size_type{0}) and
+            std::cmp_less(row_group_idx, pfm.row_groups.size()),
+          "Encountered out-of-bounds row group index for data source. Row group index: " +
+            std::to_string(row_group_idx) + ", Source index: " + std::to_string(src_idx) +
+            ", Number of row groups: " + std::to_string(pfm.row_groups.size()));
+        total_rows += pfm.row_groups[row_group_idx].num_rows;
+      }
+    });
 
   return static_cast<size_type>(total_rows);
 }

From 467a628f3fe6176e445ecad6266ac6abb527124d Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 21:53:00 +0000
Subject: [PATCH 10/23] Allow more than 2B rows

---
 cpp/include/cudf/io/experimental/hybrid_scan.hpp        | 2 +-
 cpp/src/io/parquet/experimental/hybrid_scan.cpp         | 2 +-
 cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp | 4 ++--
 cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp | 2 +-
 cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp    | 2 +-
 cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp    | 2 +-
 6 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan.hpp b/cpp/include/cudf/io/experimental/hybrid_scan.hpp
index 709acfee2804..980ab9644d3b 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan.hpp
@@ -342,7 +342,7 @@ class hybrid_scan_reader {
    * @param row_group_indices Input row groups indices
    * @return Total number of top-level rows in the row groups
    */
-  [[nodiscard]] size_type total_rows_in_row_groups(
+  [[nodiscard]] std::size_t total_rows_in_row_groups(
     cudf::host_span<size_type const> row_group_indices) const;
 
   /**
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan.cpp b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
index 3bc68cb97752..868813a1b4ed 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan.cpp
@@ -57,7 +57,7 @@ std::vector<cudf::size_type> hybrid_scan_reader::all_row_groups(
   return _impl->all_row_groups(options).front();
 }
 
-size_type hybrid_scan_reader::total_rows_in_row_groups(
+std::size_t hybrid_scan_reader::total_rows_in_row_groups(
   cudf::host_span<size_type const> row_group_indices) const
 {
   if (row_group_indices.empty()) { return 0; }
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
index 0d29fdfd358c..6e12055dacae 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
@@ -219,7 +219,7 @@ std::vector<std::vector<size_type>> aggregate_reader_metadata::all_row_groups(
   return row_groups;
 }
 
-size_type aggregate_reader_metadata::total_rows_in_row_groups(
+std::size_t aggregate_reader_metadata::total_rows_in_row_groups(
   cudf::host_span<std::vector<size_type> const> row_group_indices) const
 {
   std::size_t total_rows = 0;
@@ -240,7 +240,7 @@ size_type aggregate_reader_metadata::total_rows_in_row_groups(
       }
     });
 
-  return static_cast<size_type>(total_rows);
+  return total_rows;
 }
 
 std::tuple<std::vector<input_column_info>,
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
index 2bc10699847e..e65db678c2d1 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp
@@ -149,7 +149,7 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base {
    * @param row_group_indices Input row groups indices
    * @return Total number of top-level rows in the row groups
    */
-  [[nodiscard]] size_type total_rows_in_row_groups(
+  [[nodiscard]] std::size_t total_rows_in_row_groups(
     cudf::host_span<std::vector<size_type> const> row_group_indices) const;
 
   /**
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
index eababea05f4b..f2919c64519f 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
@@ -206,7 +206,7 @@ std::vector<std::vector<size_type>> hybrid_scan_reader_impl::all_row_groups(
   return _extended_metadata->all_row_groups(options);
 }
 
-size_type hybrid_scan_reader_impl::total_rows_in_row_groups(
+std::size_t hybrid_scan_reader_impl::total_rows_in_row_groups(
   cudf::host_span<std::vector<size_type> const> row_group_indices) const
 {
   return _extended_metadata->total_rows_in_row_groups(row_group_indices);
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
index 21d727f51be3..64eead4463f7 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
@@ -82,7 +82,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
   /**
    * @copydoc cudf::io::experimental::hybrid_scan::total_rows_in_row_groups
    */
-  [[nodiscard]] size_type total_rows_in_row_groups(
+  [[nodiscard]] std::size_t total_rows_in_row_groups(
     cudf::host_span<std::vector<size_type> const> row_group_indices) const;
 
   /**

From 1909146d843f1d376915e03bbe00629710b87f16 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 22:29:44 +0000
Subject: [PATCH 11/23] Minor bug fix

---
 cpp/src/io/parquet/experimental/page_index_filter.cu | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/cpp/src/io/parquet/experimental/page_index_filter.cu b/cpp/src/io/parquet/experimental/page_index_filter.cu
index 21ae60443f3d..7ab305174d3e 100644
--- a/cpp/src/io/parquet/experimental/page_index_filter.cu
+++ b/cpp/src/io/parquet/experimental/page_index_filter.cu
@@ -994,8 +994,11 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
                "Input row bitmask should be of type BOOL8");
 
   auto const total_rows = total_rows_in_row_groups(row_group_indices);
+  CUDF_EXPECTS(std::cmp_less_equal(total_rows, std::numeric_limits<cudf::size_type>::max()),
+               "Total rows in row groups exceed the cudf column size limit",
+               std::overflow_error);
 
-  CUDF_EXPECTS(row_mask_offset + total_rows <= row_mask.size(),
+  CUDF_EXPECTS(std::cmp_less_equal(row_mask_offset + total_rows, row_mask.size()),
                "Mismatch in total rows in input row mask and row groups",
                std::invalid_argument);
 
@@ -1091,8 +1094,8 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
   if constexpr (cuda::std::is_same_v<ColumnView, cudf::mutable_column_view>) {
     if (row_mask.nullable() and row_mask.null_count() > 0) {
       thrust::for_each(rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()),
-                       cuda::counting_iterator{row_mask_offset},
-                       cuda::counting_iterator{row_mask_offset + total_rows},
+                       cuda::counting_iterator<cudf::size_type>(row_mask_offset),
+                       cuda::counting_iterator<cudf::size_type>(row_mask_offset + total_rows),
                        [row_mask  = row_mask.template begin<bool>(),
                         null_mask = row_mask.null_mask()] __device__(auto const row_idx) {
                          if (not bit_is_set(null_mask, row_idx)) { row_mask[row_idx] = true; }
@@ -1126,7 +1129,7 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
     cudf::detail::make_device_uvector_async(host_tree_level_ptrs, stream, mr);
 
   // Build Fenwick tree levels (zeroth level is just the row mask itself)
-  auto prev_level_size = total_rows;
+  auto prev_level_size = static_cast<cudf::size_type>(total_rows);
   std::for_each(
     cuda::counting_iterator<cudf::size_type>{0},
     cuda::counting_iterator{num_levels - 1},

From 17fd247d065134b1eeefa3c8358c380ca840c4ba Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 21 May 2026 22:31:14 +0000
Subject: [PATCH 12/23] Minor

---
 cpp/src/io/parquet/experimental/page_index_filter.cu | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/cpp/src/io/parquet/experimental/page_index_filter.cu b/cpp/src/io/parquet/experimental/page_index_filter.cu
index 7ab305174d3e..0b2aec183b39 100644
--- a/cpp/src/io/parquet/experimental/page_index_filter.cu
+++ b/cpp/src/io/parquet/experimental/page_index_filter.cu
@@ -1003,7 +1003,8 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
                std::invalid_argument);
 
   // Return an empty vector if all rows are invalid or all rows are required
-  if (row_mask.null_count(row_mask_offset, row_mask_offset + total_rows, stream) == total_rows or
+  if (std::cmp_equal(row_mask.null_count(row_mask_offset, row_mask_offset + total_rows, stream),
+                     total_rows) or
       cudf::detail::all_of(row_mask.template begin<bool>() + row_mask_offset,
                            row_mask.template begin<bool>() + row_mask_offset + total_rows,
                            cuda::std::identity{},

From e013fda05e60098a1f1adb4925b1ff47bcea90fa Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Fri, 29 May 2026 19:25:16 +0000
Subject: [PATCH 13/23] Add multifile row group filtering with stats and byte
 ranges

---
 .../io/experimental/hybrid_scan_multifile.hpp | 44 ++++++++++
 .../experimental/hybrid_scan_multifile.cpp    | 26 ++++++
 .../hybrid_scan_multifile_filters_test.cpp    | 83 +++++++++++++++++++
 3 files changed, 153 insertions(+)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index 70ff792660bf..d3c0d5422dbb 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -123,6 +123,50 @@ class hybrid_scan_multifile {
    */
   void reset_column_selection() const;
 
+  /**
+   * @brief Filter the row groups using the byte range specified by [`bytes_to_skip`, `bytes_to_skip
+   * + bytes_to_read`)
+   *
+   * Filters the row groups such that only the row groups that start within the byte range are
+   * selected. Note that the last selected row group may end beyond the byte range.
+   *
+   * @param row_group_indices Input row groups indices
+   * @param options Parquet reader options
+   * @return Filtered row group indices
+   */
+  [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_byte_range(
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    parquet_reader_options const& options) const;
+
+  /**
+   * @brief Filter the input row groups using column chunk statistics
+   *
+   * @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
+   * @return Filtered row group indices
+   */
+  [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_stats(
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    parquet_reader_options const& options,
+    rmm::cuda_stream_view stream) const;
+
+  /**
+   * @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 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<std::vector<size_type> const> row_group_indices,
+                                parquet_reader_options const& options) const;
+
  private:
   std::unique_ptr<detail::hybrid_scan_reader_impl> _impl;
 };
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
index 31b3cb5a6443..59347367d223 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
@@ -57,4 +57,30 @@ size_type hybrid_scan_multifile::total_rows_in_row_groups(
 
 void hybrid_scan_multifile::reset_column_selection() const { _impl->reset_column_selection(); }
 
+std::vector<std::vector<size_type>> hybrid_scan_multifile::filter_row_groups_with_byte_range(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  parquet_reader_options const& options) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->filter_row_groups_with_byte_range(row_group_indices, options);
+}
+
+std::vector<std::vector<size_type>> hybrid_scan_multifile::filter_row_groups_with_stats(
+  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();
+  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,
+  parquet_reader_options const& options) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->secondary_filters_byte_ranges(row_group_indices, options);
+}
+
 }  // namespace cudf::io::parquet::experimental
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index 1a0c135207f4..fb291fd94bc5 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -7,11 +7,13 @@
 
 #include <cudf_test/base_fixture.hpp>
 
+#include <cudf/ast/expressions.hpp>
 #include <cudf/copying.hpp>
 #include <cudf/io/experimental/hybrid_scan_multifile.hpp>
 #include <cudf/io/parquet.hpp>
 #include <cudf/io/parquet_io_utils.hpp>
 #include <cudf/io/text/byte_range_info.hpp>
+#include <cudf/scalar/scalar.hpp>
 #include <cudf/utilities/default_stream.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 #include <cudf/utilities/span.hpp>
@@ -218,3 +220,84 @@ TEST_F(HybridScanMultifileFiltersTest, EmptySource)
   EXPECT_FALSE(page_index_byte_ranges.front().is_empty());
   EXPECT_TRUE(page_index_byte_ranges.back().is_empty());
 }
+
+TEST_F(HybridScanMultifileFiltersTest, ErrorFilterRowGroupsWithByteRanges)
+{
+  using T                    = uint32_t;
+  auto constexpr num_sources = 2;
+  srand(0xb47e);
+
+  std::vector<std::vector<char>> file_buffers;
+  file_buffers.reserve(num_sources);
+  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1>()));
+  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1>()));
+
+  auto inputs = build_multifile_inputs(file_buffers);
+
+  auto const options = cudf::io::parquet_reader_options::builder().skip_bytes(1000).build();
+  auto const reader  = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+    inputs.footer_byte_spans, options);
+
+  auto const row_group_indices = reader->all_row_groups(options);
+  ASSERT_EQ(row_group_indices.size(), num_sources);
+
+  EXPECT_THROW(std::ignore = reader->filter_row_groups_with_byte_range(row_group_indices, options),
+               std::invalid_argument);
+}
+
+TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats)
+{
+  using T                           = cudf::duration_ms;
+  auto constexpr num_sources        = 2;
+  auto constexpr rows_per_row_group = page_size_for_ordered_tests;
+
+  // Two sources, each with 4 row groups and ascending strings in col2
+  std::vector<std::vector<char>> file_buffers;
+  file_buffers.reserve(num_sources);
+  srand(0xc001);
+  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1, false>()));
+  srand(0xbeef);
+  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1, false>()));
+
+  auto inputs = build_multifile_inputs(file_buffers);
+
+  // Filter - col0 < 50 and col2 < "000010000"
+  auto literal_value0 = cudf::duration_scalar<T>(T::rep(50), true, cudf::get_default_stream());
+  auto literal0       = cudf::ast::literal(literal_value0);
+  auto col_ref0       = cudf::ast::column_reference(0);
+  auto filter1        = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref0, literal0);
+
+  auto literal_value2 = cudf::string_scalar("000010000", true, cudf::get_default_stream());
+  auto literal2       = cudf::ast::literal(literal_value2);
+  auto col_ref2       = cudf::ast::column_reference(2);
+  auto filter2        = cudf::ast::operation(cudf::ast::ast_operator::GREATER, literal2, col_ref2);
+
+  auto filter_expression =
+    cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_AND, filter1, filter2);
+
+  auto options      = cudf::io::parquet_reader_options::builder().filter(filter_expression).build();
+  auto const reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+    inputs.footer_byte_spans, options);
+
+  // Each source has 4 row groups (20000 rows / 5000 rows per row group)
+  auto input_row_group_indices = reader->all_row_groups(options);
+  ASSERT_EQ(input_row_group_indices.size(), num_sources);
+  EXPECT_EQ(reader->total_rows_in_row_groups(input_row_group_indices),
+            num_sources * 4 * rows_per_row_group);
+
+  // Each source prunes down to a single surviving row group
+  auto stats_filtered = reader->filter_row_groups_with_stats(
+    input_row_group_indices, options, cudf::get_default_stream());
+  ASSERT_EQ(stats_filtered.size(), num_sources);
+  EXPECT_TRUE(std::all_of(
+    stats_filtered.begin(), stats_filtered.end(), [](auto const& rgs) { return rgs.size() == 1; }));
+  EXPECT_EQ(reader->total_rows_in_row_groups(stats_filtered), num_sources * rows_per_row_group);
+
+  // Custom per-source indices that prune all row groups via stats, including an empty source
+  input_row_group_indices = {{1, 2}, {}};
+  stats_filtered          = reader->filter_row_groups_with_stats(
+    input_row_group_indices, options, cudf::get_default_stream());
+  ASSERT_EQ(stats_filtered.size(), num_sources);
+  EXPECT_TRUE(stats_filtered.front().empty());
+  EXPECT_TRUE(stats_filtered.back().empty());
+}

From c164897d5b305127fd2ab2e9904d1c620f444c9a Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Tue, 2 Jun 2026 01:34:15 +0000
Subject: [PATCH 14/23] Revert unneeded changes

---
 cpp/src/io/parquet/experimental/page_index_filter.cu           | 3 ---
 .../io/experimental/hybrid_scan_multifile_filters_test.cpp     | 2 +-
 2 files changed, 1 insertion(+), 4 deletions(-)

diff --git a/cpp/src/io/parquet/experimental/page_index_filter.cu b/cpp/src/io/parquet/experimental/page_index_filter.cu
index 4ea15ac9a1a3..bb2b89c56c7e 100644
--- a/cpp/src/io/parquet/experimental/page_index_filter.cu
+++ b/cpp/src/io/parquet/experimental/page_index_filter.cu
@@ -994,9 +994,6 @@ thrust::host_vector<bool> aggregate_reader_metadata::compute_data_page_mask(
                "Input row bitmask should be of type BOOL8");
 
   auto const total_rows = total_rows_in_row_groups(row_group_indices);
-  CUDF_EXPECTS(std::cmp_less_equal(total_rows, std::numeric_limits<cudf::size_type>::max()),
-               "Total rows in row groups exceed the cudf column size limit",
-               std::overflow_error);
 
   CUDF_EXPECTS(
     std::cmp_less_equal(static_cast<std::size_t>(row_mask_offset) + total_rows, row_mask.size()),
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index a9f7ab0d5ce2..1b2d0a8381dc 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -7,12 +7,12 @@
 
 #include <cudf_test/base_fixture.hpp>
 
+#include <cudf/ast/expressions.hpp>
 #include <cudf/copying.hpp>
 #include <cudf/io/experimental/hybrid_scan_multifile.hpp>
 #include <cudf/io/parquet.hpp>
 #include <cudf/io/parquet_io_utils.hpp>
 #include <cudf/io/text/byte_range_info.hpp>
-#include <cudf/scalar/scalar.hpp>
 #include <cudf/utilities/default_stream.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 #include <cudf/utilities/span.hpp>

From 7a35f6ee9ec76c43bcd9c85b875c228d17be6ac2 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Tue, 2 Jun 2026 01:42:42 +0000
Subject: [PATCH 15/23] Style fix

---
 cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index d3c0d5422dbb..f5f7b556587c 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -124,8 +124,8 @@ class hybrid_scan_multifile {
   void reset_column_selection() const;
 
   /**
-   * @brief Filter the row groups using the byte range specified by [`bytes_to_skip`, `bytes_to_skip
-   * + bytes_to_read`)
+   * @brief Filter the row groups using the byte range specified by [`bytes_to_skip`,
+   * `bytes_to_skip + bytes_to_read`)
    *
    * Filters the row groups such that only the row groups that start within the byte range are
    * selected. Note that the last selected row group may end beyond the byte range.

From 8c7b183cc9f95ba308f37440120a0e9d263e54bc Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Tue, 2 Jun 2026 02:04:53 +0000
Subject: [PATCH 16/23] Address comments

---
 .../io/parquet/experimental/hybrid_scan_helpers.cpp   | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
index 626ac249b1bd..890dc42eeee5 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
@@ -20,6 +20,7 @@
 #include <numeric>
 #include <optional>
 #include <unordered_set>
+#include <utility>
 
 namespace cudf::io::parquet::experimental::detail {
 
@@ -202,6 +203,16 @@ std::vector<std::vector<size_type>> aggregate_reader_metadata::all_row_groups(
   if (not opts_row_groups.empty()) {
     CUDF_EXPECTS(opts_row_groups.size() == per_file_metadata.size(),
                  "Row groups in parquet reader options must specify one vector per data source");
+    auto iter = cuda::zip_iterator(opts_row_groups.begin(), per_file_metadata.begin());
+    std::for_each(iter, iter + opts_row_groups.size(), [&](auto const& pair) {
+      auto const& [opts_row_groups, file_metadata] = pair;
+      auto const& row_groups                       = file_metadata.row_groups;
+      for (auto const rg_idx : opts_row_groups) {
+        CUDF_EXPECTS(rg_idx >= 0 and std::cmp_less(rg_idx, row_groups.size()),
+                     "Encountered out-of-bounds row group index for data source",
+                     std::invalid_argument);
+      }
+    });
     return opts_row_groups;
   }
 

From 140e90dfb2ae331ce3fe5d5923f21cf01ce2e62c Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Tue, 2 Jun 2026 21:13:52 +0000
Subject: [PATCH 17/23] Address review comments

---
 .../io/experimental/hybrid_scan_multifile.hpp | 16 ++++++----------
 .../experimental/hybrid_scan_helpers.cpp      |  4 ++--
 .../parquet/experimental/hybrid_scan_impl.hpp |  8 ++++----
 .../hybrid_scan_multifile_filters_test.cpp    | 19 ++++++++++++++-----
 4 files changed, 26 insertions(+), 21 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index f5f7b556587c..ef6e0eadd613 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -115,11 +115,7 @@ class hybrid_scan_multifile {
     cudf::host_span<std::vector<size_type> const> row_group_indices) const;
 
   /**
-   * @brief Resets the current column selection
-   *
-   * Resets the current column selection state forcing column re-selection in subsequent filter,
-   * byte range, setup chunking and materialization APIs. This is useful if the filter expression
-   * has been cascaded (and-ed) to include new columns.
+   * @copydoc cudf::io::experimental::hybrid_scan::reset_column_selection
    */
   void reset_column_selection() const;
 
@@ -130,9 +126,9 @@ class hybrid_scan_multifile {
    * Filters the row groups such that only the row groups that start within the byte range are
    * selected. Note that the last selected row group may end beyond the byte range.
    *
-   * @param row_group_indices Input row groups indices
+   * @param row_group_indices Input row group indices, one per source
    * @param options Parquet reader options
-   * @return Filtered row group indices
+   * @return Filtered per-source row group indices (one inner vector per source)
    */
   [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_byte_range(
     cudf::host_span<std::vector<size_type> const> row_group_indices,
@@ -141,10 +137,10 @@ class hybrid_scan_multifile {
   /**
    * @brief Filter the input row groups using column chunk statistics
    *
-   * @param row_group_indices Input row groups indices
+   * @param row_group_indices 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 Filtered row group indices
+   * @return Filtered  row group indices, one per source
    */
   [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_stats(
     cudf::host_span<std::vector<size_type> const> row_group_indices,
@@ -158,7 +154,7 @@ class hybrid_scan_multifile {
    * @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 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
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
index 890dc42eeee5..4386cd1dea42 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp
@@ -205,9 +205,9 @@ std::vector<std::vector<size_type>> aggregate_reader_metadata::all_row_groups(
                  "Row groups in parquet reader options must specify one vector per data source");
     auto iter = cuda::zip_iterator(opts_row_groups.begin(), per_file_metadata.begin());
     std::for_each(iter, iter + opts_row_groups.size(), [&](auto const& pair) {
-      auto const& [opts_row_groups, file_metadata] = pair;
+      auto const& [file_row_groups, file_metadata] = pair;
       auto const& row_groups                       = file_metadata.row_groups;
-      for (auto const rg_idx : opts_row_groups) {
+      for (auto const rg_idx : file_row_groups) {
         CUDF_EXPECTS(rg_idx >= 0 and std::cmp_less(rg_idx, row_groups.size()),
                      "Encountered out-of-bounds row group index for data source",
                      std::invalid_argument);
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
index 64eead4463f7..4dced0a168ad 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
@@ -80,7 +80,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
     parquet_reader_options const& options) const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::total_rows_in_row_groups
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::total_rows_in_row_groups
    */
   [[nodiscard]] std::size_t total_rows_in_row_groups(
     cudf::host_span<std::vector<size_type> const> row_group_indices) const;
@@ -91,14 +91,14 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
   void reset_column_selection();
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::filter_row_groups_with_byte_range
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::filter_row_groups_with_byte_range
    */
   [[nodiscard]] std::vector<std::vector<cudf::size_type>> filter_row_groups_with_byte_range(
     cudf::host_span<std::vector<size_type> const> row_group_indices,
     parquet_reader_options const& options) const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::filter_row_groups_with_stats
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::filter_row_groups_with_stats
    */
   [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_stats(
     cudf::host_span<std::vector<size_type> const> row_group_indices,
@@ -106,7 +106,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
     rmm::cuda_stream_view stream);
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::secondary_filters_byte_ranges
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::secondary_filters_byte_ranges
    */
   [[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,
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index 1b2d0a8381dc..73c0807779aa 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -233,13 +233,21 @@ TEST_F(HybridScanMultifileFiltersTest, ErrorFilterRowGroupsWithByteRanges)
 
   auto inputs = build_multifile_inputs(file_buffers);
 
-  auto const options = cudf::io::parquet_reader_options::builder().skip_bytes(1000).build();
-  auto const reader  = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+  auto options      = cudf::io::parquet_reader_options::builder().build();
+  auto const reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
     inputs.footer_byte_spans, options);
 
   auto const row_group_indices = reader->all_row_groups(options);
   ASSERT_EQ(row_group_indices.size(), num_sources);
 
+  // Setting `skip_bytes` or `num_bytes` is ambiguous when reading multiple sources. The reader is
+  // expected to throw an exception if row groups are filtered using byte range in this case.
+  options.set_skip_bytes(1000);
+  EXPECT_THROW(std::ignore = reader->filter_row_groups_with_byte_range(row_group_indices, options),
+               std::invalid_argument);
+
+  options.set_skip_bytes(0);
+  options.set_num_bytes(1000);
   EXPECT_THROW(std::ignore = reader->filter_row_groups_with_byte_range(row_group_indices, options),
                std::invalid_argument);
 }
@@ -260,7 +268,7 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats)
 
   auto inputs = build_multifile_inputs(file_buffers);
 
-  // Filter - col0 < 50 and col2 < "000010000"
+  // Filter - col0 < 50 and col2 > "000010000"
   auto literal_value0 = cudf::duration_scalar<T>(T::rep(50), true, cudf::get_default_stream());
   auto literal0       = cudf::ast::literal(literal_value0);
   auto col_ref0       = cudf::ast::column_reference(0);
@@ -288,8 +296,9 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats)
   auto stats_filtered = reader->filter_row_groups_with_stats(
     input_row_group_indices, options, cudf::get_default_stream());
   ASSERT_EQ(stats_filtered.size(), num_sources);
-  EXPECT_TRUE(std::all_of(
-    stats_filtered.begin(), stats_filtered.end(), [](auto const& rgs) { return rgs.size() == 1; }));
+  for (std::size_t i = 0; i < stats_filtered.size(); ++i) {
+    EXPECT_EQ(stats_filtered[i].size(), 1) << "Source index: " << i;
+  }
   EXPECT_EQ(reader->total_rows_in_row_groups(stats_filtered), num_sources * rows_per_row_group);
 
   // Custom per-source indices that prune all row groups via stats, including an empty source

From 9fcbb0a484b3a81f1066d61c6a5212af2f610ce5 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Wed, 3 Jun 2026 02:49:45 +0000
Subject: [PATCH 18/23] Apply suggestions

---
 .../io/experimental/hybrid_scan_multifile.hpp |  2 +-
 .../hybrid_scan_multifile_filters_test.cpp    | 35 +++++++++++--------
 2 files changed, 21 insertions(+), 16 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index ef6e0eadd613..f7b777703e78 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -140,7 +140,7 @@ class hybrid_scan_multifile {
    * @param row_group_indices 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 Filtered  row group indices, one per source
+   * @return Filtered row group indices, one per source
    */
   [[nodiscard]] std::vector<std::vector<size_type>> filter_row_groups_with_stats(
     cudf::host_span<std::vector<size_type> const> row_group_indices,
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index 73c0807779aa..29c66093b1cf 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -233,23 +233,28 @@ TEST_F(HybridScanMultifileFiltersTest, ErrorFilterRowGroupsWithByteRanges)
 
   auto inputs = build_multifile_inputs(file_buffers);
 
-  auto options      = cudf::io::parquet_reader_options::builder().build();
-  auto const reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
-    inputs.footer_byte_spans, options);
-
-  auto const row_group_indices = reader->all_row_groups(options);
-  ASSERT_EQ(row_group_indices.size(), num_sources);
-
   // Setting `skip_bytes` or `num_bytes` is ambiguous when reading multiple sources. The reader is
   // expected to throw an exception if row groups are filtered using byte range in this case.
-  options.set_skip_bytes(1000);
-  EXPECT_THROW(std::ignore = reader->filter_row_groups_with_byte_range(row_group_indices, options),
-               std::invalid_argument);
-
-  options.set_skip_bytes(0);
-  options.set_num_bytes(1000);
-  EXPECT_THROW(std::ignore = reader->filter_row_groups_with_byte_range(row_group_indices, options),
-               std::invalid_argument);
+  {
+    auto const options = cudf::io::parquet_reader_options::builder().skip_bytes(1000).build();
+    auto const reader  = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+      inputs.footer_byte_spans, options);
+    auto const row_group_indices = reader->all_row_groups(options);
+    ASSERT_EQ(row_group_indices.size(), num_sources);
+    EXPECT_THROW(
+      std::ignore = reader->filter_row_groups_with_byte_range(row_group_indices, options),
+      std::invalid_argument);
+  }
+  {
+    auto const options = cudf::io::parquet_reader_options::builder().num_bytes(1000).build();
+    auto const reader  = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+      inputs.footer_byte_spans, options);
+    auto const row_group_indices = reader->all_row_groups(options);
+    ASSERT_EQ(row_group_indices.size(), num_sources);
+    EXPECT_THROW(
+      std::ignore = reader->filter_row_groups_with_byte_range(row_group_indices, options),
+      std::invalid_argument);
+  }
 }
 
 TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats)

From a35f42901a0e52a46dfcdbea7a431429a6a4263e Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Wed, 3 Jun 2026 03:48:30 +0000
Subject: [PATCH 19/23] Style

---
 cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index f7b777703e78..8da2f535e961 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -115,7 +115,11 @@ class hybrid_scan_multifile {
     cudf::host_span<std::vector<size_type> const> row_group_indices) const;
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::reset_column_selection
+   * @brief Resets the current column selection
+   *
+   * Resets the current column selection state forcing column re-selection in subsequent filter,
+   * byte range, setup chunking and materialization APIs. This is useful if the filter expression
+   * has been cascaded (and-ed) to include new columns.
    */
   void reset_column_selection() const;
 

From 0d404b15479bea512dafff17f0eb9692dfa1f30f Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Thu, 4 Jun 2026 23:57:33 +0000
Subject: [PATCH 20/23] Multifile hybrid scan APIs for row mask construction

---
 .../io/experimental/hybrid_scan_multifile.hpp |  31 ++
 .../parquet/experimental/hybrid_scan_impl.hpp |   4 +-
 .../experimental/hybrid_scan_multifile.cpp    |  19 +
 .../hybrid_scan_multifile_filters_test.cpp    | 335 ++++++++++++++++--
 4 files changed, 356 insertions(+), 33 deletions(-)

diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
index 8da2f535e961..c556cf1b837c 100644
--- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
+++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp
@@ -167,6 +167,37 @@ class hybrid_scan_multifile {
   secondary_filters_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
                                 parquet_reader_options const& options) const;
 
+  /**
+   * @brief Builds a boolean survival column of size equal to the total number of rows in the row
+   * groups containing all `true` values
+   *
+   * @param row_group_indices Input per-source row group indices (one inner vector per source)
+   * @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 An all-true boolean (survival) column spanning all selected rows across all sources
+   */
+  [[nodiscard]] std::unique_ptr<cudf::column> build_all_true_row_mask(
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    rmm::cuda_stream_view stream,
+    rmm::device_async_resource_ref mr) const;
+
+  /**
+   * @brief Builds a boolean column indicating surviving rows using page-level statistics in the
+   * page index
+   *
+   * @param row_group_indices Input per-source row group indices (one inner vector per source)
+   * @param options Parquet reader options
+   * @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 spanning all selected rows across all sources and indicating which
+   * filter column rows survive the statistics in the page index
+   */
+  [[nodiscard]] std::unique_ptr<cudf::column> build_row_mask_with_page_index_stats(
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    parquet_reader_options const& options,
+    rmm::cuda_stream_view stream,
+    rmm::device_async_resource_ref mr) const;
+
  private:
   std::unique_ptr<detail::hybrid_scan_reader_impl> _impl;
 };
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
index d11ae1e8ddb9..a583bcb5c34f 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
@@ -131,7 +131,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
     rmm::cuda_stream_view stream);
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::build_all_true_row_mask
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::build_all_true_row_mask
    */
   [[nodiscard]] std::unique_ptr<cudf::column> build_all_true_row_mask(
     cudf::host_span<std::vector<size_type> const> row_group_indices,
@@ -139,7 +139,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
     rmm::device_async_resource_ref mr);
 
   /**
-   * @copydoc cudf::io::experimental::hybrid_scan::build_row_mask_with_page_index_stats
+   * @copydoc cudf::io::experimental::hybrid_scan_multifile::build_row_mask_with_page_index_stats
    */
   [[nodiscard]] std::unique_ptr<cudf::column> build_row_mask_with_page_index_stats(
     cudf::host_span<std::vector<size_type> const> row_group_indices,
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
index 59347367d223..1bcac3d2c22d 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp
@@ -83,4 +83,23 @@ hybrid_scan_multifile::secondary_filters_byte_ranges(
   return _impl->secondary_filters_byte_ranges(row_group_indices, options);
 }
 
+std::unique_ptr<cudf::column> hybrid_scan_multifile::build_all_true_row_mask(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->build_all_true_row_mask(row_group_indices, stream, mr);
+}
+
+std::unique_ptr<cudf::column> hybrid_scan_multifile::build_row_mask_with_page_index_stats(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->build_row_mask_with_page_index_stats(row_group_indices, options, stream, mr);
+}
+
 }  // namespace cudf::io::parquet::experimental
diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index 29c66093b1cf..c2413740a20d 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -9,6 +9,7 @@
 
 #include <cudf/ast/expressions.hpp>
 #include <cudf/copying.hpp>
+#include <cudf/detail/utilities/vector_factories.hpp>
 #include <cudf/io/experimental/hybrid_scan_multifile.hpp>
 #include <cudf/io/parquet.hpp>
 #include <cudf/io/parquet_io_utils.hpp>
@@ -17,7 +18,9 @@
 #include <cudf/utilities/memory_resource.hpp>
 #include <cudf/utilities/span.hpp>
 
+#include <algorithm>
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <vector>
 
@@ -27,26 +30,73 @@ namespace {
  * @brief Struct to hold multifile datasources, and footer buffers along with their byte spans
  */
 struct multifile_inputs {
+  /**
+   * @brief Construct datasources, datasource refs, and footer byte spans from source info
+   */
+  explicit multifile_inputs(cudf::io::source_info const& source_info)
+    : datasources{cudf::io::make_datasources(source_info)}
+  {
+    datasource_refs.reserve(datasources.size());
+    footer_buffers.reserve(datasources.size());
+    footer_byte_spans.reserve(datasources.size());
+
+    for (auto const& datasource : datasources) {
+      datasource_refs.emplace_back(*datasource);
+      footer_buffers.emplace_back(cudf::io::parquet::fetch_footer_to_host(datasource_refs.back()));
+      footer_byte_spans.emplace_back(*footer_buffers.back());
+    }
+  }
+
   std::vector<std::unique_ptr<cudf::io::datasource>> datasources;
+  std::vector<std::reference_wrapper<cudf::io::datasource>> datasource_refs;
   std::vector<std::unique_ptr<cudf::io::datasource::buffer>> footer_buffers;
   std::vector<cudf::host_span<uint8_t const>> footer_byte_spans;
 };
 
+/**
+ * @brief Construct source info from host buffers
+ */
 template <typename Buffers>
-multifile_inputs build_multifile_inputs(Buffers const& file_buffers)
+cudf::io::source_info build_source_info(Buffers const& file_buffers)
 {
-  multifile_inputs out;
-  out.datasources.reserve(file_buffers.size());
-  out.footer_buffers.reserve(file_buffers.size());
-  out.footer_byte_spans.reserve(file_buffers.size());
+  std::vector<cudf::host_span<char const>> spans;
+  spans.reserve(file_buffers.size());
   for (auto const& buf : file_buffers) {
-    out.datasources.emplace_back(cudf::io::datasource::create(cudf::host_span<std::byte const>(
-      reinterpret_cast<std::byte const*>(buf.data()), buf.size())));
-    out.footer_buffers.emplace_back(
-      cudf::io::parquet::fetch_footer_to_host(*out.datasources.back()));
-    out.footer_byte_spans.emplace_back(*out.footer_buffers.back());
+    spans.emplace_back(buf.data(), buf.size());
   }
-  return out;
+  return cudf::io::source_info(cudf::host_span<cudf::host_span<char const>>{spans});
+}
+
+/**
+ * @brief Copy fixed-width column data to a host vector
+ */
+template <typename T>
+auto host_row_mask_data(cudf::column_view const& column, rmm::cuda_stream_view stream)
+{
+  return cudf::detail::make_host_vector<T>(
+    cudf::device_span<T const>(column.data<T>(), static_cast<size_t>(column.size())), stream);
+}
+
+/**
+ * @brief Fetch and set up page indexes for all sources in a multifile reader
+ */
+void setup_page_indexes(cudf::io::parquet::experimental::hybrid_scan_multifile const& reader,
+                        multifile_inputs const& inputs)
+{
+  auto const page_index_byte_ranges = reader.page_index_byte_ranges();
+  std::vector<cudf::host_span<uint8_t const>> page_index_byte_spans;
+  page_index_byte_spans.reserve(page_index_byte_ranges.size());
+
+  auto const page_index_buffers = cudf::io::parquet::fetch_page_indexes_to_host(
+    cudf::host_span<std::reference_wrapper<cudf::io::datasource> const>{inputs.datasource_refs},
+    cudf::host_span<cudf::io::parquet::byte_range_info const>{page_index_byte_ranges});
+  std::transform(page_index_buffers.begin(),
+                 page_index_buffers.end(),
+                 std::back_inserter(page_index_byte_spans),
+                 [](auto const& buffer) { return cudf::host_span<uint8_t const>{*buffer}; });
+
+  reader.setup_page_indexes(
+    cudf::host_span<cudf::host_span<uint8_t const> const>{page_index_byte_spans});
 }
 
 /**
@@ -74,6 +124,21 @@ std::vector<char> create_empty_parquet_with_stats()
   return buffer;
 }
 
+/**
+ * @brief Build a scalar literal matching a filter column type
+ */
+template <typename T>
+auto make_scalar(cudf::size_type value, rmm::cuda_stream_view stream)
+{
+  if constexpr (cudf::is_timestamp<T>()) {
+    return cudf::timestamp_scalar<T>(T(typename T::duration(value)), true, stream);
+  } else if constexpr (cudf::is_duration<T>()) {
+    return cudf::duration_scalar<T>(T(value), true, stream);
+  } else {
+    return cudf::numeric_scalar<T>(static_cast<T>(value), true, stream);
+  }
+}
+
 }  // namespace
 
 struct HybridScanMultifileFiltersTest : public cudf::test::BaseFixture {};
@@ -91,20 +156,23 @@ TEST_F(HybridScanMultifileFiltersTest, Metadata)
   std::vector<std::vector<char>> file_buffers;
   file_buffers.reserve(num_sources);
   auto constexpr num_concat = 1;
-  srand(0xbad);
-  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, num_concat>()));
-  srand(0xf00d);
-  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, num_concat>()));
+  auto constexpr seed       = 0xbad;
+  std::transform(cuda::counting_iterator{seed},
+                 cuda::counting_iterator{seed + num_sources},
+                 std::back_inserter(file_buffers),
+                 [](auto const src_seed) {
+                   srand(src_seed);
+                   return std::get<1>(create_parquet_with_stats<T, num_concat>());
+                 });
 
   // Filtering AST - col0 < 100
-  auto literal_value =
-    cudf::timestamp_scalar<T>(T(typename T::duration(100)), true, cudf::get_default_stream());
+  auto literal_value     = make_scalar<T>(100, cudf::get_default_stream());
   auto literal           = cudf::ast::literal(literal_value);
   auto col_ref_0         = cudf::ast::column_name_reference("col0");
   auto filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref_0, literal);
 
   // Construct reader from footer bytes
-  auto inputs = build_multifile_inputs(file_buffers);
+  auto inputs = multifile_inputs(build_source_info(file_buffers));
 
   cudf::io::parquet_reader_options options =
     cudf::io::parquet_reader_options::builder().filter(filter_expression).build();
@@ -186,8 +254,6 @@ TEST_F(HybridScanMultifileFiltersTest, EmptySource)
 {
   using T = uint32_t;
 
-  srand(0xc0ffee);
-
   // Create two parquet source. First one with non-zero rows and the second one with zero rows.
   auto constexpr num_sources = 2;
   std::vector<std::vector<char>> file_buffers;
@@ -195,7 +261,7 @@ TEST_F(HybridScanMultifileFiltersTest, EmptySource)
   file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1>()));
   file_buffers.emplace_back(create_empty_parquet_with_stats<T>());
 
-  auto inputs = build_multifile_inputs(file_buffers);
+  auto inputs = multifile_inputs(build_source_info(file_buffers));
 
   cudf::io::parquet_reader_options options = cudf::io::parquet_reader_options::builder().build();
   auto const reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
@@ -224,14 +290,19 @@ TEST_F(HybridScanMultifileFiltersTest, ErrorFilterRowGroupsWithByteRanges)
 {
   using T                    = uint32_t;
   auto constexpr num_sources = 2;
-  srand(0xb47e);
 
   std::vector<std::vector<char>> file_buffers;
   file_buffers.reserve(num_sources);
-  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1>()));
-  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1>()));
+  auto constexpr seed = 0xb47e;
+  std::transform(cuda::counting_iterator{seed},
+                 cuda::counting_iterator{seed + num_sources},
+                 std::back_inserter(file_buffers),
+                 [](auto const src_seed) {
+                   srand(src_seed);
+                   return std::get<1>(create_parquet_with_stats<T, 1>());
+                 });
 
-  auto inputs = build_multifile_inputs(file_buffers);
+  auto inputs = multifile_inputs(build_source_info(file_buffers));
 
   // Setting `skip_bytes` or `num_bytes` is ambiguous when reading multiple sources. The reader is
   // expected to throw an exception if row groups are filtered using byte range in this case.
@@ -266,15 +337,19 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats)
   // Two sources, each with 4 row groups and ascending strings in col2
   std::vector<std::vector<char>> file_buffers;
   file_buffers.reserve(num_sources);
-  srand(0xc001);
-  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1, false>()));
-  srand(0xbeef);
-  file_buffers.emplace_back(std::get<1>(create_parquet_with_stats<T, 1, false>()));
+  auto constexpr seed = 0xc001;
+  std::transform(cuda::counting_iterator{seed},
+                 cuda::counting_iterator{seed + num_sources},
+                 std::back_inserter(file_buffers),
+                 [](auto const src_seed) {
+                   srand(src_seed);
+                   return std::get<1>(create_parquet_with_stats<T, 1, false>());
+                 });
 
-  auto inputs = build_multifile_inputs(file_buffers);
+  auto inputs = multifile_inputs(build_source_info(file_buffers));
 
   // Filter - col0 < 50 and col2 > "000010000"
-  auto literal_value0 = cudf::duration_scalar<T>(T::rep(50), true, cudf::get_default_stream());
+  auto literal_value0 = make_scalar<T>(50, cudf::get_default_stream());
   auto literal0       = cudf::ast::literal(literal_value0);
   auto col_ref0       = cudf::ast::column_reference(0);
   auto filter1        = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref0, literal0);
@@ -314,3 +389,201 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats)
   EXPECT_TRUE(stats_filtered.front().empty());
   EXPECT_TRUE(stats_filtered.back().empty());
 }
+
+TEST_F(HybridScanMultifileFiltersTest, BuildAllTrueRowMask)
+{
+  using T                    = uint64_t;
+  auto constexpr num_sources = 2;
+
+  std::vector<std::vector<char>> file_buffers;
+  file_buffers.reserve(num_sources);
+  auto constexpr seed = 0xa11;
+  std::transform(cuda::counting_iterator{seed},
+                 cuda::counting_iterator{seed + num_sources},
+                 std::back_inserter(file_buffers),
+                 [](auto const src_seed) {
+                   srand(src_seed);
+                   return std::get<1>(create_parquet_with_stats<T, 1>());
+                 });
+
+  auto inputs = multifile_inputs(build_source_info(file_buffers));
+
+  auto const options = cudf::io::parquet_reader_options::builder().build();
+  auto const reader  = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+    inputs.footer_byte_spans, options);
+
+  auto stream = cudf::get_default_stream();
+  auto mr     = cudf::get_current_device_resource_ref();
+
+  auto const row_group_indices = std::vector<std::vector<cudf::size_type>>{{0, 2}, {1, 3}};
+  auto const row_mask          = reader->build_all_true_row_mask(row_group_indices, stream, mr);
+
+  EXPECT_EQ(row_mask->type().id(), cudf::type_id::BOOL8);
+  EXPECT_EQ(row_mask->size(), reader->total_rows_in_row_groups(row_group_indices));
+  EXPECT_EQ(row_mask->null_count(), 0);
+
+  auto const host_row_mask = host_row_mask_data<bool>(row_mask->view(), stream);
+  auto const num_true_rows = std::count(host_row_mask.begin(), host_row_mask.end(), true);
+  EXPECT_EQ(num_true_rows, row_mask->size());
+
+  auto expected = cudf::detail::make_empty_host_vector<bool>(0, stream);
+  for (auto const src_idx : {0, 1}) {
+    auto const single_reader =
+      std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(
+        *inputs.footer_buffers[src_idx], options);
+    auto const single_row_mask =
+      single_reader->build_all_true_row_mask(row_group_indices[src_idx], stream, mr);
+    auto const host_single_row_mask = host_row_mask_data<bool>(single_row_mask->view(), stream);
+    expected.insert(expected.end(), host_single_row_mask.begin(), host_single_row_mask.end());
+  }
+  EXPECT_EQ(host_row_mask_data<bool>(row_mask->view(), stream), expected);
+
+  auto const empty_source_row_group_indices = std::vector<std::vector<cudf::size_type>>{{}, {0, 1}};
+  auto const empty_source_row_mask =
+    reader->build_all_true_row_mask(empty_source_row_group_indices, stream, mr);
+
+  EXPECT_EQ(empty_source_row_mask->size(),
+            reader->total_rows_in_row_groups(empty_source_row_group_indices));
+  auto const empty_source_host_row_mask =
+    host_row_mask_data<bool>(empty_source_row_mask->view(), stream);
+  auto const empty_source_num_true_rows =
+    std::count(empty_source_host_row_mask.begin(), empty_source_host_row_mask.end(), true);
+  EXPECT_EQ(empty_source_num_true_rows, empty_source_row_mask->size());
+}
+
+template <typename T>
+struct HybridScanMultifilePageIndexRowMaskTest : public HybridScanMultifileFiltersTest {};
+
+// Unsigned numeric types except booleans for page index stats tests
+using SignedIntegralTypesNotBool =
+  cudf::test::ContainedIn<cudf::test::Types<int8_t, int16_t, int32_t, int64_t>>;
+using PageIndexRowMaskTestTypes =
+  cudf::test::RemoveIf<SignedIntegralTypesNotBool,
+                       cudf::test::Concat<cudf::test::IntegralTypesNotBool>>;
+
+TYPED_TEST_SUITE(HybridScanMultifilePageIndexRowMaskTest, PageIndexRowMaskTestTypes);
+
+TYPED_TEST(HybridScanMultifilePageIndexRowMaskTest, BuildRowMaskWithPageIndexStats)
+{
+  using T                    = TypeParam;
+  auto constexpr num_sources = 4;
+
+  std::vector<std::vector<char>> file_buffers;
+  file_buffers.reserve(num_sources);
+  auto constexpr seed = 31337;
+  std::transform(cuda::counting_iterator{seed},
+                 cuda::counting_iterator{seed + num_sources},
+                 std::back_inserter(file_buffers),
+                 [](auto const src_seed) {
+                   srand(src_seed);
+                   return std::get<1>(create_parquet_with_stats<T, 1, false>());
+                 });
+
+  auto inputs = multifile_inputs(build_source_info(file_buffers));
+
+  auto options      = cudf::io::parquet_reader_options::builder().build();
+  auto const reader = std::make_unique<cudf::io::parquet::experimental::hybrid_scan_multifile>(
+    inputs.footer_byte_spans, options);
+
+  auto stream = cudf::get_default_stream();
+  auto mr     = cudf::get_current_device_resource_ref();
+
+  auto const input_row_group_indices = reader->all_row_groups(options);
+
+  auto const test_filter_data_pages_with_stats = [&](
+                                                   cudf::ast::operation const& filter_expression,
+                                                   cudf::size_type const expected_surviving_rows) {
+    options.set_filter(filter_expression);
+    reader->reset_column_selection();
+
+    auto const row_mask =
+      reader->build_row_mask_with_page_index_stats(input_row_group_indices, options, stream, mr);
+
+    auto const expected_num_rows = reader->total_rows_in_row_groups(input_row_group_indices);
+    EXPECT_EQ(row_mask->type().id(), cudf::type_id::BOOL8);
+    EXPECT_EQ(row_mask->size(), expected_num_rows);
+    EXPECT_EQ(row_mask->null_count(), 0);
+
+    auto const host_row_mask = host_row_mask_data<bool>(row_mask->view(), stream);
+    EXPECT_EQ(std::count(host_row_mask.begin(), host_row_mask.end(), true),
+              expected_surviving_rows);
+  };
+
+  // Calling the page-index row mask builder before setting up the page index should raise an error.
+  {
+    auto literal_value     = make_scalar<T>(100, stream);
+    auto const literal     = cudf::ast::literal(literal_value);
+    auto const col_ref     = cudf::ast::column_name_reference("col0");
+    auto filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref, literal);
+    options.set_filter(filter_expression);
+    EXPECT_THROW(std::ignore = reader->build_row_mask_with_page_index_stats(
+                   input_row_group_indices, options, stream, mr),
+                 std::runtime_error);
+  }
+
+  setup_page_indexes(*reader, inputs);
+
+  // Filtering AST - table[0] < 100
+  {
+    auto literal_value = make_scalar<T>(100, stream);
+    auto const literal = cudf::ast::literal(literal_value);
+    auto const col_ref = cudf::ast::column_name_reference("col0");
+    auto filter_expression =
+      cudf::ast::operation(cudf::ast::ast_operator::GREATER, literal, col_ref);
+    auto constexpr expected_surviving_rows =
+      num_sources * num_ordered_rows / (std::is_signed_v<T> ? 4 : 2);
+    test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows);
+  }
+
+  // Filtering AST - table[2] >= 10000
+  {
+    auto literal_value = cudf::string_scalar("000010000", true, stream);
+    auto literal       = cudf::ast::literal(literal_value);
+    auto col_ref       = cudf::ast::column_name_reference("col2");
+    auto filter_expression =
+      cudf::ast::operation(cudf::ast::ast_operator::GREATER_EQUAL, col_ref, literal);
+    auto constexpr expected_surviving_rows =
+      num_sources * num_ordered_rows / (std::is_signed_v<T> ? 4 : 2);
+    test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows);
+  }
+
+  // Filtering AST - table[0] < 50 AND table[2] < "000010000"
+  {
+    auto literal_value1 = make_scalar<T>(50, stream);
+    auto const literal1 = cudf::ast::literal(literal_value1);
+    auto const col_ref1 = cudf::ast::column_name_reference("col0");
+    auto filter_expression1 =
+      cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref1, literal1);
+
+    auto literal_value2 = cudf::string_scalar("000010000", true, stream);
+    auto literal2       = cudf::ast::literal(literal_value2);
+    auto col_ref2       = cudf::ast::column_name_reference("col2");
+    auto filter_expression2 =
+      cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref2, literal2);
+
+    auto filter_expression = cudf::ast::operation(
+      cudf::ast::ast_operator::LOGICAL_AND, filter_expression1, filter_expression2);
+    auto constexpr expected_surviving_rows = num_sources * page_size_for_ordered_tests;
+    test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows);
+  }
+
+  // Filtering AST - table[0] > 150 OR table[2] < "000005000"
+  {
+    auto literal_value1 = make_scalar<T>(150, stream);
+    auto const literal1 = cudf::ast::literal(literal_value1);
+    auto const col_ref1 = cudf::ast::column_name_reference("col0");
+    auto filter_expression1 =
+      cudf::ast::operation(cudf::ast::ast_operator::GREATER, col_ref1, literal1);
+
+    auto literal_value2 = cudf::string_scalar("000005000", true, stream);
+    auto literal2       = cudf::ast::literal(literal_value2);
+    auto col_ref2       = cudf::ast::column_name_reference("col2");
+    auto filter_expression2 =
+      cudf::ast::operation(cudf::ast::ast_operator::LESS, col_ref2, literal2);
+
+    auto filter_expression = cudf::ast::operation(
+      cudf::ast::ast_operator::LOGICAL_OR, filter_expression1, filter_expression2);
+    auto constexpr expected_surviving_rows = 2 * num_sources * page_size_for_ordered_tests;
+    test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows);
+  }
+}
\ No newline at end of file

From db305f02de05fd6f423f7da15782703640426d36 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com>
Date: Fri, 5 Jun 2026 00:06:46 +0000
Subject: [PATCH 21/23] Simplify test

---
 .../hybrid_scan_multifile_filters_test.cpp    | 50 ++++++-------------
 1 file changed, 15 insertions(+), 35 deletions(-)

diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index c2413740a20d..766548fbfb5d 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -415,40 +415,20 @@ TEST_F(HybridScanMultifileFiltersTest, BuildAllTrueRowMask)
   auto stream = cudf::get_default_stream();
   auto mr     = cudf::get_current_device_resource_ref();
 
-  auto const row_group_indices = std::vector<std::vector<cudf::size_type>>{{0, 2}, {1, 3}};
-  auto const row_mask          = reader->build_all_true_row_mask(row_group_indices, stream, mr);
-
-  EXPECT_EQ(row_mask->type().id(), cudf::type_id::BOOL8);
-  EXPECT_EQ(row_mask->size(), reader->total_rows_in_row_groups(row_group_indices));
-  EXPECT_EQ(row_mask->null_count(), 0);
-
-  auto const host_row_mask = host_row_mask_data<bool>(row_mask->view(), stream);
-  auto const num_true_rows = std::count(host_row_mask.begin(), host_row_mask.end(), true);
-  EXPECT_EQ(num_true_rows, row_mask->size());
-
-  auto expected = cudf::detail::make_empty_host_vector<bool>(0, stream);
-  for (auto const src_idx : {0, 1}) {
-    auto const single_reader =
-      std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(
-        *inputs.footer_buffers[src_idx], options);
-    auto const single_row_mask =
-      single_reader->build_all_true_row_mask(row_group_indices[src_idx], stream, mr);
-    auto const host_single_row_mask = host_row_mask_data<bool>(single_row_mask->view(), stream);
-    expected.insert(expected.end(), host_single_row_mask.begin(), host_single_row_mask.end());
-  }
-  EXPECT_EQ(host_row_mask_data<bool>(row_mask->view(), stream), expected);
-
-  auto const empty_source_row_group_indices = std::vector<std::vector<cudf::size_type>>{{}, {0, 1}};
-  auto const empty_source_row_mask =
-    reader->build_all_true_row_mask(empty_source_row_group_indices, stream, mr);
-
-  EXPECT_EQ(empty_source_row_mask->size(),
-            reader->total_rows_in_row_groups(empty_source_row_group_indices));
-  auto const empty_source_host_row_mask =
-    host_row_mask_data<bool>(empty_source_row_mask->view(), stream);
-  auto const empty_source_num_true_rows =
-    std::count(empty_source_host_row_mask.begin(), empty_source_host_row_mask.end(), true);
-  EXPECT_EQ(empty_source_num_true_rows, empty_source_row_mask->size());
+  auto test_all_true_row_mask =
+    [&](cudf::host_span<std::vector<cudf::size_type> const> row_group_indices) {
+      auto const row_mask = reader->build_all_true_row_mask(row_group_indices, stream, mr);
+
+      EXPECT_EQ(row_mask->type().id(), cudf::type_id::BOOL8);
+      EXPECT_EQ(row_mask->size(), reader->total_rows_in_row_groups(row_group_indices));
+      EXPECT_EQ(row_mask->null_count(), 0);
+    };
+
+  auto row_group_indices = std::vector<std::vector<cudf::size_type>>{{0, 2}, {1, 3}};
+  test_all_true_row_mask(row_group_indices);
+
+  row_group_indices = reader->all_row_groups(options);
+  test_all_true_row_mask(row_group_indices);
 }
 
 template <typename T>
@@ -470,7 +450,7 @@ TYPED_TEST(HybridScanMultifilePageIndexRowMaskTest, BuildRowMaskWithPageIndexSta
 
   std::vector<std::vector<char>> file_buffers;
   file_buffers.reserve(num_sources);
-  auto constexpr seed = 31337;
+  auto constexpr seed = 0xa11b;
   std::transform(cuda::counting_iterator{seed},
                  cuda::counting_iterator{seed + num_sources},
                  std::back_inserter(file_buffers),

From 24ae4765efd71d85e1d32909a67a03e528833581 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <mhaseeb@nvidia.com>
Date: Mon, 8 Jun 2026 23:43:01 +0000
Subject: [PATCH 22/23] style

---
 .../io/experimental/hybrid_scan_multifile_filters_test.cpp      | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index 766548fbfb5d..c534bc94fb4b 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -566,4 +566,4 @@ TYPED_TEST(HybridScanMultifilePageIndexRowMaskTest, BuildRowMaskWithPageIndexSta
     auto constexpr expected_surviving_rows = 2 * num_sources * page_size_for_ordered_tests;
     test_filter_data_pages_with_stats(filter_expression, expected_surviving_rows);
   }
-}
\ No newline at end of file
+}

From aedc4d2be31cc42ba5bcca0caf23e78f75f36458 Mon Sep 17 00:00:00 2001
From: Muhammad Haseeb <mhaseeb@nvidia.com>
Date: Tue, 9 Jun 2026 00:05:42 +0000
Subject: [PATCH 23/23] Update test

---
 .../io/experimental/hybrid_scan_multifile_filters_test.cpp  | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
index c534bc94fb4b..1c802680b9fa 100644
--- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp
@@ -419,9 +419,13 @@ TEST_F(HybridScanMultifileFiltersTest, BuildAllTrueRowMask)
     [&](cudf::host_span<std::vector<cudf::size_type> const> row_group_indices) {
       auto const row_mask = reader->build_all_true_row_mask(row_group_indices, stream, mr);
 
+      auto const expected_num_rows = reader->total_rows_in_row_groups(row_group_indices);
+
       EXPECT_EQ(row_mask->type().id(), cudf::type_id::BOOL8);
-      EXPECT_EQ(row_mask->size(), reader->total_rows_in_row_groups(row_group_indices));
+      EXPECT_EQ(row_mask->size(), expected_num_rows);
       EXPECT_EQ(row_mask->null_count(), 0);
+      auto const host_row_mask = host_row_mask_data<bool>(row_mask->view(), stream);
+      EXPECT_EQ(std::count(host_row_mask.begin(), host_row_mask.end(), true), expected_num_rows);
     };
 
   auto row_group_indices = std::vector<std::vector<cudf::size_type>>{{0, 2}, {1, 3}};