Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 83 additions & 0 deletions cpp/include/cudf/io/experimental/hybrid_scan.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,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
Expand All @@ -52,6 +57,72 @@ 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 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);
Comment thread
Matt711 marked this conversation as resolved.
Outdated
* @endcode
*
* @note The metadata is immutable after `setup_page_index()` has been called (or immediately after
* construction if page index setup is skipped). Readers sharing one instance may read different
* row-group ranges of the same single file concurrently; overlapping ranges produce duplicate rows.
* This handle does not support multi-source (multi-file) metadata.
Comment thread
Matt711 marked this conversation as resolved.
Outdated
*/
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);
Comment on lines +92 to +98

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

non-blocking as it needs a broader fix in a follow up. This may be misleading as the FileMetaData object is copied internally by metadata(FileMetaData&) and any subsequent operations (such as setting page index) on hybrid_scan_metadata won't affect the original FileMetaData it was constructed from.


/**
* @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;
Comment thread
wence- marked this conversation as resolved.
};

/**
* @brief The experimental parquet reader class to optimally read parquet files subject to
* highly selective filters, called a Hybrid Scan operation
Expand Down Expand Up @@ -300,6 +371,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.
Comment thread
Matt711 marked this conversation as resolved.
Outdated
*
* @param metadata Shared, pre-parsed Parquet file metadata
*/
explicit hybrid_scan_reader(hybrid_scan_metadata const& metadata);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: the type signature of this method (and I know it is similar to hybrid_scan_reader(FileMetaData const&) above) is kind of misleading. If I pass a const& to some function then, usually, I am on the hook to keep the input alive as long as the thing I've passed it to is referencing it.

But here that is not the case, because a hybrid_scan_metadata is really just a shared_ptr<detail::aggregate_reader_metadata> in a trenchcoat, and the hybrid_scan_reader ctor just reaches inside and grabs that shared pointer.

I wonder, and will summon @mhaseeb123, if it makes more sense to do a in some sense "larger" refactor:

  • Make aggregate_reader_metadata a public type
  • Store a std::shared_ptr<aggregate_reader_metadata const> _metadata in the hybrid_scan_reader.

Introduce a new ctor hybrid_scan_reader(std::shared_ptr<aggregate_reader_metadata const> reader)

An alternative would be for the hybrid_scan_reader to take a raw pointer, but then lifetime management would be on the caller, which is fiddlier.

WDYT?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about we take it by-value instead? 916cd4c

Copying the "hybrid_scan_metadata" just means copying the shared_ptr

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FYI Haseeb's out until next week, so would love to get this in sooner. I feel more comfortable doing that because hybrid scan stuff is in an experimental namespace.

@mhaseeb123 mhaseeb123 Aug 18, 2026 •

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think taking in a shared ptr of FileMetaData such as hybrid_scan_reader(shared_ptr<FileMetaData>) and/or a move hybrid_scan_reader(FileMetaData&&) makes more sense. Currently we just copy from the const ref anyway (see constructor of metadata object in hybrid_scan_helpers.cpp) so discarding original footer is safe.

Make aggregate_reader_metadata a public type

This isn't really helpful as aggregate_reader_metadata is just FileMetaData with some extra functions not really needed at public level.

Store a std::shared_ptr<aggregate_reader_metadata const> _metadata in the hybrid_scan_reader.
Introduce a new ctor hybrid_scan_reader(std::shared_ptr<aggregate_reader_metadata const> reader)

This one makes more sense to me.


/**
* @brief Destructor for the experimental parquet reader class
*/
Expand Down
25 changes: 25 additions & 0 deletions cpp/src/io/parquet/experimental/hybrid_scan.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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>(
Expand All @@ -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
Expand Down
12 changes: 10 additions & 2 deletions cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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(), has_cols_from_mismatched_sources(options));

_extended_metadata = static_cast<aggregate_reader_metadata*>(_metadata.get());
Expand All @@ -108,12 +108,20 @@ 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_metadatas,
std::make_shared<aggregate_reader_metadata>(parquet_metadatas,
options.is_enabled_use_arrow_schema(),
has_cols_from_mismatched_sources(options));
_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);
Comment thread
Matt711 marked this conversation as resolved.
_extended_metadata = static_cast<aggregate_reader_metadata*>(_metadata.get());
}

std::vector<FileMetaData> hybrid_scan_reader_impl::parquet_metadatas() const
{
return _extended_metadata->parquet_metadatas();
Expand Down
11 changes: 11 additions & 0 deletions cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Usually if one passes a shared_pointer we're not taking a borrow, but rather grabbing a new reference.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in 916cd4c

*
* @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
*/
Expand Down
4 changes: 2 additions & 2 deletions cpp/src/io/parquet/reader_impl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -533,11 +533,11 @@ reader_impl::reader_impl(std::size_t chunk_read_limit,
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(),
has_cols_from_mismatched_sources(options))
: 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(),
has_cols_from_mismatched_sources(options));
Expand Down
2 changes: 1 addition & 1 deletion cpp/src/io/parquet/reader_impl.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -548,7 +548,7 @@ 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;
std::shared_ptr<aggregate_reader_metadata> _metadata;

// Number of sources
size_t _num_sources{0};
Expand Down
1 change: 1 addition & 0 deletions cpp/tests/io/experimental/hybrid_scan_common.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -419,6 +419,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);

Expand Down
95 changes: 95 additions & 0 deletions cpp/tests/io/experimental/hybrid_scan_test.cpp

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Based on my other feedback, can you add a test that actually creates multiple readers? That would suss out the kinds of issues I was worried about above.

Original file line number Diff line number Diff line change
Expand Up @@ -985,6 +985,101 @@ 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, SharedMetadataFromFileMetaDataMatchesReadParquet)
{
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()));

// Obtain FileMetaData from an initial reader, then build shared metadata from it.
auto const footer_buffer = cudf::io::parquet::fetch_footer_to_host(*datasource);
auto const seed_reader =
std::make_unique<cudf::io::parquet::experimental::hybrid_scan_reader>(*footer_buffer, options);
auto const file_metadata = seed_reader->parquet_metadata();

auto const metadata =
std::make_shared<cudf::io::parquet::experimental::hybrid_scan_metadata>(file_metadata, options);

// Two readers sharing the FileMetaData-derived metadata each produce the correct table.
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;
};

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;
Expand Down
4 changes: 3 additions & 1 deletion python/pylibcudf/pylibcudf/io/experimental/__init__.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,16 @@
# 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

from pylibcudf.io.experimental.hybrid_scan import (
HybridScanMetadata,
HybridScanReader,
UseDataPageMask,
)
from pylibcudf.io.parquet_metadata import FileMetaData

__all__ = [
"FileMetaData", # backwards compatibility
"HybridScanMetadata",
"HybridScanReader",
"UseDataPageMask",
]
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# 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

from libc.stdint cimport uint8_t
Expand All @@ -13,6 +13,7 @@ from pylibcudf.io.parquet cimport ParquetReaderOptions
from pylibcudf.io.parquet_metadata cimport FileMetaData as c_FileMetaData
from pylibcudf.io.types cimport TableWithMetadata
from pylibcudf.libcudf.io.hybrid_scan cimport (
hybrid_scan_metadata as cpp_hybrid_scan_metadata,
hybrid_scan_reader as cpp_hybrid_scan_reader,
use_data_page_mask,
)
Expand All @@ -23,7 +24,13 @@ from pylibcudf.libcudf.utilities.span cimport device_span
cdef device_span[const_uint8_t] _get_device_span(object obj) except *


cdef class HybridScanMetadata:
cdef unique_ptr[cpp_hybrid_scan_metadata] c_obj


cdef class HybridScanReader:
cdef unique_ptr[cpp_hybrid_scan_reader] c_obj
cdef Stream _stream
cdef DeviceMemoryResource mr
cdef object _filter_chunk_data
cdef object _payload_chunk_data
Loading
Loading