Enable page-level payload I/O in hybrid scan

cpp/examples/hybrid_scan_io/CMakeLists.txt

@@ -50,6 +50,7 @@ add_hybrid_scan_example(hybrid_scan_io hybrid_scan_io.cpp)
 add_hybrid_scan_example(hybrid_scan_pipeline hybrid_scan_pipeline.cpp)
 add_hybrid_scan_example(hybrid_scan_multifile_single_step hybrid_scan_multifile_single_step.cpp)
 add_hybrid_scan_example(hybrid_scan_multifile_two_step hybrid_scan_multifile_two_step.cpp)
+add_hybrid_scan_example(mint1t_hybrid_scan mint1t_hybrid_scan.cpp)
 
 # Install the example.parquet file
 install(FILES ${CMAKE_CURRENT_LIST_DIR}/example.parquet

cpp/examples/hybrid_scan_io/io_utils.cpp

@@ -3,6 +3,8 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include "io_utils.hpp"
+
 #include <cudf/io/datasource.hpp>
 #include <cudf/io/parquet_io_utils.hpp>
 #include <cudf/io/text/byte_range_info.hpp>
@@ -39,3 +41,73 @@ fetch_byte_ranges_async(cudf::io::datasource& datasource,
   // Using libcudf utility but may have custom implementation in the future
   return cudf::io::parquet::fetch_byte_ranges_to_device_async(datasource, byte_ranges, stream, mr);
 }
+
+multifile_inputs::multifile_inputs(cudf::io::source_info const& source_info)
+  : datasources{cudf::io::make_datasources(source_info)}
+{
+  datasource_refs.reserve(datasources.size());
+  std::transform(datasources.begin(),
+                 datasources.end(),
+                 std::back_inserter(datasource_refs),
+                 [](auto const& datasource) { return std::ref(*datasource); });
+}
+
+void multifile_inputs::fetch_footers()
+{
+  footer_buffers = cudf::io::parquet::fetch_footers_to_host(datasource_refs);
+  footer_byte_spans.clear();
+  footer_byte_spans.reserve(footer_buffers.size());
+  std::transform(footer_buffers.begin(),
+                 footer_buffers.end(),
+                 std::back_inserter(footer_byte_spans),
+                 [](auto const& buffer) { return cudf::host_span<uint8_t const>{*buffer}; });
+}
+
+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);
+  for (auto range_index = std::size_t{0}; range_index < byte_ranges.size(); ++range_index) {
+    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 source index");
+    byte_ranges_per_source[source_index].push_back(byte_ranges[range_index]);
+  }
+  return byte_ranges_per_source;
+}
+
+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());
+  return fetch_multisource_device_data(inputs, byte_ranges_per_source, stream, mr);
+}
+
+multisource_device_data fetch_multisource_device_data(
+  multifile_inputs const& inputs,
+  std::vector<std::vector<cudf::io::text::byte_range_info>> const& byte_ranges_per_source,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr)
+{
+  auto [buffers, per_source_spans, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async(
+    inputs.datasource_refs, 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)};
+}

cpp/examples/hybrid_scan_io/io_utils.hpp

@@ -57,3 +57,56 @@ fetch_byte_ranges_async(cudf::io::datasource& datasource,
                         cudf::host_span<cudf::io::text::byte_range_info const> byte_ranges,
                         rmm::cuda_stream_view stream,
                         rmm::device_async_resource_ref mr);
+
+/**
+ * @brief Owns the datasources, footer buffers, and byte spans for a multifile read.
+ *
+ * Call `fetch_footers()` after construction. Keeping datasource and footer setup separate allows
+ * examples to time those operations independently.
+ */
+struct multifile_inputs {
+  explicit multifile_inputs(cudf::io::source_info const& source_info);
+
+  void fetch_footers();
+
+  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 Owns multifile device buffers and the corresponding per-source and flattened spans.
+ */
+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 Regroups flattened byte ranges using the source map returned by Hybrid Scan.
+ */
+[[nodiscard]] 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);
+
+/**
+ * @brief Fetches source-mapped multifile byte ranges and flattens the resulting device spans.
+ */
+[[nodiscard]] 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);
+
+/**
+ * @brief Fetches source-grouped multifile byte ranges.
+ */
+[[nodiscard]] multisource_device_data fetch_multisource_device_data(
+  multifile_inputs const& inputs,
+  std::vector<std::vector<cudf::io::text::byte_range_info>> const& byte_ranges_per_source,
+  rmm::cuda_stream_view stream,
+  rmm::device_async_resource_ref mr);