diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
index f2919c64519f..0f0fd7de6d96 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
@@ -185,8 +185,8 @@ void hybrid_scan_reader_impl::select_columns(read_columns_mode read_columns_mode
     _is_all_columns_selected     = false;
   }
 
-  // Reset the rows processed so far
-  _rows_processed_so_far = 0;
+  // Reset the materialization step flag
+  _output_chunk_produced = false;
 
   CUDF_EXPECTS(_input_columns.size() > 0 and _output_buffers.size() > 0, "No columns selected");
 
@@ -381,6 +381,10 @@ std::unique_ptr<cudf::column> hybrid_scan_reader_impl::build_all_true_row_mask(
   CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered");
 
   auto const num_rows = total_rows_in_row_groups(row_group_indices);
+  CUDF_EXPECTS(num_rows < std::numeric_limits<cudf::size_type>::max(),
+               "Total rows in row groups exceed the cudf's column size limit. Retry with a smaller "
+               "set of row groups",
+               std::invalid_argument);
   auto true_scalar =
     cudf::numeric_scalar<bool>(true, true, stream, cudf::get_current_device_resource_ref());
   return cudf::make_column_from_scalar(true_scalar, num_rows, stream, mr);
@@ -510,7 +514,7 @@ table_with_metadata hybrid_scan_reader_impl::materialize_filter_columns(
   auto data_page_mask = thrust::host_vector<bool>{};
   if (mask_data_pages == use_data_page_mask::YES) {
     data_page_mask = _extended_metadata->compute_data_page_mask(
-      row_mask, row_group_indices, _input_columns, _rows_processed_so_far, stream);
+      row_mask, row_group_indices, _input_columns, _row_mask_offset, stream);
   }
 
   prepare_data(read_mode::READ_ALL, row_group_indices, column_chunk_data, data_page_mask);
@@ -537,7 +541,7 @@ table_with_metadata hybrid_scan_reader_impl::materialize_payload_columns(
   auto data_page_mask = thrust::host_vector<bool>{};
   if (not row_mask.is_empty() and mask_data_pages == use_data_page_mask::YES) {
     data_page_mask = _extended_metadata->compute_data_page_mask(
-      row_mask, row_group_indices, _input_columns, _rows_processed_so_far, stream);
+      row_mask, row_group_indices, _input_columns, _row_mask_offset, stream);
   }
 
   prepare_data(read_mode::READ_ALL, row_group_indices, column_chunk_data, data_page_mask);
@@ -565,8 +569,8 @@ table_with_metadata hybrid_scan_reader_impl::materialize_all_columns(
   // Use the main reader's function
   auto result = reader_impl::read_chunk_internal(read_mode::READ_ALL);
 
-  // base read_chunk_internal() does not update the _rows_processed_so_far
-  _rows_processed_so_far += result.tbl->num_rows();
+  // base read_chunk_internal() does not update the output chunk produced flag
+  _output_chunk_produced = true;
 
   return result;
 }
@@ -599,7 +603,7 @@ void hybrid_scan_reader_impl::setup_chunking_for_filter_columns(
   auto data_page_mask = thrust::host_vector<bool>{};
   if (mask_data_pages == use_data_page_mask::YES) {
     data_page_mask = _extended_metadata->compute_data_page_mask(
-      row_mask, row_group_indices, _input_columns, _rows_processed_so_far, stream);
+      row_mask, row_group_indices, _input_columns, _row_mask_offset, stream);
   }
 
   prepare_data(read_mode::CHUNKED_READ, row_group_indices, column_chunk_data, data_page_mask);
@@ -648,7 +652,7 @@ void hybrid_scan_reader_impl::setup_chunking_for_payload_columns(
   auto data_page_mask = thrust::host_vector<bool>{};
   if (not row_mask.is_empty() and mask_data_pages == use_data_page_mask::YES) {
     data_page_mask = _extended_metadata->compute_data_page_mask(
-      row_mask, row_group_indices, _input_columns, _rows_processed_so_far, stream);
+      row_mask, row_group_indices, _input_columns, _row_mask_offset, stream);
   }
 
   prepare_data(read_mode::CHUNKED_READ, row_group_indices, column_chunk_data, data_page_mask);
@@ -713,8 +717,8 @@ table_with_metadata hybrid_scan_reader_impl::materialize_all_columns_chunk()
   // Use the main reader's function for reading all columns
   auto result = reader_impl::read_chunk_internal(read_mode::CHUNKED_READ);
 
-  // base read_chunk_internal() does not update the _rows_processed_so_far
-  _rows_processed_so_far += result.tbl->num_rows();
+  // base read_chunk_internal() does not update the output chunk produced flag
+  _output_chunk_produced = true;
 
   return result;
 }
@@ -789,6 +793,7 @@ bool hybrid_scan_reader_impl::has_next_table_chunk()
 
 void hybrid_scan_reader_impl::reset_internal_state()
 {
+  _row_mask_offset   = 0;
   _file_itm_data     = file_intermediate_data{};
   _file_preprocessed = false;
   _has_page_index    = false;
@@ -1039,6 +1044,12 @@ table_with_metadata hybrid_scan_reader_impl::finalize_output(
   // If the input row mask is empty, return the table as is.
   if (row_mask.is_empty()) { return {std::move(read_table), std::move(out_metadata)}; }
 
+  // Get the current row mask offset
+  auto const mask_offset = _row_mask_offset;
+  // Update the row mask offset and the output chunk produced flag
+  _row_mask_offset += read_table->num_rows();
+  _output_chunk_produced = true;
+
   // For filter columns, apply the filter expression and update the input row mask
   if constexpr (std::is_same_v<RowMaskView, cudf::mutable_column_view>) {
     CUDF_EXPECTS(read_columns_mode == read_columns_mode::FILTER_COLUMNS, "Invalid read mode");
@@ -1055,9 +1066,6 @@ table_with_metadata hybrid_scan_reader_impl::finalize_output(
     auto output_table = cudf::detail::apply_mask(
       read_table->view(), *final_row_mask, cudf::detail::mask_type::RETENTION, _stream, _mr);
 
-    auto const mask_offset = _rows_processed_so_far;
-    _rows_processed_so_far += read_table->num_rows();
-
     // Update the input row mask to reflect the final row mask.
     update_row_mask(final_row_mask->view(), row_mask, mask_offset, _stream);
 
@@ -1068,9 +1076,6 @@ table_with_metadata hybrid_scan_reader_impl::finalize_output(
   else {
     CUDF_EXPECTS(read_columns_mode == read_columns_mode::PAYLOAD_COLUMNS, "Invalid read mode");
 
-    auto const mask_offset = _rows_processed_so_far;
-    _rows_processed_so_far += read_table->num_rows();
-
     CUDF_EXPECTS(mask_offset + read_table->num_rows() <= row_mask.size(),
                  "Encountered invalid sized row mask to apply");
     auto effective_row_mask =
diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
index 64eead4463f7..cb7798790030 100644
--- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
+++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
@@ -523,7 +523,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
    */
   [[nodiscard]] bool is_first_output_chunk() const
   {
-    return _file_itm_data._output_chunk_count == 0 and _rows_processed_so_far == 0;
+    return _file_itm_data._output_chunk_count == 0 and not _output_chunk_produced;
   }
 
  private:
@@ -531,7 +531,8 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
 
   std::optional<std::vector<std::string>> _filter_columns_names;
 
-  cudf::size_type _rows_processed_so_far{0};
+  cudf::size_type _row_mask_offset{0};
+  bool _output_chunk_produced{false};
 
   bool _use_pandas_metadata{false};
 
diff --git a/cpp/src/io/parquet/experimental/page_index_filter.cu b/cpp/src/io/parquet/experimental/page_index_filter.cu
index bb2b89c56c7e..1cf4995bf66a 100644
--- a/cpp/src/io/parquet/experimental/page_index_filter.cu
+++ b/cpp/src/io/parquet/experimental/page_index_filter.cu
@@ -870,20 +870,11 @@ std::unique_ptr<cudf::column> aggregate_reader_metadata::build_row_mask_with_pag
                std::runtime_error);
 
   // Total number of rows
-  auto const total_rows = std::accumulate(
-    cuda::counting_iterator<std::size_t>{0},
-    cuda::counting_iterator{row_group_indices.size()},
-    std::size_t{0},
-    [&](auto sum, auto const src_index) {
-      auto const& rg_indices = row_group_indices[src_index];
-      return std::accumulate(
-        rg_indices.begin(), rg_indices.end(), sum, [&](auto subsum, auto const rg_index) {
-          CUDF_EXPECTS(subsum + per_file_metadata[src_index].row_groups[rg_index].num_rows <=
-                         std::numeric_limits<size_type>::max(),
-                       "Total rows exceed the maximum value");
-          return subsum + per_file_metadata[src_index].row_groups[rg_index].num_rows;
-        });
-    });
+  auto const total_rows = total_rows_in_row_groups(row_group_indices);
+  CUDF_EXPECTS(std::cmp_less_equal(total_rows, std::numeric_limits<size_type>::max()),
+               "Total rows in row groups exceed the cudf's column size limit. Retry with a smaller "
+               "set of row groups",
+               std::invalid_argument);
 
   auto const num_columns = output_dtypes.size();
 
@@ -994,11 +985,15 @@ 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<size_type>::max()),
+               "Total rows in row groups exceed the cudf's column size limit. Retry with a smaller "
+               "set of row groups",
+               std::invalid_argument);
 
   CUDF_EXPECTS(
-    std::cmp_less_equal(static_cast<std::size_t>(row_mask_offset) + total_rows, row_mask.size()),
+    std::cmp_less_equal(row_mask_offset + total_rows, row_mask.size()),
     "Encountered a mismatch in number of rows in the row group pass and the row mask size",
-    std::invalid_argument);
+    std::overflow_error);
 
   // Return an empty vector if all rows are invalid or all rows are required
   if (std::cmp_equal(row_mask.null_count(row_mask_offset, row_mask_offset + total_rows, stream),
diff --git a/cpp/tests/io/experimental/hybrid_scan_composer.cpp b/cpp/tests/io/experimental/hybrid_scan_composer.cpp
index b3541eb221b7..5365988220d9 100644
--- a/cpp/tests/io/experimental/hybrid_scan_composer.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_composer.cpp
@@ -5,11 +5,13 @@
 
 #include "hybrid_scan_composer.hpp"
 
+#include <cudf/column/column_view.hpp>
 #include <cudf/concatenate.hpp>
 #include <cudf/io/experimental/hybrid_scan.hpp>
 #include <cudf/io/parquet.hpp>
 #include <cudf/io/parquet_io_utils.hpp>
 #include <cudf/io/text/byte_range_info.hpp>
+#include <cudf/null_mask.hpp>
 #include <cudf/utilities/memory_resource.hpp>
 
 #include <rmm/mr/aligned_resource_adaptor.hpp>
@@ -48,15 +50,14 @@ std::unique_ptr<hybrid_scan_reader> setup_reader(cudf::io::datasource& datasourc
 }
 
 /**
- * @brief Apply hybrid scan filters
+ * @brief Apply hybrid scan row group filters
  *
  * @param datasource Input datasource
  * @param options Reader options
  * @param stream CUDA stream
  * @param mr Device memory resource
  *
- * @return A tuple of the reader, filtered row group indices, and row mask and data page mask from
- * data page pruning
+ * @return Filtered row group indices
  */
 auto apply_hybrid_scan_filters(cudf::io::datasource& datasource,
                                hybrid_scan_reader const& reader,
@@ -120,20 +121,8 @@ auto apply_hybrid_scan_filters(cudf::io::datasource& datasource,
     current_row_group_indices = bloom_filtered_row_group_indices;
   }
 
-  // Build row mask using page index stats or all true if no filter is provided
-  auto row_mask = [&]() {
-    if (options.get_filter().has_value()) {
-      return reader.build_row_mask_with_page_index_stats(
-        current_row_group_indices, options, stream, mr);
-    } else {
-      return reader.build_all_true_row_mask(current_row_group_indices, stream, mr);
-    }
-  }();
-
-  std::vector<cudf::size_type> final_row_group_indices(current_row_group_indices.begin(),
-                                                       current_row_group_indices.end());
-
-  return std::tuple{std::move(final_row_group_indices), std::move(row_mask)};
+  return std::vector<cudf::size_type>(current_row_group_indices.begin(),
+                                      current_row_group_indices.end());
 }
 
 /*
@@ -182,10 +171,17 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> hybrid_sc
   auto const reader = setup_reader(datasource, options);
   auto reader_ref   = std::ref(*reader);
 
-  auto [filtered_row_group_indices, row_mask] =
+  auto const filtered_row_group_indices =
     apply_hybrid_scan_filters(datasource, reader_ref, options, stream, mr);
 
-  auto current_row_group_indices = cudf::host_span<cudf::size_type>(filtered_row_group_indices);
+  auto const current_row_group_indices =
+    cudf::host_span<cudf::size_type const>(filtered_row_group_indices);
+
+  // Build a row mask for the filtered row groups using page index stats, or all true if no filter
+  auto row_mask =
+    options.get_filter().has_value()
+      ? reader->build_row_mask_with_page_index_stats(current_row_group_indices, options, stream, mr)
+      : reader->build_all_true_row_mask(current_row_group_indices, stream, mr);
 
   // Get column chunk byte ranges from the reader
   auto const filter_column_chunk_byte_ranges =
@@ -198,11 +194,11 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> hybrid_sc
   filter_col_tasks.get();
 
   // Materialize the table with only the filter columns
-  auto row_mask_mutable_view = row_mask->mutable_view();
+  auto row_mask_view = row_mask->mutable_view();
   auto [filter_table, filter_metadata] =
     reader->materialize_filter_columns(current_row_group_indices,
                                        filter_col_data,
-                                       row_mask_mutable_view,
+                                       row_mask_view,
                                        cudf::io::parquet::experimental::use_data_page_mask::YES,
                                        options,
                                        stream,
@@ -222,7 +218,7 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> hybrid_sc
   auto [payload_table, payload_metadata] =
     reader->materialize_payload_columns(current_row_group_indices,
                                         payload_col_data,
-                                        row_mask->view(),
+                                        row_mask_view,
                                         cudf::io::parquet::experimental::use_data_page_mask::YES,
                                         options,
                                         stream,
@@ -251,97 +247,97 @@ std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> chunked_h
   auto const reader = setup_reader(datasource, options);
   auto reader_ref   = std::ref(*reader);
 
-  auto [filtered_row_group_indices, row_mask] =
+  auto const filtered_row_group_indices =
     apply_hybrid_scan_filters(datasource, reader_ref, options, stream, mr);
 
-  auto current_row_group_indices = cudf::host_span<cudf::size_type>(filtered_row_group_indices);
-
-  // Helper to split the materialization of filter columns into chunks
-  auto tables          = std::vector<std::unique_ptr<cudf::table>>{};
-  auto filter_metadata = cudf::io::table_metadata{};
-  auto const materialize_filter_columns =
-    [&](cudf::host_span<cudf::size_type const> row_group_indices) {
-      // Get column chunk byte ranges from the reader and fetch device buffers
-      auto const filter_column_chunk_byte_ranges =
-        reader->filter_column_chunks_byte_ranges(row_group_indices, options);
-      auto [filter_col_buffers, filter_col_data, filter_col_tasks] =
-        cudf::io::parquet::fetch_byte_ranges_to_device_async(
-          datasource, filter_column_chunk_byte_ranges, stream, mr);
-      filter_col_tasks.get();
-
-      // Setup chunking for filter columns and materialize the columns
-      reader->setup_chunking_for_filter_columns(
-        1024,
-        10240,
-        row_group_indices,
-        row_mask->view(),
-        cudf::io::parquet::experimental::use_data_page_mask::YES,
-        filter_col_data,
-        options,
-        stream,
-        mr);
-
-      auto row_mask_mutable_view = row_mask->mutable_view();
-      while (reader->has_next_table_chunk()) {
-        auto chunk = reader->materialize_filter_columns_chunk(row_mask_mutable_view);
-        tables.push_back(std::move(chunk.tbl));
-        filter_metadata = std::move(chunk.metadata);
-      }
-    };
-
-  if (current_row_group_indices.size() > 1) {
-    auto const row_group_split = current_row_group_indices.size() / 2;
-    materialize_filter_columns(current_row_group_indices.subspan(0, row_group_split));
-    materialize_filter_columns(current_row_group_indices.subspan(
-      row_group_split, current_row_group_indices.size() - row_group_split));
-  } else {
-    materialize_filter_columns(current_row_group_indices);
-  }
-
-  auto filter_table = concatenate_tables(std::move(tables), stream, mr);
+  auto const current_row_group_indices =
+    cudf::host_span<cudf::size_type const>(filtered_row_group_indices);
+
+  // Build a row mask for the filtered row groups using page index stats, or all true if no filter
+  auto row_mask =
+    options.get_filter().has_value()
+      ? reader->build_row_mask_with_page_index_stats(current_row_group_indices, options, stream, mr)
+      : reader->build_all_true_row_mask(current_row_group_indices, stream, mr);
+
+  auto filter_tables  = std::vector<std::unique_ptr<cudf::table>>{};
+  auto payload_tables = std::vector<std::unique_ptr<cudf::table>>{};
+
+  // Helper to materialize filter and payload columns for a row group pass
+  std::size_t rows_materialized = 0;
+  auto const materialize_pass   = [&](cudf::host_span<cudf::size_type const> row_group_indices) {
+    // Sliced row mask view for the current pass
+    auto const rows_in_pass = reader->total_rows_in_row_groups(row_group_indices);
+    auto* null_mask         = row_mask->nullable() ? row_mask->mutable_view().null_mask() : nullptr;
+    auto const slice_null_count =
+      cudf::null_count(null_mask, rows_materialized, rows_materialized + rows_in_pass, stream);
+    auto row_mask_view = cudf::mutable_column_view(row_mask->type(),
+                                                   rows_in_pass,
+                                                   row_mask->mutable_view().data<bool>(),
+                                                   null_mask,
+                                                   slice_null_count,
+                                                   rows_materialized);
+
+    // Materialize filter columns
+    auto const filter_byte_ranges =
+      reader->filter_column_chunks_byte_ranges(row_group_indices, options);
+    auto [filter_buffers, filter_data, filter_tasks] =
+      cudf::io::parquet::fetch_byte_ranges_to_device_async(
+        datasource, filter_byte_ranges, stream, mr);
+    filter_tasks.get();
+
+    reader->setup_chunking_for_filter_columns(
+      1024,
+      10240,
+      row_group_indices,
+      row_mask_view,
+      cudf::io::parquet::experimental::use_data_page_mask::YES,
+      filter_data,
+      options,
+      stream,
+      mr);
+
+    while (reader->has_next_table_chunk()) {
+      filter_tables.push_back(reader->materialize_filter_columns_chunk(row_mask_view).tbl);
+    }
 
-  // Helper to split the materialization of payload columns into chunks
-  tables.clear();
-  auto payload_metadata = cudf::io::table_metadata{};
-  auto const materialize_payload_columns =
-    [&](cudf::host_span<cudf::size_type const> row_group_indices) {
-      // Get column chunk byte ranges from the reader and fetch device buffers
-      auto const payload_column_chunk_byte_ranges =
-        reader->payload_column_chunks_byte_ranges(row_group_indices, options);
-      auto [payload_col_buffers, payload_col_data, payload_col_tasks] =
-        cudf::io::parquet::fetch_byte_ranges_to_device_async(
-          datasource, payload_column_chunk_byte_ranges, stream, mr);
-      payload_col_tasks.get();
-
-      // Setup chunking for payload columns and materialize the table
-      reader->setup_chunking_for_payload_columns(
-        1024,
-        10240,
-        row_group_indices,
-        row_mask->view(),
-        cudf::io::parquet::experimental::use_data_page_mask::YES,
-        payload_col_data,
-        options,
-        stream,
-        mr);
+    // Materialize payload columns
+    auto const payload_byte_ranges =
+      reader->payload_column_chunks_byte_ranges(row_group_indices, options);
+    auto [payload_buffers, payload_data, payload_tasks] =
+      cudf::io::parquet::fetch_byte_ranges_to_device_async(
+        datasource, payload_byte_ranges, stream, mr);
+    payload_tasks.get();
+
+    reader->setup_chunking_for_payload_columns(
+      1024,
+      10240,
+      row_group_indices,
+      row_mask_view,
+      cudf::io::parquet::experimental::use_data_page_mask::YES,
+      payload_data,
+      options,
+      stream,
+      mr);
+
+    while (reader->has_next_table_chunk()) {
+      payload_tables.push_back(reader->materialize_payload_columns_chunk(row_mask_view).tbl);
+    }
 
-      while (reader->has_next_table_chunk()) {
-        auto chunk = reader->materialize_payload_columns_chunk(row_mask->view());
-        tables.push_back(std::move(chunk.tbl));
-        payload_metadata = std::move(chunk.metadata);
-      }
-    };
+    // Update the number of rows materialized
+    rows_materialized += rows_in_pass;
+  };
 
   if (current_row_group_indices.size() > 1) {
     auto const row_group_split = current_row_group_indices.size() / 2;
-    materialize_payload_columns(current_row_group_indices.subspan(0, row_group_split));
-    materialize_payload_columns(current_row_group_indices.subspan(
+    materialize_pass(current_row_group_indices.subspan(0, row_group_split));
+    materialize_pass(current_row_group_indices.subspan(
       row_group_split, current_row_group_indices.size() - row_group_split));
   } else {
-    materialize_payload_columns(current_row_group_indices);
+    materialize_pass(current_row_group_indices);
   }
 
-  auto payload_table = concatenate_tables(std::move(tables), stream, mr);
+  auto filter_table  = concatenate_tables(std::move(filter_tables), stream, mr);
+  auto payload_table = concatenate_tables(std::move(payload_tables), stream, mr);
 
   return std::tuple{std::move(filter_table), std::move(payload_table)};
 }
@@ -365,10 +361,11 @@ std::unique_ptr<cudf::table> hybrid_scan_single_step(
   auto const reader = setup_reader(datasource, options);
   auto reader_ref   = std::ref(*reader);
 
-  auto [filtered_row_group_indices, _ /* row_mask */] =
+  auto const filtered_row_group_indices =
     apply_hybrid_scan_filters(datasource, reader_ref, options, stream, mr);
 
-  auto current_row_group_indices = cudf::host_span<cudf::size_type>(filtered_row_group_indices);
+  auto const current_row_group_indices =
+    cudf::host_span<cudf::size_type const>(filtered_row_group_indices);
 
   // Get all column chunk byte ranges from the reader
   auto const all_column_chunk_byte_ranges =
@@ -405,7 +402,7 @@ std::unique_ptr<cudf::table> chunked_hybrid_scan_single_step(
   auto const reader = setup_reader(datasource, options);
   auto reader_ref   = std::ref(*reader);
 
-  auto [filtered_row_group_indices, _ /*row_mask*/] =
+  auto filtered_row_group_indices =
     apply_hybrid_scan_filters(datasource, reader_ref, options, stream, mr);
 
   auto current_row_group_indices = cudf::host_span<cudf::size_type>(filtered_row_group_indices);
diff --git a/cpp/tests/io/experimental/hybrid_scan_test.cpp b/cpp/tests/io/experimental/hybrid_scan_test.cpp
index a672ebb61491..05e7fba3d844 100644
--- a/cpp/tests/io/experimental/hybrid_scan_test.cpp
+++ b/cpp/tests/io/experimental/hybrid_scan_test.cpp
@@ -923,6 +923,115 @@ TEST_F(HybridScanTest, StructChildFilterColumn)
     std::invalid_argument);
 }
 
+TEST_F(HybridScanTest, ChunkedReadRowMaskPerPass)
+{
+  using T = uint32_t;
+
+  // A multi-row-group table so that the row groups can be split into more than one pass
+  auto constexpr num_concat            = 4;
+  auto [written_table, parquet_buffer] = create_parquet_with_stats<T, num_concat>();
+
+  // Filtering AST - col0 < 100
+  auto literal_value     = cudf::numeric_scalar<T>(100);
+  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);
+
+  auto const stream = cudf::get_default_stream();
+  auto const mr     = cudf::get_current_device_resource_ref();
+
+  auto const options =
+    cudf::io::parquet_reader_options::builder().filter(filter_expression).build();
+
+  auto datasource = cudf::io::datasource::create(cudf::host_span<std::byte const>(
+    reinterpret_cast<std::byte const*>(parquet_buffer.data()), parquet_buffer.size()));
+
+  auto const footer_buffer = cudf::io::parquet::fetch_footer_to_host(*datasource);
+  auto reader =
+    std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(*footer_buffer, options);
+
+  auto const page_index_byte_range = reader->page_index_byte_range();
+  if (not page_index_byte_range.is_empty()) {
+    auto const page_index_buffer =
+      cudf::io::parquet::fetch_page_index_to_host(*datasource, page_index_byte_range);
+    reader->setup_page_index(*page_index_buffer);
+  }
+
+  auto const all_row_groups = reader->all_row_groups(options);
+
+  std::vector<std::unique_ptr<cudf::table>> filter_tables;
+  std::vector<std::unique_ptr<cudf::table>> payload_tables;
+
+  auto const materialize_pass = [&](cudf::host_span<cudf::size_type const> row_group_indices) {
+    // Build a per-pass row mask for just this pass's row groups
+    auto row_mask =
+      reader->build_row_mask_with_page_index_stats(row_group_indices, options, stream, mr);
+    auto row_mask_view = row_mask->mutable_view();
+
+    // Materialize filter columns for this pass
+    auto const filter_byte_ranges =
+      reader->filter_column_chunks_byte_ranges(row_group_indices, options);
+    auto [filter_buffers, filter_data, filter_tasks] =
+      cudf::io::parquet::fetch_byte_ranges_to_device_async(
+        *datasource, filter_byte_ranges, stream, mr);
+    filter_tasks.get();
+
+    filter_tables.push_back(
+      reader
+        ->materialize_filter_columns(row_group_indices,
+                                     filter_data,
+                                     row_mask_view,
+                                     cudf::io::parquet::experimental::use_data_page_mask::YES,
+                                     options,
+                                     stream,
+                                     mr)
+        .tbl);
+
+    // Materialize payload columns for this pass using the same per-pass row mask
+    auto const payload_byte_ranges =
+      reader->payload_column_chunks_byte_ranges(row_group_indices, options);
+    auto [payload_buffers, payload_data, payload_tasks] =
+      cudf::io::parquet::fetch_byte_ranges_to_device_async(
+        *datasource, payload_byte_ranges, stream, mr);
+    payload_tasks.get();
+
+    payload_tables.push_back(
+      reader
+        ->materialize_payload_columns(row_group_indices,
+                                      payload_data,
+                                      row_mask_view,
+                                      cudf::io::parquet::experimental::use_data_page_mask::YES,
+                                      options,
+                                      stream,
+                                      mr)
+        .tbl);
+  };
+
+  // Split the row groups into two passes and materialize each pass independently
+  auto const row_group_span = cudf::host_span<cudf::size_type const>(all_row_groups);
+  ASSERT_GT(row_group_span.size(), 1);
+  auto const split = row_group_span.size() / 2;
+  materialize_pass(row_group_span.subspan(0, split));
+  materialize_pass(row_group_span.subspan(split, row_group_span.size() - split));
+
+  auto const filter_table = cudf::concatenate(
+    std::vector<cudf::table_view>{filter_tables[0]->view(), filter_tables[1]->view()});
+  auto const payload_table = cudf::concatenate(
+    std::vector<cudf::table_view>{payload_tables[0]->view(), payload_tables[1]->view()});
+
+  // Read the same file with the mainline parquet reader for comparison
+  auto const expected_options =
+    cudf::io::parquet_reader_options::builder(
+      cudf::io::source_info(cudf::host_span<char>(parquet_buffer.data(), parquet_buffer.size())))
+      .filter(filter_expression)
+      .build();
+  auto const expected = cudf::io::read_parquet(expected_options, stream).tbl;
+
+  // The filter column (col0) and payload columns (col1, col2) must match the mainline reader
+  CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected->select({0}), filter_table->view());
+  CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected->select({1, 2}), payload_table->view());
+}
+
 TEST_F(HybridScanTest, RowGroupPassesMatchesChunkedReader)
 {
   auto constexpr num_rg      = 10;