From bfce4898ad695af531251bdd93566361a05a801a Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Fri, 12 Jun 2026 11:41:19 +0200 Subject: [PATCH 01/18] [WIP] Add dictionary page filtering and byte range retrieval to hybrid scan multifile reader Need to refact testing code reused. --- .../io/experimental/hybrid_scan_multifile.hpp | 29 ++ .../experimental/hybrid_scan_helpers.cpp | 46 +-- .../parquet/experimental/hybrid_scan_impl.cpp | 13 + .../parquet/experimental/hybrid_scan_impl.hpp | 7 + .../experimental/hybrid_scan_multifile.cpp | 19 ++ .../hybrid_scan_multifile_filters_test.cpp | 262 ++++++++++++++++++ 6 files changed, 357 insertions(+), 19 deletions(-) diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp index 8da2f535e961..9bf0a037fa7b 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp @@ -167,6 +167,35 @@ class hybrid_scan_multifile { secondary_filters_byte_ranges(cudf::host_span const> row_group_indices, parquet_reader_options const& options) const; + /** + * @brief Get byte ranges of column chunk dictionary pages for row group pruning + * + * @param row_group_indices Input row group indices, one per source + * @param options Parquet reader options + * @return Vector of byte ranges of column chunks with dictionary pages subject to the filter + * predicate, ordered source-major then row-group then dictionary column + */ + [[nodiscard]] std::vector dictionary_pages_byte_ranges( + cudf::host_span const> row_group_indices, + parquet_reader_options const& options) const; + + /** + * @brief Filter the row groups using column chunk dictionary pages + * + * @param dictionary_page_data Device spans of dictionary page data of column chunks with an + * (in)equality predicate, ordered to match the dictionary page byte + * ranges returned by `dictionary_pages_byte_ranges` + * @param row_group_indices Input row group indices, one per source + * @param options Parquet reader options + * @param stream CUDA stream used for device memory operations and kernel launches + * @return Filtered per-source row group indices (one inner vector per source) + */ + [[nodiscard]] std::vector> filter_row_groups_with_dictionary_pages( + cudf::host_span const> dictionary_page_data, + cudf::host_span const> row_group_indices, + parquet_reader_options const& options, + rmm::cuda_stream_view stream) const; + private: std::unique_ptr _impl; }; diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index e61b4ad09bcb..f2e4b222f6be 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -477,6 +477,9 @@ std::vector aggregate_reader_metadata::get_dictionary_page_byte // Flag to check if we have at least one valid dictionary page auto have_dictionary_pages = false; + // Cache each dictionary column's chunk offset across sources and row groups + std::vector> colchunk_offsets(dictionary_col_schemas.size()); + // For all sources std::for_each( cuda::counting_iterator{0}, @@ -484,31 +487,36 @@ std::vector aggregate_reader_metadata::get_dictionary_page_byte [&](auto const src_index) { // Get all row group indices in the data source auto const& rg_indices = row_group_indices[src_index]; - std::optional 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(row_group.columns.size()); + // For all dictionary column chunks (kept inner to preserve the source-major emission order) 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) + + cuda::counting_iterator{0}, + cuda::counting_iterator{dictionary_col_schemas.size()}, + [&](auto const col) { + // Map the schema index to this source (for `allow_mismatched_pq_schemas`) + auto const mapped_schema_idx = + map_schema_index(dictionary_col_schemas[col], static_cast(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) { + auto const it = std::find_if( + row_group.columns.begin(), + row_group.columns.end(), + [mapped_schema_idx](auto const& c) { return c.schema_idx == mapped_schema_idx; }); + CUDF_EXPECTS(it != row_group.columns.end(), + "Column chunk with schema index " + std::to_string(mapped_schema_idx) + " not found in row group", std::invalid_argument); - colchunk_iter_offset = std::distance(row_group.columns.begin(), colchunk_iter); + colchunk_offset = + static_cast(std::distance(row_group.columns.begin(), it)); } - 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 diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index 0f0fd7de6d96..5344782baa65 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -297,6 +297,19 @@ hybrid_scan_reader_impl::secondary_filters_byte_ranges( return {bloom_filter_bytes, dictionary_page_bytes}; } +std::vector hybrid_scan_reader_impl::dictionary_pages_byte_ranges( + cudf::host_span const> row_group_indices, + parquet_reader_options const& options) +{ + CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered"); + auto [expr_conv, output_dtypes] = prepare_filter_and_output_types(options); + + return _extended_metadata->get_dictionary_page_bytes(row_group_indices, + output_dtypes, + _output_column_schemas, + expr_conv.get_converted_expr().value()); +} + std::vector> hybrid_scan_reader_impl::filter_row_groups_with_dictionary_pages( cudf::host_span const> dictionary_page_data, diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index d11ae1e8ddb9..01ab7e7e2d46 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -112,6 +112,13 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { secondary_filters_byte_ranges(cudf::host_span const> row_group_indices, parquet_reader_options const& options); + /** + * @copydoc cudf::io::experimental::hybrid_scan_multifile::dictionary_pages_byte_ranges + */ + [[nodiscard]] std::vector dictionary_pages_byte_ranges( + cudf::host_span const> row_group_indices, + parquet_reader_options const& options); + /** * @copydoc cudf::io::experimental::hybrid_scan::filter_row_groups_with_dictionary_pages */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp index 59347367d223..82d70815d0b4 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp @@ -83,4 +83,23 @@ hybrid_scan_multifile::secondary_filters_byte_ranges( return _impl->secondary_filters_byte_ranges(row_group_indices, options); } +std::vector hybrid_scan_multifile::dictionary_pages_byte_ranges( + cudf::host_span const> row_group_indices, + parquet_reader_options const& options) const +{ + CUDF_FUNC_RANGE(); + return _impl->dictionary_pages_byte_ranges(row_group_indices, options); +} + +std::vector> hybrid_scan_multifile::filter_row_groups_with_dictionary_pages( + cudf::host_span const> dictionary_page_data, + cudf::host_span const> row_group_indices, + parquet_reader_options const& options, + rmm::cuda_stream_view stream) const +{ + CUDF_FUNC_RANGE(); + return _impl->filter_row_groups_with_dictionary_pages( + dictionary_page_data, row_group_indices, options, stream); +} + } // namespace cudf::io::parquet::experimental diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index 29c66093b1cf..b53ff5016cd6 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -13,12 +13,20 @@ #include #include #include +#include +#include #include +#include #include #include +#include + +#include +#include #include #include +#include #include namespace { @@ -74,6 +82,127 @@ std::vector create_empty_parquet_with_stats() return buffer; } +/** + * @brief Writes `tbl` to a Parquet host buffer with the given top-level column names, column-chunk + * statistics + page index (STATISTICS_COLUMN) and ALWAYS dictionary encoding. + * + * Used to build reordered/mismatched per-source schemas that exercise the dictionary-page pruning + * path under `allow_mismatched_pq_schemas`. + */ +[[nodiscard]] std::vector write_mismatched_source(cudf::table_view const& tbl, + std::vector const& names) +{ + cudf::io::table_input_metadata md{tbl}; + for (std::size_t i = 0; i < names.size(); ++i) { + md.column_metadata[i].set_name(names[i]); + } + std::vector buffer; + auto out_opts = cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&buffer}, tbl) + .metadata(std::move(md)) + .stats_level(cudf::io::statistics_freq::STATISTICS_COLUMN) + .dictionary_policy(cudf::io::dictionary_policy::ALWAYS) + .build(); + cudf::io::write_parquet(out_opts); + return buffer; +} + +/** + * @brief Fetches and sets up the per-source page index on the reader. + * + * The page index buffers and their host spans are stored in the caller-provided vectors so that + * they outlive subsequent reader calls that rely on the page index. + */ +void setup_multifile_page_index( + cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, + multifile_inputs& inputs, + std::vector>& page_index_buffers, + std::vector>& page_index_byte_spans) +{ + auto const page_index_byte_ranges = reader.page_index_byte_ranges(); + auto const num_sources = inputs.datasources.size(); + page_index_buffers.reserve(num_sources); + page_index_byte_spans.reserve(num_sources); + + auto iter = cuda::zip_iterator(page_index_byte_ranges.begin(), inputs.datasources.begin()); + std::for_each(iter, iter + num_sources, [&](auto const& pair) { + auto const& [pgidx_byte_range, datasource] = pair; + page_index_buffers.emplace_back( + cudf::io::parquet::fetch_page_index_to_host(*datasource, pgidx_byte_range)); + page_index_byte_spans.emplace_back(*page_index_buffers.back()); + }); + + reader.setup_page_indexes( + cudf::host_span const>{page_index_byte_spans}); +} + +/** + * @brief Filter input row groups using column chunk dictionaries via the experimental parquet + * reader for hybrid scan (multi-file) + * + * Multi-file counterpart of the single-file `filter_row_groups_with_dictionaries` helper, kept as + * close to it as possible. The dictionary page byte ranges are flat and source-major, so each + * source's slice is fetched from its own datasource before filtering. + * + * @param inputs Multi-file datasources + * @param reader Hybrid scan multi-file reader + * @param input_row_group_indices Input per-source row group indices + * @param options Parquet reader options + * @param stream CUDA stream + * @param mr Device memory resource + * + * @return Vector of per-source dictionary-filtered row group indices + */ +auto filter_row_groups_with_dictionaries( + multifile_inputs& inputs, + cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, + std::vector> const& input_row_group_indices, + cudf::io::parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) +{ + // Get dictionary page byte ranges from the reader + auto const dict_page_byte_ranges = + reader.dictionary_pages_byte_ranges(input_row_group_indices, options); + + // If we have dictionary page byte ranges, filter row groups with dictionary pages + std::vector> dict_page_filtered_row_group_indices; + dict_page_filtered_row_group_indices.reserve(input_row_group_indices.size()); + + CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); + + // Dictionary page byte ranges are flat and source-major, so derive the dictionary column count + // and fetch each source's slice from its own datasource + std::size_t total_row_groups = 0; + for (auto const& rgs : input_row_group_indices) { + total_row_groups += rgs.size(); + } + auto const num_dictionary_columns = dict_page_byte_ranges.size() / total_row_groups; + + // Fetch dictionary page buffers from the input file buffers + std::vector dict_page_buffers; + std::vector> dict_page_data; + std::size_t offset = 0; + for (std::size_t src = 0; src < input_row_group_indices.size(); ++src) { + auto const count = input_row_group_indices[src].size() * num_dictionary_columns; + std::vector const src_ranges( + dict_page_byte_ranges.begin() + offset, dict_page_byte_ranges.begin() + offset + count); + offset += count; + + auto [buffers, data, task] = cudf::io::parquet::fetch_byte_ranges_to_device_async( + *inputs.datasources[src], src_ranges, stream, mr); + task.get(); + for (auto& buffer : buffers) { + dict_page_buffers.emplace_back(std::move(buffer)); + } + dict_page_data.insert(dict_page_data.end(), data.begin(), data.end()); + } + + dict_page_filtered_row_group_indices = reader.filter_row_groups_with_dictionary_pages( + dict_page_data, input_row_group_indices, options, stream); + + return dict_page_filtered_row_group_indices; +} + } // namespace struct HybridScanMultifileFiltersTest : public cudf::test::BaseFixture {}; @@ -314,3 +443,136 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats) EXPECT_TRUE(stats_filtered.front().empty()); EXPECT_TRUE(stats_filtered.back().empty()); } + +// Matched-schema real dictionary pruning across two sources. Both sources share the same schema but +// `col2` is a per-source constant string ("0100" in source A, "0200" in source B). With an equality +// predicate `col2 == "0100"` source A keeps all of its row groups while source B is fully pruned. +TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) +{ + using T = uint32_t; + auto constexpr num_sources = 2; + auto constexpr num_row_groups = 4; + auto constexpr rows_per_row_group = page_size_for_ordered_tests; + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); + + // Each source has 4 row groups (20000 rows / 5000 rows per row group) and is dictionary encoded + // under `dictionary_policy::ALWAYS`. `col2` is a per-source constant string. + std::vector> file_buffers; + file_buffers.reserve(num_sources); + srand(0xd1c7); + file_buffers.emplace_back(std::get<1>(create_parquet_with_stats(100))); // col2 == "0100" + srand(0xfeed); + file_buffers.emplace_back(std::get<1>(create_parquet_with_stats(200))); // col2 == "0200" + + auto inputs = build_multifile_inputs(file_buffers); + + // Filter - col2 == "0100" (present only in source A's dictionary) + auto literal_value = cudf::string_scalar("0100", true, stream); + auto literal = cudf::ast::literal(literal_value); + auto col_ref = cudf::ast::column_name_reference("col2"); + auto filter = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col_ref, literal); + + auto options = cudf::io::parquet_reader_options::builder().filter(filter).build(); + auto const reader = std::make_unique( + inputs.footer_byte_spans, options); + + // Page index is needed to detect dictionary-only encoded pages + std::vector> page_index_buffers; + std::vector> page_index_byte_spans; + setup_multifile_page_index(*reader, inputs, page_index_buffers, page_index_byte_spans); + + auto const input_row_group_indices = reader->all_row_groups(options); + ASSERT_EQ(input_row_group_indices.size(), num_sources); + EXPECT_EQ(reader->total_rows_in_row_groups(input_row_group_indices), + num_sources * num_row_groups * rows_per_row_group); + + auto const dict_filtered = filter_row_groups_with_dictionaries( + inputs, *reader, input_row_group_indices, options, stream, mr); + + // Source A keeps all 4 row groups (col2 == "0100"); source B is fully pruned (only "0200") + ASSERT_EQ(dict_filtered.size(), num_sources); + EXPECT_EQ(dict_filtered.front(), (std::vector{0, 1, 2, 3})); + EXPECT_TRUE(dict_filtered.back().empty()); +} + +// Mismatched-schema regression for the dictionary-page pruning path under +// `allow_mismatched_pq_schemas`. `get_dictionary_page_bytes` used to resolve column chunks by the +// raw zeroth-source `schema_idx` (no `map_schema_index`), so a reordered source read the wrong +// column chunk. Here `price` is schema 2 in source A but schema 3 in source B; the buggy lookup +// reads source B's int64 `id` (schema 2) instead of its double `price`. With `price == 40` (present +// only in source B) the correct result keeps source B and prunes source A. +TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollision) +{ + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); + + // Use enough low-cardinality rows that the writer actually emits dictionary-encoded pages even + // under `dictionary_policy::ALWAYS` (a tiny all-unique column falls back to PLAIN because the + // dictionary would not save space). `price == 40` is present only in source B's `price` column. + auto constexpr num_rows = cudf::size_type{600}; + std::array const price_a_cycle{50.0, 150.0, 75.0}; // no 40 + std::array const price_b_cycle{40.0, 200.0, 99.0}; // has 40 + std::array const cat_cycle{"x", "y", "z"}; + + std::vector id_a_vals(num_rows); + std::vector id_b_vals(num_rows); + std::vector price_a_vals(num_rows); + std::vector price_b_vals(num_rows); + std::vector category_b_vals(num_rows); + for (cudf::size_type i = 0; i < num_rows; ++i) { + id_a_vals[i] = (i % 3) + 1; // {1, 2, 3} + id_b_vals[i] = 1000 + (i % 3); // {1000, 1001, 1002} + price_a_vals[i] = price_a_cycle[i % 3]; + price_b_vals[i] = price_b_cycle[i % 3]; + category_b_vals[i] = cat_cycle[i % 3]; + } + cudf::test::fixed_width_column_wrapper const id_a(id_a_vals.begin(), id_a_vals.end()); + cudf::test::fixed_width_column_wrapper const price_a(price_a_vals.begin(), + price_a_vals.end()); + cudf::test::strings_column_wrapper const category_b(category_b_vals.begin(), + category_b_vals.end()); + cudf::test::fixed_width_column_wrapper const id_b(id_b_vals.begin(), id_b_vals.end()); + cudf::test::fixed_width_column_wrapper const price_b(price_b_vals.begin(), + price_b_vals.end()); + + std::vector> file_buffers; + file_buffers.emplace_back( + write_mismatched_source(cudf::table_view{{id_a, price_a}}, {"id", "price"})); + file_buffers.emplace_back(write_mismatched_source(cudf::table_view{{category_b, id_b, price_b}}, + {"category", "id", "price"})); + + auto inputs = build_multifile_inputs(file_buffers); + + // Filter - price == 40 (dictionary pruning participates for equality predicates) + auto literal_value = cudf::numeric_scalar(40.0, true, stream); + auto literal = cudf::ast::literal(literal_value); + auto col_ref = cudf::ast::column_name_reference("price"); + auto filter = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col_ref, literal); + + auto options = cudf::io::parquet_reader_options::builder() + .allow_mismatched_pq_schemas(true) + .column_names({"id", "price"}) + .filter(filter) + .build(); + + auto const reader = std::make_unique( + cudf::host_span const>{inputs.footer_byte_spans}, options); + + // Page index is needed to detect dictionary-only encoded pages + std::vector> page_index_buffers; + std::vector> page_index_byte_spans; + setup_multifile_page_index(*reader, inputs, page_index_buffers, page_index_byte_spans); + + auto const input_row_group_indices = reader->all_row_groups(options); + ASSERT_EQ(input_row_group_indices.size(), 2); + + auto const dict_filtered = filter_row_groups_with_dictionaries( + inputs, *reader, input_row_group_indices, options, stream, mr); + + // Correct behavior: source A is pruned (no price == 40), source B survives (price == 40 present) + ASSERT_EQ(dict_filtered.size(), 2); + EXPECT_TRUE(dict_filtered.front().empty()) << "Source A should be pruned (no price == 40)"; + EXPECT_EQ(dict_filtered.back(), (std::vector{0})) + << "Source B should survive (price == 40 present)"; +} From 187036c4929c1155d883def82ccae0b274fed662 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Fri, 12 Jun 2026 12:50:10 +0200 Subject: [PATCH 02/18] Refactor hybrid scan multifile filters test to simplify row group filtering --- .../hybrid_scan_multifile_filters_test.cpp | 54 ++++++------------- 1 file changed, 16 insertions(+), 38 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index b53ff5016cd6..5072fde4d2e6 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -139,13 +139,8 @@ void setup_multifile_page_index( * @brief Filter input row groups using column chunk dictionaries via the experimental parquet * reader for hybrid scan (multi-file) * - * Multi-file counterpart of the single-file `filter_row_groups_with_dictionaries` helper, kept as - * close to it as possible. The dictionary page byte ranges are flat and source-major, so each - * source's slice is fetched from its own datasource before filtering. - * * @param inputs Multi-file datasources * @param reader Hybrid scan multi-file reader - * @param input_row_group_indices Input per-source row group indices * @param options Parquet reader options * @param stream CUDA stream * @param mr Device memory resource @@ -155,11 +150,13 @@ void setup_multifile_page_index( auto filter_row_groups_with_dictionaries( multifile_inputs& inputs, cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, - std::vector> const& input_row_group_indices, cudf::io::parquet_reader_options const& options, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { + // Get all row groups from the reader + auto const input_row_group_indices = reader.all_row_groups(options); + // Get dictionary page byte ranges from the reader auto const dict_page_byte_ranges = reader.dictionary_pages_byte_ranges(input_row_group_indices, options); @@ -444,20 +441,14 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithStats) EXPECT_TRUE(stats_filtered.back().empty()); } -// Matched-schema real dictionary pruning across two sources. Both sources share the same schema but -// `col2` is a per-source constant string ("0100" in source A, "0200" in source B). With an equality -// predicate `col2 == "0100"` source A keeps all of its row groups while source B is fully pruned. TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) { - using T = uint32_t; - auto constexpr num_sources = 2; - auto constexpr num_row_groups = 4; - auto constexpr rows_per_row_group = page_size_for_ordered_tests; - auto stream = cudf::get_default_stream(); - auto mr = cudf::get_current_device_resource_ref(); + using T = uint32_t; + auto constexpr num_sources = 2; + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); - // Each source has 4 row groups (20000 rows / 5000 rows per row group) and is dictionary encoded - // under `dictionary_policy::ALWAYS`. `col2` is a per-source constant string. + // 2 sources, each `dictionary_policy::ALWAYS` with a per-source constant `col2` std::vector> file_buffers; file_buffers.reserve(num_sources); srand(0xd1c7); @@ -482,13 +473,8 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) std::vector> page_index_byte_spans; setup_multifile_page_index(*reader, inputs, page_index_buffers, page_index_byte_spans); - auto const input_row_group_indices = reader->all_row_groups(options); - ASSERT_EQ(input_row_group_indices.size(), num_sources); - EXPECT_EQ(reader->total_rows_in_row_groups(input_row_group_indices), - num_sources * num_row_groups * rows_per_row_group); - - auto const dict_filtered = filter_row_groups_with_dictionaries( - inputs, *reader, input_row_group_indices, options, stream, mr); + auto const dict_filtered = + filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); // Source A keeps all 4 row groups (col2 == "0100"); source B is fully pruned (only "0200") ASSERT_EQ(dict_filtered.size(), num_sources); @@ -496,20 +482,13 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) EXPECT_TRUE(dict_filtered.back().empty()); } -// Mismatched-schema regression for the dictionary-page pruning path under -// `allow_mismatched_pq_schemas`. `get_dictionary_page_bytes` used to resolve column chunks by the -// raw zeroth-source `schema_idx` (no `map_schema_index`), so a reordered source read the wrong -// column chunk. Here `price` is schema 2 in source A but schema 3 in source B; the buggy lookup -// reads source B's int64 `id` (schema 2) instead of its double `price`. With `price == 40` (present -// only in source B) the correct result keeps source B and prunes source A. TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollision) { auto stream = cudf::get_default_stream(); auto mr = cudf::get_current_device_resource_ref(); - // Use enough low-cardinality rows that the writer actually emits dictionary-encoded pages even - // under `dictionary_policy::ALWAYS` (a tiny all-unique column falls back to PLAIN because the - // dictionary would not save space). `price == 40` is present only in source B's `price` column. + // Low-cardinality `price` so the writer emits dictionary pages under ALWAYS; `price == 40` is in + // source B only auto constexpr num_rows = cudf::size_type{600}; std::array const price_a_cycle{50.0, 150.0, 75.0}; // no 40 std::array const price_b_cycle{40.0, 200.0, 99.0}; // has 40 @@ -536,6 +515,8 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio cudf::test::fixed_width_column_wrapper const price_b(price_b_vals.begin(), price_b_vals.end()); + // Reordered schemas: `price` is schema 2 in source A but schema 3 in source B, so dictionary + // pruning must map schema indices per source std::vector> file_buffers; file_buffers.emplace_back( write_mismatched_source(cudf::table_view{{id_a, price_a}}, {"id", "price"})); @@ -564,11 +545,8 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio std::vector> page_index_byte_spans; setup_multifile_page_index(*reader, inputs, page_index_buffers, page_index_byte_spans); - auto const input_row_group_indices = reader->all_row_groups(options); - ASSERT_EQ(input_row_group_indices.size(), 2); - - auto const dict_filtered = filter_row_groups_with_dictionaries( - inputs, *reader, input_row_group_indices, options, stream, mr); + auto const dict_filtered = + filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); // Correct behavior: source A is pruned (no price == 40), source B survives (price == 40 present) ASSERT_EQ(dict_filtered.size(), 2); From 01159105bb77be65fecf1e1de403fede9d7ba423 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Fri, 12 Jun 2026 16:59:47 +0200 Subject: [PATCH 03/18] Refactor setup_multifile_page_index to simplify buffer management in hybrid scan multifile filters test --- .../hybrid_scan_multifile_filters_test.cpp | 46 +++++++++---------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index 5072fde4d2e6..632daeeb0ae7 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -25,6 +25,8 @@ #include #include #include +#include +#include #include #include #include @@ -108,28 +110,30 @@ std::vector create_empty_parquet_with_stats() /** * @brief Fetches and sets up the per-source page index on the reader. - * - * The page index buffers and their host spans are stored in the caller-provided vectors so that - * they outlive subsequent reader calls that rely on the page index. */ void setup_multifile_page_index( - cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, - multifile_inputs& inputs, - std::vector>& page_index_buffers, - std::vector>& page_index_byte_spans) + cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, multifile_inputs& inputs) { + // Reference wrappers to the datasources, in source order + std::vector> datasource_refs; + datasource_refs.reserve(inputs.datasources.size()); + std::transform(inputs.datasources.begin(), + inputs.datasources.end(), + std::back_inserter(datasource_refs), + [](auto& datasource) { return std::ref(*datasource); }); + + // Fetch all per-source page index buffers in one batch auto const page_index_byte_ranges = reader.page_index_byte_ranges(); - auto const num_sources = inputs.datasources.size(); - page_index_buffers.reserve(num_sources); - page_index_byte_spans.reserve(num_sources); + auto const page_index_buffers = + cudf::io::parquet::fetch_page_indexes_to_host(datasource_refs, page_index_byte_ranges); - auto iter = cuda::zip_iterator(page_index_byte_ranges.begin(), inputs.datasources.begin()); - std::for_each(iter, iter + num_sources, [&](auto const& pair) { - auto const& [pgidx_byte_range, datasource] = pair; - page_index_buffers.emplace_back( - cudf::io::parquet::fetch_page_index_to_host(*datasource, pgidx_byte_range)); - page_index_byte_spans.emplace_back(*page_index_buffers.back()); - }); + // Set up the page index on the reader from the fetched buffers + std::vector> page_index_byte_spans; + page_index_byte_spans.reserve(page_index_buffers.size()); + std::transform(page_index_buffers.begin(), + page_index_buffers.end(), + std::back_inserter(page_index_byte_spans), + [](auto const& buffer) { return cudf::host_span{*buffer}; }); reader.setup_page_indexes( cudf::host_span const>{page_index_byte_spans}); @@ -469,9 +473,7 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) inputs.footer_byte_spans, options); // Page index is needed to detect dictionary-only encoded pages - std::vector> page_index_buffers; - std::vector> page_index_byte_spans; - setup_multifile_page_index(*reader, inputs, page_index_buffers, page_index_byte_spans); + setup_multifile_page_index(*reader, inputs); auto const dict_filtered = filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); @@ -541,9 +543,7 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio cudf::host_span const>{inputs.footer_byte_spans}, options); // Page index is needed to detect dictionary-only encoded pages - std::vector> page_index_buffers; - std::vector> page_index_byte_spans; - setup_multifile_page_index(*reader, inputs, page_index_buffers, page_index_byte_spans); + setup_multifile_page_index(*reader, inputs); auto const dict_filtered = filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); From cddcce338eef9a88f0054020b024dc7ad2d36fe1 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Fri, 12 Jun 2026 18:07:41 +0200 Subject: [PATCH 04/18] Enhance create_parquet_with_stats to support customizable column names and order. Update hybrid_scan_multifile_filters_test to validate mismatched schema handling for dictionary pruning. Remove unused write_mismatched_source function to streamline code. --- .../io/experimental/hybrid_scan_common.hpp | 33 ++++- .../hybrid_scan_multifile_filters_test.cpp | 116 +++++++----------- 2 files changed, 72 insertions(+), 77 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index a64b9facb471..3614f50d7c7a 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -9,6 +9,7 @@ #include +#include #include #include #include @@ -18,6 +19,8 @@ #include #include +#include +#include /** * @brief Creates a strings column with a constant stringified value between 0 and 9999 @@ -92,6 +95,11 @@ cudf::test::fixed_width_column_wrapper descending_low_cardinality() * * @param str_col_value Value for the constant string column used when IsConstantStrings is true * @param compression Compression type + * @param column_names Top-level column names assigned in `column_order` order (default + * {"col0", "col1", "col2"}) + * @param column_order Physical emit order of the base [col0, col1, col2] columns (default + * {0, 1, 2}). Reordering emits the same logical columns at different schema positions, which + * is used to build mismatched per-source schemas for the row-group filtering tests. * @param stream CUDA stream * * @return Tuple of table and Parquet host buffer @@ -101,9 +109,11 @@ template auto create_parquet_with_stats( - cudf::size_type str_col_value = 100, - cudf::io::compression_type compression = cudf::io::compression_type::AUTO, - rmm::cuda_stream_view stream = cudf::get_default_stream()) + cudf::size_type str_col_value = 100, + cudf::io::compression_type compression = cudf::io::compression_type::AUTO, + std::vector column_names = {"col0", "col1", "col2"}, + std::vector column_order = {0, 1, 2}, + rmm::cuda_stream_view stream = cudf::get_default_stream()) { static_assert(NumTableConcats >= 1, "Concatenated table must contain at least one table"); @@ -157,12 +167,23 @@ auto create_parquet_with_stats( output = table_view{{columns[0]->view(), columns[1]->view(), columns[2]->view()}}; } + // Reorder the base [col0, col1, col2] columns into the requested physical order, naming them in + // that new order. Reordering lets callers emit the same logical columns at different schema + // positions across sources (a "mismatched schema"), which exercises the per-source schema-index + // mapping in the row-group filtering paths. The defaults leave the table and names unchanged. + std::vector reordered_columns; + reordered_columns.reserve(column_order.size()); + for (auto const col_idx : column_order) { + reordered_columns.emplace_back(output.column(col_idx)); + } + output = table_view{reordered_columns}; + auto table = cudf::concatenate(std::vector(NumTableConcats, output)); output = table->view(); cudf::io::table_input_metadata output_metadata(output); - output_metadata.column_metadata[0].set_name("col0"); - output_metadata.column_metadata[1].set_name("col1"); - output_metadata.column_metadata[2].set_name("col2"); + for (std::size_t i = 0; i < column_names.size(); ++i) { + output_metadata.column_metadata[i].set_name(column_names[i]); + } std::vector buffer; cudf::io::parquet_writer_options out_opts = diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index 632daeeb0ae7..7d34f1368220 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -23,7 +23,6 @@ #include #include -#include #include #include #include @@ -84,30 +83,6 @@ std::vector create_empty_parquet_with_stats() return buffer; } -/** - * @brief Writes `tbl` to a Parquet host buffer with the given top-level column names, column-chunk - * statistics + page index (STATISTICS_COLUMN) and ALWAYS dictionary encoding. - * - * Used to build reordered/mismatched per-source schemas that exercise the dictionary-page pruning - * path under `allow_mismatched_pq_schemas`. - */ -[[nodiscard]] std::vector write_mismatched_source(cudf::table_view const& tbl, - std::vector const& names) -{ - cudf::io::table_input_metadata md{tbl}; - for (std::size_t i = 0; i < names.size(); ++i) { - md.column_metadata[i].set_name(names[i]); - } - std::vector buffer; - auto out_opts = cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&buffer}, tbl) - .metadata(std::move(md)) - .stats_level(cudf::io::statistics_freq::STATISTICS_COLUMN) - .dictionary_policy(cudf::io::dictionary_policy::ALWAYS) - .build(); - cudf::io::write_parquet(out_opts); - return buffer; -} - /** * @brief Fetches and sets up the per-source page index on the reader. */ @@ -484,73 +459,72 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) EXPECT_TRUE(dict_filtered.back().empty()); } +// Mismatched-schema regression for the dictionary-page pruning path under +// `allow_mismatched_pq_schemas`. `get_dictionary_page_bytes` used to resolve column chunks by the +// raw zeroth-source `schema_idx` (without `map_schema_index`), so a reordered source read the wrong +// column chunk. +// +// Both sources hold the same three `create_parquet_with_stats` columns (col0: ascending, col1: +// low-cardinality descending, col2: constant string), but source B emits them physically reordered +// as {col2, col0, col1}. So the predicate column `col2` is schema index 3 in source A (the zeroth +// source) yet schema index 1 in source B. The buggy raw-index lookup reads source B's schema index +// 3 -- `col1`, a duration column -- instead of its `col2` string dictionary, garbling/throwing on +// the type mismatch. With the fix, `col2` maps per source: source A (col2 == "0200") is pruned and +// source B (col2 == "0100") survives. A chrono `T` is used so the collided `col1` is itself +// dictionary-encoded, and hence actually fetched by the buggy path. TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollision) { - auto stream = cudf::get_default_stream(); - auto mr = cudf::get_current_device_resource_ref(); - - // Low-cardinality `price` so the writer emits dictionary pages under ALWAYS; `price == 40` is in - // source B only - auto constexpr num_rows = cudf::size_type{600}; - std::array const price_a_cycle{50.0, 150.0, 75.0}; // no 40 - std::array const price_b_cycle{40.0, 200.0, 99.0}; // has 40 - std::array const cat_cycle{"x", "y", "z"}; - - std::vector id_a_vals(num_rows); - std::vector id_b_vals(num_rows); - std::vector price_a_vals(num_rows); - std::vector price_b_vals(num_rows); - std::vector category_b_vals(num_rows); - for (cudf::size_type i = 0; i < num_rows; ++i) { - id_a_vals[i] = (i % 3) + 1; // {1, 2, 3} - id_b_vals[i] = 1000 + (i % 3); // {1000, 1001, 1002} - price_a_vals[i] = price_a_cycle[i % 3]; - price_b_vals[i] = price_b_cycle[i % 3]; - category_b_vals[i] = cat_cycle[i % 3]; - } - cudf::test::fixed_width_column_wrapper const id_a(id_a_vals.begin(), id_a_vals.end()); - cudf::test::fixed_width_column_wrapper const price_a(price_a_vals.begin(), - price_a_vals.end()); - cudf::test::strings_column_wrapper const category_b(category_b_vals.begin(), - category_b_vals.end()); - cudf::test::fixed_width_column_wrapper const id_b(id_b_vals.begin(), id_b_vals.end()); - cudf::test::fixed_width_column_wrapper const price_b(price_b_vals.begin(), - price_b_vals.end()); - - // Reordered schemas: `price` is schema 2 in source A but schema 3 in source B, so dictionary - // pruning must map schema indices per source + using T = cudf::duration_ms; + auto constexpr num_sources = 2; + auto stream = cudf::get_default_stream(); + auto mr = cudf::get_current_device_resource_ref(); + + // Source A: default column order/names, col2 == "0200" (pruned by the filter). + // Source B: same columns emitted as {col2, col0, col1}, col2 == "0100" (survives). The reorder + // moves `col2` to a different schema index than in source A, which is what triggers the bug. std::vector> file_buffers; - file_buffers.emplace_back( - write_mismatched_source(cudf::table_view{{id_a, price_a}}, {"id", "price"})); - file_buffers.emplace_back(write_mismatched_source(cudf::table_view{{category_b, id_b, price_b}}, - {"category", "id", "price"})); + file_buffers.reserve(num_sources); + srand(0xd1c7); + file_buffers.emplace_back(std::get<1>(create_parquet_with_stats(200))); + srand(0xfeed); + file_buffers.emplace_back(std::get<1>(create_parquet_with_stats( + 100, cudf::io::compression_type::AUTO, {"col2", "col0", "col1"}, {2, 0, 1}))); auto inputs = build_multifile_inputs(file_buffers); - // Filter - price == 40 (dictionary pruning participates for equality predicates) - auto literal_value = cudf::numeric_scalar(40.0, true, stream); + // Filter - col2 == "0100" + auto literal_value = cudf::string_scalar("0100", true, stream); auto literal = cudf::ast::literal(literal_value); - auto col_ref = cudf::ast::column_name_reference("price"); + auto col_ref = cudf::ast::column_name_reference("col2"); auto filter = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col_ref, literal); auto options = cudf::io::parquet_reader_options::builder() .allow_mismatched_pq_schemas(true) - .column_names({"id", "price"}) + .column_names({"col2"}) .filter(filter) .build(); auto const reader = std::make_unique( cudf::host_span const>{inputs.footer_byte_spans}, options); + // Guard the premise of the test: the reorder must genuinely differ the per-source schemas + // (source A's first column is `col0`, source B's first column is `col2`), otherwise there is no + // schema-index mismatch for the dictionary pruning path to get wrong. + auto const metadatas = reader->parquet_metadatas(); + ASSERT_EQ(metadatas.size(), num_sources); + EXPECT_EQ(metadatas.front().schema.at(1).name, "col0"); + EXPECT_EQ(metadatas.back().schema.at(1).name, "col2"); + // Page index is needed to detect dictionary-only encoded pages setup_multifile_page_index(*reader, inputs); auto const dict_filtered = filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); - // Correct behavior: source A is pruned (no price == 40), source B survives (price == 40 present) - ASSERT_EQ(dict_filtered.size(), 2); - EXPECT_TRUE(dict_filtered.front().empty()) << "Source A should be pruned (no price == 40)"; - EXPECT_EQ(dict_filtered.back(), (std::vector{0})) - << "Source B should survive (price == 40 present)"; + // Correct behavior: source A is pruned (col2 == "0200"), source B survives (col2 == "0100"). The + // buggy raw-index path instead reads source B's `col1` duration dictionary for `col2`. + ASSERT_EQ(dict_filtered.size(), num_sources); + EXPECT_TRUE(dict_filtered.front().empty()) << "Source A should be pruned (col2 == \"0200\")"; + EXPECT_EQ(dict_filtered.back(), (std::vector{0, 1, 2, 3})) + << "Source B should survive (col2 == \"0100\")"; } From f63224128188927253758b0ab2758f35fca47620 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Sat, 13 Jun 2026 09:31:47 +0200 Subject: [PATCH 05/18] Refactor column chunk offset retrieval in hybrid scan helpers to use a dedicated function for improved clarity and maintainability. Update error message formatting in page index filter utilities to utilize std::format for better readability. --- .../io/parquet/experimental/hybrid_scan_helpers.cpp | 12 ++---------- .../parquet/experimental/page_index_filter_utils.cu | 8 ++++---- 2 files changed, 6 insertions(+), 14 deletions(-) diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index f2e4b222f6be..9f311c962fa1 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -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 #include @@ -503,16 +504,7 @@ std::vector aggregate_reader_metadata::get_dictionary_page_byte 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) { - auto const it = std::find_if( - row_group.columns.begin(), - row_group.columns.end(), - [mapped_schema_idx](auto const& c) { return c.schema_idx == mapped_schema_idx; }); - CUDF_EXPECTS(it != row_group.columns.end(), - "Column chunk with schema index " + std::to_string(mapped_schema_idx) + - " not found in row group", - std::invalid_argument); - colchunk_offset = - static_cast(std::distance(row_group.columns.begin(), it)); + colchunk_offset = find_colchunk_iter_offset(row_group, mapped_schema_idx); } auto const& col_chunk = row_group.columns[colchunk_offset.value()]; diff --git a/cpp/src/io/parquet/experimental/page_index_filter_utils.cu b/cpp/src/io/parquet/experimental/page_index_filter_utils.cu index 57ce7f852e52..2eef122e840f 100644 --- a/cpp/src/io/parquet/experimental/page_index_filter_utils.cu +++ b/cpp/src/io/parquet/experimental/page_index_filter_utils.cu @@ -18,6 +18,7 @@ #include #include +#include #include #include @@ -29,10 +30,9 @@ size_type find_colchunk_iter_offset(RowGroup const& row_group, size_type schema_ 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); + CUDF_EXPECTS(colchunk_iter != row_group.columns.end(), + std::format("Column chunk with schema index {} not found in row group", schema_idx), + std::invalid_argument); return std::distance(row_group.columns.begin(), colchunk_iter); } From a76277e3aae7c4a1d42736daa4685c0d35499000 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Mon, 15 Jun 2026 23:45:57 +0200 Subject: [PATCH 06/18] Refactor dictionary page byte range functions to return pairs of vectors Updated the `dictionary_pages_byte_ranges` method across multiple files to return a pair of vectors: one for the byte ranges of dictionary pages and another for their corresponding source indices. This change enhances the functionality and clarity of the data returned, facilitating better handling of dictionary page filtering in hybrid scan operations. --- .../io/experimental/hybrid_scan_multifile.hpp | 10 ++--- .../experimental/hybrid_scan_helpers.cpp | 12 ++++-- .../experimental/hybrid_scan_helpers.hpp | 14 ++++--- .../parquet/experimental/hybrid_scan_impl.cpp | 21 ++++++---- .../parquet/experimental/hybrid_scan_impl.hpp | 6 +-- .../experimental/hybrid_scan_multifile.cpp | 3 +- .../hybrid_scan_multifile_filters_test.cpp | 42 +++++++------------ 7 files changed, 54 insertions(+), 54 deletions(-) diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp index cee98a7f2c42..d6d892f595a2 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp @@ -233,12 +233,12 @@ class hybrid_scan_multifile { * * @param row_group_indices Input row group indices, one per source * @param options Parquet reader options - * @return Vector of byte ranges of column chunks with dictionary pages subject to the filter - * predicate, ordered source-major then row-group then dictionary column + * @return Pair of flattened byte ranges to column chunk dictionary pages subject to the filter + * predicate and their corresponding source indices */ - [[nodiscard]] std::vector dictionary_pages_byte_ranges( - cudf::host_span const> row_group_indices, - parquet_reader_options const& options) const; + [[nodiscard]] std::pair, std::vector> + dictionary_pages_byte_ranges(cudf::host_span const> row_group_indices, + parquet_reader_options const& options) const; /** * @brief Filter the row groups using column chunk dictionary pages diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index 9f311c962fa1..53aa8fc3a7ae 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -444,7 +444,8 @@ std::vector aggregate_reader_metadata::get_bloom_filter_bytes( return bloom_filter_bytes; } -std::vector aggregate_reader_metadata::get_dictionary_page_bytes( +std::pair, std::vector> +aggregate_reader_metadata::dictionary_pages_byte_ranges( cudf::host_span const> row_group_indices, host_span output_dtypes, host_span output_column_schemas, @@ -462,7 +463,7 @@ std::vector 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 @@ -475,6 +476,10 @@ std::vector aggregate_reader_metadata::get_dictionary_page_byte std::vector dictionary_page_bytes; dictionary_page_bytes.reserve(num_chunks); + // Association between each dictionary page byte range and its source + std::vector dictionary_page_source_map; + dictionary_page_source_map.reserve(num_chunks); + // Flag to check if we have at least one valid dictionary page auto have_dictionary_pages = false; @@ -564,13 +569,14 @@ std::vector aggregate_reader_metadata::get_dictionary_page_byte } dictionary_page_bytes.emplace_back(dictionary_offset, dictionary_size); + dictionary_page_source_map.emplace_back(static_cast(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> diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp index e65db678c2d1..5c494060bfb2 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp @@ -238,13 +238,15 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base { * @param output_column_schemas schema indices of output columns * @param filter AST expression to filter row groups based on dictionary pages * - * @return Byte ranges of dictionary pages, one input column chunk with (in)equality predicate + * @return A pair of vectors containing dictionary page byte ranges and corresponding source + * indices */ - [[nodiscard]] std::vector get_dictionary_page_bytes( - cudf::host_span const> row_group_indices, - cudf::host_span output_dtypes, - cudf::host_span output_column_schemas, - std::reference_wrapper filter); + [[nodiscard]] std::pair, + std::vector> + dictionary_pages_byte_ranges(cudf::host_span const> row_group_indices, + cudf::host_span output_dtypes, + cudf::host_span output_column_schemas, + std::reference_wrapper filter); /** * @brief Filter the row groups using dictionaries based on predicate filter diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index 5344782baa65..b8dc19328437 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -289,25 +289,28 @@ hybrid_scan_reader_impl::secondary_filters_byte_ranges( _output_column_schemas, expr_conv.get_converted_expr().value()); auto const dictionary_page_bytes = - _extended_metadata->get_dictionary_page_bytes(row_group_indices, - output_dtypes, - _output_column_schemas, - expr_conv.get_converted_expr().value()); + _extended_metadata + ->dictionary_pages_byte_ranges(row_group_indices, + output_dtypes, + _output_column_schemas, + expr_conv.get_converted_expr().value()) + .first; return {bloom_filter_bytes, dictionary_page_bytes}; } -std::vector hybrid_scan_reader_impl::dictionary_pages_byte_ranges( +std::pair, std::vector> +hybrid_scan_reader_impl::dictionary_pages_byte_ranges( cudf::host_span const> row_group_indices, parquet_reader_options const& options) { CUDF_EXPECTS(not row_group_indices.empty(), "Empty input row group indices encountered"); auto [expr_conv, output_dtypes] = prepare_filter_and_output_types(options); - return _extended_metadata->get_dictionary_page_bytes(row_group_indices, - output_dtypes, - _output_column_schemas, - expr_conv.get_converted_expr().value()); + return _extended_metadata->dictionary_pages_byte_ranges(row_group_indices, + output_dtypes, + _output_column_schemas, + expr_conv.get_converted_expr().value()); } std::vector> diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index fd012b5ccfbf..cde91da596be 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -115,9 +115,9 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { /** * @copydoc cudf::io::experimental::hybrid_scan_multifile::dictionary_pages_byte_ranges */ - [[nodiscard]] std::vector dictionary_pages_byte_ranges( - cudf::host_span const> row_group_indices, - parquet_reader_options const& options); + [[nodiscard]] std::pair, std::vector> + dictionary_pages_byte_ranges(cudf::host_span const> row_group_indices, + parquet_reader_options const& options); /** * @copydoc cudf::io::experimental::hybrid_scan::filter_row_groups_with_dictionary_pages diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp index 8ad53743994d..89584a353b3f 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp @@ -122,7 +122,8 @@ table_with_metadata hybrid_scan_multifile::materialize_all_columns( return _impl->materialize_all_columns(row_group_indices, column_chunk_data, options, stream, mr); } -std::vector hybrid_scan_multifile::dictionary_pages_byte_ranges( +std::pair, std::vector> +hybrid_scan_multifile::dictionary_pages_byte_ranges( cudf::host_span const> row_group_indices, parquet_reader_options const& options) const { diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index e3100be483c6..32f2211933d6 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -137,33 +137,23 @@ auto filter_row_groups_with_dictionaries( // Get all row groups from the reader auto const input_row_group_indices = reader.all_row_groups(options); - // Get dictionary page byte ranges from the reader - auto const dict_page_byte_ranges = + // Get dictionary page byte ranges and their corresponding source indices from the reader + auto const [dict_page_byte_ranges, dict_page_source_map] = reader.dictionary_pages_byte_ranges(input_row_group_indices, options); - - // If we have dictionary page byte ranges, filter row groups with dictionary pages - std::vector> dict_page_filtered_row_group_indices; - dict_page_filtered_row_group_indices.reserve(input_row_group_indices.size()); - CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); - // Dictionary page byte ranges are flat and source-major, so derive the dictionary column count - // and fetch each source's slice from its own datasource - std::size_t total_row_groups = 0; - for (auto const& rgs : input_row_group_indices) { - total_row_groups += rgs.size(); - } - auto const num_dictionary_columns = dict_page_byte_ranges.size() / total_row_groups; - - // Fetch dictionary page buffers from the input file buffers + // Fetch each source's dictionary page byte ranges from its own datasource, grouping the flat + // byte ranges by the parallel source map std::vector dict_page_buffers; std::vector> dict_page_data; - std::size_t offset = 0; - for (std::size_t src = 0; src < input_row_group_indices.size(); ++src) { - auto const count = input_row_group_indices[src].size() * num_dictionary_columns; - std::vector const src_ranges( - dict_page_byte_ranges.begin() + offset, dict_page_byte_ranges.begin() + offset + count); - offset += count; + std::size_t range_idx = 0; + while (range_idx < dict_page_byte_ranges.size()) { + auto const src = dict_page_source_map[range_idx]; + std::vector src_ranges; + for (; range_idx < dict_page_byte_ranges.size() and dict_page_source_map[range_idx] == src; + ++range_idx) { + src_ranges.emplace_back(dict_page_byte_ranges[range_idx]); + } auto [buffers, data, task] = cudf::io::parquet::fetch_byte_ranges_to_device_async( *inputs.datasources[src], src_ranges, stream, mr); @@ -174,10 +164,8 @@ auto filter_row_groups_with_dictionaries( dict_page_data.insert(dict_page_data.end(), data.begin(), data.end()); } - dict_page_filtered_row_group_indices = reader.filter_row_groups_with_dictionary_pages( + return reader.filter_row_groups_with_dictionary_pages( dict_page_data, input_row_group_indices, options, stream); - - return dict_page_filtered_row_group_indices; } } // namespace @@ -624,7 +612,7 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) srand(0xfeed); file_buffers.emplace_back(std::get<1>(create_parquet_with_stats(200))); // col2 == "0200" - auto inputs = build_multifile_inputs(file_buffers); + auto inputs = multifile_inputs(build_source_info(file_buffers)); // Filter - col2 == "0100" (present only in source A's dictionary) auto literal_value = cudf::string_scalar("0100", true, stream); @@ -679,7 +667,7 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio file_buffers.emplace_back(std::get<1>(create_parquet_with_stats( 100, cudf::io::compression_type::AUTO, {"col2", "col0", "col1"}, {2, 0, 1}))); - auto inputs = build_multifile_inputs(file_buffers); + auto inputs = multifile_inputs(build_source_info(file_buffers)); // Filter - col2 == "0100" auto literal_value = cudf::string_scalar("0100", true, stream); From d2ba16daf4609284a7abbd355f3e6fe6f0109a79 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Mon, 22 Jun 2026 10:31:11 +0200 Subject: [PATCH 07/18] Refactor comments and add validation checks in hybrid scan helpers and tests - Removed redundant comments in `hybrid_scan_helpers.cpp` for clarity. - Added validation checks in `create_parquet_with_stats` to ensure column names and order are consistent. - Updated comments in `hybrid_scan_multifile_filters_test.cpp` for better readability and understanding of filtering logic. --- .../experimental/hybrid_scan_helpers.cpp | 10 +++---- .../io/experimental/hybrid_scan_common.hpp | 20 +++++++++++-- .../hybrid_scan_multifile_filters_test.cpp | 30 ++++--------------- 3 files changed, 27 insertions(+), 33 deletions(-) diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index 53aa8fc3a7ae..adc254da920b 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -476,13 +476,13 @@ aggregate_reader_metadata::dictionary_pages_byte_ranges( std::vector dictionary_page_bytes; dictionary_page_bytes.reserve(num_chunks); + // 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 dictionary_page_source_map; dictionary_page_source_map.reserve(num_chunks); - // Flag to check if we have at least one valid dictionary page - auto have_dictionary_pages = false; - // Cache each dictionary column's chunk offset across sources and row groups std::vector> colchunk_offsets(dictionary_col_schemas.size()); @@ -497,12 +497,12 @@ aggregate_reader_metadata::dictionary_pages_byte_ranges( 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]; auto const num_col_chunks = static_cast(row_group.columns.size()); - // For all dictionary column chunks (kept inner to preserve the source-major emission order) + // For all dictionary column chunks std::for_each( cuda::counting_iterator{0}, cuda::counting_iterator{dictionary_col_schemas.size()}, [&](auto const col) { - // Map the schema index to this source (for `allow_mismatched_pq_schemas`) + // Map the schema index to this source auto const mapped_schema_idx = map_schema_index(dictionary_col_schemas[col], static_cast(src_index)); auto& colchunk_offset = colchunk_offsets[col]; diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index ae670ecc78c7..93b93c0a530b 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -14,12 +14,14 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -165,6 +167,20 @@ auto create_parquet_with_stats( rmm::cuda_stream_view stream = cudf::get_default_stream()) { static_assert(NumTableConcats >= 1, "Concatenated table must contain at least one table"); + CUDF_EXPECTS(column_names.size() == column_order.size(), + "Column names and column order must have the same size"); + CUDF_EXPECTS(column_order.size() == 3, "Column order must include all three test columns"); + CUDF_EXPECTS(std::all_of(column_order.begin(), column_order.end(), [](auto const col_idx) { + return col_idx >= 0 and col_idx < 3; + }), + "Column order contains an out-of-bounds column index"); + CUDF_EXPECTS(std::all_of(cuda::counting_iterator{0}, + cuda::counting_iterator{3}, + [&](auto const col_idx) { + return std::count(column_order.begin(), column_order.end(), col_idx) == + 1; + }), + "Column order must be a permutation of the three test columns"); auto col0 = testdata::ascending(); auto col1 = []() { @@ -217,9 +233,7 @@ auto create_parquet_with_stats( } // Reorder the base [col0, col1, col2] columns into the requested physical order, naming them in - // that new order. Reordering lets callers emit the same logical columns at different schema - // positions across sources (a "mismatched schema"), which exercises the per-source schema-index - // mapping in the row-group filtering paths. The defaults leave the table and names unchanged. + // that new order. std::vector reordered_columns; reordered_columns.reserve(column_order.size()); for (auto const col_idx : column_order) { diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index dfdf2b33795c..bfdfe91001e4 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -136,10 +136,7 @@ auto filter_row_groups_with_dictionaries( rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { - // Get all row groups from the reader auto const input_row_group_indices = reader.all_row_groups(options); - - // Get dictionary page byte ranges and their corresponding source indices from the reader auto const [dict_page_byte_ranges, dict_page_source_map] = reader.dictionary_pages_byte_ranges(input_row_group_indices, options); CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); @@ -767,7 +764,7 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) auto inputs = multifile_inputs(build_source_info(file_buffers)); - // Filter - col2 == "0100" (present only in source A's dictionary) + // Filter: `col2 == "0100"` (present only in source A's dictionary) auto literal_value = cudf::string_scalar("0100", true, stream); auto literal = cudf::ast::literal(literal_value); auto col_ref = cudf::ast::column_name_reference("col2"); @@ -789,19 +786,6 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) EXPECT_TRUE(dict_filtered.back().empty()); } -// Mismatched-schema regression for the dictionary-page pruning path under -// `allow_mismatched_pq_schemas`. `get_dictionary_page_bytes` used to resolve column chunks by the -// raw zeroth-source `schema_idx` (without `map_schema_index`), so a reordered source read the wrong -// column chunk. -// -// Both sources hold the same three `create_parquet_with_stats` columns (col0: ascending, col1: -// low-cardinality descending, col2: constant string), but source B emits them physically reordered -// as {col2, col0, col1}. So the predicate column `col2` is schema index 3 in source A (the zeroth -// source) yet schema index 1 in source B. The buggy raw-index lookup reads source B's schema index -// 3 -- `col1`, a duration column -- instead of its `col2` string dictionary, garbling/throwing on -// the type mismatch. With the fix, `col2` maps per source: source A (col2 == "0200") is pruned and -// source B (col2 == "0100") survives. A chrono `T` is used so the collided `col1` is itself -// dictionary-encoded, and hence actually fetched by the buggy path. TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollision) { using T = cudf::duration_ms; @@ -810,8 +794,7 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio auto mr = cudf::get_current_device_resource_ref(); // Source A: default column order/names, col2 == "0200" (pruned by the filter). - // Source B: same columns emitted as {col2, col0, col1}, col2 == "0100" (survives). The reorder - // moves `col2` to a different schema index than in source A, which is what triggers the bug. + // Source B: same columns emitted as {col2, col0, col1}, col2 == "0100" (survives). std::vector> file_buffers; file_buffers.reserve(num_sources); srand(0xd1c7); @@ -822,7 +805,7 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio auto inputs = multifile_inputs(build_source_info(file_buffers)); - // Filter - col2 == "0100" + // Filter: `col2 == "0100"` auto literal_value = cudf::string_scalar("0100", true, stream); auto literal = cudf::ast::literal(literal_value); auto col_ref = cudf::ast::column_name_reference("col2"); @@ -837,9 +820,7 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio auto const reader = std::make_unique( cudf::host_span const>{inputs.footer_byte_spans}, options); - // Guard the premise of the test: the reorder must genuinely differ the per-source schemas - // (source A's first column is `col0`, source B's first column is `col2`), otherwise there is no - // schema-index mismatch for the dictionary pruning path to get wrong. + // Ensure the reorder genuinely differs the per-source schemas auto const metadatas = reader->parquet_metadatas(); ASSERT_EQ(metadatas.size(), num_sources); EXPECT_EQ(metadatas.front().schema.at(1).name, "col0"); @@ -851,8 +832,7 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio auto const dict_filtered = filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); - // Correct behavior: source A is pruned (col2 == "0200"), source B survives (col2 == "0100"). The - // buggy raw-index path instead reads source B's `col1` duration dictionary for `col2`. + // Source A is pruned (col2 == "0200"), source B survives (col2 == "0100"). ASSERT_EQ(dict_filtered.size(), num_sources); EXPECT_TRUE(dict_filtered.front().empty()) << "Source A should be pruned (col2 == \"0200\")"; EXPECT_EQ(dict_filtered.back(), (std::vector{0, 1, 2, 3})) From 94251cd2acc6eef4893f477021835cabc2592aba Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Mon, 22 Jun 2026 16:20:38 +0200 Subject: [PATCH 08/18] Update copyright notices to include "AFFILIATES" in multiple files - Modified copyright statements in various source and header files to reflect the inclusion of "NVIDIA CORPORATION & AFFILIATES." - Updated files include `hybrid_scan_multifile.hpp`, `hybrid_scan_helpers.cpp`, `hybrid_scan_helpers.hpp`, `hybrid_scan_impl.cpp`, `hybrid_scan_impl.hpp`, `page_index_filter_utils.cu`, and test files related to hybrid scans. --- .../cudf/io/experimental/hybrid_scan_multifile.hpp | 2 +- cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp | 2 +- cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp | 2 +- cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp | 2 +- cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp | 2 +- cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp | 2 +- .../io/parquet/experimental/page_index_filter_utils.cu | 2 +- cpp/tests/io/experimental/hybrid_scan_common.hpp | 8 ++++---- .../experimental/hybrid_scan_multifile_filters_test.cpp | 2 +- 9 files changed, 12 insertions(+), 12 deletions(-) diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp index 09ac9515e207..f3e129c59d13 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index adc254da920b..f407197cfa72 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -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 */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp index 5c494060bfb2..e03e8bf9da3f 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp @@ -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 */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index ef9ee4a18ea0..885ef10683e5 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -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 */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index 0cc2a0f9d88d..7c7158a0e26a 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -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 */ diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp index edad1cd53494..580c813abe06 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_multifile.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ diff --git a/cpp/src/io/parquet/experimental/page_index_filter_utils.cu b/cpp/src/io/parquet/experimental/page_index_filter_utils.cu index 2eef122e840f..fcd6dae68a99 100644 --- a/cpp/src/io/parquet/experimental/page_index_filter_utils.cu +++ b/cpp/src/io/parquet/experimental/page_index_filter_utils.cu @@ -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 */ diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index 93b93c0a530b..cc34d99ab677 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -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 */ @@ -170,9 +170,9 @@ auto create_parquet_with_stats( CUDF_EXPECTS(column_names.size() == column_order.size(), "Column names and column order must have the same size"); CUDF_EXPECTS(column_order.size() == 3, "Column order must include all three test columns"); - CUDF_EXPECTS(std::all_of(column_order.begin(), column_order.end(), [](auto const col_idx) { - return col_idx >= 0 and col_idx < 3; - }), + CUDF_EXPECTS(std::all_of(column_order.begin(), + column_order.end(), + [](auto const col_idx) { return col_idx >= 0 and col_idx < 3; }), "Column order contains an out-of-bounds column index"); CUDF_EXPECTS(std::all_of(cuda::counting_iterator{0}, cuda::counting_iterator{3}, diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index bfdfe91001e4..813d7e339602 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ From a163660f4742ec1f2682f573fd23485b5143b7fe Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Tue, 23 Jun 2026 14:54:41 +0200 Subject: [PATCH 09/18] Remove out-of-bounds check from `create_parquet_with_stats` and update comment in `setup_multifile_page_index` for clarity --- cpp/tests/io/experimental/hybrid_scan_common.hpp | 4 ---- .../io/experimental/hybrid_scan_multifile_filters_test.cpp | 2 +- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index cc34d99ab677..948c1a4bc54b 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -170,10 +170,6 @@ auto create_parquet_with_stats( CUDF_EXPECTS(column_names.size() == column_order.size(), "Column names and column order must have the same size"); CUDF_EXPECTS(column_order.size() == 3, "Column order must include all three test columns"); - CUDF_EXPECTS(std::all_of(column_order.begin(), - column_order.end(), - [](auto const col_idx) { return col_idx >= 0 and col_idx < 3; }), - "Column order contains an out-of-bounds column index"); CUDF_EXPECTS(std::all_of(cuda::counting_iterator{0}, cuda::counting_iterator{3}, [&](auto const col_idx) { diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index 813d7e339602..fc3c017a0d9d 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -87,7 +87,7 @@ auto make_scalar(cudf::size_type value, rmm::cuda_stream_view stream) } /** - * @brief Fetches and sets up the per-source page index on the reader. + * @brief Fetches and sets up the per-source page index on the reader */ void setup_multifile_page_index( cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, multifile_inputs& inputs) From 4a34b6c621f0a7764b2d40449d9abf8329856ece Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Wed, 24 Jun 2026 22:43:24 +0200 Subject: [PATCH 10/18] Update copyright notice and refactor hybrid scan multifile tests - Updated copyright notice in header files to include "AFFILIATES". - Introduced a new function `group_byte_ranges_by_source` to group byte ranges by their source indices. - Refactored existing tests to utilize the new grouping function, improving code clarity and maintainability. - Removed redundant setup function for page indexes, streamlining the test setup process. --- .../hybrid_scan_multifile_common.hpp | 36 ++++++++- .../hybrid_scan_multifile_filters_test.cpp | 76 +++++-------------- .../hybrid_scan_multifile_test.cpp | 29 +------ 3 files changed, 56 insertions(+), 85 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_common.hpp b/cpp/tests/io/experimental/hybrid_scan_multifile_common.hpp index 2d033539324e..af55f34a5fec 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_common.hpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -8,11 +8,16 @@ #include #include #include +#include +#include +#include #include #include +#include #include #include +#include #include /** @@ -77,3 +82,32 @@ inline void setup_page_indexes(cudf::io::parquet::experimental::hybrid_scan_mult reader.setup_page_indexes( cudf::host_span const>{page_index_byte_spans}); } + +/** + * @brief Group flattened byte ranges by their corresponding source indices + * + * @param byte_ranges_and_source_map Pair of flattened byte ranges and their parallel source map + * @param num_sources Number of sources + * @return Vector of byte ranges, one inner vector per source + */ +inline std::vector> group_byte_ranges_by_source( + std::pair, std::vector> const& + byte_ranges_and_source_map, + std::size_t num_sources) +{ + auto const& [byte_ranges, source_map] = byte_ranges_and_source_map; + CUDF_EXPECTS(byte_ranges.size() == source_map.size(), "Invalid source map size"); + + auto byte_ranges_per_source = + std::vector>(num_sources); + std::for_each(byte_ranges.begin(), + byte_ranges.end(), + [&, range_index = std::size_t{0}](auto const& range) mutable { + auto const source_index = source_map[range_index++]; + CUDF_EXPECTS(source_index >= 0 and static_cast(source_index) < + byte_ranges_per_source.size(), + "Invalid byte range source index"); + byte_ranges_per_source[source_index].push_back(range); + }); + return byte_ranges_per_source; +} diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index fc3c017a0d9d..fffbd774b91a 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -27,7 +27,6 @@ #include #include -#include #include #include #include @@ -86,37 +85,6 @@ auto make_scalar(cudf::size_type value, rmm::cuda_stream_view stream) } } -/** - * @brief Fetches and sets up the per-source page index on the reader - */ -void setup_multifile_page_index( - cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, multifile_inputs& inputs) -{ - // Reference wrappers to the datasources, in source order - std::vector> datasource_refs; - datasource_refs.reserve(inputs.datasources.size()); - std::transform(inputs.datasources.begin(), - inputs.datasources.end(), - std::back_inserter(datasource_refs), - [](auto& datasource) { return std::ref(*datasource); }); - - // Fetch all per-source page index buffers in one batch - auto const page_index_byte_ranges = reader.page_index_byte_ranges(); - auto const page_index_buffers = - cudf::io::parquet::fetch_page_indexes_to_host(datasource_refs, page_index_byte_ranges); - - // Set up the page index on the reader from the fetched buffers - std::vector> page_index_byte_spans; - page_index_byte_spans.reserve(page_index_buffers.size()); - std::transform(page_index_buffers.begin(), - page_index_buffers.end(), - std::back_inserter(page_index_byte_spans), - [](auto const& buffer) { return cudf::host_span{*buffer}; }); - - reader.setup_page_indexes( - cudf::host_span const>{page_index_byte_spans}); -} - /** * @brief Filter input row groups using column chunk dictionaries via the experimental parquet * reader for hybrid scan (multi-file) @@ -130,37 +98,31 @@ void setup_multifile_page_index( * @return Vector of per-source dictionary-filtered row group indices */ auto filter_row_groups_with_dictionaries( - multifile_inputs& inputs, + multifile_inputs const& inputs, cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, cudf::io::parquet_reader_options const& options, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { auto const input_row_group_indices = reader.all_row_groups(options); - auto const [dict_page_byte_ranges, dict_page_source_map] = - reader.dictionary_pages_byte_ranges(input_row_group_indices, options); - CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); - // Fetch each source's dictionary page byte ranges from its own datasource, grouping the flat - // byte ranges by the parallel source map - std::vector dict_page_buffers; - std::vector> dict_page_data; - std::size_t range_idx = 0; - while (range_idx < dict_page_byte_ranges.size()) { - auto const src = dict_page_source_map[range_idx]; - std::vector src_ranges; - for (; range_idx < dict_page_byte_ranges.size() and dict_page_source_map[range_idx] == src; - ++range_idx) { - src_ranges.emplace_back(dict_page_byte_ranges[range_idx]); - } + auto const dict_pages = reader.dictionary_pages_byte_ranges(input_row_group_indices, options); + CUDF_EXPECTS(dict_pages.first.size() > 0, "No dictionary page byte ranges found"); - auto [buffers, data, task] = cudf::io::parquet::fetch_byte_ranges_to_device_async( - *inputs.datasources[src], src_ranges, stream, mr); - task.get(); - for (auto& buffer : buffers) { - dict_page_buffers.emplace_back(std::move(buffer)); - } - dict_page_data.insert(dict_page_data.end(), data.begin(), data.end()); + auto const dict_page_ranges_per_source = + group_byte_ranges_by_source(dict_pages, inputs.datasources.size()); + auto [dict_page_buffers, dict_page_data_per_source, task] = + cudf::io::parquet::fetch_byte_ranges_to_device_async( + inputs.datasource_refs, + cudf::host_span const>{ + dict_page_ranges_per_source}, + stream, + mr); + task.get(); + + std::vector> dict_page_data; + for (auto const& source_dict_pages : dict_page_data_per_source) { + dict_page_data.insert(dict_page_data.end(), source_dict_pages.begin(), source_dict_pages.end()); } return reader.filter_row_groups_with_dictionary_pages( @@ -775,7 +737,7 @@ TEST_F(HybridScanMultifileFiltersTest, FilterRowGroupsWithDictionaryPages) inputs.footer_byte_spans, options); // Page index is needed to detect dictionary-only encoded pages - setup_multifile_page_index(*reader, inputs); + setup_page_indexes(*reader, inputs); auto const dict_filtered = filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); @@ -827,7 +789,7 @@ TEST_F(HybridScanMultifileFiltersTest, MismatchedSchemaDictionaryPruningCollisio EXPECT_EQ(metadatas.back().schema.at(1).name, "col2"); // Page index is needed to detect dictionary-only encoded pages - setup_multifile_page_index(*reader, inputs); + setup_page_indexes(*reader, inputs); auto const dict_filtered = filter_row_groups_with_dictionaries(inputs, *reader, options, stream, mr); diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp index b3c927f320a0..041ca74b8fcf 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_test.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -32,31 +32,6 @@ namespace { -/** - * @brief Group flattened column chunk byte ranges by their corresponding sources - */ -std::vector> column_chunks_byte_ranges_per_source( - std::pair, std::vector> const& - byte_ranges_and_source_map, - std::size_t num_sources) -{ - auto const& [byte_ranges, source_map] = byte_ranges_and_source_map; - CUDF_EXPECTS(byte_ranges.size() == source_map.size(), "Invalid source map size"); - - auto byte_ranges_per_source = - std::vector>(num_sources); - std::for_each(byte_ranges.begin(), - byte_ranges.end(), - [&, range_index = std::size_t{0}](auto const& range) mutable { - auto const source_index = source_map[range_index++]; - CUDF_EXPECTS(source_index >= 0 and static_cast(source_index) < - byte_ranges_per_source.size(), - "Invalid byte range source index"); - byte_ranges_per_source[source_index].push_back(range); - }); - return byte_ranges_per_source; -} - /** * @brief Helper to test multifile hybrid scan single-shot materialization * @@ -111,7 +86,7 @@ void test_hybrid_scan_multifile(std::vector const& columns, auto reader = cudf::io::parquet::experimental::hybrid_scan_multifile{inputs.footer_byte_spans, options}; auto const row_groups = reader.all_row_groups(options); - auto const byte_ranges_per_source = column_chunks_byte_ranges_per_source( + auto const byte_ranges_per_source = group_byte_ranges_by_source( reader.all_column_chunks_byte_ranges(row_groups, options), inputs.datasources.size()); auto [column_chunk_buffers, column_chunks_per_source, tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async( From eb2dbd4d286624a7bbe41c1459ed22984f04db78 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Mon, 29 Jun 2026 09:39:16 +0200 Subject: [PATCH 11/18] Refactor column chunk offset retrieval in parquet reader This commit introduces a new helper function, `find_colchunk_iter_offset`, to streamline the process of finding the offset of a column chunk by its schema index within a row group. The function replaces direct usage of `std::find_if` in multiple locations, enhancing code readability and maintainability. Additionally, unnecessary includes have been removed to clean up the codebase. --- .../experimental/hybrid_scan_helpers.cpp | 2 +- .../experimental/page_index_filter_utils.cu | 13 +---------- .../experimental/page_index_filter_utils.hpp | 11 +--------- cpp/src/io/parquet/predicate_pushdown.cpp | 13 +---------- cpp/src/io/parquet/reader_impl_helpers.cpp | 22 ++++++++++++------- cpp/src/io/parquet/reader_impl_helpers.hpp | 9 ++++++++ 6 files changed, 27 insertions(+), 43 deletions(-) diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index f407197cfa72..ebd3b66a819c 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -9,7 +9,6 @@ #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 #include @@ -32,6 +31,7 @@ using metadata_base = parquet::detail::metadata; using io::detail::inline_column_buffer; using parquet::detail::CompactProtocolReader; using parquet::detail::equality_literals_collector; +using parquet::detail::find_colchunk_iter_offset; using parquet::detail::input_column_info; using parquet::detail::row_group_info; using text::byte_range_info; diff --git a/cpp/src/io/parquet/experimental/page_index_filter_utils.cu b/cpp/src/io/parquet/experimental/page_index_filter_utils.cu index fcd6dae68a99..892e5dfecf64 100644 --- a/cpp/src/io/parquet/experimental/page_index_filter_utils.cu +++ b/cpp/src/io/parquet/experimental/page_index_filter_utils.cu @@ -18,23 +18,12 @@ #include #include -#include #include #include namespace cudf::io::parquet::experimental::detail { -size_type find_colchunk_iter_offset(RowGroup const& row_group, size_type 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(), - std::format("Column chunk with schema index {} not found in row group", schema_idx), - std::invalid_argument); - return std::distance(row_group.columns.begin(), colchunk_iter); -} +using parquet::detail::find_colchunk_iter_offset; bool compute_has_page_index(cudf::host_span file_metadatas, cudf::host_span const> row_group_indices) diff --git a/cpp/src/io/parquet/experimental/page_index_filter_utils.hpp b/cpp/src/io/parquet/experimental/page_index_filter_utils.hpp index 5a8335777d6d..9b41f1ae02c7 100644 --- a/cpp/src/io/parquet/experimental/page_index_filter_utils.hpp +++ b/cpp/src/io/parquet/experimental/page_index_filter_utils.hpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2025, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -20,15 +20,6 @@ namespace cudf::io::parquet::experimental::detail { using metadata_base = parquet::detail::metadata; -/** - * @brief Find the offset of the column chunk with the given schema index in the row group - * - * @param row_group Row group - * @param schema_idx Schema index - * @return Offset of the column chunk iterator - */ -[[nodiscard]] size_type find_colchunk_iter_offset(RowGroup const& row_group, size_type schema_idx); - /** * @brief Compute if the page index is present in all parquet data sources for all columns * diff --git a/cpp/src/io/parquet/predicate_pushdown.cpp b/cpp/src/io/parquet/predicate_pushdown.cpp index 9aab5ae708cd..35b4b407e2ad 100644 --- a/cpp/src/io/parquet/predicate_pushdown.cpp +++ b/cpp/src/io/parquet/predicate_pushdown.cpp @@ -23,7 +23,6 @@ #include #include -#include #include #include #include @@ -155,17 +154,7 @@ bool aggregate_reader_metadata::any_row_group_stats_available( if (cached_offset < 0 or cached_offset >= num_col_chunks or first_row_group.columns[cached_offset].schema_idx != mapped_schema_idx) { - auto const it = std::find_if( - first_row_group.columns.begin(), - first_row_group.columns.end(), - [mapped_schema_idx](ColumnChunk const& c) { return c.schema_idx == mapped_schema_idx; }); - CUDF_EXPECTS( - it != first_row_group.columns.end(), - std::format( - "Column chunk with schema index {} not found in source {}", mapped_schema_idx, src_idx), - std::invalid_argument); - colchunk_offset = - static_cast(std::distance(first_row_group.columns.begin(), it)); + colchunk_offset = find_colchunk_iter_offset(first_row_group, mapped_schema_idx); } if (colchunk_has_stats(first_row_group.columns[colchunk_offset.value()])) { return true; } diff --git a/cpp/src/io/parquet/reader_impl_helpers.cpp b/cpp/src/io/parquet/reader_impl_helpers.cpp index 667449fdb9ae..dac5b8b8fffe 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.cpp +++ b/cpp/src/io/parquet/reader_impl_helpers.cpp @@ -56,6 +56,18 @@ std::size_t derive_pass_read_limit(std::size_t chunk_read_limit) return pass_read_limit; } +size_type find_colchunk_iter_offset(RowGroup const& row_group, size_type 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(), + std::format("Column chunk with schema index {} not found in row group", schema_idx), + std::invalid_argument); + return std::distance(row_group.columns.begin(), colchunk_iter); +} + namespace flatbuf = cudf::io::parquet::flatbuf; namespace { @@ -1086,14 +1098,8 @@ ColumnChunkMetaData const& aggregate_reader_metadata::get_column_metadata(size_t // Map schema index to the provided source file index schema_idx = map_schema_index(schema_idx, src_idx); - auto col = - std::find_if(per_file_metadata[src_idx].row_groups[row_group_index].columns.begin(), - per_file_metadata[src_idx].row_groups[row_group_index].columns.end(), - [schema_idx](ColumnChunk const& col) { return col.schema_idx == schema_idx; }); - CUDF_EXPECTS(col != std::end(per_file_metadata[src_idx].row_groups[row_group_index].columns), - "Found no metadata for schema index", - std::range_error); - return col->meta_data; + auto const& row_group = per_file_metadata[src_idx].row_groups[row_group_index]; + return row_group.columns[find_colchunk_iter_offset(row_group, schema_idx)].meta_data; } std::vector> diff --git a/cpp/src/io/parquet/reader_impl_helpers.hpp b/cpp/src/io/parquet/reader_impl_helpers.hpp index 9417416c1e57..67448c45a24c 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.hpp +++ b/cpp/src/io/parquet/reader_impl_helpers.hpp @@ -107,6 +107,15 @@ struct row_group_info { */ [[nodiscard]] std::size_t derive_pass_read_limit(std::size_t chunk_read_limit); +/** + * @brief Find the offset of the column chunk with the given schema index in the row group + * + * @param row_group Row group + * @param schema_idx Schema index + * @return Offset of the column chunk within the row group's columns + */ +[[nodiscard]] size_type find_colchunk_iter_offset(RowGroup const& row_group, size_type schema_idx); + /** * @brief Class for parsing dataset metadata */ From e6009c1a7cc2e2fb1b6f82a02baa64d39627789a Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Tue, 30 Jun 2026 23:10:29 +0200 Subject: [PATCH 12/18] Refactor hybrid scan reader implementation - Removed the unused `dictionary_pages_byte_ranges` method from `hybrid_scan_reader_impl`. - Updated `create_parquet_with_stats` to accept `column_names` and `column_order` as parameters, ensuring they match in size and contain all required columns. - Added checks to validate the `column_order` as a permutation of the specified columns. - Enhanced the output metadata to set column names based on the provided `column_names` vector. --- .../parquet/experimental/hybrid_scan_impl.hpp | 7 -- .../io/experimental/hybrid_scan_common.cpp | 33 +++++- .../io/experimental/hybrid_scan_common.hpp | 102 +----------------- 3 files changed, 31 insertions(+), 111 deletions(-) diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index 9a29894b5bf0..4eef0c7363aa 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -114,13 +114,6 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { secondary_filters_byte_ranges(cudf::host_span const> row_group_indices, parquet_reader_options const& options); - /** - * @copydoc cudf::io::experimental::hybrid_scan_multifile::dictionary_pages_byte_ranges - */ - [[nodiscard]] std::pair, std::vector> - dictionary_pages_byte_ranges(cudf::host_span const> row_group_indices, - parquet_reader_options const& options); - /** * @copydoc cudf::io::parquet::experimental::hybrid_scan_multifile::dictionary_pages_byte_ranges */ diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index 99ee37e6c333..f2bf8a4b3246 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -232,9 +232,21 @@ template , std::vector> create_parquet_with_stats( cudf::size_type str_col_value, cudf::io::compression_type compression, + std::vector column_names, + std::vector column_order, rmm::cuda_stream_view stream) { static_assert(NumTableConcats >= 1, "Concatenated table must contain at least one table"); + CUDF_EXPECTS(column_names.size() == column_order.size(), + "Column names and column order must have the same size"); + CUDF_EXPECTS(column_order.size() == 3, "Column order must include all three test columns"); + CUDF_EXPECTS(std::all_of(cuda::counting_iterator{0}, + cuda::counting_iterator{3}, + [&](auto const col_idx) { + return std::count(column_order.begin(), column_order.end(), col_idx) == + 1; + }), + "Column order must be a permutation of the three test columns"); auto col0 = testdata::ascending(); auto col1 = []() { @@ -291,12 +303,21 @@ std::pair, std::vector> create_parquet_with_s output = table_view{{columns[0]->view(), columns[1]->view(), columns[2]->view()}}; } + // Reorder the base [col0, col1, col2] columns into the requested physical order, naming them in + // that new order. + std::vector reordered_columns; + reordered_columns.reserve(column_order.size()); + for (auto const col_idx : column_order) { + reordered_columns.emplace_back(output.column(col_idx)); + } + output = table_view{reordered_columns}; + auto table = cudf::concatenate(std::vector(NumTableConcats, output), stream); output = table->view(); cudf::io::table_input_metadata output_metadata(output); - output_metadata.column_metadata[0].set_name("col0"); - output_metadata.column_metadata[1].set_name("col1"); - output_metadata.column_metadata[2].set_name("col2"); + for (std::size_t i = 0; i < column_names.size(); ++i) { + output_metadata.column_metadata[i].set_name(column_names[i]); + } std::vector buffer; cudf::io::parquet_writer_options out_opts = @@ -321,7 +342,11 @@ std::pair, std::vector> create_parquet_with_s #define INSTANTIATE_CREATE_PARQUET_WITH_STATS(T, NUM_CONCATS, CONSTANT_STRINGS, NULLABLE) \ template std::pair, std::vector> \ create_parquet_with_stats( \ - cudf::size_type, cudf::io::compression_type, rmm::cuda_stream_view) + cudf::size_type, \ + cudf::io::compression_type, \ + std::vector, \ + std::vector, \ + rmm::cuda_stream_view) #define INSTANTIATE_CREATE_PARQUET_WITH_STATS_DICT(T) \ INSTANTIATE_CREATE_PARQUET_WITH_STATS(T, 1, true, false); \ diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index 326d2600c517..99e80a0d6969 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -20,9 +20,9 @@ #include #include +#include #include #include -#include #include #include #include @@ -146,102 +146,4 @@ template column_names = {"col0", "col1", "col2"}, std::vector column_order = {0, 1, 2}, - rmm::cuda_stream_view stream = cudf::get_default_stream()) -{ - static_assert(NumTableConcats >= 1, "Concatenated table must contain at least one table"); - CUDF_EXPECTS(column_names.size() == column_order.size(), - "Column names and column order must have the same size"); - CUDF_EXPECTS(column_order.size() == 3, "Column order must include all three test columns"); - CUDF_EXPECTS(std::all_of(cuda::counting_iterator{0}, - cuda::counting_iterator{3}, - [&](auto const col_idx) { - return std::count(column_order.begin(), column_order.end(), col_idx) == - 1; - }), - "Column order must be a permutation of the three test columns"); - - auto col0 = testdata::ascending(); - auto col1 = []() { - if constexpr (cudf::is_chrono()) { - return descending_low_cardinality(); - } else { - return testdata::descending(); - } - }(); - - auto col2 = [&]() { - if constexpr (IsConstantStrings) { - return constant_strings(str_col_value); // constant stringified value - } else { - return testdata::ascending(); // ascending strings - } - }(); - - // Output table view - auto output = table_view{{col0, col1, col2}}; - - // Add nullmasks to the columns if specified - std::vector> columns; - if constexpr (IsNullable) { - std::mt19937 gen(0xc0ffee); - std::bernoulli_distribution bn(0.7f); - auto valids = - cudf::detail::make_counting_transform_iterator(0, [&](int index) { return bn(gen); }); - auto const num_rows = static_cast(col0).size(); - - columns.emplace_back(col0.release()); - auto [nullmask, nullcount] = cudf::test::detail::make_null_mask(valids, valids + num_rows); - columns.back()->set_null_mask(std::move(nullmask), nullcount); - - columns.emplace_back(col1.release()); - std::tie(nullmask, nullcount) = - cudf::test::detail::make_null_mask(valids + num_rows, valids + 2 * num_rows); - columns.back()->set_null_mask(std::move(nullmask), nullcount); - - columns.emplace_back(col2.release()); - std::tie(nullmask, nullcount) = - cudf::test::detail::make_null_mask(valids + 2 * num_rows, valids + 3 * num_rows); - columns.back()->set_null_mask(std::move(nullmask), nullcount); - - // Purge non-empty nulls from the strings column only - cudf::purge_nonempty_nulls(columns.back()->view()); - - // Update the output table view with the nullable columns - output = table_view{{columns[0]->view(), columns[1]->view(), columns[2]->view()}}; - } - - // Reorder the base [col0, col1, col2] columns into the requested physical order, naming them in - // that new order. - std::vector reordered_columns; - reordered_columns.reserve(column_order.size()); - for (auto const col_idx : column_order) { - reordered_columns.emplace_back(output.column(col_idx)); - } - output = table_view{reordered_columns}; - - auto table = cudf::concatenate(std::vector(NumTableConcats, output)); - output = table->view(); - cudf::io::table_input_metadata output_metadata(output); - for (std::size_t i = 0; i < column_names.size(); ++i) { - output_metadata.column_metadata[i].set_name(column_names[i]); - } - - std::vector buffer; - cudf::io::parquet_writer_options out_opts = - cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&buffer}, output) - .metadata(std::move(output_metadata)) - .row_group_size_rows(page_size_for_ordered_tests) - .max_page_size_rows(page_size_for_ordered_tests / 5) - .compression(compression) - .dictionary_policy(cudf::io::dictionary_policy::ALWAYS) - .stats_level(cudf::io::statistics_freq::STATISTICS_COLUMN); - - if constexpr (NumTableConcats > 1) { - out_opts.set_row_group_size_rows(num_ordered_rows); - out_opts.set_max_page_size_rows(page_size_for_ordered_tests); - } - - cudf::io::write_parquet(out_opts); - - return std::pair{std::move(table), std::move(buffer)}; -} + rmm::cuda_stream_view stream = cudf::get_default_stream()); From da507b5eaa05117a3b131bc10e7a38da62ae393a Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Thu, 2 Jul 2026 11:14:11 +0200 Subject: [PATCH 13/18] Add hybrid scan read-amplification benchmark to nvbench This commit introduces a new benchmark for measuring read-amplification in the hybrid scan implementation. The benchmark is configured in the CMakeLists.txt file and points to the relevant source file for the hybrid scan amplification tests. --- .../io/experimental/hybrid_scan_multifile.hpp | 6 +- .../parquet/experimental/hybrid_scan_impl.hpp | 2 +- cpp/src/io/parquet/reader_impl_helpers.hpp | 5 +- .../io/experimental/hybrid_scan_common.cpp | 108 ++++++++++++++++-- .../io/experimental/hybrid_scan_common.hpp | 26 +++++ .../experimental/hybrid_scan_filters_test.cpp | 57 --------- .../hybrid_scan_multifile_filters_test.cpp | 44 ------- 7 files changed, 132 insertions(+), 116 deletions(-) diff --git a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp index 8e865cd30c75..3bdf5d53f12f 100644 --- a/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp +++ b/cpp/include/cudf/io/experimental/hybrid_scan_multifile.hpp @@ -453,7 +453,7 @@ class hybrid_scan_multifile { /** * @brief Get byte ranges of column chunk dictionary pages for row group pruning * - * @param row_group_indices Input row group indices, one per source + * @param row_group_indices Span of vectors of input row group indices, one per source * @param options Parquet reader options * @return Pair of flattened byte ranges to column chunk dictionary pages subject to the filter * predicate and their corresponding source indices @@ -468,10 +468,10 @@ class hybrid_scan_multifile { * @param dictionary_page_data Device spans of dictionary page data of column chunks with an * (in)equality predicate, ordered to match the dictionary page byte * ranges returned by `dictionary_pages_byte_ranges` - * @param row_group_indices Input row group indices, one per source + * @param row_group_indices Span of vectors of input row group indices, one per source * @param options Parquet reader options * @param stream CUDA stream used for device memory operations and kernel launches - * @return Filtered per-source row group indices (one inner vector per source) + * @return Vector of vectors of filtered row group indices, one per source */ [[nodiscard]] std::vector> filter_row_groups_with_dictionary_pages( cudf::host_span const> dictionary_page_data, diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp index c39d72708668..9a3d2501dd14 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.hpp @@ -290,7 +290,7 @@ class hybrid_scan_reader_impl : public parquet::detail::reader_impl { * @throws std::invalid_argument if @p row_group_indices.size() is all empty or not equal to the * number of input datasources * - * @param row_group_indices Input row group indices, one per source + * @param row_group_indices Span of vectors of input row group indices, one per source * @param total_row_groups Total number of row groups across all sources * @param pass_read_limit Memory limit to read and decompress row * group data diff --git a/cpp/src/io/parquet/reader_impl_helpers.hpp b/cpp/src/io/parquet/reader_impl_helpers.hpp index 67448c45a24c..c26fc7c07617 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.hpp +++ b/cpp/src/io/parquet/reader_impl_helpers.hpp @@ -110,8 +110,11 @@ struct row_group_info { /** * @brief Find the offset of the column chunk with the given schema index in the row group * + * @note For mismatched schemas, the caller must pre-map `schema_idx` to the row group's source via + * `aggregate_reader_metadata::map_schema_index`; this function does not remap it. + * * @param row_group Row group - * @param schema_idx Schema index + * @param schema_idx Schema index, already mapped to the row group's source * @return Offset of the column chunk within the row group's columns */ [[nodiscard]] size_type find_colchunk_iter_offset(RowGroup const& row_group, size_type schema_idx); diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index f2bf8a4b3246..3e3cca0e55e3 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -27,6 +27,7 @@ #include #include #include +#include namespace { @@ -228,6 +229,86 @@ std::unique_ptr concatenate_tables(std::vector +auto filter_row_groups_with_dictionaries_impl(InputType& inputs, + ReaderType const& reader, + cudf::io::parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) +{ + // Reset column selection so the helper is safe to call repeatedly on the same reader; the + // single-file tests reuse one reader across many filter expressions. + reader.reset_column_selection(); + auto const row_group_indices = reader.all_row_groups(options); + + if constexpr (std::is_same_v) { + // Dictionary page byte ranges carry a per-range source map used to regroup them by source. + auto const dict_pages = reader.dictionary_pages_byte_ranges(row_group_indices, options); + CUDF_EXPECTS(dict_pages.first.size() > 0, "No dictionary page byte ranges found"); + + auto const dict_page_ranges_per_source = + group_byte_ranges_by_source(dict_pages, inputs.datasources.size()); + auto [dict_page_buffers, dict_page_data_per_source, task] = + cudf::io::parquet::fetch_byte_ranges_to_device_async( + inputs.datasource_refs, + cudf::host_span const>{ + dict_page_ranges_per_source}, + stream, + mr); + task.get(); + + std::vector> dict_page_data; + for (auto const& source_dict_pages : dict_page_data_per_source) { + dict_page_data.insert( + dict_page_data.end(), source_dict_pages.begin(), source_dict_pages.end()); + } + return reader.filter_row_groups_with_dictionary_pages( + dict_page_data, row_group_indices, options, stream); + } else { + // `secondary_filters_byte_ranges().second` is the single-file equivalent of the multi-file + // `dictionary_pages_byte_ranges()`. + auto const current = cudf::host_span{row_group_indices}; + auto const dict_page_byte_ranges = + reader.secondary_filters_byte_ranges(current, options).second; + CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); + + auto [dict_page_buffers, dict_page_data, dict_page_tasks] = + cudf::io::parquet::fetch_byte_ranges_to_device_async( + inputs, dict_page_byte_ranges, stream, mr); + dict_page_tasks.get(); + return reader.filter_row_groups_with_dictionary_pages(dict_page_data, current, options, stream); + } +} + +} // namespace + +std::vector filter_row_groups_with_dictionaries( + cudf::io::datasource& datasource, + cudf::io::parquet::experimental::hybrid_scan_reader const& reader, + cudf::ast::operation const& filter_expression, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) +{ + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + return filter_row_groups_with_dictionaries_impl(datasource, reader, options, stream, mr); +} + +std::vector> filter_row_groups_with_dictionaries( + multifile_inputs const& inputs, + cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, + cudf::io::parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) +{ + return filter_row_groups_with_dictionaries_impl(inputs, reader, options, stream, mr); +} + template std::pair, std::vector> create_parquet_with_stats( cudf::size_type str_col_value, @@ -237,16 +318,23 @@ std::pair, std::vector> create_parquet_with_s rmm::cuda_stream_view stream) { static_assert(NumTableConcats >= 1, "Concatenated table must contain at least one table"); - CUDF_EXPECTS(column_names.size() == column_order.size(), - "Column names and column order must have the same size"); - CUDF_EXPECTS(column_order.size() == 3, "Column order must include all three test columns"); - CUDF_EXPECTS(std::all_of(cuda::counting_iterator{0}, - cuda::counting_iterator{3}, - [&](auto const col_idx) { - return std::count(column_order.begin(), column_order.end(), col_idx) == - 1; - }), - "Column order must be a permutation of the three test columns"); + + // Common default-layout callers skip these checks. + static auto const default_column_names = std::vector{"col0", "col1", "col2"}; + static auto const default_column_order = std::vector{0, 1, 2}; + if (column_names != default_column_names or column_order != default_column_order) { + CUDF_EXPECTS(column_names.size() == column_order.size(), + "Column names and column order must have the same size"); + CUDF_EXPECTS(column_order.size() == default_column_order.size(), + "Column order must include all three test columns"); + CUDF_EXPECTS(std::all_of(default_column_order.begin(), + default_column_order.end(), + [&](auto const col_idx) { + return std::count( + column_order.begin(), column_order.end(), col_idx) == 1; + }), + "Column order must be a permutation of the three test columns"); + } auto col0 = testdata::ascending(); auto col1 = []() { diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index 99e80a0d6969..ba64b69234d3 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -5,9 +5,11 @@ #pragma once +#include #include #include #include +#include #include #include #include @@ -147,3 +149,27 @@ template column_names = {"col0", "col1", "col2"}, std::vector column_order = {0, 1, 2}, rmm::cuda_stream_view stream = cudf::get_default_stream()); + +/** + * @brief Prune row groups using column chunk dictionaries via the single-file hybrid scan reader + * + * @return Dictionary-filtered row group indices + */ +[[nodiscard]] std::vector filter_row_groups_with_dictionaries( + cudf::io::datasource& datasource, + cudf::io::parquet::experimental::hybrid_scan_reader const& reader, + cudf::ast::operation const& filter_expression, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr); + +/** + * @brief Prune row groups using column chunk dictionaries via the multi-file hybrid scan reader + * + * @return Per-source dictionary-filtered row group indices + */ +[[nodiscard]] std::vector> filter_row_groups_with_dictionaries( + multifile_inputs const& inputs, + cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, + cudf::io::parquet_reader_options const& options, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr); diff --git a/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp index 4a76bd507dd1..ac244f17157d 100644 --- a/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp @@ -27,63 +27,6 @@ #include #include -namespace { - -/** - * @brief Filter input row groups using column chunk dictionaries via the experimental parquet - * reader for hybrid scan - * - * @param datasource Input datasource - * @param reader Hybrid scan reader - * @param filter_expression Filter expression - * @param stream CUDA stream - * @param mr Device memory resource - * - * @return Vector of dictionary-filtered row group indices - */ -auto filter_row_groups_with_dictionaries( - cudf::io::datasource& datasource, - cudf::io::parquet::experimental::hybrid_scan_reader const& reader, - cudf::ast::operation const& filter_expression, - rmm::cuda_stream_view stream, - rmm::device_async_resource_ref mr) -{ - // Reader options - cudf::io::parquet_reader_options options = - cudf::io::parquet_reader_options::builder().filter(filter_expression); - - reader.reset_column_selection(); - - // Get all row groups from the reader - auto input_row_group_indices = reader.all_row_groups(options); - - // Span to track current row group indices - auto current_row_group_indices = cudf::host_span(input_row_group_indices); - - // Get dictionary page byte ranges from the reader - auto const dict_page_byte_ranges = - std::get<1>(reader.secondary_filters_byte_ranges(current_row_group_indices, options)); - - // If we have dictionary page byte ranges, filter row groups with dictionary pages - std::vector dict_page_filtered_row_group_indices; - dict_page_filtered_row_group_indices.reserve(current_row_group_indices.size()); - - CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); - - // Fetch dictionary page buffers from the input file buffer - auto [dict_page_buffers, dict_page_data, dict_page_tasks] = - cudf::io::parquet::fetch_byte_ranges_to_device_async( - datasource, dict_page_byte_ranges, stream, mr); - dict_page_tasks.get(); - - dict_page_filtered_row_group_indices = reader.filter_row_groups_with_dictionary_pages( - dict_page_data, current_row_group_indices, options, stream); - - return dict_page_filtered_row_group_indices; -} - -} // namespace - // Base test fixture for tests struct HybridScanFiltersTest : public cudf::test::BaseFixture {}; diff --git a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp index 97ab17ffe855..baf05e6188b9 100644 --- a/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_multifile_filters_test.cpp @@ -85,50 +85,6 @@ auto make_scalar(cudf::size_type value, rmm::cuda_stream_view stream) } } -/** - * @brief Filter input row groups using column chunk dictionaries via the experimental parquet - * reader for hybrid scan (multi-file) - * - * @param inputs Multi-file datasources - * @param reader Hybrid scan multi-file reader - * @param options Parquet reader options - * @param stream CUDA stream - * @param mr Device memory resource - * - * @return Vector of per-source dictionary-filtered row group indices - */ -auto filter_row_groups_with_dictionaries( - multifile_inputs const& inputs, - cudf::io::parquet::experimental::hybrid_scan_multifile const& reader, - cudf::io::parquet_reader_options const& options, - rmm::cuda_stream_view stream, - rmm::device_async_resource_ref mr) -{ - auto const input_row_group_indices = reader.all_row_groups(options); - - auto const dict_pages = reader.dictionary_pages_byte_ranges(input_row_group_indices, options); - CUDF_EXPECTS(dict_pages.first.size() > 0, "No dictionary page byte ranges found"); - - auto const dict_page_ranges_per_source = - group_byte_ranges_by_source(dict_pages, inputs.datasources.size()); - auto [dict_page_buffers, dict_page_data_per_source, task] = - cudf::io::parquet::fetch_byte_ranges_to_device_async( - inputs.datasource_refs, - cudf::host_span const>{ - dict_page_ranges_per_source}, - stream, - mr); - task.get(); - - std::vector> dict_page_data; - for (auto const& source_dict_pages : dict_page_data_per_source) { - dict_page_data.insert(dict_page_data.end(), source_dict_pages.begin(), source_dict_pages.end()); - } - - return reader.filter_row_groups_with_dictionary_pages( - dict_page_data, input_row_group_indices, options, stream); -} - } // namespace struct HybridScanMultifileFiltersTest : public cudf::test::BaseFixture {}; From 8b01cc4067196f193c7b69651b78c84f137f969f Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Thu, 2 Jul 2026 14:07:45 +0200 Subject: [PATCH 14/18] updates the documentation and error handling for schema indices in the page index filtering functions. --- cpp/src/io/parquet/experimental/page_index_filter.cu | 12 ++++++++++++ cpp/tests/io/experimental/hybrid_scan_common.cpp | 10 ++++------ cpp/tests/io/experimental/hybrid_scan_common.hpp | 8 ++++++-- 3 files changed, 22 insertions(+), 8 deletions(-) diff --git a/cpp/src/io/parquet/experimental/page_index_filter.cu b/cpp/src/io/parquet/experimental/page_index_filter.cu index d0052d1799e0..ec7a3cd3b10f 100644 --- a/cpp/src/io/parquet/experimental/page_index_filter.cu +++ b/cpp/src/io/parquet/experimental/page_index_filter.cu @@ -855,6 +855,12 @@ std::unique_ptr aggregate_reader_metadata::build_row_mask_with_pag // Return if empty row group indices if (row_group_indices.empty()) { return cudf::make_empty_column(cudf::type_id::BOOL8); } + // TODO(#22900): remove this guard once this path maps schema indices per source. It currently + // reuses one source's schema index for every source, so it is correct only when schemas match. + CUDF_EXPECTS(schema_idx_maps.empty(), + "Page index statistics filtering does not support mismatched Parquet schemas yet", + std::invalid_argument); + // Check if we have page index for all columns in all row groups auto const has_page_index = compute_has_page_index(per_file_metadata, row_group_indices); @@ -1007,6 +1013,12 @@ thrust::host_vector aggregate_reader_metadata::compute_data_page_mask( return thrust::host_vector(0, stream); } + // TODO(#22900): remove this guard once this path maps schema indices per source. It currently + // reuses one source's schema index for every source, so it is correct only when schemas match. + CUDF_EXPECTS(schema_idx_maps.empty(), + "Data page masking does not support mismatched Parquet schemas yet", + std::invalid_argument); + // Collect column schema indices from the input columns. auto column_schema_indices = std::vector(input_columns.size()); std::transform( diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index 3e3cca0e55e3..3850b8c2af78 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -320,15 +320,13 @@ std::pair, std::vector> create_parquet_with_s static_assert(NumTableConcats >= 1, "Concatenated table must contain at least one table"); // Common default-layout callers skip these checks. - static auto const default_column_names = std::vector{"col0", "col1", "col2"}; - static auto const default_column_order = std::vector{0, 1, 2}; - if (column_names != default_column_names or column_order != default_column_order) { + if (column_names != default_test_column_names or column_order != default_test_column_order) { CUDF_EXPECTS(column_names.size() == column_order.size(), "Column names and column order must have the same size"); - CUDF_EXPECTS(column_order.size() == default_column_order.size(), + CUDF_EXPECTS(column_order.size() == default_test_column_order.size(), "Column order must include all three test columns"); - CUDF_EXPECTS(std::all_of(default_column_order.begin(), - default_column_order.end(), + CUDF_EXPECTS(std::all_of(default_test_column_order.begin(), + default_test_column_order.end(), [&](auto const col_idx) { return std::count( column_order.begin(), column_order.end(), col_idx) == 1; diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index ba64b69234d3..24a67c91bf0a 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -112,6 +112,10 @@ void setup_page_indexes(cudf::io::parquet::experimental::hybrid_scan_multifile c rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr); +/// Default top-level column names and physical emit order used by `create_parquet_with_stats`. +inline std::vector const default_test_column_names{"col0", "col1", "col2"}; +inline std::vector const default_test_column_order{0, 1, 2}; + /** * @brief Creates a table and writes it to Parquet host buffer with column level statistics * @@ -146,8 +150,8 @@ template , std::vector> create_parquet_with_stats( cudf::size_type str_col_value = 100, cudf::io::compression_type compression = cudf::io::compression_type::AUTO, - std::vector column_names = {"col0", "col1", "col2"}, - std::vector column_order = {0, 1, 2}, + std::vector column_names = default_test_column_names, + std::vector column_order = default_test_column_order, rmm::cuda_stream_view stream = cudf::get_default_stream()); /** From ae55fc538ac97dc7bfc84e09969c11cb529ef7bc Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Thu, 2 Jul 2026 16:57:55 +0200 Subject: [PATCH 15/18] Refactor dictionary page byte range handling in hybrid scan implementation This commit simplifies the handling of dictionary page byte ranges in the `filter_row_groups_with_dictionaries_impl` function. It removes unnecessary comments and consolidates the logic for fetching byte ranges, ensuring consistency in the use of row group indices across different reader types. This enhances code clarity and maintainability. --- .../io/experimental/hybrid_scan_common.cpp | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index 3850b8c2af78..51f3d0f07d0a 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -240,14 +240,11 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { - // Reset column selection so the helper is safe to call repeatedly on the same reader; the - // single-file tests reuse one reader across many filter expressions. reader.reset_column_selection(); auto const row_group_indices = reader.all_row_groups(options); if constexpr (std::is_same_v) { - // Dictionary page byte ranges carry a per-range source map used to regroup them by source. auto const dict_pages = reader.dictionary_pages_byte_ranges(row_group_indices, options); CUDF_EXPECTS(dict_pages.first.size() > 0, "No dictionary page byte ranges found"); @@ -255,11 +252,7 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs, group_byte_ranges_by_source(dict_pages, inputs.datasources.size()); auto [dict_page_buffers, dict_page_data_per_source, task] = cudf::io::parquet::fetch_byte_ranges_to_device_async( - inputs.datasource_refs, - cudf::host_span const>{ - dict_page_ranges_per_source}, - stream, - mr); + inputs.datasource_refs, dict_page_ranges_per_source, stream, mr); task.get(); std::vector> dict_page_data; @@ -267,21 +260,21 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs, dict_page_data.insert( dict_page_data.end(), source_dict_pages.begin(), source_dict_pages.end()); } + return reader.filter_row_groups_with_dictionary_pages( dict_page_data, row_group_indices, options, stream); } else { - // `secondary_filters_byte_ranges().second` is the single-file equivalent of the multi-file - // `dictionary_pages_byte_ranges()`. - auto const current = cudf::host_span{row_group_indices}; auto const dict_page_byte_ranges = - reader.secondary_filters_byte_ranges(current, options).second; + reader.secondary_filters_byte_ranges(row_group_indices, options).second; CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); auto [dict_page_buffers, dict_page_data, dict_page_tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async( inputs, dict_page_byte_ranges, stream, mr); dict_page_tasks.get(); - return reader.filter_row_groups_with_dictionary_pages(dict_page_data, current, options, stream); + + return reader.filter_row_groups_with_dictionary_pages( + dict_page_data, row_group_indices, options, stream); } } From a07b7b55ead0d46b6c5b2ac6e6795616dde309e4 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Tue, 7 Jul 2026 12:39:49 +0200 Subject: [PATCH 16/18] Refactor and clarify Parquet reader helper functions - Updated documentation in `reader_impl_helpers.hpp` to specify that the function finds the offset of the column chunk in the specified row group. - Simplified validation checks in `create_parquet_with_stats` by removing redundant conditions and directly enforcing that `column_order` must include all three test columns. - Replaced default test column names and order with inline definitions for clarity in `hybrid_scan_common.hpp`. These changes enhance code readability and maintainability while ensuring correct validation logic in the Parquet creation process. --- cpp/src/io/parquet/reader_impl_helpers.hpp | 6 +++--- .../io/experimental/hybrid_scan_common.cpp | 17 +++-------------- .../io/experimental/hybrid_scan_common.hpp | 11 +++++------ 3 files changed, 11 insertions(+), 23 deletions(-) diff --git a/cpp/src/io/parquet/reader_impl_helpers.hpp b/cpp/src/io/parquet/reader_impl_helpers.hpp index c26fc7c07617..e83ff8d0f6f1 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.hpp +++ b/cpp/src/io/parquet/reader_impl_helpers.hpp @@ -108,10 +108,10 @@ struct row_group_info { [[nodiscard]] std::size_t derive_pass_read_limit(std::size_t chunk_read_limit); /** - * @brief Find the offset of the column chunk with the given schema index in the row group + * @brief Find the offset of the column chunk with the given schema index in the specified row group * - * @note For mismatched schemas, the caller must pre-map `schema_idx` to the row group's source via - * `aggregate_reader_metadata::map_schema_index`; this function does not remap it. + * @note For mismatched schemas, `schema_idx` must be pre-mapped to the row group's source using + * `map_schema_index`. * * @param row_group Row group * @param schema_idx Schema index, already mapped to the row group's source diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index 51f3d0f07d0a..598c0ec2ffe1 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -312,20 +312,9 @@ std::pair, std::vector> create_parquet_with_s { static_assert(NumTableConcats >= 1, "Concatenated table must contain at least one table"); - // Common default-layout callers skip these checks. - if (column_names != default_test_column_names or column_order != default_test_column_order) { - CUDF_EXPECTS(column_names.size() == column_order.size(), - "Column names and column order must have the same size"); - CUDF_EXPECTS(column_order.size() == default_test_column_order.size(), - "Column order must include all three test columns"); - CUDF_EXPECTS(std::all_of(default_test_column_order.begin(), - default_test_column_order.end(), - [&](auto const col_idx) { - return std::count( - column_order.begin(), column_order.end(), col_idx) == 1; - }), - "Column order must be a permutation of the three test columns"); - } + CUDF_EXPECTS(column_names.size() == column_order.size(), + "Column names and column order must have the same size"); + CUDF_EXPECTS(column_order.size() == 3, "Column order must include all three test columns"); auto col0 = testdata::ascending(); auto col1 = []() { diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index 24a67c91bf0a..736985a96eed 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -112,10 +112,6 @@ void setup_page_indexes(cudf::io::parquet::experimental::hybrid_scan_multifile c rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr); -/// Default top-level column names and physical emit order used by `create_parquet_with_stats`. -inline std::vector const default_test_column_names{"col0", "col1", "col2"}; -inline std::vector const default_test_column_order{0, 1, 2}; - /** * @brief Creates a table and writes it to Parquet host buffer with column level statistics * @@ -141,6 +137,9 @@ inline std::vector const default_test_column_order{0, 1, 2}; * is used to build mismatched per-source schemas for the row-group filtering tests. * @param stream CUDA stream * + * @note `column_order` must be a permutation of {0, 1, 2}; passing a non-permutation (e.g. + * {0, 0, 1}) is undefined behavior. + * * @return Tuple of table and Parquet host buffer */ template , std::vector> create_parquet_with_stats( cudf::size_type str_col_value = 100, cudf::io::compression_type compression = cudf::io::compression_type::AUTO, - std::vector column_names = default_test_column_names, - std::vector column_order = default_test_column_order, + std::vector column_names = {"col0", "col1", "col2"}, + std::vector column_order = {0, 1, 2}, rmm::cuda_stream_view stream = cudf::get_default_stream()); /** From 7aa70cfc655acb5c42ea73848f3e8d28a606db48 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Tue, 7 Jul 2026 18:34:55 +0200 Subject: [PATCH 17/18] Refactor filter_row_groups_with_dictionaries to use parquet_reader_options - Updated the `filter_row_groups_with_dictionaries` function to accept `parquet_reader_options` instead of `filter_expression`. - Adjusted related test cases in `hybrid_scan_filters_test.cpp` to build options using the new `parquet_reader_options` structure. These changes improve the flexibility and clarity of the filtering mechanism in the hybrid scan functionality. --- .../io/experimental/hybrid_scan_common.cpp | 4 +- .../io/experimental/hybrid_scan_common.hpp | 2 +- .../experimental/hybrid_scan_filters_test.cpp | 117 +++++++++++------- 3 files changed, 73 insertions(+), 50 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index 598c0ec2ffe1..00c79d298ebd 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -283,12 +283,10 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs, std::vector filter_row_groups_with_dictionaries( cudf::io::datasource& datasource, cudf::io::parquet::experimental::hybrid_scan_reader const& reader, - cudf::ast::operation const& filter_expression, + cudf::io::parquet_reader_options const& options, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr) { - auto const options = - cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); return filter_row_groups_with_dictionaries_impl(datasource, reader, options, stream, mr); } diff --git a/cpp/tests/io/experimental/hybrid_scan_common.hpp b/cpp/tests/io/experimental/hybrid_scan_common.hpp index 736985a96eed..8a8493977a4b 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.hpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.hpp @@ -161,7 +161,7 @@ template filter_row_groups_with_dictionaries( cudf::io::datasource& datasource, cudf::io::parquet::experimental::hybrid_scan_reader const& reader, - cudf::ast::operation const& filter_expression, + cudf::io::parquet_reader_options const& options, rmm::cuda_stream_view stream, rmm::device_async_resource_ref mr); diff --git a/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp b/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp index ac244f17157d..95a5677b4372 100644 --- a/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_filters_test.cpp @@ -964,9 +964,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) auto filter_expression = cudf::ast::operation(cudf::ast::ast_operator::NOT_EQUAL, col0_ref, uint_literal); constexpr size_t expected_row_groups = 4; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -977,9 +978,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) auto filter_expression = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, uint_literal, col0_ref); constexpr size_t expected_row_groups = 0; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -991,9 +993,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::operation(cudf::ast::ast_operator::NOT_EQUAL, col2_ref, str_literal); constexpr size_t expected_row_groups = 0; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1005,9 +1008,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col2_ref, str_literal); constexpr size_t expected_row_groups = 4; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1026,9 +1030,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_AND, uint_filter_expression, str_filter_expression); constexpr size_t expected_row_groups = 4; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1046,9 +1051,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_AND, uint_filter_expression, uint_filter_expression2); constexpr size_t expected_row_groups = 4; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1066,9 +1072,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_AND, uint_filter_expression, uint_filter_expression2); constexpr size_t expected_row_groups = 1; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1086,9 +1093,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_OR, str_filter_expression, str_filter_expression2); constexpr size_t expected_row_groups = 4; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1107,9 +1115,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_OR, uint_filter_expression, str_filter_expression); constexpr size_t expected_row_groups = 4; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1128,9 +1137,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_AND, uint_filter_expression, str_filter_expression); constexpr size_t expected_row_groups = 0; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1154,9 +1164,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_OR, composed_filter_expression, uint_filter_expression3); constexpr size_t expected_row_groups = 3; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1180,9 +1191,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_OR, composed_filter_expression, uint_filter_expression3); constexpr size_t expected_row_groups = 4; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1206,9 +1218,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::ast_operator::LOGICAL_AND, composed_filter_expression, str_filter_expression3); constexpr size_t expected_row_groups = 0; + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); EXPECT_EQ( - filter_row_groups_with_dictionaries(datasource_ref, reader_ref, filter_expression, stream, mr) - .size(), + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr).size(), expected_row_groups); } @@ -1220,8 +1233,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) auto rhs = cudf::ast::operation(cudf::ast::ast_operator::NOT_EQUAL, col0_ref, col2_ref); auto const filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_AND, lhs, rhs); - auto const result = filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr); + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + auto const result = + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr); auto const expected = std::vector{1}; EXPECT_EQ(result, expected); } @@ -1232,8 +1247,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) auto uint_literal = cudf::ast::literal(uint_literal_value); auto inner = cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col0_ref, uint_literal); auto const filter_expression = cudf::ast::operation(cudf::ast::ast_operator::NOT, inner); - auto const result = filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr); + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + auto const result = + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr); auto const expected = std::vector{0, 2, 3}; EXPECT_EQ(result, expected); } @@ -1250,8 +1267,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) cudf::ast::operation(cudf::ast::ast_operator::NULL_EQUAL, col0_ref, literal_100); auto const filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_AND, not_eq_50, null_eq_100); - auto const result = filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr); + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + auto const result = + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr); auto const expected = std::vector{0, 2, 3}; EXPECT_EQ(result, expected); } @@ -1268,8 +1287,10 @@ TEST_F(HybridScanFiltersTest, FilterRowGroupsWithDictionary) auto not_eq_str = cudf::ast::operation(cudf::ast::ast_operator::NOT, eq_str); auto const filter_expression = cudf::ast::operation(cudf::ast::ast_operator::LOGICAL_OR, not_eq_50, not_eq_str); - auto const result = filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr); + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + auto const result = + filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr); auto const expected = std::vector{0, 2, 3}; EXPECT_EQ(result, expected); } @@ -1326,9 +1347,9 @@ TYPED_TEST(RowGroupFilteringWithDictTest, FilterFewLiteralsTyped) auto mr = cudf::get_current_device_resource_ref(); // Input datasource - auto const datasource = cudf::io::datasource::create(cudf::host_span( + auto const datasource = cudf::io::datasource::create(cudf::host_span( reinterpret_cast(buffer.data()), buffer.size())); - auto datasource_ref = std::ref(*datasource); + auto const datasource_ref = std::ref(*datasource); // Hybrid scan reader auto options = cudf::io::parquet_reader_options::builder().build(); @@ -1381,8 +1402,9 @@ TYPED_TEST(RowGroupFilteringWithDictTest, FilterFewLiteralsTyped) cudf::ast::operation(cudf::ast::ast_operator::EQUAL, col_ref, literal); // Check the results - EXPECT_EQ(filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr), + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + EXPECT_EQ(filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr), expected_row_groups); } @@ -1403,8 +1425,9 @@ TYPED_TEST(RowGroupFilteringWithDictTest, FilterFewLiteralsTyped) cudf::ast::operation(cudf::ast::ast_operator::NOT_EQUAL, col_name, literal); // Check the results - EXPECT_EQ(filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr), + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + EXPECT_EQ(filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr), expected_row_groups); } } @@ -1451,9 +1474,9 @@ TYPED_TEST(RowGroupFilteringWithDictTest, FilterManyLiteralsTyped) auto mr = cudf::get_current_device_resource_ref(); // Input datasource - auto const datasource = cudf::io::datasource::create(cudf::host_span( + auto const datasource = cudf::io::datasource::create(cudf::host_span( reinterpret_cast(buffer.data()), buffer.size())); - auto datasource_ref = std::ref(*datasource); + auto const datasource_ref = std::ref(*datasource); // Hybrid scan reader auto options = cudf::io::parquet_reader_options::builder().build(); @@ -1557,8 +1580,9 @@ TYPED_TEST(RowGroupFilteringWithDictTest, FilterManyLiteralsTyped) cudf::ast::ast_operator::LOGICAL_OR, filter_expression12, filter_expression3); // Check the results - EXPECT_EQ(filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr), + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + EXPECT_EQ(filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr), expected_row_groups); } @@ -1590,8 +1614,9 @@ TYPED_TEST(RowGroupFilteringWithDictTest, FilterManyLiteralsTyped) cudf::ast::ast_operator::LOGICAL_AND, filter_expression12, filter_expression3); // Check the results - EXPECT_EQ(filter_row_groups_with_dictionaries( - datasource_ref, reader_ref, filter_expression, stream, mr), + auto const options = + cudf::io::parquet_reader_options::builder().filter(filter_expression).build(); + EXPECT_EQ(filter_row_groups_with_dictionaries(datasource_ref, reader_ref, options, stream, mr), expected_row_groups); } } From eea8f1f86e53b07c1695b935adb0d905859e3ab3 Mon Sep 17 00:00:00 2001 From: Qi Chen Date: Wed, 15 Jul 2026 10:15:16 +0200 Subject: [PATCH 18/18] Mark unused variables in hybrid_scan_common.cpp to avoid compiler warnings --- cpp/tests/io/experimental/hybrid_scan_common.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/cpp/tests/io/experimental/hybrid_scan_common.cpp b/cpp/tests/io/experimental/hybrid_scan_common.cpp index 00c79d298ebd..339009ecf7e5 100644 --- a/cpp/tests/io/experimental/hybrid_scan_common.cpp +++ b/cpp/tests/io/experimental/hybrid_scan_common.cpp @@ -250,7 +250,7 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs, auto const dict_page_ranges_per_source = group_byte_ranges_by_source(dict_pages, inputs.datasources.size()); - auto [dict_page_buffers, dict_page_data_per_source, task] = + [[maybe_unused]] auto [dict_page_buffers, dict_page_data_per_source, task] = cudf::io::parquet::fetch_byte_ranges_to_device_async( inputs.datasource_refs, dict_page_ranges_per_source, stream, mr); task.get(); @@ -268,7 +268,7 @@ auto filter_row_groups_with_dictionaries_impl(InputType& inputs, reader.secondary_filters_byte_ranges(row_group_indices, options).second; CUDF_EXPECTS(dict_page_byte_ranges.size() > 0, "No dictionary page byte ranges found"); - auto [dict_page_buffers, dict_page_data, dict_page_tasks] = + [[maybe_unused]] auto [dict_page_buffers, dict_page_data, dict_page_tasks] = cudf::io::parquet::fetch_byte_ranges_to_device_async( inputs, dict_page_byte_ranges, stream, mr); dict_page_tasks.get();