Dispatch SplitScan to HybridScanReader in the streaming engine

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

@@ -26,6 +26,11 @@ namespace cudf::io::parquet::experimental::detail {
  *        Hybrid Scan operation.
  */
 class hybrid_scan_reader_impl;
+
+/**
+ * @brief Internal parsed Parquet file metadata for the Hybrid Scan reader.
+ */
+class aggregate_reader_metadata;
 }  // namespace cudf::io::parquet::experimental::detail
 
 //! Using `byte_range_info` from cudf::io::text
@@ -49,6 +54,73 @@ enum class use_data_page_mask : bool {
   NO  = false  ///< Do not compute or use a data page mask
 };
 
+/**
+ * @brief Shareable, pre-parsed Parquet file metadata for the Hybrid Scan reader
+ *
+ * Parses the Parquet file metadata once so that multiple `hybrid_scan_reader` instances reading the
+ * same file can borrow it rather than each re-parsing and copying the (potentially large) row group
+ * metadata. The intended use is to read disjoint row-group ranges of a single file: construct one
+ * `hybrid_scan_metadata` per file and pass it to as many readers as there are ranges.
+ *
+ * @code{.cpp}
+ * // Parse the metadata once
+ * auto metadata = std::make_shared<parquet::experimental::hybrid_scan_metadata>(*footer_buffer,
+ *                                                                               options);
+ * // Construct lightweight readers that share it
+ * auto reader_a = std::make_unique<parquet::experimental::hybrid_scan_reader>(*metadata);
+ * auto reader_b = std::make_unique<parquet::experimental::hybrid_scan_reader>(*metadata);
+ * @endcode
+ *
+ * @note The metadata is immutable once constructed. Readers sharing one instance must read disjoint
+ * row-group ranges of the same single file; such reads do not mutate the shared metadata, so they
+ * may run concurrently. This handle does not support multi-source (multi-file) metadata.
+ */
+class hybrid_scan_metadata {
+ public:
+  /**
+   * @brief Parse and own Parquet file metadata from a span of footer bytes
+   *
+   * @param footer_bytes Host span of Parquet file footer bytes
+   * @param options Parquet reader options
+   */
+  hybrid_scan_metadata(cudf::host_span<uint8_t const> footer_bytes,
+                       parquet_reader_options const& options);
+
+  /**
+   * @brief Own Parquet file metadata from a pre-populated `FileMetaData`
+   *
+   * @param parquet_metadata Pre-populated Parquet file metadata
+   * @param options Parquet reader options
+   */
+  hybrid_scan_metadata(FileMetaData const& parquet_metadata, parquet_reader_options const& options);
+
+  /**
+   * @brief Destructor for the shared Parquet metadata
+   */
+  ~hybrid_scan_metadata();
+
+  hybrid_scan_metadata(hybrid_scan_metadata const&) = default;  ///< Copy constructor
+  hybrid_scan_metadata(hybrid_scan_metadata&&)      = default;  ///< Move constructor
+
+  /**
+   * @brief Copy assignment operator
+   *
+   * @return Reference to this object
+   */
+  hybrid_scan_metadata& operator=(hybrid_scan_metadata const&) = default;
+
+  /**
+   * @brief Move assignment operator
+   *
+   * @return Reference to this object
+   */
+  hybrid_scan_metadata& operator=(hybrid_scan_metadata&&) = default;
+
+ private:
+  std::shared_ptr<detail::aggregate_reader_metadata> _metadata;
+  friend class hybrid_scan_reader;
+};
+
 /**
  * @brief The experimental parquet reader class to optimally read parquet files subject to
  *        highly selective filters, called a Hybrid Scan operation
@@ -297,6 +369,18 @@ class hybrid_scan_reader {
   explicit hybrid_scan_reader(FileMetaData const& parquet_metadata,
                               parquet_reader_options const& options);
 
+  /**
+   * @brief Constructor that borrows shared, pre-parsed Parquet file metadata
+   *
+   * Constructs a reader that shares `metadata` instead of parsing and copying the file metadata
+   * again. Use this to read disjoint row-group ranges of a single file without paying the metadata
+   * copy per reader. The reader options that govern reading (column selection, filter, ...) are
+   * supplied per call to the individual read methods.
+   *
+   * @param metadata Shared, pre-parsed Parquet file metadata
+   */
+  explicit hybrid_scan_reader(hybrid_scan_metadata const& metadata);
+
   /**
    * @brief Destructor for the experimental parquet reader class
    */

cpp/include/cudf/io/parquet_io_utils.hpp

@@ -152,6 +152,28 @@ fetch_byte_ranges_to_device_async(
   rmm::cuda_stream_view stream,
   rmm::device_async_resource_ref mr);
 
+/**
+ * @brief Fetches byte ranges from a single datasource into device buffers, blocking until complete
+ *
+ * @ingroup io_utils
+ *
+ * Convenience wrapper around `fetch_byte_ranges_to_device_async` that waits for the
+ * returned future before returning.
+ *
+ * @param datasource Input datasource
+ * @param byte_ranges Byte ranges to fetch
+ * @param stream CUDA stream
+ * @param mr Device memory resource
+ *
+ * @return A pair containing the device buffers and the device spans of the fetched data
+ */
+[[nodiscard]] std::pair<std::vector<rmm::device_buffer>,
+                        std::vector<cudf::device_span<uint8_t const>>>
+fetch_byte_ranges_to_device(cudf::io::datasource& datasource,
+                            cudf::host_span<byte_range_info const> byte_ranges,
+                            rmm::cuda_stream_view stream,
+                            rmm::device_async_resource_ref mr);
+
 /** @} */  // end of group
 }  // namespace io::parquet
 }  // namespace CUDF_EXPORT cudf

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

@@ -13,6 +13,26 @@
 
 namespace cudf::io::parquet::experimental {
 
+hybrid_scan_metadata::hybrid_scan_metadata(cudf::host_span<uint8_t const> footer_bytes,
+                                           parquet_reader_options const& options)
+  : _metadata{std::make_shared<detail::aggregate_reader_metadata>(
+      std::vector<cudf::host_span<uint8_t const>>{footer_bytes},
+      options.is_enabled_use_arrow_schema(),
+      options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas())}
+{
+}
+
+hybrid_scan_metadata::hybrid_scan_metadata(FileMetaData const& parquet_metadata,
+                                           parquet_reader_options const& options)
+  : _metadata{std::make_shared<detail::aggregate_reader_metadata>(
+      std::vector<FileMetaData>{parquet_metadata},
+      options.is_enabled_use_arrow_schema(),
+      options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas())}
+{
+}
+
+hybrid_scan_metadata::~hybrid_scan_metadata() = default;
+
 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>(
@@ -27,6 +47,11 @@ hybrid_scan_reader::hybrid_scan_reader(FileMetaData const& parquet_metadata,
 {
 }
 
+hybrid_scan_reader::hybrid_scan_reader(hybrid_scan_metadata const& metadata)
+  : _impl{std::make_unique<detail::hybrid_scan_reader_impl>(metadata._metadata)}
+{
+}
+
 hybrid_scan_reader::~hybrid_scan_reader() = default;
 
 [[nodiscard]] text::byte_range_info hybrid_scan_reader::page_index_byte_range() const

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

@@ -98,7 +98,7 @@ 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)
 {
-  _metadata = std::make_unique<aggregate_reader_metadata>(
+  _metadata = std::make_shared<aggregate_reader_metadata>(
     footer_bytes,
     options.is_enabled_use_arrow_schema(),
     options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas());
@@ -109,13 +109,21 @@ hybrid_scan_reader_impl::hybrid_scan_reader_impl(
 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>(
+  _metadata = std::make_shared<aggregate_reader_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());
 }
 
+hybrid_scan_reader_impl::hybrid_scan_reader_impl(
+  std::shared_ptr<aggregate_reader_metadata> metadata)
+{
+  CUDF_EXPECTS(metadata != nullptr, "Shared parquet metadata must not be null");
+  _metadata          = std::move(metadata);
+  _extended_metadata = static_cast<aggregate_reader_metadata*>(_metadata.get());
+}
+
 std::vector<FileMetaData> hybrid_scan_reader_impl::parquet_metadatas() const
 {
   return _extended_metadata->parquet_metadatas();

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

@@ -59,6 +59,17 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl {
   explicit hybrid_scan_reader_impl(cudf::host_span<FileMetaData const> parquet_metadatas,
                                    parquet_reader_options const& options);
 
+  /**
+   * @brief Constructor that shares pre-parsed Parquet metadata
+   *
+   * Borrows an already-constructed `aggregate_reader_metadata` instead of parsing and copying the
+   * file metadata again. Multiple single-file readers can share one metadata object, avoiding a
+   * per-reader copy of the (potentially large) row group metadata.
+   *
+   * @param metadata Shared, pre-parsed Parquet file metadata. Must not be null.
+   */
+  explicit hybrid_scan_reader_impl(std::shared_ptr<aggregate_reader_metadata> metadata);
+
   /**
    * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::parquet_metadatas
    */

cpp/src/io/parquet/io_utils/parquet_io_utils.cpp

@@ -471,4 +471,17 @@ fetch_byte_ranges_to_device_async(
     mr);
 }
 
+std::pair<std::vector<rmm::device_buffer>, std::vector<cudf::device_span<uint8_t const>>>
+fetch_byte_ranges_to_device(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)
+{
+  CUDF_FUNC_RANGE();
+  auto [buffers, spans, fut] =
+    fetch_byte_ranges_to_device_async(datasource, byte_ranges, stream, mr);
+  fut.get();
+  return {std::move(buffers), std::move(spans)};
+}
+
 }  // namespace cudf::io::parquet

cpp/src/io/parquet/reader_impl.cpp

@@ -522,12 +522,12 @@ reader_impl::reader_impl(std::size_t chunk_read_limit,
   // Open and parse the source dataset metadata
   CUDF_EXPECTS(file_metadatas.empty() or file_metadatas.size() == _sources.size(),
                "Encountered a mismatch in the number of provided data sources and metadatas");
-  _metadata = file_metadatas.empty() ? std::make_unique<aggregate_reader_metadata>(
+  _metadata = file_metadatas.empty() ? std::make_shared<aggregate_reader_metadata>(
                                          _sources,
                                          options.is_enabled_use_arrow_schema(),
                                          options.get_column_names().has_value() and
                                            options.is_enabled_allow_mismatched_pq_schemas())
-                                     : std::make_unique<aggregate_reader_metadata>(
+                                     : std::make_shared<aggregate_reader_metadata>(
                                          std::forward<std::vector<FileMetaData>>(file_metadatas),
                                          options.is_enabled_use_arrow_schema(),
                                          options.get_column_names().has_value() and

cpp/src/io/parquet/reader_impl.hpp

@@ -489,7 +489,8 @@ class reader_impl {
   named_to_reference_converter _expr_conv{std::nullopt, table_metadata{}, true};
 
   std::vector<std::unique_ptr<datasource>> _sources;
-  std::unique_ptr<aggregate_reader_metadata> _metadata;
+  // shared so experimental hybrid scan readers can share one copy across single-file readers
+  std::shared_ptr<aggregate_reader_metadata> _metadata;
 
   // Number of sources
   size_t _num_sources{0};

cpp/tests/io/experimental/hybrid_scan_common.cpp

@@ -328,6 +328,7 @@ std::pair<std::unique_ptr<cudf::table>, std::vector<char>> create_parquet_with_s
   INSTANTIATE_CREATE_PARQUET_WITH_STATS(T, 1, true, true)
 
 INSTANTIATE_CREATE_PARQUET_WITH_STATS(uint32_t, 4, true, false);
+INSTANTIATE_CREATE_PARQUET_WITH_STATS(int32_t, 2, true, false);
 INSTANTIATE_CREATE_PARQUET_WITH_STATS(cudf::timestamp_ms, 2, true, false);
 INSTANTIATE_CREATE_PARQUET_WITH_STATS(cudf::duration_ms, 2, true, false);
 

cpp/tests/io/experimental/hybrid_scan_test.cpp

@@ -878,6 +878,52 @@ TEST_F(HybridScanTest, StructChildFilterColumn)
     std::invalid_argument);
 }
 
+TEST_F(HybridScanTest, SharedMetadataReaderMatchesReadParquet)
+{
+  using T                              = int32_t;
+  auto constexpr num_concat            = 2;
+  auto [written_table, parquet_buffer] = create_parquet_with_stats<T, num_concat>();
+
+  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().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);
+
+  // Parse the file metadata once and share it across independent readers.
+  auto const metadata = std::make_shared<cudf::io::parquet::experimental::hybrid_scan_metadata>(
+    *footer_buffer, options);
+
+  // Read all columns (single step) through a reader that borrows the shared metadata.
+  auto const read_all_columns = [&] {
+    auto const reader =
+      std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(*metadata);
+    auto const row_groups   = reader->all_row_groups(options);
+    auto const chunk_ranges = reader->all_column_chunks_byte_ranges(row_groups, options);
+    auto [buffers, data, tasks] =
+      cudf::io::parquet::fetch_byte_ranges_to_device_async(*datasource, chunk_ranges, stream, mr);
+    tasks.get();
+    return reader->materialize_all_columns(row_groups, data, options, stream, mr).tbl;
+  };
+
+  // Two readers sharing one metadata instance each produce the same table as the main reader.
+  auto const table_a = read_all_columns();
+  auto const table_b = read_all_columns();
+
+  auto const expected =
+    cudf::io::read_parquet(
+      cudf::io::parquet_reader_options::builder(
+        cudf::io::source_info(cudf::host_span<char>(parquet_buffer.data(), parquet_buffer.size())))
+        .build(),
+      stream)
+      .tbl;
+
+  CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected->view(), table_a->view());
+  CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected->view(), table_b->view());
+}
+
 TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups)
 {
   using cudf::io::parquet::experimental::use_data_page_mask;

cpp/tests/strings/contains_tests.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 

cpp/tests/strings/findall_tests.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 

cpp/tests/strings/replace_regex_tests.cpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
 

docs/cudf/source/pylibcudf/api_docs/io/experimental.rst

@@ -0,0 +1,8 @@
+============
+Experimental
+============
+
+APIs in this namespace are experimental and may change without warning in the future.
+
+.. automodule:: pylibcudf.io.experimental
+   :members:

docs/cudf/source/pylibcudf/api_docs/io/index.rst

@@ -17,9 +17,11 @@ I/O Functions
 
     avro
     csv
+    experimental
     json
     orc
     parquet
+    parquet_io_utils
     parquet_metadata
     text
     timezone

docs/cudf/source/pylibcudf/api_docs/io/parquet_io_utils.rst

@@ -0,0 +1,6 @@
+================
+Parquet IO Utils
+================
+
+.. automodule:: pylibcudf.io.parquet_io_utils
+   :members:

python/cudf_polars/cudf_polars/dsl/ir.py

@@ -49,7 +49,10 @@
 from cudf_polars.dsl.expressions.base import ExecutionContext
 from cudf_polars.dsl.nodebase import Node
 from cudf_polars.dsl.to_ast import _DECIMAL_IDS, to_ast, to_parquet_filter
-from cudf_polars.dsl.tracing import log_do_evaluate, nvtx_annotate_cudf_polars
+from cudf_polars.dsl.tracing import (
+    log_do_evaluate,
+    nvtx_annotate_cudf_polars,
+)
 from cudf_polars.dsl.utils.reshape import broadcast
 from cudf_polars.dsl.utils.windows import (
     offsets_to_windows,