-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Fix Parquet page-index read ranges for optional indexes #24001
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -11,19 +11,26 @@ | |||||
| #include <cudf/ast/expressions.hpp> | ||||||
| #include <cudf/detail/utilities/integer_utils.hpp> | ||||||
| #include <cudf/io/datasource.hpp> | ||||||
| #include <cudf/io/experimental/hybrid_scan.hpp> | ||||||
| #include <cudf/io/parquet.hpp> | ||||||
| #include <cudf/io/parquet_io_utils.hpp> | ||||||
| #include <cudf/io/parquet_metadata.hpp> | ||||||
| #include <cudf/scalar/scalar.hpp> | ||||||
| #include <cudf/utilities/default_stream.hpp> | ||||||
|
|
||||||
| #include <cuda/iterator> | ||||||
|
|
||||||
| #include <nvbench/nvbench.cuh> | ||||||
| #include <src/io/parquet/compact_protocol_reader.hpp> | ||||||
| #include <src/io/parquet/compact_protocol_writer.hpp> | ||||||
|
|
||||||
| #include <algorithm> | ||||||
| #include <atomic> | ||||||
| #include <cctype> | ||||||
| #include <cstddef> | ||||||
| #include <cstring> | ||||||
| #include <limits> | ||||||
| #include <memory> | ||||||
| #include <string> | ||||||
| #include <utility> | ||||||
| #include <vector> | ||||||
|
|
@@ -84,6 +91,87 @@ auto write_file_data(cudf::size_type num_cols, | |||||
| return source_sink; | ||||||
| } | ||||||
|
|
||||||
| /** @brief Counts logical host-read bytes without including filesystem or page-cache effects. */ | ||||||
| class PageIndexCountingDatasource : public cudf::io::datasource { | ||||||
| public: | ||||||
| explicit PageIndexCountingDatasource(std::vector<char> const& data) | ||||||
| : source_{cudf::io::datasource::create(cudf::host_span<std::byte const>{ | ||||||
| reinterpret_cast<std::byte const*>(data.data()), data.size()})} | ||||||
| { | ||||||
| } | ||||||
|
|
||||||
| std::unique_ptr<buffer> host_read(std::size_t offset, std::size_t size) override | ||||||
| { | ||||||
| auto result = source_->host_read(offset, size); | ||||||
| bytes_read_ += result->size(); | ||||||
| return result; | ||||||
| } | ||||||
|
|
||||||
| std::size_t host_read(std::size_t offset, std::size_t size, uint8_t* dst) override | ||||||
| { | ||||||
| auto const result = source_->host_read(offset, size, dst); | ||||||
| bytes_read_ += result; | ||||||
| return result; | ||||||
| } | ||||||
|
|
||||||
| [[nodiscard]] std::size_t size() const override { return source_->size(); } | ||||||
| [[nodiscard]] std::size_t bytes_read() const { return bytes_read_.load(); } | ||||||
| void reset() { bytes_read_ = 0; } | ||||||
|
|
||||||
| private: | ||||||
| std::unique_ptr<cudf::io::datasource> source_; | ||||||
| std::atomic<std::size_t> bytes_read_{0}; | ||||||
| }; | ||||||
|
|
||||||
| /** @brief Reuses the metadata benchmark input, removing only optional index references. */ | ||||||
| std::vector<char> make_optional_index_data(cudf::size_type num_cols, | ||||||
| cudf::size_type num_row_groups, | ||||||
| std::string const& layout) | ||||||
| { | ||||||
| auto source_sink = write_file_data(num_cols, num_row_groups, io_type::HOST_BUFFER, true); | ||||||
| auto sources = cudf::io::make_datasources(source_sink.make_source_info()); | ||||||
| auto const footer_buffer = cudf::io::parquet::fetch_footer_to_host(*sources.front()); | ||||||
| cudf::io::parquet::FileMetaData metadata; | ||||||
| cudf::io::parquet::detail::CompactProtocolReader cp(footer_buffer->data(), footer_buffer->size()); | ||||||
| cp.read(&metadata); | ||||||
| CUDF_EXPECTS(layout == "none" or layout == "offset_only" or layout == "mixed" or layout == "both", | ||||||
| "Unexpected page index layout"); | ||||||
| for (auto& rg : metadata.row_groups) { | ||||||
| for (auto& col : rg.columns) { | ||||||
| if (layout == "none" or layout == "offset_only") { | ||||||
| col.column_index_offset = 0; | ||||||
| col.column_index_length = 0; | ||||||
| } | ||||||
| if (layout == "none") { | ||||||
| col.offset_index_offset = 0; | ||||||
| col.offset_index_length = 0; | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
| if (layout == "mixed") { | ||||||
| metadata.row_groups.front().columns.front().column_index_offset = 0; | ||||||
| metadata.row_groups.front().columns.front().column_index_length = 0; | ||||||
| } | ||||||
|
|
||||||
| auto const original = sources.front()->host_read(0, sources.front()->size()); | ||||||
| auto const begin = reinterpret_cast<char const*>(original->data()); | ||||||
| std::vector<char> data(begin, begin + original->size()); | ||||||
| cudf::io::parquet::file_ender_s ender; | ||||||
| CUDF_EXPECTS(data.size() >= sizeof(ender), "Invalid Parquet benchmark input"); | ||||||
| std::memcpy(&ender, data.data() + data.size() - sizeof(ender), sizeof(ender)); | ||||||
| CUDF_EXPECTS(ender.footer_len <= data.size() - sizeof(ender), "Invalid Parquet benchmark footer"); | ||||||
| data.resize(data.size() - sizeof(ender) - ender.footer_len); | ||||||
| // Keep unused index bytes in place so every layout has the same data-page offsets. | ||||||
| std::vector<uint8_t> footer; | ||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The code in lines 159-171 is repeated almost verbatim in |
||||||
| cudf::io::parquet::detail::CompactProtocolWriter writer(&footer); | ||||||
| writer.write(metadata); | ||||||
| data.insert(data.end(), footer.begin(), footer.end()); | ||||||
| ender.footer_len = static_cast<uint32_t>(footer.size()); | ||||||
| auto const ender_bytes = reinterpret_cast<char const*>(&ender); | ||||||
| data.insert(data.end(), ender_bytes, ender_bytes + sizeof(ender)); | ||||||
|
Comment on lines
+170
to
+171
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is there a reason you use |
||||||
| return data; | ||||||
| } | ||||||
|
|
||||||
| // Combines `operands` into a balanced AST tree using `op`: pairing adjacent operands gives a tree | ||||||
| // of depth ceil(log2(n)) rather than the n-deep chain a left fold would produce. | ||||||
| [[nodiscard]] cudf::ast::expression const* reduce_balanced( | ||||||
|
|
@@ -343,6 +431,57 @@ void BM_parquet_filter_name_resolution(nvbench::state& state) | |||||
| mem_stats_logger.peak_memory_usage(), "peak_memory_usage", "peak_memory_usage"); | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * @brief Measures metadata parsing, index range calculation and index loading for both readers. | ||||||
| * | ||||||
| * Reports logical host-read bytes per invocation alongside metadata latency. Input generation is | ||||||
| * excluded from timing; host-buffer sources isolate metadata work from storage and cache behavior. | ||||||
| */ | ||||||
| void BM_parquet_page_index_metadata(nvbench::state& state) | ||||||
| { | ||||||
| auto const num_cols = static_cast<cudf::size_type>(state.get_int64("num_cols")); | ||||||
| auto const num_row_groups = static_cast<cudf::size_type>(state.get_int64("num_row_groups")); | ||||||
| auto const hybrid = state.get_string("reader") == "hybrid"; | ||||||
| auto const data = make_optional_index_data(num_cols, num_row_groups, state.get_string("layout")); | ||||||
| auto source = std::make_unique<PageIndexCountingDatasource>(data); | ||||||
| auto const counter = source.get(); | ||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
|
||||||
| std::vector<std::unique_ptr<cudf::io::datasource>> sources; | ||||||
| sources.emplace_back(std::move(source)); | ||||||
| auto const options = cudf::io::parquet_reader_options::builder().use_arrow_schema(false).build(); | ||||||
| state.set_cuda_stream(nvbench::make_cuda_stream_view(cudf::get_default_stream().get())); | ||||||
|
|
||||||
| state.exec( | ||||||
| nvbench::exec_tag::sync | nvbench::exec_tag::timer, [&](nvbench::launch&, auto& timer) { | ||||||
| counter->reset(); | ||||||
| timer.start(); | ||||||
| if (hybrid) { | ||||||
| auto const footer = cudf::io::parquet::fetch_footer_to_host(*counter); | ||||||
| auto const reader = cudf::io::parquet::experimental::hybrid_scan_reader{*footer, options}; | ||||||
| auto const range = reader.page_index_byte_range(); | ||||||
| if (not range.is_empty()) { | ||||||
| auto const indexes = cudf::io::parquet::fetch_page_index_to_host(*counter, range); | ||||||
| reader.setup_page_index(*indexes); | ||||||
| } | ||||||
| } else { | ||||||
| auto const metadata = cudf::io::read_parquet_footers(sources); | ||||||
| CUDF_EXPECTS(std::cmp_equal(metadata.front().row_groups.size(), num_row_groups), | ||||||
| "Unexpected number of row groups"); | ||||||
| } | ||||||
| timer.stop(); | ||||||
| }); | ||||||
|
|
||||||
| state.add_buffer_size(counter->bytes_read(), "host_bytes_read", "Logical host bytes read"); | ||||||
| state.add_buffer_size(data.size(), "file_size", "Parquet file size"); | ||||||
| } | ||||||
|
|
||||||
| NVBENCH_BENCH(BM_parquet_page_index_metadata) | ||||||
| .set_name("parquet_page_index_metadata") | ||||||
| .set_min_samples(4) | ||||||
| .add_string_axis("layout", {"none", "offset_only", "mixed", "both"}) | ||||||
| .add_string_axis("reader", {"regular", "hybrid"}) | ||||||
| .add_int64_axis("num_cols", {4, 16}) | ||||||
| .add_int64_axis("num_row_groups", {10, 100}); | ||||||
|
|
||||||
| NVBENCH_BENCH(BM_parquet_read_footer) | ||||||
| .set_name("parquet_read_footer") | ||||||
| .set_min_samples(4) | ||||||
|
|
||||||
| Original file line number | Diff line number | Diff line change | ||||||||
|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -37,6 +37,7 @@ | |||||||||
| #include <functional> | ||||||||||
| #include <future> | ||||||||||
| #include <iterator> | ||||||||||
| #include <limits> | ||||||||||
| #include <numeric> | ||||||||||
| #include <optional> | ||||||||||
| #include <regex> | ||||||||||
|
|
@@ -46,6 +47,34 @@ | |||||||||
|
|
||||||||||
| namespace cudf::io::parquet::detail { | ||||||||||
|
|
||||||||||
| // Compute the page index (column index and/or offset index) byte range | ||||||||||
| text::byte_range_info page_index_byte_range(FileMetaData const& file_metadata) | ||||||||||
| { | ||||||||||
| int64_t min_offset = std::numeric_limits<int64_t>::max(); | ||||||||||
| int64_t max_offset = 0; | ||||||||||
| auto const include_index = [&](int64_t offset, int32_t length) { | ||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Naming nit: |
||||||||||
| if (offset > 0 and length > 0) { | ||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [Optional] I'm a big fan of guard clauses for readability. Can we invert this test and turn it into an early return, i.e.: if (offset <= 0 or length <= 0) { return; }
CUDF_EXPECTS(…);
min_offset = …
…? |
||||||||||
| CUDF_EXPECTS(offset <= std::numeric_limits<int64_t>::max() - length, | ||||||||||
| "Parquet page index range exceeds the supported offset range", | ||||||||||
| std::invalid_argument); | ||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why isn't this just |
||||||||||
| min_offset = std::min(min_offset, offset); | ||||||||||
| max_offset = std::max(max_offset, offset + length); | ||||||||||
|
coderabbitai[bot] marked this conversation as resolved.
|
||||||||||
| } | ||||||||||
| }; | ||||||||||
|
|
||||||||||
| // Indexes are optional for each column chunk. The first and last chunks need not have either | ||||||||||
| // index, so inspect all chunks to include every index that setup_page_index will parse. | ||||||||||
| for (auto const& row_group : file_metadata.row_groups) { | ||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is this rewrite simply because the indexes are optional? Can it still assume sorted offsets? If it can, it might be more efficient to iterate forward to find |
||||||||||
| for (auto const& column : row_group.columns) { | ||||||||||
| include_index(column.column_index_offset, column.column_index_length); | ||||||||||
| include_index(column.offset_index_offset, column.offset_index_length); | ||||||||||
| } | ||||||||||
| } | ||||||||||
|
|
||||||||||
| return max_offset > min_offset ? text::byte_range_info{min_offset, max_offset - min_offset} | ||||||||||
| : text::byte_range_info{}; | ||||||||||
|
Comment on lines
+74
to
+75
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If we change the
Suggested change
|
||||||||||
| } | ||||||||||
|
|
||||||||||
| std::size_t derive_pass_read_limit(std::size_t chunk_read_limit) | ||||||||||
| { | ||||||||||
| if (chunk_read_limit == 0) { return 0; } | ||||||||||
|
|
@@ -529,19 +558,22 @@ metadata::metadata(datasource* source, bool read_page_indexes) | |||||||||
| auto const has_strings = std::any_of( | ||||||||||
| schema.begin(), schema.end(), [](auto const& elem) { return elem.type == Type::BYTE_ARRAY; }); | ||||||||||
|
|
||||||||||
| if (read_page_indexes and has_strings and not row_groups.empty() and | ||||||||||
| not row_groups.front().columns.empty()) { | ||||||||||
| // column index and offset index are encoded back to back. | ||||||||||
| // the first column of the first row group will have the first column index, the last | ||||||||||
| // column of the last row group will have the final offset index. | ||||||||||
| int64_t const min_offset = row_groups.front().columns.front().column_index_offset; | ||||||||||
| auto const& last_col = row_groups.back().columns.back(); | ||||||||||
| int64_t const max_offset = last_col.offset_index_offset + last_col.offset_index_length; | ||||||||||
|
|
||||||||||
| if (max_offset > min_offset) { | ||||||||||
| size_t const length = max_offset - min_offset; | ||||||||||
| auto const page_idx_buf = source->host_read(min_offset, length); | ||||||||||
| setup_page_index({page_idx_buf->data(), length}, min_offset); | ||||||||||
| // Without offset indexes the decode paths cannot use column-index-derived information. | ||||||||||
| auto const has_offset_index = | ||||||||||
| std::any_of(row_groups.begin(), row_groups.end(), [](auto const& rg) { | ||||||||||
| return std::any_of(rg.columns.begin(), rg.columns.end(), [](auto const& col) { | ||||||||||
| return col.offset_index_offset > 0 and col.offset_index_length > 0; | ||||||||||
| }); | ||||||||||
| }); | ||||||||||
|
Comment on lines
+562
to
+567
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This computation is done unconditionally, even though it's only used when both Also, As written, the |
||||||||||
|
|
||||||||||
| if (read_page_indexes and has_strings and has_offset_index) { | ||||||||||
| auto const page_index_range = page_index_byte_range(*this); | ||||||||||
| if (not page_index_range.is_empty()) { | ||||||||||
| auto const page_idx_buf = | ||||||||||
| source->host_read(page_index_range.offset(), page_index_range.size()); | ||||||||||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should we check the resulting range against |
||||||||||
| CUDF_EXPECTS(std::cmp_equal(page_idx_buf->size(), page_index_range.size()), | ||||||||||
| "Encountered an invalid page index buffer"); | ||||||||||
| setup_page_index({page_idx_buf->data(), page_idx_buf->size()}, page_index_range.offset()); | ||||||||||
| } | ||||||||||
| } | ||||||||||
|
|
||||||||||
|
|
||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Just curious, why is
layouta string? Could (should) it be an enum, or, better yet, a composition of flags (see #24001 (comment))?And why isn't there a "column_only" benchmark?