Repository navigation
Add multifile dictionary pruning support for hybrid scan parquet reader #22866
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
Changes from 10 commits
bfce489
187036c
0115910
cddcce3
f632241
451367d
a76277e
291125f
d2ba16d
94251cd
a163660
865ab76
506ee8c
4a34b6c
b7b1791
91f421e
e5d97d5
eb2dbd4
2cd38dc
e6009c1
2f7d23a
da507b5
cca34ec
8b01cc4
ae55fc5
d1e2b89
2fc348d
a07b7b5
7aa70cf
f13cac8
eea8f1f
7d7271e
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 |
|---|---|---|
| @@ -1,5 +1,5 @@ | ||
| /* | ||
| * 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 | ||
| */ | ||
|
|
||
|
|
@@ -9,6 +9,7 @@ | |
| #include "io/parquet/compact_protocol_reader.hpp" | ||
| #include "io/parquet/expression_transform_helpers.hpp" | ||
| #include "io/parquet/reader_impl_helpers.hpp" | ||
| #include "page_index_filter_utils.hpp" | ||
|
|
||
| #include <cudf/detail/nvtx/ranges.hpp> | ||
| #include <cudf/logger.hpp> | ||
|
|
@@ -443,7 +444,8 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_bloom_filter_bytes( | |
| return bloom_filter_bytes; | ||
| } | ||
|
|
||
| std::vector<byte_range_info> aggregate_reader_metadata::get_dictionary_page_bytes( | ||
| std::pair<std::vector<byte_range_info>, std::vector<cudf::size_type>> | ||
| aggregate_reader_metadata::dictionary_pages_byte_ranges( | ||
| cudf::host_span<std::vector<cudf::size_type> const> row_group_indices, | ||
| host_span<data_type const> output_dtypes, | ||
| host_span<cudf::size_type const> output_column_schemas, | ||
|
|
@@ -461,7 +463,7 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_dictionary_page_byte | |
| std::back_inserter(dictionary_col_schemas), | ||
| [](auto& dict_literals) { return not dict_literals.empty(); }); | ||
|
|
||
| // No (in)equality literals found, return empty vector | ||
| // No (in)equality literals found, return empty vectors | ||
| if (dictionary_col_schemas.empty()) { return {}; } | ||
|
|
||
| // Compute total number of input row groups | ||
|
|
@@ -477,38 +479,41 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_dictionary_page_byte | |
| // Flag to check if we have at least one valid dictionary page | ||
| auto have_dictionary_pages = false; | ||
|
|
||
| // Association between each dictionary page byte range and its source | ||
| std::vector<cudf::size_type> dictionary_page_source_map; | ||
| dictionary_page_source_map.reserve(num_chunks); | ||
|
|
||
| // Cache each dictionary column's chunk offset across sources and row groups | ||
| std::vector<std::optional<size_type>> colchunk_offsets(dictionary_col_schemas.size()); | ||
|
|
||
| // For all sources | ||
| std::for_each( | ||
| cuda::counting_iterator<std::size_t>{0}, | ||
| cuda::counting_iterator{row_group_indices.size()}, | ||
| [&](auto const src_index) { | ||
| // Get all row group indices in the data source | ||
| auto const& rg_indices = row_group_indices[src_index]; | ||
| std::optional<size_type> colchunk_iter_offset{}; | ||
| // For all row groups | ||
| std::for_each(rg_indices.cbegin(), rg_indices.cend(), [&](auto const rg_index) { | ||
| auto const& row_group = per_file_metadata[src_index].row_groups[rg_index]; | ||
| // For all column chunks | ||
| auto const& row_group = per_file_metadata[src_index].row_groups[rg_index]; | ||
| auto const num_col_chunks = static_cast<size_type>(row_group.columns.size()); | ||
|
Comment on lines
475
to
+485
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. Validate row-group/source bounds before indexing metadata vectors. At Line 498, Suggested fix std::pair<std::vector<byte_range_info>, std::vector<cudf::size_type>>
aggregate_reader_metadata::dictionary_pages_byte_ranges(
cudf::host_span<std::vector<cudf::size_type> const> row_group_indices,
host_span<data_type const> output_dtypes,
host_span<cudf::size_type const> output_column_schemas,
std::reference_wrapper<ast::expression const> filter)
{
+ CUDF_EXPECTS(row_group_indices.size() == per_file_metadata.size(),
+ "Row group indices must provide one vector per input source",
+ std::invalid_argument);
+ for (std::size_t src_index = 0; src_index < row_group_indices.size(); ++src_index) {
+ auto const num_row_groups = per_file_metadata[src_index].row_groups.size();
+ for (auto const rg_index : row_group_indices[src_index]) {
+ CUDF_EXPECTS(std::cmp_greater_equal(rg_index, 0) and
+ std::cmp_less(rg_index, num_row_groups),
+ "Encountered out-of-bounds row group index for data source",
+ std::invalid_argument);
+ }
+ }
+
// Collect (in)equality literals for each input table column
auto const literals = dictionary_literals_collector{filter.get(), output_dtypes}.get_literals();🤖 Prompt for AI Agents |
||
| // For all dictionary column chunks | ||
| std::for_each( | ||
| dictionary_col_schemas.begin(), | ||
| dictionary_col_schemas.end(), | ||
| [&](auto const& schema_idx) { | ||
| // Get the column chunk iterator | ||
| if (not colchunk_iter_offset.has_value() or | ||
| row_group.columns[colchunk_iter_offset.value()].schema_idx != schema_idx) { | ||
| auto const& colchunk_iter = std::find_if( | ||
| row_group.columns.begin(), row_group.columns.end(), [schema_idx](auto const& col) { | ||
| return col.schema_idx == schema_idx; | ||
| }); | ||
| CUDF_EXPECTS(colchunk_iter != row_group.columns.end(), | ||
| "Column chunk with schema index " + std::to_string(schema_idx) + | ||
| " not found in row group", | ||
| std::invalid_argument); | ||
| colchunk_iter_offset = std::distance(row_group.columns.begin(), colchunk_iter); | ||
| cuda::counting_iterator<std::size_t>{0}, | ||
| cuda::counting_iterator{dictionary_col_schemas.size()}, | ||
| [&](auto const col) { | ||
| // Map the schema index to this source | ||
| auto const mapped_schema_idx = | ||
| map_schema_index(dictionary_col_schemas[col], static_cast<int>(src_index)); | ||
| auto& colchunk_offset = colchunk_offsets[col]; | ||
| auto const cached_offset = colchunk_offset.value_or(-1); | ||
| if (cached_offset < 0 or cached_offset >= num_col_chunks or | ||
| row_group.columns[cached_offset].schema_idx != mapped_schema_idx) { | ||
| colchunk_offset = find_colchunk_iter_offset(row_group, mapped_schema_idx); | ||
|
qbacpey marked this conversation as resolved.
Resolved
|
||
| } | ||
| auto const colchunk_iter = row_group.columns.begin() + colchunk_iter_offset.value(); | ||
| auto const& col_chunk = *colchunk_iter; | ||
| auto const& col_meta = col_chunk.meta_data; | ||
|
|
||
| auto const& col_chunk = row_group.columns[colchunk_offset.value()]; | ||
| auto const& col_meta = col_chunk.meta_data; | ||
|
|
||
| // Make sure that we have page index and the column chunk doesn't have any | ||
| // non-dictionary encoded pages | ||
|
|
@@ -564,13 +569,14 @@ std::vector<byte_range_info> aggregate_reader_metadata::get_dictionary_page_byte | |
| } | ||
|
|
||
| dictionary_page_bytes.emplace_back(dictionary_offset, dictionary_size); | ||
| dictionary_page_source_map.emplace_back(static_cast<size_type>(src_index)); | ||
| }); | ||
| }); | ||
| }); | ||
|
|
||
| if (not have_dictionary_pages) { return {}; } | ||
|
|
||
| return dictionary_page_bytes; | ||
| return {std::move(dictionary_page_bytes), std::move(dictionary_page_source_map)}; | ||
| } | ||
|
|
||
| std::vector<std::vector<cudf::size_type>> | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.