diff --git a/cpp/include/cudf/io/experimental/hybrid_scan.hpp b/cpp/include/cudf/io/experimental/hybrid_scan.hpp index bf8991b8557b..772b65e62fc9 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan.hpp @@ -32,6 +32,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 @@ -53,6 +58,70 @@ 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 share 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 = parquet::experimental::hybrid_scan_metadata{*footer_buffer, options}; + * // Construct lightweight readers that share it + * auto reader_a = std::make_unique(metadata); + * auto reader_b = std::make_unique(metadata); + * @endcode + * + * @note The metadata is immutable after `setup_page_index()` has been called (or immediately after + * construction if page index setup is skipped). Concurrent usage by multiple readers is thread + * safe. 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 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 _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 @@ -301,6 +370,15 @@ class hybrid_scan_reader { explicit hybrid_scan_reader(FileMetaData const& parquet_metadata, parquet_reader_options const& options); + /** + * @brief Constructor that takes shared ownership of pre-parsed Parquet file metadata + * + * Constructs a reader that shares the pre-parsed metadata object. + * + * @param metadata Shared, pre-parsed Parquet file metadata + */ + explicit hybrid_scan_reader(hybrid_scan_metadata metadata); + /** * @brief Destructor for the experimental parquet reader class */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan.cpp b/cpp/src/io/parquet/experimental/hybrid_scan.cpp index a97135ca8480..09faac8c261a 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan.cpp +++ b/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 footer_bytes, + parquet_reader_options const& options) + : _metadata{std::make_shared( + std::vector>{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( + std::vector{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 footer_bytes, parquet_reader_options const& options) : _impl{std::make_unique( @@ -27,6 +47,11 @@ hybrid_scan_reader::hybrid_scan_reader(FileMetaData const& parquet_metadata, { } +hybrid_scan_reader::hybrid_scan_reader(hybrid_scan_metadata metadata) + : _impl{std::make_unique(std::move(metadata._metadata))} +{ +} + hybrid_scan_reader::~hybrid_scan_reader() = default; [[nodiscard]] text::byte_range_info hybrid_scan_reader::page_index_byte_range() const diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index 96ae63b61409..813f8e011d49 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -122,7 +122,7 @@ hybrid_scan_reader_impl::hybrid_scan_reader_impl( cudf::host_span const> footer_bytes, parquet_reader_options const& options) { - _metadata = std::make_unique( + _metadata = std::make_shared( footer_bytes, options.is_enabled_use_arrow_schema(), has_cols_from_mismatched_sources(options)); _extended_metadata = static_cast(_metadata.get()); @@ -132,12 +132,20 @@ hybrid_scan_reader_impl::hybrid_scan_reader_impl( cudf::host_span parquet_metadatas, parquet_reader_options const& options) { _metadata = - std::make_unique(parquet_metadatas, + std::make_shared(parquet_metadatas, options.is_enabled_use_arrow_schema(), has_cols_from_mismatched_sources(options)); _extended_metadata = static_cast(_metadata.get()); } +hybrid_scan_reader_impl::hybrid_scan_reader_impl( + std::shared_ptr metadata) +{ + CUDF_EXPECTS(metadata != nullptr, "Shared parquet metadata must not be null"); + _metadata = std::move(metadata); + _extended_metadata = static_cast(_metadata.get()); +} + std::vector hybrid_scan_reader_impl::parquet_metadatas() const { return _extended_metadata->parquet_metadatas(); diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index a6588f5c2ee8..f19edcdb0bfd 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -58,6 +58,13 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { explicit hybrid_scan_reader_impl(cudf::host_span parquet_metadatas, parquet_reader_options const& options); + /** + * @brief Constructor that takes shared ownership of pre-parsed Parquet metadata + * + * @param metadata Shared, pre-parsed Parquet file metadata. Must not be null. + */ + explicit hybrid_scan_reader_impl(std::shared_ptr metadata); + /** * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::parquet_metadatas */ diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index bbcd6ec05f21..06076778f6ee 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -550,11 +550,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( + _metadata = file_metadatas.empty() ? std::make_shared( _sources, options.is_enabled_use_arrow_schema(), has_cols_from_mismatched_sources(options)) - : std::make_unique( + : std::make_shared( std::forward>(file_metadatas), options.is_enabled_use_arrow_schema(), has_cols_from_mismatched_sources(options)); diff --git a/cpp/src/io/parquet/reader_impl.hpp b/cpp/src/io/parquet/reader_impl.hpp index 4c8cf0003b52..63a2a9b4ec84 100644 --- a/cpp/src/io/parquet/reader_impl.hpp +++ b/cpp/src/io/parquet/reader_impl.hpp @@ -571,7 +571,7 @@ class reader_impl { named_to_reference_converter _expr_conv{std::nullopt, table_metadata{}, true}; std::vector> _sources; - std::unique_ptr _metadata; + std::shared_ptr _metadata; // Number of sources size_t _num_sources{0}; diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index f478c09e2eac..2c9498408df2 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -420,6 +420,7 @@ std::pair, std::vector> 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); diff --git a/cpp/tests/io/experimental/hybrid_scan_test.cpp b/cpp/tests/io/experimental/hybrid_scan_test.cpp index 6607c941e51c..b4b8c212f7f0 100644 --- a/cpp/tests/io/experimental/hybrid_scan_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_test.cpp @@ -985,6 +985,175 @@ 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(); + + 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( + reinterpret_cast(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 = + 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(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(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(); + + 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( + reinterpret_cast(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(*footer_buffer, options); + auto const file_metadata = seed_reader->parquet_metadata(); + + auto const metadata = + 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(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(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, SharedMetadataConcurrentReadersMatchReadParquet) +{ + using T = uint32_t; + auto constexpr num_concat = 4; + auto [written_table, parquet_buffer] = create_parquet_with_stats(); + + 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( + reinterpret_cast(parquet_buffer.data()), parquet_buffer.size())); + auto const footer_buffer = cudf::io::parquet::fetch_footer_to_host(*datasource); + + // Parse the metadata once and share it, by value, across two readers that are alive at the + // same time, each responsible for a disjoint range of the file's row groups. + auto const metadata = + cudf::io::parquet::experimental::hybrid_scan_metadata{*footer_buffer, options}; + auto const reader_a = + std::make_unique(metadata); + auto const reader_b = + std::make_unique(metadata); + + // Only one of the readers sharing this metadata sets up the page index. + auto const page_index_byte_range = reader_a->page_index_byte_range(); + ASSERT_FALSE(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_a->setup_page_index(*page_index_buffer); + + // The page index materialized through `reader_a` must be visible through `reader_b` since both + // readers share the same underlying metadata. + auto const metadata_from_b = reader_b->parquet_metadata(); + ASSERT_GT(metadata_from_b.row_groups.size(), 1); + for (auto const& row_group : metadata_from_b.row_groups) { + for (auto const& column_chunk : row_group.columns) { + EXPECT_TRUE(column_chunk.column_index.has_value()); + EXPECT_TRUE(column_chunk.offset_index.has_value()); + } + } + + // Split the row groups into two disjoint ranges and read each range through a different reader + // sharing the metadata, with both readers alive and used concurrently. + auto const all_row_groups = reader_a->all_row_groups(options); + auto const split = all_row_groups.size() / 2; + auto const row_groups_a = + std::vector(all_row_groups.begin(), all_row_groups.begin() + split); + auto const row_groups_b = + std::vector(all_row_groups.begin() + split, all_row_groups.end()); + + auto const materialize = [&](auto const& reader, auto const& row_group_indices) { + auto const chunk_ranges = reader->all_column_chunks_byte_ranges(row_group_indices, 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_group_indices, data, options, stream, mr).tbl; + }; + + auto const table_a = materialize(reader_a, row_groups_a); + auto const table_b = materialize(reader_b, row_groups_b); + + auto const table = + cudf::concatenate(std::vector{table_a->view(), table_b->view()}); + auto const expected = + cudf::io::read_parquet( + cudf::io::parquet_reader_options::builder( + cudf::io::source_info(cudf::host_span(parquet_buffer.data(), parquet_buffer.size()))) + .build(), + stream) + .tbl; + + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected->view(), table->view()); +} + TEST_F(HybridScanTest, AllRowsPrunedReportsInputRowGroups) { using cudf::io::parquet::experimental::use_data_page_mask; diff --git a/python/pylibcudf/pylibcudf/io/experimental/__init__.py b/python/pylibcudf/pylibcudf/io/experimental/__init__.py index 6c64231eb1e9..ef1ca25b7cff 100644 --- a/python/pylibcudf/pylibcudf/io/experimental/__init__.py +++ b/python/pylibcudf/pylibcudf/io/experimental/__init__.py @@ -1,7 +1,8 @@ -# 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, ) @@ -9,6 +10,7 @@ __all__ = [ "FileMetaData", # backwards compatibility + "HybridScanMetadata", "HybridScanReader", "UseDataPageMask", ] diff --git a/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pxd b/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pxd index a19cd7db8bf8..a999fc5bc7c2 100644 --- a/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pxd +++ b/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pxd @@ -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 @@ -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, ) @@ -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 diff --git a/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyi b/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyi index f95dc8b054d3..498a458b9f7d 100644 --- a/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyi +++ b/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyi @@ -1,6 +1,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +from collections.abc import Sequence from enum import IntEnum from rmm.pylibrmm.memory_resource import DeviceMemoryResource @@ -10,6 +11,7 @@ from pylibcudf.io.parquet import ParquetReaderOptions from pylibcudf.io.parquet_metadata import FileMetaData from pylibcudf.io.text import ByteRangeInfo from pylibcudf.io.types import TableWithMetadata +from pylibcudf.span import Span from pylibcudf.utils import CudaStreamLike try: @@ -21,6 +23,16 @@ class UseDataPageMask(IntEnum): YES = 1 NO = 0 +class HybridScanMetadata: + @staticmethod + def from_footer_bytes( + footer_bytes: Buffer, options: ParquetReaderOptions + ) -> HybridScanMetadata: ... + @staticmethod + def from_parquet_metadata( + metadata: FileMetaData, options: ParquetReaderOptions + ) -> HybridScanMetadata: ... + class HybridScanReader: def __init__( self, footer_bytes: Buffer, options: ParquetReaderOptions @@ -29,6 +41,8 @@ class HybridScanReader: def from_parquet_metadata( metadata: FileMetaData, options: ParquetReaderOptions ) -> HybridScanReader: ... + @staticmethod + def from_metadata(metadata: HybridScanMetadata) -> HybridScanReader: ... def parquet_metadata(self) -> FileMetaData: ... def page_index_byte_range(self) -> ByteRangeInfo: ... def setup_page_index(self, page_index_bytes: Buffer) -> None: ... @@ -48,18 +62,24 @@ class HybridScanReader: ) -> tuple[list[ByteRangeInfo], list[ByteRangeInfo]]: ... def filter_row_groups_with_dictionary_pages( self, - dictionary_page_data: list, + dictionary_page_data: Sequence[Span], row_group_indices: list[int], options: ParquetReaderOptions, stream: CudaStreamLike | None = None, ) -> list[int]: ... def filter_row_groups_with_bloom_filters( self, - bloom_filter_data: list, + bloom_filter_data: Sequence[Span], row_group_indices: list[int], options: ParquetReaderOptions, stream: CudaStreamLike | None = None, ) -> list[int]: ... + def build_all_true_row_mask( + self, + row_group_indices: list[int], + stream: CudaStreamLike | None = None, + mr: DeviceMemoryResource | None = None, + ) -> Column: ... def build_row_mask_with_page_index_stats( self, row_group_indices: list[int], @@ -73,7 +93,7 @@ class HybridScanReader: def materialize_filter_columns( self, row_group_indices: list[int], - column_chunk_data: list, + column_chunk_data: Sequence[Span], row_mask: Column, mask_data_pages: UseDataPageMask, options: ParquetReaderOptions, @@ -86,7 +106,7 @@ class HybridScanReader: def materialize_payload_columns( self, row_group_indices: list[int], - column_chunk_data: list, + column_chunk_data: Sequence[Span], row_mask: Column, mask_data_pages: UseDataPageMask, options: ParquetReaderOptions, @@ -99,7 +119,7 @@ class HybridScanReader: def materialize_all_columns( self, row_group_indices: list[int], - column_chunk_data: list, + column_chunk_data: Sequence[Span], options: ParquetReaderOptions, stream: CudaStreamLike | None = None, mr: DeviceMemoryResource | None = None, @@ -111,7 +131,7 @@ class HybridScanReader: row_group_indices: list[int], row_mask: Column, mask_data_pages: UseDataPageMask, - column_chunk_data: list, + column_chunk_data: Sequence[Span], options: ParquetReaderOptions, stream: CudaStreamLike | None = None, mr: DeviceMemoryResource | None = None, @@ -127,7 +147,7 @@ class HybridScanReader: row_group_indices: list[int], row_mask: Column, mask_data_pages: UseDataPageMask, - column_chunk_data: list, + column_chunk_data: Sequence[Span], options: ParquetReaderOptions, stream: CudaStreamLike | None = None, mr: DeviceMemoryResource | None = None, diff --git a/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyx b/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyx index ed878c2647bb..4cfc0214f4d8 100644 --- a/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyx +++ b/python/pylibcudf/pylibcudf/io/experimental/hybrid_scan.pyx @@ -4,6 +4,7 @@ from cython.operator cimport dereference from libc.stdint cimport uint8_t, uintptr_t from libc.stddef cimport size_t +from libcpp cimport bool from libcpp.memory cimport make_unique, unique_ptr from libcpp.pair cimport pair from libcpp.span cimport span as std_span @@ -16,6 +17,7 @@ from rmm.pylibrmm.stream cimport Stream from pylibcudf.column cimport Column from pylibcudf.io.parquet cimport ParquetReaderOptions from pylibcudf.io.parquet_metadata cimport FileMetaData as c_FileMetaData +from pylibcudf.libcudf.io.parquet_schema cimport FileMetaData as cpp_FileMetaData from pylibcudf.io.text cimport ByteRangeInfo from pylibcudf.io.types cimport TableWithMetadata from pylibcudf.libcudf.column.column cimport column @@ -24,6 +26,7 @@ from pylibcudf.libcudf.io.hybrid_scan cimport ( const_device_span_const_uint8_t, const_size_type, const_uint8_t, + hybrid_scan_metadata as cpp_hybrid_scan_metadata, hybrid_scan_reader as cpp_hybrid_scan_reader, use_data_page_mask as cpp_use_data_page_mask, ) @@ -46,7 +49,7 @@ import pylibcudf.libcudf.io.hybrid_scan UseDataPageMask = pylibcudf.libcudf.io.hybrid_scan.use_data_page_mask -__all__ = ["FileMetaData", "HybridScanReader", "UseDataPageMask"] +__all__ = ["FileMetaData", "HybridScanMetadata", "HybridScanReader", "UseDataPageMask"] cdef device_span[const_uint8_t] _get_device_span(object obj) except *: @@ -60,6 +63,83 @@ cdef device_span[const_uint8_t] _get_device_span(object obj) except *: obj.size) +cdef class HybridScanMetadata: + """Shareable, pre-parsed Parquet file metadata for the hybrid scan reader. + + This class enables parsing the metadata of a Parquet file once, then + constructing multiple :class:`HybridScanReader` instances that share it + (one per row-group range of the file) instead of each re-parsing and copying + the metadata. + + For details, see :cpp:class:`cudf::io::parquet::experimental::hybrid_scan_metadata` + + Examples + -------- + >>> import pylibcudf as plc + >>> metadata = plc.io.experimental.HybridScanMetadata.from_parquet_metadata( + ... file_metadata, options) + >>> reader = plc.io.experimental.HybridScanReader.from_metadata(metadata) + """ + + def __init__(self): + raise ValueError( + "HybridScanMetadata cannot be constructed directly. " + "Use from_footer_bytes() or from_parquet_metadata()." + ) + + @staticmethod + def from_footer_bytes( + const uint8_t[::1] footer_bytes, + ParquetReaderOptions options + ): + """Parse shareable metadata from Parquet footer bytes. + + Parameters + ---------- + footer_bytes : Buffer + Parquet file footer bytes + options : ParquetReaderOptions + Parquet reader options + + Returns + ------- + HybridScanMetadata + """ + cdef HybridScanMetadata result = HybridScanMetadata.__new__(HybridScanMetadata) + cdef const uint8_t* footer_ptr = 0 + if len(footer_bytes) > 0: + footer_ptr = &footer_bytes[0] + with nogil: + result.c_obj = make_unique[cpp_hybrid_scan_metadata]( + host_span[const_uint8_t](footer_ptr, len(footer_bytes)), + options.c_obj + ) + return result + + @staticmethod + def from_parquet_metadata(c_FileMetaData metadata, ParquetReaderOptions options): + """Build shareable metadata from a pre-populated ``FileMetaData``. + + Parameters + ---------- + metadata : FileMetaData + Pre-populated Parquet file metadata + options : ParquetReaderOptions + Parquet reader options + + Returns + ------- + HybridScanMetadata + """ + cdef HybridScanMetadata result = HybridScanMetadata.__new__(HybridScanMetadata) + with nogil: + result.c_obj = make_unique[cpp_hybrid_scan_metadata]( + dereference(metadata.c_obj), + options.c_obj + ) + return result + + cdef class HybridScanReader: """Experimental Parquet reader optimized for highly selective filters. @@ -90,13 +170,17 @@ cdef class HybridScanReader: def __init__( self, - const uint8_t[::1] footer_bytes: Buffer, - ParquetReaderOptions options, + const uint8_t[::1] footer_bytes, + ParquetReaderOptions options ): - self.c_obj = make_unique[cpp_hybrid_scan_reader]( - host_span[const_uint8_t](&footer_bytes[0], len(footer_bytes)), - options.c_obj - ) + cdef const uint8_t* footer_ptr = 0 + if len(footer_bytes) > 0: + footer_ptr = &footer_bytes[0] + with nogil: + self.c_obj = make_unique[cpp_hybrid_scan_reader]( + host_span[const_uint8_t](footer_ptr, len(footer_bytes)), + options.c_obj + ) @staticmethod def from_parquet_metadata( @@ -116,10 +200,36 @@ cdef class HybridScanReader: HybridScanReader """ cdef HybridScanReader reader = HybridScanReader.__new__(HybridScanReader) - reader.c_obj = make_unique[cpp_hybrid_scan_reader]( - dereference(metadata.c_obj), - options.c_obj - ) + with nogil: + reader.c_obj = make_unique[cpp_hybrid_scan_reader]( + dereference(metadata.c_obj), + options.c_obj + ) + return reader + + @staticmethod + def from_metadata(HybridScanMetadata metadata not None): + """Create a HybridScanReader that shares pre-parsed metadata. + + Constructs a lightweight reader that borrows ``metadata`` instead of + re-parsing and copying the file metadata. Use one shared + :class:`HybridScanMetadata` to read row-group ranges of a single file. + Overlapping row-group ranges across readers produce duplicate rows. + + Parameters + ---------- + metadata : HybridScanMetadata + Shared, pre-parsed Parquet file metadata + + Returns + ------- + HybridScanReader + """ + cdef HybridScanReader reader = HybridScanReader.__new__(HybridScanReader) + with nogil: + reader.c_obj = make_unique[cpp_hybrid_scan_reader]( + dereference(metadata.c_obj.get()) + ) return reader def parquet_metadata(self) -> FileMetaData: @@ -145,7 +255,9 @@ cdef class HybridScanReader: ByteRangeInfo Byte range of the page index """ - cdef byte_range_info info = self.c_obj.get()[0].page_index_byte_range() + cdef byte_range_info info + with nogil: + info = self.c_obj.get()[0].page_index_byte_range() return ByteRangeInfo(info.offset(), info.size()) def setup_page_index( @@ -158,9 +270,13 @@ cdef class HybridScanReader: page_index_bytes : Buffer Parquet page index buffer bytes """ - self.c_obj.get()[0].setup_page_index( - host_span[const_uint8_t](&page_index_bytes[0], len(page_index_bytes)) - ) + cdef const uint8_t* page_index_ptr = 0 + if len(page_index_bytes) > 0: + page_index_ptr = &page_index_bytes[0] + with nogil: + self.c_obj.get()[0].setup_page_index( + host_span[const_uint8_t](page_index_ptr, len(page_index_bytes)) + ) def all_row_groups(self, ParquetReaderOptions options) -> list[int]: """Get all available row groups from the parquet file. @@ -175,9 +291,9 @@ cdef class HybridScanReader: list[int] List of row group indices """ - cdef vector[size_type] row_groups = self.c_obj.get()[0].all_row_groups( - options.c_obj - ) + cdef vector[size_type] row_groups + with nogil: + row_groups = self.c_obj.get()[0].all_row_groups(options.c_obj) return list(row_groups) def total_rows_in_row_groups( @@ -196,9 +312,12 @@ cdef class HybridScanReader: Total number of top-level rows """ cdef vector[size_type] indices_vec = row_group_indices - return self.c_obj.get()[0].total_rows_in_row_groups( - std_span[const_size_type](indices_vec.data(), indices_vec.size()) - ) + cdef size_type result + with nogil: + result = self.c_obj.get()[0].total_rows_in_row_groups( + std_span[const_size_type](indices_vec.data(), indices_vec.size()) + ) + return result def reset_column_selection(self) -> None: """Reset the column selection state. @@ -206,7 +325,8 @@ cdef class HybridScanReader: Resets the internal column selection state forcing re-selection of columns in subsequent filter and read operations """ - self.c_obj.get()[0].reset_column_selection() + with nogil: + self.c_obj.get()[0].reset_column_selection() def filter_row_groups_with_stats( self, @@ -232,15 +352,15 @@ cdef class HybridScanReader: """ cdef Stream _stream = _get_stream(stream) cdef vector[size_type] indices_vec = row_group_indices - cdef vector[size_type] filtered = ( - self.c_obj.get()[0].filter_row_groups_with_stats( + cdef vector[size_type] filtered + with nogil: + filtered = move(self.c_obj.get()[0].filter_row_groups_with_stats( std_span[const_size_type]( indices_vec.data(), indices_vec.size() ), options.c_obj, _stream.view().value() - ) - ) + )) return list(filtered) def secondary_filters_byte_ranges( @@ -263,11 +383,13 @@ cdef class HybridScanReader: Tuple of (bloom_filter_ranges, dictionary_page_ranges) """ cdef vector[size_type] indices_vec = row_group_indices - cdef pair[vector[byte_range_info], vector[byte_range_info]] ranges = \ - self.c_obj.get()[0].secondary_filters_byte_ranges( + cdef pair[vector[byte_range_info], vector[byte_range_info]] ranges + cdef cpp_hybrid_scan_reader* reader_ptr = self.c_obj.get() + with nogil: + ranges = move(reader_ptr.secondary_filters_byte_ranges( std_span[const_size_type](indices_vec.data(), indices_vec.size()), options.c_obj - ) + )) bloom_ranges = [ ByteRangeInfo(r.offset(), r.size()) for r in ranges.first @@ -288,7 +410,7 @@ cdef class HybridScanReader: Parameters ---------- - dictionary_page_data : list + dictionary_page_data : Sequence Span-like objects containing dictionary page data row_group_indices : list[int] Input row group indices @@ -309,15 +431,16 @@ cdef class HybridScanReader: cdef vector[size_type] indices_vec = row_group_indices - cdef vector[size_type] filtered = \ - self.c_obj.get()[0].filter_row_groups_with_dictionary_pages( + cdef vector[size_type] filtered + with nogil: + filtered = move(self.c_obj.get()[0].filter_row_groups_with_dictionary_pages( std_span[const_device_span_const_uint8_t]( spans_vec.data(), spans_vec.size() ), std_span[const_size_type](indices_vec.data(), indices_vec.size()), options.c_obj, _stream.view().value() - ) + )) return list(filtered) def filter_row_groups_with_bloom_filters( @@ -331,7 +454,7 @@ cdef class HybridScanReader: Parameters ---------- - bloom_filter_data : list + bloom_filter_data : Sequence Span-like objects containing bloom filter data row_group_indices : list[int] Input row group indices @@ -352,17 +475,52 @@ cdef class HybridScanReader: cdef vector[size_type] indices_vec = row_group_indices - cdef vector[size_type] filtered = \ - self.c_obj.get()[0].filter_row_groups_with_bloom_filters( + cdef vector[size_type] filtered + with nogil: + filtered = move(self.c_obj.get()[0].filter_row_groups_with_bloom_filters( std_span[const_device_span_const_uint8_t]( spans_vec.data(), spans_vec.size() ), std_span[const_size_type](indices_vec.data(), indices_vec.size()), options.c_obj, _stream.view().value() - ) + )) return list(filtered) + def build_all_true_row_mask( + self, + list row_group_indices, + object stream=None, + DeviceMemoryResource mr=None + ): + """Build an all-true boolean survival column for the given row groups. + + Parameters + ---------- + row_group_indices : list[int] + Input row group indices + stream : Stream, optional + CUDA stream + mr : DeviceMemoryResource, optional + Device memory resource + + Returns + ------- + Column + All-true boolean column with one entry per row across all row groups + """ + cdef vector[size_type] indices_vec = row_group_indices + cdef Stream _stream = _get_stream(stream) + mr = _get_memory_resource(mr) + cdef unique_ptr[column] c_result + with nogil: + c_result = move(self.c_obj.get()[0].build_all_true_row_mask( + std_span[const_size_type](indices_vec.data(), indices_vec.size()), + _stream.view().value(), + mr.get_mr() + )) + return Column.from_libcudf(move(c_result), _stream, mr) + def build_row_mask_with_page_index_stats( self, list row_group_indices: list[int], @@ -391,13 +549,14 @@ cdef class HybridScanReader: cdef vector[size_type] indices_vec = row_group_indices cdef Stream _stream = _get_stream(stream) mr = _get_memory_resource(mr) - cdef unique_ptr[column] c_result = \ - self.c_obj.get()[0].build_row_mask_with_page_index_stats( + cdef unique_ptr[column] c_result + with nogil: + c_result = move(self.c_obj.get()[0].build_row_mask_with_page_index_stats( std_span[const_size_type](indices_vec.data(), indices_vec.size()), options.c_obj, _stream.view().value(), mr.get_mr() - ) + )) return Column.from_libcudf(move(c_result), _stream, mr) def filter_column_chunks_byte_ranges( @@ -420,11 +579,12 @@ cdef class HybridScanReader: Byte ranges to column chunks of filter columns """ cdef vector[size_type] indices_vec = row_group_indices - cdef vector[byte_range_info] ranges = \ - self.c_obj.get()[0].filter_column_chunks_byte_ranges( + cdef vector[byte_range_info] ranges + with nogil: + ranges = move(self.c_obj.get()[0].filter_column_chunks_byte_ranges( std_span[const_size_type](indices_vec.data(), indices_vec.size()), options.c_obj - ) + )) return [ByteRangeInfo(r.offset(), r.size()) for r in ranges] def materialize_filter_columns( @@ -443,7 +603,7 @@ cdef class HybridScanReader: ---------- row_group_indices : list[int] Input row group indices - column_chunk_data : list + column_chunk_data : Sequence Span-like objects containing column chunk data of filter columns row_mask : Column Mutable boolean column indicating surviving rows @@ -470,8 +630,9 @@ cdef class HybridScanReader: spans_vec.push_back(_get_device_span(span)) cdef mutable_column_view mask_view = row_mask.mutable_view() - cdef table_with_metadata c_result = \ - self.c_obj.get()[0].materialize_filter_columns( + cdef table_with_metadata c_result + with nogil: + c_result = move(self.c_obj.get()[0].materialize_filter_columns( std_span[const_size_type](indices_vec.data(), indices_vec.size()), std_span[const_device_span_const_uint8_t]( spans_vec.data(), spans_vec.size() @@ -481,7 +642,7 @@ cdef class HybridScanReader: options.c_obj, _stream.view().value(), mr.get_mr() - ) + )) return TableWithMetadata.from_libcudf(c_result, _stream, mr) def payload_column_chunks_byte_ranges( @@ -504,11 +665,12 @@ cdef class HybridScanReader: Byte ranges to column chunks of payload columns """ cdef vector[size_type] indices_vec = row_group_indices - cdef vector[byte_range_info] ranges = \ - self.c_obj.get()[0].payload_column_chunks_byte_ranges( + cdef vector[byte_range_info] ranges + with nogil: + ranges = move(self.c_obj.get()[0].payload_column_chunks_byte_ranges( std_span[const_size_type](indices_vec.data(), indices_vec.size()), options.c_obj - ) + )) return [ByteRangeInfo(r.offset(), r.size()) for r in ranges] def materialize_payload_columns( @@ -527,7 +689,7 @@ cdef class HybridScanReader: ---------- row_group_indices : list[int] Input row group indices - column_chunk_data : list + column_chunk_data : Sequence Span-like objects containing column chunk data of payload columns row_mask : Column Boolean column indicating surviving rows @@ -554,8 +716,9 @@ cdef class HybridScanReader: spans_vec.push_back(_get_device_span(span)) cdef column_view mask_view = row_mask.view() - cdef table_with_metadata c_result = \ - self.c_obj.get()[0].materialize_payload_columns( + cdef table_with_metadata c_result + with nogil: + c_result = move(self.c_obj.get()[0].materialize_payload_columns( std_span[const_size_type](indices_vec.data(), indices_vec.size()), std_span[const_device_span_const_uint8_t]( spans_vec.data(), spans_vec.size() @@ -565,7 +728,7 @@ cdef class HybridScanReader: options.c_obj, _stream.view().value(), mr.get_mr() - ) + )) return TableWithMetadata.from_libcudf(c_result, _stream, mr) def all_column_chunks_byte_ranges( @@ -588,11 +751,12 @@ cdef class HybridScanReader: Byte ranges to column chunks of all columns """ cdef vector[size_type] indices_vec = row_group_indices - cdef vector[byte_range_info] ranges = \ - self.c_obj.get()[0].all_column_chunks_byte_ranges( + cdef vector[byte_range_info] ranges + with nogil: + ranges = move(self.c_obj.get()[0].all_column_chunks_byte_ranges( std_span[const_size_type](indices_vec.data(), indices_vec.size()), options.c_obj - ) + )) return [ByteRangeInfo(r.offset(), r.size()) for r in ranges] def materialize_all_columns( @@ -609,7 +773,7 @@ cdef class HybridScanReader: ---------- row_group_indices : list[int] Input row group indices - column_chunk_data : list + column_chunk_data : Sequence Span-like objects containing column chunk data of all columns options : ParquetReaderOptions Parquet reader options @@ -630,8 +794,9 @@ cdef class HybridScanReader: mr = _get_memory_resource(mr) for span in column_chunk_data: spans_vec.push_back(_get_device_span(span)) - cdef table_with_metadata c_result = \ - self.c_obj.get()[0].materialize_all_columns( + cdef table_with_metadata c_result + with nogil: + c_result = move(self.c_obj.get()[0].materialize_all_columns( std_span[const_size_type](indices_vec.data(), indices_vec.size()), std_span[const_device_span_const_uint8_t]( spans_vec.data(), spans_vec.size() @@ -639,7 +804,7 @@ cdef class HybridScanReader: options.c_obj, _stream.view().value(), mr.get_mr() - ) + )) return TableWithMetadata.from_libcudf(c_result, _stream, mr) def setup_chunking_for_filter_columns( @@ -649,7 +814,7 @@ cdef class HybridScanReader: list row_group_indices: list[int], Column row_mask, cpp_use_data_page_mask mask_data_pages, - list column_chunk_data, + object column_chunk_data, ParquetReaderOptions options, object stream: CudaStreamLike | None = None, DeviceMemoryResource mr=None @@ -668,7 +833,7 @@ cdef class HybridScanReader: Boolean column indicating surviving rows mask_data_pages : UseDataPageMask Whether to use a data page mask - column_chunk_data : list + column_chunk_data : Sequence Span-like objects containing column chunk data of filter columns options : ParquetReaderOptions Parquet reader options @@ -685,21 +850,24 @@ cdef class HybridScanReader: self._stream = _get_stream(stream) self.mr = _get_memory_resource(mr) + # keep reference to avoid use-after-free of device spans + self._filter_chunk_data = column_chunk_data cdef column_view mask_view = row_mask.view() - self.c_obj.get()[0].setup_chunking_for_filter_columns( - chunk_read_limit, - pass_read_limit, - std_span[const_size_type](indices_vec.data(), indices_vec.size()), - mask_view, - mask_data_pages, - std_span[const_device_span_const_uint8_t]( - spans_vec.data(), spans_vec.size() - ), - options.c_obj, - self._stream.view().value(), - self.mr.get_mr() - ) + with nogil: + self.c_obj.get()[0].setup_chunking_for_filter_columns( + chunk_read_limit, + pass_read_limit, + std_span[const_size_type](indices_vec.data(), indices_vec.size()), + mask_view, + mask_data_pages, + std_span[const_device_span_const_uint8_t]( + spans_vec.data(), spans_vec.size() + ), + options.c_obj, + self._stream.view().value(), + self.mr.get_mr() + ) def materialize_filter_columns_chunk( self, @@ -717,10 +885,15 @@ cdef class HybridScanReader: Table chunk of materialized filter columns and metadata """ cdef mutable_column_view mask_view = row_mask.mutable_view() - cdef table_with_metadata c_result = \ - self.c_obj.get()[0].materialize_filter_columns_chunk( + cdef table_with_metadata c_result + cdef bool more_chunks + with nogil: + c_result = move(self.c_obj.get()[0].materialize_filter_columns_chunk( mask_view - ) + )) + more_chunks = self.c_obj.get()[0].has_next_table_chunk() + if not more_chunks: + self._filter_chunk_data = None return TableWithMetadata.from_libcudf( c_result, self._stream, self.mr ) @@ -732,7 +905,7 @@ cdef class HybridScanReader: list row_group_indices: list[int], Column row_mask, cpp_use_data_page_mask mask_data_pages, - list column_chunk_data, + object column_chunk_data, ParquetReaderOptions options, object stream: CudaStreamLike | None = None, DeviceMemoryResource mr=None @@ -751,7 +924,7 @@ cdef class HybridScanReader: Boolean column indicating surviving rows mask_data_pages : UseDataPageMask Whether to use a data page mask - column_chunk_data : list + column_chunk_data : Sequence Span-like objects containing column chunk data of payload columns options : ParquetReaderOptions Parquet reader options @@ -768,21 +941,23 @@ cdef class HybridScanReader: self._stream = _get_stream(stream) self.mr = _get_memory_resource(mr) + self._payload_chunk_data = column_chunk_data cdef column_view mask_view = row_mask.view() - self.c_obj.get()[0].setup_chunking_for_payload_columns( - chunk_read_limit, - pass_read_limit, - std_span[const_size_type](indices_vec.data(), indices_vec.size()), - mask_view, - mask_data_pages, - std_span[const_device_span_const_uint8_t]( - spans_vec.data(), spans_vec.size() - ), - options.c_obj, - self._stream.view().value(), - self.mr.get_mr() - ) + with nogil: + self.c_obj.get()[0].setup_chunking_for_payload_columns( + chunk_read_limit, + pass_read_limit, + std_span[const_size_type](indices_vec.data(), indices_vec.size()), + mask_view, + mask_data_pages, + std_span[const_device_span_const_uint8_t]( + spans_vec.data(), spans_vec.size() + ), + options.c_obj, + self._stream.view().value(), + self.mr.get_mr() + ) def materialize_payload_columns_chunk( self, @@ -800,10 +975,15 @@ cdef class HybridScanReader: Table chunk of materialized payload columns and metadata """ cdef column_view mask_view = row_mask.view() - cdef table_with_metadata c_result = \ - self.c_obj.get()[0].materialize_payload_columns_chunk( + cdef table_with_metadata c_result + cdef bool more_chunks + with nogil: + c_result = move(self.c_obj.get()[0].materialize_payload_columns_chunk( mask_view - ) + )) + more_chunks = self.c_obj.get()[0].has_next_table_chunk() + if not more_chunks: + self._payload_chunk_data = None return TableWithMetadata.from_libcudf( c_result, self._stream, self.mr ) @@ -839,12 +1019,15 @@ cdef class HybridScanReader: If ``row_group_indices`` is empty. """ cdef vector[size_type] indices_vec = row_group_indices - return self.c_obj.get()[0].construct_row_group_passes( - std_span[const_size_type]( - indices_vec.data(), indices_vec.size() - ), - pass_read_limit - ) + cdef vector[vector[size_type]] passes + with nogil: + passes = move(self.c_obj.get()[0].construct_row_group_passes( + std_span[const_size_type]( + indices_vec.data(), indices_vec.size() + ), + pass_read_limit + )) + return passes def has_next_table_chunk(self) -> bool: """Check if there is any parquet data left to read. @@ -854,7 +1037,10 @@ cdef class HybridScanReader: bool True if there is data left to read """ - return self.c_obj.get()[0].has_next_table_chunk() + cdef bool result + with nogil: + result = self.c_obj.get()[0].has_next_table_chunk() + return result UseDataPageMask.__str__ = UseDataPageMask.__repr__ diff --git a/python/pylibcudf/pylibcudf/libcudf/io/hybrid_scan.pxd b/python/pylibcudf/pylibcudf/libcudf/io/hybrid_scan.pxd index 36201d545de6..7a5aec269a56 100644 --- a/python/pylibcudf/pylibcudf/libcudf/io/hybrid_scan.pxd +++ b/python/pylibcudf/pylibcudf/libcudf/io/hybrid_scan.pxd @@ -30,6 +30,17 @@ cdef extern from "cudf/io/experimental/hybrid_scan.hpp" \ YES NO + cdef cppclass hybrid_scan_metadata: + hybrid_scan_metadata( + host_span[const_uint8_t] footer_bytes, + const parquet_reader_options& options + ) except +libcudf_exception_handler + + hybrid_scan_metadata( + const FileMetaData& parquet_metadata, + const parquet_reader_options& options + ) except +libcudf_exception_handler + cdef cppclass hybrid_scan_reader: hybrid_scan_reader( host_span[const_uint8_t] footer_bytes, @@ -41,6 +52,10 @@ cdef extern from "cudf/io/experimental/hybrid_scan.hpp" \ const parquet_reader_options& options ) except +libcudf_exception_handler + hybrid_scan_reader( + hybrid_scan_metadata metadata + ) except +libcudf_exception_handler + FileMetaData parquet_metadata() except +libcudf_exception_handler byte_range_info page_index_byte_range() except +libcudf_exception_handler @@ -86,6 +101,12 @@ cdef extern from "cudf/io/experimental/hybrid_scan.hpp" \ cudaStream_t stream ) except +libcudf_exception_handler + unique_ptr[column] build_all_true_row_mask( + std_span[const_size_type] row_group_indices, + cudaStream_t stream, + device_async_resource_ref mr + ) except +libcudf_exception_handler + unique_ptr[column] build_row_mask_with_page_index_stats( std_span[const_size_type] row_group_indices, const parquet_reader_options& options,