Add hybrid scan multifile two-step materializers

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -198,6 +198,87 @@ class hybrid_scan_multifile {
     rmm::cuda_stream_view stream,
     rmm::device_async_resource_ref mr) const;
 
+  /**
+   * @brief Get byte ranges of column chunks of filter columns
+   *
+   * Byte ranges are flattened in source order. Within each source, byte ranges follow the selected
+   * row group and column chunk order used by `row_group_indices` and `options`. The returned source
+   * map has one source index per byte range and can be used to regroup byte ranges by datasource
+   * before fetching.
+   *
+   * @param row_group_indices Input row group indices, one inner vector per source
+   * @param options Parquet reader options
+   * @return Pair of flattened byte ranges to column chunks of filter columns and their
+   *         corresponding source indices
+   */
+  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+  filter_column_chunks_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
+                                   parquet_reader_options const& options) const;
+
+  /**
+   * @brief Materializes filter columns and updates the input row mask to only the rows that exist
+   * in the output table
+   *
+   * @param row_group_indices Input row group indices, one inner vector per source
+   * @param column_chunk_data Flattened device spans of filter column chunk data returned in the
+   * same order as `filter_column_chunks_byte_ranges`
+   * @param[in,out] row_mask Mutable boolean column spanning all selected rows across all sources
+   * and indicating surviving rows from page pruning
+   * @param mask_data_pages Whether to build and use a data page mask using the row mask
+   * @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 device memory for the output table
+   * @return Table of materialized filter columns and metadata
+   */
+  [[nodiscard]] table_with_metadata materialize_filter_columns(
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    cudf::host_span<cudf::device_span<uint8_t const> const> column_chunk_data,
+    cudf::mutable_column_view& row_mask,
+    use_data_page_mask mask_data_pages,
+    parquet_reader_options const& options,
+    rmm::cuda_stream_view stream,
+    rmm::device_async_resource_ref mr) const;
+
+  /**
+   * @brief Get byte ranges of column chunks of payload columns
+   *
+   * Byte ranges are flattened in source order. Within each source, byte ranges follow the selected
+   * row group and column chunk order used by `row_group_indices` and `options`. The returned source
+   * map has one source index per byte range and can be used to regroup byte ranges by datasource
+   * before fetching.
+   *
+   * @param row_group_indices Input row group indices, one inner vector per source
+   * @param options Parquet reader options
+   * @return Pair of flattened byte ranges to column chunks of payload columns and their
+   *         corresponding source indices
+   */
+  [[nodiscard]] std::pair<std::vector<byte_range_info>, std::vector<size_type>>
+  payload_column_chunks_byte_ranges(cudf::host_span<std::vector<size_type> const> row_group_indices,
+                                    parquet_reader_options const& options) const;
+
+  /**
+   * @brief Materialize payload columns and applies the row mask to the output table
+   *
+   * @param row_group_indices Input row group indices, one inner vector per source
+   * @param column_chunk_data Flattened device spans of payload column chunk data returned in the
+   *        same order as `payload_column_chunks_byte_ranges`
+   * @param row_mask Boolean column spanning all selected rows across all sources and indicating
+   *        which rows need to be read. All rows read if empty
+   * @param mask_data_pages Whether to build and use a data page mask using the row mask
+   * @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 device memory for the output table
+   * @return Table of materialized payload columns and metadata
+   */
+  [[nodiscard]] table_with_metadata materialize_payload_columns(
+    cudf::host_span<std::vector<size_type> const> row_group_indices,
+    cudf::host_span<cudf::device_span<uint8_t const> const> column_chunk_data,
+    cudf::column_view const& row_mask,
+    use_data_page_mask mask_data_pages,
+    parquet_reader_options const& options,
+    rmm::cuda_stream_view stream,
+    rmm::device_async_resource_ref mr) const;
+
   /**
    * @brief Get byte ranges of column chunks of all (or selected) columns
    *

cpp/src/io/parquet/experimental/hybrid_scan.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -370,6 +370,8 @@ table_with_metadata hybrid_scan_reader::materialize_all_columns_chunk() const
 std::vector<std::vector<cudf::size_type>> hybrid_scan_reader::construct_row_group_passes(
   cudf::host_span<cudf::size_type const> row_group_indices, std::size_t pass_read_limit) const
 {
+  CUDF_FUNC_RANGE();
+
   auto const total_row_groups = row_group_indices.size();
 
   CUDF_EXPECTS(

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -105,6 +105,52 @@ std::unique_ptr<cudf::column> hybrid_scan_multifile::build_row_mask_with_page_in
   return _impl->build_row_mask_with_page_index_stats(row_group_indices, options, stream, mr);
 }
 
+std::pair<std::vector<text::byte_range_info>, std::vector<size_type>>
+hybrid_scan_multifile::filter_column_chunks_byte_ranges(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  parquet_reader_options const& options) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->filter_column_chunks_byte_ranges(row_group_indices, options);
+}
+
+table_with_metadata hybrid_scan_multifile::materialize_filter_columns(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  cudf::host_span<cudf::device_span<uint8_t const> const> column_chunk_data,
+  cudf::mutable_column_view& row_mask,
+  use_data_page_mask mask_data_pages,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->materialize_filter_columns(
+    row_group_indices, column_chunk_data, row_mask, mask_data_pages, options, stream, mr);
+}
+
+std::pair<std::vector<text::byte_range_info>, std::vector<size_type>>
+hybrid_scan_multifile::payload_column_chunks_byte_ranges(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  parquet_reader_options const& options) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->payload_column_chunks_byte_ranges(row_group_indices, options);
+}
+
+table_with_metadata hybrid_scan_multifile::materialize_payload_columns(
+  cudf::host_span<std::vector<size_type> const> row_group_indices,
+  cudf::host_span<cudf::device_span<uint8_t const> const> column_chunk_data,
+  cudf::column_view const& row_mask,
+  use_data_page_mask mask_data_pages,
+  parquet_reader_options const& options,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr) const
+{
+  CUDF_FUNC_RANGE();
+  return _impl->materialize_payload_columns(
+    row_group_indices, column_chunk_data, row_mask, mask_data_pages, options, stream, mr);
+}
+
 std::pair<std::vector<text::byte_range_info>, std::vector<size_type>>
 hybrid_scan_multifile::all_column_chunks_byte_ranges(
   cudf::host_span<std::vector<size_type> const> row_group_indices,
@@ -129,6 +175,8 @@ std::vector<std::vector<std::vector<size_type>>> hybrid_scan_multifile::construc
   cudf::host_span<std::vector<size_type> const> row_group_indices,
   std::size_t pass_read_limit) const
 {
+  CUDF_FUNC_RANGE();
+
   auto const total_row_groups =
     std::accumulate(row_group_indices.begin(),
                     row_group_indices.end(),

cpp/tests/CMakeLists.txt

@@ -1,6 +1,6 @@
 # =============================================================================
 # cmake-format: off
-# SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION.
+# SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 # cmake-format: on
 # =============================================================================
@@ -348,6 +348,7 @@ ConfigureTest(
 ConfigureTest(
   HYBRID_SCAN_TEST
   io/experimental/hybrid_scan_composer.cpp
+  io/experimental/hybrid_scan_multifile_composer.cpp
   io/experimental/hybrid_scan_filters_test.cpp
   io/experimental/hybrid_scan_multifile_filters_test.cpp
   io/experimental/hybrid_scan_multifile_test.cpp

cpp/tests/io/experimental/hybrid_scan_multifile_composer.cpp

@@ -0,0 +1,164 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "hybrid_scan_multifile_composer.hpp"
+
+#include "hybrid_scan_multifile_common.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/error.hpp>
+
+#include <rmm/device_buffer.hpp>
+
+#include <algorithm>
+#include <cstddef>
+#include <memory>
+#include <vector>
+
+using cudf::io::parquet::experimental::use_data_page_mask;
+
+namespace {
+
+/**
+ * @brief Device buffers and spans from multiple input sources
+ */
+struct multisource_device_data {
+  std::vector<rmm::device_buffer> buffers;
+  std::vector<std::vector<cudf::device_span<uint8_t const>>> per_source_spans;
+  std::vector<cudf::device_span<uint8_t const>> flat_spans;
+};
+
+/**
+ * @brief Groups a flat byte range list by source using the specified source map
+ */
+std::vector<std::vector<cudf::io::text::byte_range_info>> group_byte_ranges_by_source(
+  std::pair<std::vector<cudf::io::text::byte_range_info>, std::vector<cudf::size_type>> const&
+    byte_ranges_and_source_map,
+  std::size_t num_sources)
+{
+  auto const& [byte_ranges, source_map] = byte_ranges_and_source_map;
+  CUDF_EXPECTS(byte_ranges.size() == source_map.size(), "Invalid source map size");
+
+  auto byte_ranges_per_source =
+    std::vector<std::vector<cudf::io::text::byte_range_info>>(num_sources);
+  std::for_each(byte_ranges.begin(),
+                byte_ranges.end(),
+                [&, range_index = std::size_t{0}](auto const& range) mutable {
+                  auto const source_index = source_map[range_index++];
+                  CUDF_EXPECTS(source_index >= 0 and static_cast<std::size_t>(source_index) <
+                                                       byte_ranges_per_source.size(),
+                               "Invalid byte range source index");
+                  byte_ranges_per_source[source_index].push_back(range);
+                });
+  return byte_ranges_per_source;
+}
+
+/**
+ * @brief Fetches byte ranges from multiple sources and returns per-source and flattened spans
+ */
+multisource_device_data fetch_multisource_device_data(
+  multifile_inputs const& inputs,
+  std::pair<std::vector<cudf::io::text::byte_range_info>, std::vector<cudf::size_type>> const&
+    byte_ranges_and_source_map,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  auto const byte_ranges_per_source =
+    group_byte_ranges_by_source(byte_ranges_and_source_map, inputs.datasources.size());
+  auto [buffers, per_source_spans, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
+    inputs.datasource_refs,
+    cudf::host_span<std::vector<cudf::io::text::byte_range_info> const>{byte_ranges_per_source},
+    stream,
+    mr);
+  tasks.get();
+
+  auto flat_spans = std::vector<cudf::device_span<uint8_t const>>{};
+  for (auto const& source_spans : per_source_spans) {
+    flat_spans.insert(flat_spans.end(), source_spans.begin(), source_spans.end());
+  }
+
+  return {std::move(buffers), std::move(per_source_spans), std::move(flat_spans)};
+}
+
+}  // namespace
+
+std::tuple<std::unique_ptr<cudf::table>, std::unique_ptr<cudf::table>> hybrid_scan_multifile(
+  cudf::io::source_info const& source_info,
+  cudf::ast::operation const& filter_expression,
+  std::optional<std::vector<std::string>> const& payload_column_names,
+  bool case_sensitive_names,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  auto options = cudf::io::parquet_reader_options::builder()
+                   .filter(filter_expression)
+                   .case_sensitive_names(case_sensitive_names)
+                   .build();
+  if (payload_column_names.has_value()) { options.set_column_names(payload_column_names.value()); }
+
+  auto inputs = multifile_inputs(source_info);
+  auto reader =
+    cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, options};
+  setup_page_indexes(reader, inputs);
+
+  auto const input_row_groups = reader.all_row_groups(options);
+  auto const row_groups = reader.filter_row_groups_with_stats(input_row_groups, options, stream);
+  auto row_mask = reader.build_row_mask_with_page_index_stats(row_groups, options, stream, mr);
+
+  auto filter_column_chunks = fetch_multisource_device_data(
+    inputs, reader.filter_column_chunks_byte_ranges(row_groups, options), stream, mr);
+  auto row_mask_view = row_mask->mutable_view();
+  auto filter_result = reader.materialize_filter_columns(row_groups,
+                                                         filter_column_chunks.flat_spans,
+                                                         row_mask_view,
+                                                         use_data_page_mask::YES,
+                                                         options,
+                                                         stream,
+                                                         mr);
+
+  auto payload_column_chunks = fetch_multisource_device_data(
+    inputs, reader.payload_column_chunks_byte_ranges(row_groups, options), stream, mr);
+  auto payload_result = reader.materialize_payload_columns(row_groups,
+                                                           payload_column_chunks.flat_spans,
+                                                           row_mask_view,
+                                                           use_data_page_mask::YES,
+                                                           options,
+                                                           stream,
+                                                           mr);
+
+  return std::tuple{std::move(filter_result.tbl), std::move(payload_result.tbl)};
+}
+
+std::unique_ptr<cudf::table> hybrid_scan_multifile_single_step(
+  cudf::io::source_info const& source_info,
+  cudf::ast::operation const& filter_expression,
+  std::optional<std::vector<std::string>> const& column_names,
+  bool case_sensitive_names,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  auto options = cudf::io::parquet_reader_options::builder()
+                   .filter(filter_expression)
+                   .case_sensitive_names(case_sensitive_names)
+                   .build();
+  if (column_names.has_value()) { options.set_column_names(column_names.value()); }
+
+  auto inputs = multifile_inputs(source_info);
+  auto reader =
+    cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, options};
+  setup_page_indexes(reader, inputs);
+
+  auto const input_row_groups = reader.all_row_groups(options);
+  auto const row_groups = reader.filter_row_groups_with_stats(input_row_groups, options, stream);
+
+  auto all_column_chunks = fetch_multisource_device_data(
+    inputs, reader.all_column_chunks_byte_ranges(row_groups, options), stream, mr);
+  return reader
+    .materialize_all_columns(row_groups, all_column_chunks.flat_spans, options, stream, mr)
+    .tbl;
+}