From 08f1e363b24235cb95eb8779e093cc6c075da644 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com> Date: Wed, 24 Jun 2026 00:17:16 +0000 Subject: [PATCH 1/6] Select Parquet columns by field ID --- cpp/include/cudf/io/parquet.hpp | 46 +++- cpp/src/io/parquet/column_path_helpers.cpp | 18 ++ cpp/src/io/parquet/column_path_helpers.hpp | 13 + .../experimental/hybrid_scan_helpers.cpp | 12 +- .../experimental/hybrid_scan_helpers.hpp | 4 +- .../parquet/experimental/hybrid_scan_impl.cpp | 23 +- .../parquet/expression_transform_helpers.cpp | 94 ++++--- .../parquet/expression_transform_helpers.hpp | 2 +- cpp/src/io/parquet/reader_impl.cpp | 67 +++-- cpp/src/io/parquet/reader_impl_helpers.cpp | 68 +++-- cpp/src/io/parquet/reader_impl_helpers.hpp | 6 +- cpp/tests/io/parquet_common.cpp | 6 +- cpp/tests/io/parquet_common.hpp | 8 +- cpp/tests/io/parquet_reader_test.cpp | 235 +++++++++++++----- 14 files changed, 445 insertions(+), 157 deletions(-) diff --git a/cpp/include/cudf/io/parquet.hpp b/cpp/include/cudf/io/parquet.hpp index 712001685d55..89dee89568e2 100644 --- a/cpp/include/cudf/io/parquet.hpp +++ b/cpp/include/cudf/io/parquet.hpp @@ -66,11 +66,14 @@ class parquet_reader_options_builder; class parquet_reader_options { source_info _source; + // Column selection options. Only one of these must be set at a time. + // Path in schema of column names to read; `nullopt` is all std::optional> _column_names; - // Indices of top-level columns to read; `nullopt` is all (cannot be used alongside - // `_column_names`) + // Indices of top-level columns to read; `nullopt` is all std::optional> _column_indices; + // Parquet field IDs of columns/fields to read; `nullopt` is all + std::optional> _column_field_ids; // List of individual row groups to read (ignored if empty) std::vector> _row_groups; @@ -253,6 +256,13 @@ class parquet_reader_options { */ [[nodiscard]] auto const& get_column_indices() const { return _column_indices; } + /** + * @brief Returns Parquet field IDs of columns/fields to be read, if set. + * + * @return Parquet field IDs of columns/fields to be read; `nullopt` if the option is not set + */ + [[nodiscard]] auto const& get_column_field_ids() const { return _column_field_ids; } + /** * @brief Returns list of individual row groups to be read. * @@ -357,6 +367,8 @@ class parquet_reader_options { { CUDF_EXPECTS(not _column_indices.has_value(), "Cannot select columns by indices and names simultaneously"); + CUDF_EXPECTS(not _column_field_ids.has_value(), + "Cannot select columns by field IDs and names simultaneously"); _column_names = std::move(column_names); } @@ -375,9 +387,26 @@ class parquet_reader_options { { CUDF_EXPECTS(not _column_names.has_value(), "Cannot select columns by indices and names simultaneously"); + CUDF_EXPECTS(not _column_field_ids.has_value(), + "Cannot select columns by field IDs and indices simultaneously"); _column_indices = std::move(col_indices); } + /** + * @brief Sets the Parquet field IDs of columns/fields to be read from all input sources. + * + * @param column_field_ids A vector of Parquet field IDs to attempt to read from each input + * source. + */ + void set_column_field_ids(std::vector column_field_ids) + { + CUDF_EXPECTS(not _column_names.has_value(), + "Cannot select columns by field IDs and names simultaneously"); + CUDF_EXPECTS(not _column_indices.has_value(), + "Cannot select columns by field IDs and indices simultaneously"); + _column_field_ids = std::move(column_field_ids); + } + /** * @brief Specifies which row groups to read from each input source. * @@ -604,6 +633,19 @@ class parquet_reader_options_builder { return *this; } + /** + * @brief Sets the Parquet field IDs of columns/fields to be read from all input sources. + * + * @param column_field_ids A vector of Parquet field IDs to attempt to read from each input + * source. + * @return this for chaining + */ + parquet_reader_options_builder& column_field_ids(std::vector column_field_ids) + { + options.set_column_field_ids(std::move(column_field_ids)); + return *this; + } + /** * @copydoc parquet_reader_options::set_row_groups * @return this for chaining diff --git a/cpp/src/io/parquet/column_path_helpers.cpp b/cpp/src/io/parquet/column_path_helpers.cpp index 240a06ff103e..e2de8084feef 100644 --- a/cpp/src/io/parquet/column_path_helpers.cpp +++ b/cpp/src/io/parquet/column_path_helpers.cpp @@ -5,15 +5,33 @@ #include "column_path_helpers.hpp" +#include + #include #include #include #include +#include #include #include +#include +#include namespace cudf::io::parquet::detail { +std::string column_path_from_index(std::span schema_tree, int schema_idx) +{ + std::vector path; + for (auto idx = schema_idx; idx > 0; idx = schema_tree[idx].parent_idx) { + path.push_back(schema_tree[idx].name); + } + + return std::accumulate( + path.rbegin() + 1, path.rend(), path.back(), [](auto path_so_far, auto const& elem_name) { + return std::move(path_so_far) + "." + elem_name; + }); +} + std::string normalize_column_path(std::string_view col_path, bool case_sensitive_names) { if (case_sensitive_names) { return std::string{col_path}; } diff --git a/cpp/src/io/parquet/column_path_helpers.hpp b/cpp/src/io/parquet/column_path_helpers.hpp index 418721cea0e2..dde8efb5ddc7 100644 --- a/cpp/src/io/parquet/column_path_helpers.hpp +++ b/cpp/src/io/parquet/column_path_helpers.hpp @@ -5,7 +5,10 @@ #pragma once +#include + #include +#include #include #include #include @@ -13,6 +16,16 @@ namespace cudf::io::parquet::detail { +/** + * @brief Gets the dot-separated path for a schema element. + * + * @param schema_tree The schema tree describing the file structure + * @param schema_idx Index of the schema element + * @return Dot-separated schema path from the root child to the schema element + */ +[[nodiscard]] std::string column_path_from_index(std::span schema_tree, + int schema_idx); + /** * @brief Returns a normalized (lowercased) column name or path when case-insensitive matching is * enabled diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp index e61b4ad09bcb..25c5198c594c 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.cpp @@ -271,7 +271,8 @@ aggregate_reader_metadata::select_payload_columns( bool ignore_missing_columns, type_id timestamp_type_id, type_id decimal_type_id, - bool case_sensitive_names) + bool case_sensitive_names, + bool match_schema_by_field_id) { // If neither payload nor filter columns are specified, select all columns if (not payload_column_names.has_value() and not filter_column_names.has_value()) { @@ -283,7 +284,8 @@ aggregate_reader_metadata::select_payload_columns( ignore_missing_columns, timestamp_type_id, decimal_type_id, - case_sensitive_names); + case_sensitive_names, + match_schema_by_field_id); } std::vector valid_payload_columns; @@ -318,7 +320,8 @@ aggregate_reader_metadata::select_payload_columns( ignore_missing_columns, timestamp_type_id, decimal_type_id, - case_sensitive_names); + case_sensitive_names, + match_schema_by_field_id); } // Else if only filter columns are specified, select all columns that do not appear in the @@ -348,7 +351,8 @@ aggregate_reader_metadata::select_payload_columns( ignore_missing_columns, timestamp_type_id, decimal_type_id, - case_sensitive_names); + case_sensitive_names, + match_schema_by_field_id); } 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..47c699da31e2 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_helpers.hpp @@ -163,6 +163,7 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base { * @param timestamp_type_id Type conversion parameter * @param decimal_type_id Type conversion parameter * @param case_sensitive_names Boolean indicating if column names are case sensitive + * @param match_schema_by_field_id Whether to match schema by field ID * * @return input column information, output column buffers, list of output column schema * indices @@ -176,7 +177,8 @@ class aggregate_reader_metadata : public aggregate_reader_metadata_base { bool ignore_missing_columns, type_id timestamp_type_id, type_id decimal_type_id, - bool case_sensitive_names); + bool case_sensitive_names, + bool match_schema_by_field_id); /** * @brief Filters row groups such that only the row groups that start within the byte range diff --git a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp index 665d431593fa..0a73c428b750 100644 --- a/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp +++ b/cpp/src/io/parquet/experimental/hybrid_scan_impl.cpp @@ -68,10 +68,11 @@ hybrid_scan_reader_impl::hybrid_scan_reader_impl( cudf::host_span const> footer_bytes, parquet_reader_options const& options) { + auto const has_cols_from_mismatched_srcs = + (options.get_column_names().has_value() or options.get_column_field_ids().has_value()) and + options.is_enabled_allow_mismatched_pq_schemas(); _metadata = std::make_unique( - footer_bytes, - options.is_enabled_use_arrow_schema(), - options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas()); + footer_bytes, options.is_enabled_use_arrow_schema(), has_cols_from_mismatched_srcs); _extended_metadata = static_cast(_metadata.get()); } @@ -79,10 +80,11 @@ hybrid_scan_reader_impl::hybrid_scan_reader_impl( hybrid_scan_reader_impl::hybrid_scan_reader_impl( cudf::host_span parquet_metadatas, parquet_reader_options const& options) { + auto const has_cols_from_mismatched_srcs = + (options.get_column_names().has_value() or options.get_column_field_ids().has_value()) and + options.is_enabled_allow_mismatched_pq_schemas(); _metadata = std::make_unique( - parquet_metadatas, - options.is_enabled_use_arrow_schema(), - options.get_column_names().has_value() and options.is_enabled_allow_mismatched_pq_schemas()); + parquet_metadatas, options.is_enabled_use_arrow_schema(), has_cols_from_mismatched_srcs); _extended_metadata = static_cast(_metadata.get()); } @@ -132,7 +134,8 @@ void hybrid_scan_reader_impl::select_columns(read_columns_mode read_columns_mode options.is_enabled_ignore_missing_columns(), _options.timestamp_type.id(), _options.decimal_width, - _options.case_sensitive_names); + _options.case_sensitive_names, + options.get_column_field_ids().has_value()); _is_all_columns_selected = true; _is_filter_columns_selected = false; @@ -157,7 +160,8 @@ void hybrid_scan_reader_impl::select_columns(read_columns_mode read_columns_mode ignore_missing_columns, _options.timestamp_type.id(), _options.decimal_width, - _options.case_sensitive_names); + _options.case_sensitive_names, + options.get_column_field_ids().has_value()); _is_filter_columns_selected = true; _is_payload_columns_selected = false; @@ -178,7 +182,8 @@ void hybrid_scan_reader_impl::select_columns(read_columns_mode read_columns_mode options.is_enabled_ignore_missing_columns(), _options.timestamp_type.id(), _options.decimal_width, - _options.case_sensitive_names); + _options.case_sensitive_names, + options.get_column_field_ids().has_value()); _is_payload_columns_selected = true; _is_filter_columns_selected = false; diff --git a/cpp/src/io/parquet/expression_transform_helpers.cpp b/cpp/src/io/parquet/expression_transform_helpers.cpp index f7931c98adb5..340d0f6bd7bd 100644 --- a/cpp/src/io/parquet/expression_transform_helpers.cpp +++ b/cpp/src/io/parquet/expression_transform_helpers.cpp @@ -18,6 +18,10 @@ #include +#include +#include +#include + namespace cudf::io::parquet::detail { namespace { @@ -235,17 +239,20 @@ void names_from_expression::visit_operands( [[nodiscard]] std::unordered_map map_column_indices_to_names( cudf::io::parquet_reader_options const& options, - std::vector const& schema_tree, + std::span schema_tree, bool case_sensitive_names) { std::unordered_map column_indices_to_names; - auto const& selected_column_names = options.get_column_names(); - auto const& selected_column_indices = options.get_column_indices(); + auto const& selected_column_names = options.get_column_names(); + auto const& selected_column_indices = options.get_column_indices(); + auto const& selected_column_field_ids = options.get_column_field_ids(); - CUDF_EXPECTS( - not(selected_column_names.has_value() and selected_column_indices.has_value()), - "Parquet reader encountered column selection by both names and indices simultaneously"); + CUDF_EXPECTS(static_cast(selected_column_names.has_value()) + + static_cast(selected_column_indices.has_value()) + + static_cast(selected_column_field_ids.has_value()) <= + 1, + "Parquet reader encountered multiple column selection modes"); // Map counting indices to the selected column by names if (selected_column_names.has_value()) { @@ -257,32 +264,59 @@ void names_from_expression::visit_operands( return std::make_pair(col_index, normalize_column_path(col_name, case_sensitive_names)); }); - } else { - // Map selected top-level column indices to their names from the schema tree + } + // Map selected top-level column indices to their names from the schema tree + else if (selected_column_indices.has_value()) { auto const& root = schema_tree.front(); - if (selected_column_indices.has_value()) { - std::transform(selected_column_indices->begin(), - selected_column_indices->end(), - cuda::counting_iterator{0}, - std::inserter(column_indices_to_names, column_indices_to_names.end()), - [&](auto selected_col_idx, auto const mapped_col_idx) { - auto const schema_idx = root.children_idx[selected_col_idx]; - return std::make_pair( - mapped_col_idx, - normalize_column_path(schema_tree[schema_idx].name, case_sensitive_names)); - }); - } else { - // Map all top-level column indices to their names from the schema tree - std::for_each( - cuda::counting_iterator{0}, - cuda::counting_iterator{static_cast(root.children_idx.size())}, - [&](auto col_idx) { - auto const schema_idx = root.children_idx[col_idx]; - column_indices_to_names.insert( - {col_idx, normalize_column_path(schema_tree[schema_idx].name, case_sensitive_names)}); - }); - } + std::transform(selected_column_indices->begin(), + selected_column_indices->end(), + cuda::counting_iterator{0}, + std::inserter(column_indices_to_names, column_indices_to_names.end()), + [&](auto selected_col_idx, auto const mapped_col_idx) { + CUDF_EXPECTS( + std::cmp_less(selected_col_idx, root.children_idx.size()), + "Encountered an invalid col index in the top-level column selection", + std::invalid_argument); + auto const schema_idx = root.children_idx[selected_col_idx]; + return std::make_pair( + mapped_col_idx, + normalize_column_path(schema_tree[schema_idx].name, case_sensitive_names)); + }); + } + // Map selected field ids to column paths from the schema tree + else if (selected_column_field_ids.has_value()) { + std::transform( + selected_column_field_ids->begin(), + selected_column_field_ids->end(), + cuda::counting_iterator{0}, + std::inserter(column_indices_to_names, column_indices_to_names.end()), + [&](auto const& field_id, auto const mapped_col_idx) { + auto const schema_iter = + std::find_if(schema_tree.begin() + 1, schema_tree.end(), [field_id](auto const& schema) { + return schema.field_id.has_value() and schema.field_id.value() == field_id; + }); + CUDF_EXPECTS(schema_iter != schema_tree.end(), + "Encountered a non-existent Parquet field ID in selected columns", + std::invalid_argument); + auto const schema_idx = static_cast(std::distance(schema_tree.begin(), schema_iter)); + return std::make_pair(mapped_col_idx, + normalize_column_path(column_path_from_index(schema_tree, schema_idx), + case_sensitive_names)); + }); + } + // Map all top-level column indices to their names from the schema tree + else { + auto const& root = schema_tree.front(); + + std::for_each( + cuda::counting_iterator{0}, + cuda::counting_iterator{static_cast(root.children_idx.size())}, + [&](auto col_idx) { + auto const schema_idx = root.children_idx[col_idx]; + column_indices_to_names.insert( + {col_idx, normalize_column_path(schema_tree[schema_idx].name, case_sensitive_names)}); + }); } return column_indices_to_names; diff --git a/cpp/src/io/parquet/expression_transform_helpers.hpp b/cpp/src/io/parquet/expression_transform_helpers.hpp index 8af1722f215c..29ea58a8ef0f 100644 --- a/cpp/src/io/parquet/expression_transform_helpers.hpp +++ b/cpp/src/io/parquet/expression_transform_helpers.hpp @@ -321,7 +321,7 @@ class equality_literals_collector : public ast::detail::expression_transformer { */ [[nodiscard]] std::unordered_map map_column_indices_to_names( cudf::io::parquet_reader_options const& options, - std::vector const& schema_tree, + std::span schema_tree, bool case_sensitive_names); /** diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index eb00fafa6896..e103f633190b 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -5,6 +5,7 @@ #include "reader_impl.hpp" +#include "column_path_helpers.hpp" #include "error.hpp" #include "runtime/context.hpp" @@ -520,16 +521,17 @@ reader_impl::reader_impl(std::size_t chunk_read_limit, // Open and parse the source dataset metadata CUDF_EXPECTS(file_metadatas.empty() or file_metadatas.size() == _sources.size(), "Encountered a mismatch in the number of provided data sources and metadatas"); - _metadata = file_metadatas.empty() ? std::make_unique( - _sources, - options.is_enabled_use_arrow_schema(), - options.get_column_names().has_value() and - options.is_enabled_allow_mismatched_pq_schemas()) - : std::make_unique( - std::forward>(file_metadatas), - options.is_enabled_use_arrow_schema(), - options.get_column_names().has_value() and - options.is_enabled_allow_mismatched_pq_schemas()); + + auto const has_cols_from_mismatched_srcs = + (options.get_column_names().has_value() or options.get_column_field_ids().has_value()) and + options.is_enabled_allow_mismatched_pq_schemas(); + _metadata = file_metadatas.empty() + ? std::make_unique( + _sources, options.is_enabled_use_arrow_schema(), has_cols_from_mismatched_srcs) + : std::make_unique( + std::forward>(file_metadatas), + options.is_enabled_use_arrow_schema(), + has_cols_from_mismatched_srcs); // Number of input sources _num_sources = _sources.size(); @@ -545,8 +547,10 @@ reader_impl::reader_impl(std::size_t chunk_read_limit, get_column_projection(options, options.is_enabled_ignore_missing_columns()); std::optional> filter_only_columns_names; - if (options.get_filter().has_value() and - (options.get_column_names().has_value() or options.get_column_indices().has_value())) { + auto const has_column_selection = options.get_column_names().has_value() or + options.get_column_indices().has_value() or + options.get_column_field_ids().has_value(); + if (options.get_filter().has_value() and has_column_selection) { // list, struct, dictionary are not supported by AST filter yet. // extract columns not present in get_column_names() & keep count to remove at end. filter_only_columns_names = get_column_names_in_expression( @@ -561,7 +565,8 @@ reader_impl::reader_impl(std::size_t chunk_read_limit, options.is_enabled_ignore_missing_columns(), _options.timestamp_type.id(), _options.decimal_width, - _options.case_sensitive_names); + _options.case_sensitive_names, + options.get_column_field_ids().has_value()); // Save the states of the output buffers for reuse in `chunk_read()`. std::transform( @@ -817,22 +822,25 @@ std::vector reader_impl::calculate_output_num_rows_per_source(size_t con std::optional> reader_impl::get_column_projection( parquet_reader_options const& options, bool ignore_missing_columns) const { - auto const has_column_names = options.get_column_names().has_value(); - auto const has_column_indices = options.get_column_indices().has_value(); + auto const has_column_names = options.get_column_names().has_value(); + auto const has_column_indices = options.get_column_indices().has_value(); + auto const has_column_field_ids = options.get_column_field_ids().has_value(); - CUDF_EXPECTS( - not(has_column_names and has_column_indices), - "Parquet reader encountered column selection by both names and indices simultaneously"); + auto constexpr max_allowed_col_selection_modes = 1; + CUDF_EXPECTS(static_cast(has_column_names) + static_cast(has_column_indices) + + static_cast(has_column_field_ids) <= + max_allowed_col_selection_modes, + "Parquet reader encountered more than one column selection mode"); // No column selection specified. Return nullopt indicating all columns to be selected - if (not has_column_names and not has_column_indices) { + if (not has_column_names and not has_column_indices and not has_column_field_ids) { return std::nullopt; } else if (has_column_names) { return options.get_column_names(); - } else { + } else if (has_column_indices) { std::vector col_names; auto const& top_level_schema_indices = _metadata->get_schema(0).children_idx; - for (auto const index : options.get_column_indices().value_or(std::vector{})) { + for (auto const index : options.get_column_indices().value()) { auto const is_valid_index = std::cmp_greater_equal(index, 0) and std::cmp_less(index, top_level_schema_indices.size()); CUDF_EXPECTS(ignore_missing_columns or is_valid_index, @@ -842,6 +850,23 @@ std::optional> reader_impl::get_column_projection( } } return std::make_optional(std::move(col_names)); + } else { + std::vector col_names; + auto const& schema_tree = _metadata->get_schema_tree(); + for (auto const field_id : options.get_column_field_ids().value()) { + auto const schema_iter = + std::find_if(schema_tree.cbegin() + 1, schema_tree.cend(), [field_id](auto const& schema) { + return schema.field_id.has_value() and schema.field_id.value() == field_id; + }); + CUDF_EXPECTS(ignore_missing_columns or schema_iter != schema_tree.end(), + "Encountered a non-existent Parquet field ID in selected columns", + std::invalid_argument); + if (schema_iter != schema_tree.end()) { + auto const schema_idx = static_cast(std::distance(schema_tree.cbegin(), schema_iter)); + col_names.emplace_back(column_path_from_index(schema_tree, schema_idx)); + } + } + return std::make_optional(std::move(col_names)); } } diff --git a/cpp/src/io/parquet/reader_impl_helpers.cpp b/cpp/src/io/parquet/reader_impl_helpers.cpp index 7208e2e32091..778f535e1967 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.cpp +++ b/cpp/src/io/parquet/reader_impl_helpers.cpp @@ -1661,7 +1661,8 @@ aggregate_reader_metadata::select_columns( bool ignore_missing_columns, type_id timestamp_type_id, type_id decimal_type_id, - bool case_sensitive_names) + bool case_sensitive_names, + bool match_schema_by_field_id) { auto const find_schema_child = [&](SchemaElement const& schema_elem, std::string_view name, int const pfm_idx = 0) { @@ -1678,6 +1679,37 @@ aggregate_reader_metadata::select_columns( : -1; }; + auto const find_schema_child_by_field_id = + [&](SchemaElement const& schema_elem, int32_t field_id, int const pfm_idx = 0) { + auto const& col_schema_idx = std::find_if( + schema_elem.children_idx.cbegin(), + schema_elem.children_idx.cend(), + [&](size_t col_schema_idx) { + auto const& child_schema = get_schema(col_schema_idx, pfm_idx); + return child_schema.field_id.has_value() and child_schema.field_id.value() == field_id; + }); + + return (col_schema_idx != schema_elem.children_idx.end()) + ? static_cast(*col_schema_idx) + : -1; + }; + + auto const find_schema_child_for_mapping = [&](SchemaElement const& src_schema_elem, + SchemaElement const& dst_schema_elem, + std::string_view name, + int const pfm_idx) { + auto const src_child_idx = find_schema_child(src_schema_elem, name); + if (match_schema_by_field_id and src_child_idx != -1) { + auto const& src_child = get_schema(src_child_idx); + if (src_child.field_id.has_value()) { + auto const dst_child_idx = + find_schema_child_by_field_id(dst_schema_elem, src_child.field_id.value(), pfm_idx); + if (dst_child_idx != -1) { return dst_child_idx; } + } + } + return find_schema_child(dst_schema_elem, name, pfm_idx); + }; + std::vector output_columns; std::vector input_columns; std::vector nesting; @@ -1789,9 +1821,15 @@ aggregate_reader_metadata::select_columns( }; // Compares two schema elements to be equal except their number of children - auto const equal_to_except_num_children = [](SchemaElement const& lhs, SchemaElement const& rhs) { + auto const equal_to_except_num_children = [match_schema_by_field_id](SchemaElement const& lhs, + SchemaElement const& rhs) { + // Match by field ID if enabled, otherwise match by name + auto const names_match = + (match_schema_by_field_id and lhs.field_id.has_value() and rhs.field_id.has_value()) + ? lhs.field_id == rhs.field_id + : lhs.name == rhs.name; return lhs.type == rhs.type and lhs.converted_type == rhs.converted_type and - lhs.type_length == rhs.type_length and lhs.name == rhs.name and + lhs.type_length == rhs.type_length and names_match and lhs.decimal_scale == rhs.decimal_scale and lhs.decimal_precision == rhs.decimal_precision and lhs.field_id == rhs.field_id; }; @@ -1863,14 +1901,13 @@ aggregate_reader_metadata::select_columns( [&](auto const& child_col_name_info) { // Ensure that each named child column exists in the destination schema tree for the // paths to align up. An out_of_range error otherwise. - CUDF_EXPECTS( - find_schema_child(dst_schema_elem, child_col_name_info.name, pfm_idx) != -1, - "Encountered mismatching schema tree depths across data sources", - std::out_of_range); - map_column(&child_col_name_info, - find_schema_child(src_schema_elem, child_col_name_info.name), - find_schema_child(dst_schema_elem, child_col_name_info.name, pfm_idx), - pfm_idx); + auto const src_child_idx = find_schema_child(src_schema_elem, child_col_name_info.name); + auto const dst_child_idx = find_schema_child_for_mapping( + src_schema_elem, dst_schema_elem, child_col_name_info.name, pfm_idx); + CUDF_EXPECTS(dst_child_idx != -1, + "Encountered mismatching schema tree depths across data sources", + std::out_of_range); + map_column(&child_col_name_info, src_child_idx, dst_child_idx, pfm_idx); }); } }; @@ -2027,14 +2064,13 @@ aggregate_reader_metadata::select_columns( auto const& dst_root = get_schema(0, pfm_idx); // Ensure that each top level column exists in the destination schema // tree. An out_of_range error is thrown otherwise. + auto const dst_col_schema_idx = + find_schema_child_for_mapping(root, dst_root, col.name, pfm_idx); CUDF_EXPECTS( - find_schema_child(dst_root, col.name, pfm_idx) != -1, + dst_col_schema_idx != -1, "Encountered mismatching schema tree depths across data sources", std::out_of_range); - map_column(&col, - top_level_col_schema_idx, - find_schema_child(dst_root, col.name, pfm_idx), - pfm_idx); + map_column(&col, top_level_col_schema_idx, dst_col_schema_idx, pfm_idx); }); } } diff --git a/cpp/src/io/parquet/reader_impl_helpers.hpp b/cpp/src/io/parquet/reader_impl_helpers.hpp index c9c42a1bd52c..eb049095f6d8 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.hpp +++ b/cpp/src/io/parquet/reader_impl_helpers.hpp @@ -602,6 +602,9 @@ class aggregate_reader_metadata { * @param ignore_missing_columns Whether to ignore non-existent projected columns * @param timestamp_type_id Type conversion parameter * @param decimal_type_id Type conversion parameter + * @param case_sensitive_names Whether column name matching is case sensitive + * @param match_schema_by_field_id Whether multi-source schema matching should use Parquet field + * IDs * * @return input column information, output column buffers, list of output column schema * indices @@ -616,7 +619,8 @@ class aggregate_reader_metadata { bool ignore_missing_columns, type_id timestamp_type_id, type_id decimal_type_id, - bool case_sensitive_names); + bool case_sensitive_names, + bool match_schema_by_field_id); }; } // namespace cudf::io::parquet::detail diff --git a/cpp/tests/io/parquet_common.cpp b/cpp/tests/io/parquet_common.cpp index 3f4696480ce5..a531f7f15cff 100644 --- a/cpp/tests/io/parquet_common.cpp +++ b/cpp/tests/io/parquet_common.cpp @@ -27,12 +27,16 @@ cudf::test::TempDirTestEnvironment* const temp_env = std::string write_parquet_temp_file(cudf::table_view const& tbl, std::string_view const filename, - std::vector const& column_names) + std::vector column_names, + std::vector field_ids) { cudf::io::table_input_metadata md{tbl}; for (std::size_t i = 0; i < column_names.size(); ++i) { md.column_metadata[i].set_name(column_names[i]); } + for (std::size_t i = 0; i < field_ids.size(); ++i) { + md.column_metadata[i].set_parquet_field_id(field_ids[i]); + } auto const path = temp_env->get_temp_filepath(std::string{filename}); cudf::io::parquet_writer_options opts = cudf::io::parquet_writer_options::builder(cudf::io::sink_info{path}, tbl) diff --git a/cpp/tests/io/parquet_common.hpp b/cpp/tests/io/parquet_common.hpp index 0eca6b0605c0..415b13964140 100644 --- a/cpp/tests/io/parquet_common.hpp +++ b/cpp/tests/io/parquet_common.hpp @@ -36,10 +36,10 @@ extern cudf::test::TempDirTestEnvironment* const temp_env; // Writes `tbl` to a temp Parquet file. If `column_names` is non-empty, it sets the top-level // column names in the file metadata. -[[nodiscard]] std::string write_parquet_temp_file( - cudf::table_view const& tbl, - std::string_view const filename, - std::vector const& column_names = {}); +[[nodiscard]] std::string write_parquet_temp_file(cudf::table_view const& tbl, + std::string_view const filename, + std::vector column_names = {}, + std::vector field_ids = {}); // TODO: Replace with `NumericTypes` when unsigned support is added. Issue #5352 using SupportedTypes = cudf::test::Types; diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index 8276fea5ecb7..9617cf179038 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -378,7 +378,20 @@ TEST_F(ParquetReaderTest, ReorderedColumns) auto d = cudf::test::strings_column_wrapper{"ducks", "sheep", "cows", "fish", "birds", "ants"}; cudf::table_view tbl{{a, b, c, d}}; - auto filepath = write_parquet_temp_file(tbl, "ReorderedColumns3.parquet", {"a", "b", "c", "d"}); + auto filepath = temp_env->get_temp_filepath("ReorderedColumns3.parquet"); + cudf::io::table_input_metadata md(tbl); + md.column_metadata[0].set_name("a"); + md.column_metadata[0].set_parquet_field_id(10); + md.column_metadata[1].set_name("b"); + md.column_metadata[1].set_parquet_field_id(11); + md.column_metadata[2].set_name("c"); + md.column_metadata[2].set_parquet_field_id(12); + md.column_metadata[3].set_name("d"); + md.column_metadata[3].set_parquet_field_id(13); + cudf::io::parquet_writer_options opts = + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, tbl) + .metadata(std::move(md)); + cudf::io::write_parquet(opts); { // read them out of order using indices @@ -393,6 +406,29 @@ TEST_F(ParquetReaderTest, ReorderedColumns) CUDF_TEST_EXPECT_COLUMNS_EQUAL(result.tbl->view().column(3), c); } + { + // read them out of order using Parquet field IDs + cudf::io::parquet_reader_options read_opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}) + .column_field_ids({13, 10, 11, 12}); + auto result = cudf::io::read_parquet(read_opts); + + CUDF_TEST_EXPECT_COLUMNS_EQUAL(result.tbl->view().column(0), d); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(result.tbl->view().column(1), a); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(result.tbl->view().column(2), b); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(result.tbl->view().column(3), c); + } + + { + // missing Parquet field IDs are errors when ignore_missing_columns is disabled + cudf::io::parquet_reader_options read_opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}) + .column_field_ids({999}) + .ignore_missing_columns(false); + + EXPECT_THROW(cudf::io::read_parquet(read_opts), std::invalid_argument); + } + { // read them out of order cudf::io::parquet_reader_options read_opts = @@ -462,10 +498,15 @@ TEST_F(ParquetReaderTest, SelectNestedColumn) cudf::io::table_input_metadata input_metadata(input); input_metadata.column_metadata[0].set_name("being"); + input_metadata.column_metadata[0].set_parquet_field_id(1); input_metadata.column_metadata[0].child(0).set_name("human?"); + input_metadata.column_metadata[0].child(0).set_parquet_field_id(2); input_metadata.column_metadata[0].child(1).set_name("particulars"); + input_metadata.column_metadata[0].child(1).set_parquet_field_id(3); input_metadata.column_metadata[0].child(1).child(0).set_name("weight"); + input_metadata.column_metadata[0].child(1).child(0).set_parquet_field_id(4); input_metadata.column_metadata[0].child(1).child(1).set_name("age"); + input_metadata.column_metadata[0].child(1).child(1).set_parquet_field_id(5); auto filepath = temp_env->get_temp_filepath("SelectNestedColumn.parquet"); cudf::io::parquet_writer_options args = @@ -474,15 +515,8 @@ TEST_F(ParquetReaderTest, SelectNestedColumn) cudf::io::write_parquet(args); { // Test selecting a single leaf from the table - cudf::io::parquet_reader_options read_args = - cudf::io::parquet_reader_options::builder(cudf::io::source_info(filepath)) - .column_names({"being.particulars.age"}); - auto const result = cudf::io::read_parquet(read_args); - - auto expect_ages_col = cudf::test::fixed_width_column_wrapper{ - {48, 27, 25, 31, 351, 351}, {true, true, true, true, true, false}}; auto expect_s_1 = - cudf::test::structs_column_wrapper{{expect_ages_col}, {true, true, true, true, false, true}}; + cudf::test::structs_column_wrapper{{ages_col}, {true, true, true, true, false, true}}; auto expect_s_2 = cudf::test::structs_column_wrapper{{expect_s_1}, {false, true, true, true, true, true}} .release(); @@ -493,24 +527,24 @@ TEST_F(ParquetReaderTest, SelectNestedColumn) expected_metadata.column_metadata[0].child(0).set_name("particulars"); expected_metadata.column_metadata[0].child(0).child(0).set_name("age"); + cudf::io::parquet_reader_options read_args = + cudf::io::parquet_reader_options::builder(cudf::io::source_info(filepath)) + .column_names({"being.particulars.age"}); + auto result = cudf::io::read_parquet(read_args); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + cudf::test::expect_metadata_equal(expected_metadata, result.metadata); + + // Test selecting a single leaf by Parquet field ID + read_args = cudf::io::parquet_reader_options::builder(cudf::io::source_info(filepath)) + .column_field_ids({5}); + result = cudf::io::read_parquet(read_args); CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); cudf::test::expect_metadata_equal(expected_metadata, result.metadata); } { // Test selecting a non-leaf and expecting all hierarchy from that node onwards - cudf::io::parquet_reader_options read_args = - cudf::io::parquet_reader_options::builder(cudf::io::source_info(filepath)) - .column_names({"being.particulars"}); - auto const result = cudf::io::read_parquet(read_args); - - auto expected_weights_col = - cudf::test::fixed_width_column_wrapper{1.1, 2.4, 5.3, 8.0, 9.6, 6.9}; - - auto expected_ages_col = cudf::test::fixed_width_column_wrapper{ - {48, 27, 25, 31, 351, 351}, {true, true, true, true, true, false}}; - - auto expected_s_1 = cudf::test::structs_column_wrapper{ - {expected_weights_col, expected_ages_col}, {true, true, true, true, false, true}}; + auto expected_s_1 = cudf::test::structs_column_wrapper{{weights_col, ages_col}, + {true, true, true, true, false, true}}; auto expect_s_2 = cudf::test::structs_column_wrapper{{expected_s_1}, {false, true, true, true, true, true}} @@ -523,6 +557,17 @@ TEST_F(ParquetReaderTest, SelectNestedColumn) expected_metadata.column_metadata[0].child(0).child(0).set_name("weight"); expected_metadata.column_metadata[0].child(0).child(1).set_name("age"); + cudf::io::parquet_reader_options read_args = + cudf::io::parquet_reader_options::builder(cudf::io::source_info(filepath)) + .column_names({"being.particulars"}); + auto result = cudf::io::read_parquet(read_args); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + cudf::test::expect_metadata_equal(expected_metadata, result.metadata); + + // Test selecting a non-leaf by Parquet field ID + read_args = cudf::io::parquet_reader_options::builder(cudf::io::source_info(filepath)) + .column_field_ids({3}); + result = cudf::io::read_parquet(read_args); CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); cudf::test::expect_metadata_equal(expected_metadata, result.metadata); } @@ -533,19 +578,10 @@ TEST_F(ParquetReaderTest, SelectNestedColumn) .column_names({"being.particulars.age", "being.particulars.weight", "being.human?"}); auto const result = cudf::io::read_parquet(read_args); - auto expected_weights_col = - cudf::test::fixed_width_column_wrapper{1.1, 2.4, 5.3, 8.0, 9.6, 6.9}; - - auto expected_ages_col = cudf::test::fixed_width_column_wrapper{ - {48, 27, 25, 31, 351, 351}, {true, true, true, true, true, false}}; - - auto expected_is_human_col = cudf::test::fixed_width_column_wrapper{ - {true, true, false, false, false, false}, {true, true, false, true, true, false}}; - - auto expect_s_1 = cudf::test::structs_column_wrapper{{expected_ages_col, expected_weights_col}, + auto expect_s_1 = cudf::test::structs_column_wrapper{{ages_col, weights_col}, {true, true, true, true, false, true}}; - auto expect_s_2 = cudf::test::structs_column_wrapper{{expect_s_1, expected_is_human_col}, + auto expect_s_2 = cudf::test::structs_column_wrapper{{expect_s_1, is_human_col}, {false, true, true, true, true, true}} .release(); @@ -1416,8 +1452,11 @@ auto create_parquet_with_stats(std::string const& filename) cudf::io::table_input_metadata expected_metadata(expected); expected_metadata.column_metadata[0].set_name("col_uint32"); + expected_metadata.column_metadata[0].set_parquet_field_id(10); expected_metadata.column_metadata[1].set_name("col_int64"); + expected_metadata.column_metadata[1].set_parquet_field_id(11); expected_metadata.column_metadata[2].set_name("col_double"); + expected_metadata.column_metadata[2].set_parquet_field_id(12); auto const filepath = temp_env->get_temp_filepath(filename); const cudf::io::parquet_writer_options out_opts = @@ -1458,7 +1497,7 @@ TEST_F(ParquetReaderTest, FilterIdentity) TEST_F(ParquetReaderTest, FilterWithColumnProjection) { - // col_uint32, col_int64, col_double + // col_uint32 (field_id: 10), col_int64 (field_id: 11), col_double (field_id: 12) auto [src, filepath] = create_parquet_with_stats("FilterWithColumnProjection.parquet"); auto val = cudf::numeric_scalar{10}; auto lit = cudf::ast::literal{val}; @@ -1488,6 +1527,14 @@ TEST_F(ParquetReaderTest, FilterWithColumnProjection) .filter(read_expr); result = cudf::io::read_parquet(read_opts); CUDF_TEST_EXPECT_TABLES_EQUAL(*result.tbl, *expected); + + // Repeat but select columns using field IDs instead of names + read_opts = cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}) + .column_field_ids({12}) + .case_sensitive_names(false) + .filter(read_expr); + result = cudf::io::read_parquet(read_opts); + CUDF_TEST_EXPECT_TABLES_EQUAL(*result.tbl, *expected); } { // column_reference in parquet filter (indices as per order of column projection) @@ -1508,6 +1555,12 @@ TEST_F(ParquetReaderTest, FilterWithColumnProjection) .column_indices({2, 0}) .filter(read_ref_expr); CUDF_TEST_EXPECT_TABLES_EQUAL(*(cudf::io::read_parquet(read_opts).tbl), *expected); + + // Repeat but select columns using field IDs instead of names + read_opts = cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}) + .column_field_ids({12, 10}) + .filter(read_ref_expr); + CUDF_TEST_EXPECT_TABLES_EQUAL(*(cudf::io::read_parquet(read_opts).tbl), *expected); } // Error cases @@ -1526,6 +1579,12 @@ TEST_F(ParquetReaderTest, FilterWithColumnProjection) .column_indices({2, 0}) .filter(read_ref_expr); EXPECT_ANY_THROW(cudf::io::read_parquet(read_opts)); + + // Repeat but select columns using field IDs instead of names + read_opts = cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath}) + .column_field_ids({12, 10}) + .filter(read_ref_expr); + EXPECT_ANY_THROW(cudf::io::read_parquet(read_opts)); } } } @@ -4787,63 +4846,92 @@ TEST_F(ParquetReaderTest, MismatchedSchemaFilterColumnCollision) auto const price_a = column_wrapper{50.0, 150.0, 75.0}; cudf::table_view const table_a{{id_a, price_a}}; auto const path_a = - write_parquet_temp_file(table_a, "MismatchCollisionA.parquet", {"id", "price"}); + write_parquet_temp_file(table_a, "MismatchCollisionA.parquet", {"id", "price"}, {1, 2}); auto const category_b = column_wrapper{"x", "y", "z"}; auto const id_b = column_wrapper{1000, 1001, 1002}; auto const price_b = column_wrapper{40.0, 200.0, 99.0}; cudf::table_view const table_b{{category_b, id_b, price_b}}; - auto const path_b = - write_parquet_temp_file(table_b, "MismatchCollisionB.parquet", {"category", "id", "price"}); + auto const path_b = write_parquet_temp_file( + table_b, "MismatchCollisionB.parquet", {"category", "id", "price"}, {3, 1, 2}); auto value = cudf::numeric_scalar(100.0); auto lit = cudf::ast::literal(value); auto col = cudf::ast::column_name_reference("price"); auto filter = cudf::ast::operation(cudf::ast::ast_operator::LESS, col, lit); - auto const opts = - cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) - .allow_mismatched_pq_schemas(true) - .column_names({"id", "price"}) - .filter(filter) - .build(); auto const exp_id = column_wrapper{1, 3, 1000, 1002}; auto const exp_price = column_wrapper{50.0, 75.0, 40.0, 99.0}; cudf::table_view const expected{{exp_id, exp_price}}; - auto const result = cudf::io::read_parquet(opts); - CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + + { + auto const opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) + .allow_mismatched_pq_schemas(true) + .column_names({"id", "price"}) + .filter(filter) + .build(); + auto const result = cudf::io::read_parquet(opts); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + } + { + auto const opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) + .allow_mismatched_pq_schemas(true) + .column_field_ids({1, 2}) + .filter(filter) + .build(); + auto const result = cudf::io::read_parquet(opts); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + } } TEST_F(ParquetReaderTest, MismatchedSchemaFilterSameTypeCollisionWrongResult) { - // Same-type collision: source 0's int64 `a` lands on source 1's int64 `b`. + // Same-type collision: source 0's int64 `a` (field_id: 1) lands on source 1's int64 `b` + // (field_id: 2). auto const a_a = column_wrapper{1, 2, 3}; auto const b_a = column_wrapper{7, 8, 9}; cudf::table_view const table_a{{a_a, b_a}}; - auto const path_a = write_parquet_temp_file(table_a, "MismatchSameTypeA.parquet", {"a", "b"}); + auto const path_a = + write_parquet_temp_file(table_a, "MismatchSameTypeA.parquet", {"a", "b"}, {1, 2}); auto const b_b = column_wrapper{1000, 2000, 3000}; // B's `b` lands at a's index auto const a_b = column_wrapper{10, 20, 30}; // B's `a` (all < 100) cudf::table_view const table_b{{b_b, a_b}}; - auto const path_b = write_parquet_temp_file(table_b, "MismatchSameTypeB.parquet", {"b", "a"}); + auto const path_b = + write_parquet_temp_file(table_b, "MismatchSameTypeB.parquet", {"b", "a"}, {2, 1}); auto value = cudf::numeric_scalar(100); auto lit = cudf::ast::literal(value); auto col = cudf::ast::column_name_reference("a"); auto filter = cudf::ast::operation(cudf::ast::ast_operator::LESS, col, lit); - auto const opts = - cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) - .allow_mismatched_pq_schemas(true) - .column_names({"a", "b"}) - .filter(filter) - .build(); // All rows have a < 100, so none should be pruned. auto const exp_a = column_wrapper{1, 2, 3, 10, 20, 30}; auto const exp_b = column_wrapper{7, 8, 9, 1000, 2000, 3000}; cudf::table_view const expected{{exp_a, exp_b}}; - auto const result = cudf::io::read_parquet(opts); - CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + + { + auto const opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) + .allow_mismatched_pq_schemas(true) + .column_names({"a", "b"}) + .filter(filter) + .build(); + auto const result = cudf::io::read_parquet(opts); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + } + { + auto const opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) + .allow_mismatched_pq_schemas(true) + .column_field_ids({1, 2}) + .filter(filter) + .build(); + auto const result = cudf::io::read_parquet(opts); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + } } TEST_F(ParquetReaderTest, MismatchedSchemaFilterOnlyColumnCollision) @@ -4853,28 +4941,41 @@ TEST_F(ParquetReaderTest, MismatchedSchemaFilterOnlyColumnCollision) auto const price_a = column_wrapper{10.0, 200.0, 30.0}; cudf::table_view const table_a{{id_a, price_a}}; auto const path_a = - write_parquet_temp_file(table_a, "MismatchFilterOnlyA.parquet", {"id", "price"}); + write_parquet_temp_file(table_a, "MismatchFilterOnlyA.parquet", {"id", "price"}, {10, 11}); auto const category_b = column_wrapper{"x", "y", "z"}; auto const id_b = column_wrapper{1000, 1001, 1002}; auto const price_b = column_wrapper{40.0, 500.0, 60.0}; cudf::table_view const table_b{{category_b, id_b, price_b}}; - auto const path_b = - write_parquet_temp_file(table_b, "MismatchFilterOnlyB.parquet", {"category", "id", "price"}); + auto const path_b = write_parquet_temp_file( + table_b, "MismatchFilterOnlyB.parquet", {"category", "id", "price"}, {12, 10, 11}); auto value = cudf::numeric_scalar(100.0); auto lit = cudf::ast::literal(value); auto col = cudf::ast::column_name_reference("price"); auto filter = cudf::ast::operation(cudf::ast::ast_operator::LESS, col, lit); - auto const opts = - cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) - .allow_mismatched_pq_schemas(true) - .column_names({"id"}) // `price` is filter-only - .filter(filter) - .build(); auto const exp_id = column_wrapper{1, 3, 1000, 1002}; cudf::table_view const expected{{exp_id}}; - auto const result = cudf::io::read_parquet(opts); - CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + + { + auto const opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) + .allow_mismatched_pq_schemas(true) + .column_names({"id"}) // `price` is filter-only + .filter(filter) + .build(); + auto const result = cudf::io::read_parquet(opts); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + } + { + auto const opts = + cudf::io::parquet_reader_options::builder(cudf::io::source_info{{path_a, path_b}}) + .allow_mismatched_pq_schemas(true) + .column_field_ids({10}) // `price` is filter-only + .filter(filter) + .build(); + auto const result = cudf::io::read_parquet(opts); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, result.tbl->view()); + } } From d74fd4da7294be5c3d8194d7795ac45db249e2f0 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com> Date: Wed, 24 Jun 2026 00:22:08 +0000 Subject: [PATCH 2/6] Python bindings for Parquet column selection by field ID --- python/pylibcudf/pylibcudf/io/parquet.pxd | 2 + python/pylibcudf/pylibcudf/io/parquet.pyi | 5 ++ python/pylibcudf/pylibcudf/io/parquet.pyx | 43 +++++++++++++++-- .../pylibcudf/libcudf/io/parquet.pxd | 8 +++- python/pylibcudf/tests/io/test_parquet.py | 47 +++++++++++++++++++ 5 files changed, 101 insertions(+), 4 deletions(-) diff --git a/python/pylibcudf/pylibcudf/io/parquet.pxd b/python/pylibcudf/pylibcudf/io/parquet.pxd index fd3cc3f16676..a1e0c8e33dfd 100644 --- a/python/pylibcudf/pylibcudf/io/parquet.pxd +++ b/python/pylibcudf/pylibcudf/io/parquet.pxd @@ -49,6 +49,7 @@ cdef class ParquetReaderOptions: cpdef void set_columns(self, list col_names) cpdef void set_column_names(self, list col_names) cpdef void set_column_indices(self, list col_indices) + cpdef void set_column_field_ids(self, list column_field_ids) cpdef void set_filter(self, Expression filter) cpdef void set_source(self, SourceInfo src) cpdef bool is_enabled_use_jit_filter(self) @@ -68,6 +69,7 @@ cdef class ParquetReaderOptionsBuilder: cpdef ParquetReaderOptionsBuilder columns(self, list col_names) cpdef ParquetReaderOptionsBuilder column_names(self, list col_names) cpdef ParquetReaderOptionsBuilder column_indices(self, list col_indices) + cpdef ParquetReaderOptionsBuilder column_field_ids(self, list column_field_ids) cpdef ParquetReaderOptionsBuilder use_jit_filter(self, bool use_jit_filter) cpdef ParquetReaderOptionsBuilder case_sensitive_names(self, bool val) cpdef ParquetReaderOptionsBuilder decimal_width(self, type_id width) diff --git a/python/pylibcudf/pylibcudf/io/parquet.pyi b/python/pylibcudf/pylibcudf/io/parquet.pyi index a79e9cf3f5b1..d43d6f308028 100644 --- a/python/pylibcudf/pylibcudf/io/parquet.pyi +++ b/python/pylibcudf/pylibcudf/io/parquet.pyi @@ -30,6 +30,7 @@ class ParquetReaderOptions: def set_columns(self, col_names: list[str]): ... def set_column_names(self, col_names: list[str]): ... def set_column_indices(self, col_indices: list[int]): ... + def set_column_field_ids(self, column_field_ids: list[int]): ... def set_filter(self, filter: Expression): ... def set_source(self, src: SourceInfo) -> None: ... def is_enabled_use_jit_filter(self) -> bool: ... @@ -45,6 +46,10 @@ class ParquetReaderOptionsBuilder: def allow_mismatched_pq_schemas(self, val: bool) -> Self: ... def ignore_missing_columns(self, val: bool) -> Self: ... def use_arrow_schema(self, val: bool) -> Self: ... + def columns(self, col_names: list[str]) -> Self: ... + def column_names(self, col_names: list[str]) -> Self: ... + def column_indices(self, col_indices: list[int]) -> Self: ... + def column_field_ids(self, column_field_ids: list[int]) -> Self: ... def use_jit_filter(self, use_jit_filter: bool) -> Self: ... def case_sensitive_names(self, val: bool) -> Self: ... def decimal_width(self, width: TypeId) -> Self: ... diff --git a/python/pylibcudf/pylibcudf/io/parquet.pyx b/python/pylibcudf/pylibcudf/io/parquet.pyx index d43d956960f1..d782c9cf3305 100644 --- a/python/pylibcudf/pylibcudf/io/parquet.pyx +++ b/python/pylibcudf/pylibcudf/io/parquet.pyx @@ -3,7 +3,7 @@ from cython.operator cimport dereference import warnings -from libc.stdint cimport int64_t, uint8_t +from libc.stdint cimport int32_t, int64_t, uint8_t from libcpp cimport bool from libcpp.memory cimport unique_ptr, make_unique @@ -250,7 +250,7 @@ cdef class ParquetReaderOptions: Parameters ---------- - col_names : list + col_indices : list List of top-level column indices Returns @@ -262,6 +262,24 @@ cdef class ParquetReaderOptions: vec.push_back(idx) self.c_obj.set_column_indices(vec) + cpdef void set_column_field_ids(self, list column_field_ids): + """ + Sets Parquet field IDs of the columns/fields to be read. + + Parameters + ---------- + column_field_ids : list + List of Parquet field IDs + + Returns + ------- + None + """ + cdef vector[int32_t] vec + for field_id in column_field_ids: + vec.push_back(field_id) + self.c_obj.set_column_field_ids(vec) + cpdef void set_filter(self, Expression filter): """ Sets AST based filter for predicate pushdown. @@ -464,7 +482,7 @@ cdef class ParquetReaderOptionsBuilder: Parameters ---------- - col_names : list[int] + col_indices : list[int] List of top-level column indices Returns @@ -477,6 +495,25 @@ cdef class ParquetReaderOptionsBuilder: self.c_obj.column_indices(vec) return self + cpdef ParquetReaderOptionsBuilder column_field_ids(self, list column_field_ids): + """ + Sets Parquet field IDs of the columns/fields to be read. + + Parameters + ---------- + column_field_ids : list[int] + List of Parquet field IDs + + Returns + ------- + ParquetReaderOptionsBuilder + """ + cdef vector[int32_t] vec + for field_id in column_field_ids: + vec.push_back(field_id) + self.c_obj.column_field_ids(vec) + return self + cpdef ParquetReaderOptionsBuilder use_jit_filter(self, bool use_jit_filter): """ Sets whether to use JIT compilation for filtering. diff --git a/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd b/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd index a6a1ca3e0f91..d0e23cedad79 100644 --- a/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd +++ b/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: Copyright (c) 2020-2026, NVIDIA CORPORATION. # SPDX-License-Identifier: Apache-2.0 -from libc.stdint cimport int64_t, uint8_t +from libc.stdint cimport int32_t, int64_t, uint8_t from libcpp cimport bool from libcpp.functional cimport reference_wrapper from libcpp.map cimport map @@ -53,6 +53,9 @@ cdef extern from "cudf/io/parquet.hpp" namespace "cudf::io" nogil: void set_column_indices( vector[size_type] col_indices ) except +libcudf_exception_handler + void set_column_field_ids( + vector[int32_t] column_field_ids + ) except +libcudf_exception_handler void set_num_rows(int64_t val) except +libcudf_exception_handler void set_row_groups( vector[vector[size_type]] row_grp @@ -89,6 +92,9 @@ cdef extern from "cudf/io/parquet.hpp" namespace "cudf::io" nogil: parquet_reader_options_builder& column_indices( vector[size_type] col_indices ) except +libcudf_exception_handler + parquet_reader_options_builder& column_field_ids( + vector[int32_t] column_field_ids + ) except +libcudf_exception_handler parquet_reader_options_builder& row_groups( vector[vector[size_type]] row_grp ) except +libcudf_exception_handler diff --git a/python/pylibcudf/tests/io/test_parquet.py b/python/pylibcudf/tests/io/test_parquet.py index 88ba587d8e88..45ead06214a3 100644 --- a/python/pylibcudf/tests/io/test_parquet.py +++ b/python/pylibcudf/tests/io/test_parquet.py @@ -106,6 +106,53 @@ def test_read_parquet_basic( assert res.num_row_groups_after_bloom_filter is None +def test_read_parquet_column_field_ids(binary_source_or_sink): + schema = pa.schema( + [ + pa.field( + "col_int64", + pa.int64(), + metadata={b"PARQUET:field_id": b"10"}, + ), + pa.field( + "col_string", + pa.string(), + metadata={b"PARQUET:field_id": b"20"}, + ), + pa.field( + "col_bool", + pa.bool_(), + metadata={b"PARQUET:field_id": b"30"}, + ), + ] + ) + pa_table = pa.Table.from_arrays( + [ + pa.array([1, 2, 3], type=pa.int64()), + pa.array(["a", "b", "c"], type=pa.string()), + pa.array([True, False, True], type=pa.bool_()), + ], + schema=schema, + ) + source = make_source( + binary_source_or_sink, pa_table, **_COMMON_PARQUET_SOURCE_KWARGS + ) + source_info = plc.io.SourceInfo([source]) + options = ( + plc.io.parquet.ParquetReaderOptions.builder(source_info) + .column_field_ids([30, 10]) + .build() + ) + + res = plc.io.parquet.read_parquet(options) + + assert_table_and_meta_eq( + pa_table.select(["col_bool", "col_int64"]), + res, + check_field_nullability=False, + ) + + @pytest.mark.parametrize("if_prune_rowgroup,result", [(True, 0), (False, 1)]) def test_read_parquet_filters_metadata(tmp_path, if_prune_rowgroup, result): col_list = list(range(1, 10)) From febd9c0e62cb693bd8d9f0523be1ee82645591d3 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb Date: Mon, 20 Jul 2026 17:32:10 +0000 Subject: [PATCH 3/6] Merge conflicts --- .gitignore | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/.gitignore b/.gitignore index 180a6a286e2a..1ef218c80962 100644 --- a/.gitignore +++ b/.gitignore @@ -102,6 +102,12 @@ cpp/doxygen/xml #Java target +# Local Maven repo created by java/ci/build-in-docker.sh +.m2/ + +# Recreated on every Maven `validate` phase by gmaven-plugin:1.5 +java/bin/ + # Translations *.mo *.pot @@ -181,3 +187,7 @@ compile_commands.json # pytest artifacts rmm_log.txt python/cudf/cudf_pandas_tests/data/rmm_log.txt + +# Quent traces +logs/*.ndjson + From a0f8ca372a5d27684fac80531446e743fcbfff7c Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb Date: Mon, 20 Jul 2026 17:32:41 +0000 Subject: [PATCH 4/6] Merge conflicts --- .gitignore | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index 1ef218c80962..6413dc27f22d 100644 --- a/.gitignore +++ b/.gitignore @@ -189,5 +189,4 @@ rmm_log.txt python/cudf/cudf_pandas_tests/data/rmm_log.txt # Quent traces -logs/*.ndjson - +logs/*.ndjson \ No newline at end of file From ce63c86824a1b36ad8ab0a7ba965e8b561f9e2e4 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb <14217455+mhaseeb123@users.noreply.github.com> Date: Mon, 20 Jul 2026 10:32:59 -0700 Subject: [PATCH 5/6] Discard changes to .gitignore --- .gitignore | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 6413dc27f22d..cb4b70da0503 100644 --- a/.gitignore +++ b/.gitignore @@ -189,4 +189,4 @@ rmm_log.txt python/cudf/cudf_pandas_tests/data/rmm_log.txt # Quent traces -logs/*.ndjson \ No newline at end of file +logs/*.ndjson From ceb429655923fdd24bbebb3699fe2e281ca22e13 Mon Sep 17 00:00:00 2001 From: Muhammad Haseeb Date: Mon, 20 Jul 2026 17:48:31 +0000 Subject: [PATCH 6/6] Copyright headers --- python/pylibcudf/pylibcudf/io/parquet.pxd | 2 +- python/pylibcudf/pylibcudf/io/parquet.pyi | 2 +- python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd | 2 +- python/pylibcudf/tests/io/test_parquet.py | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/python/pylibcudf/pylibcudf/io/parquet.pxd b/python/pylibcudf/pylibcudf/io/parquet.pxd index a1e0c8e33dfd..58b186647286 100644 --- a/python/pylibcudf/pylibcudf/io/parquet.pxd +++ b/python/pylibcudf/pylibcudf/io/parquet.pxd @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION. +# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 from libc.stdint cimport int64_t, uint8_t diff --git a/python/pylibcudf/pylibcudf/io/parquet.pyi b/python/pylibcudf/pylibcudf/io/parquet.pyi index d43d6f308028..8564b270e583 100644 --- a/python/pylibcudf/pylibcudf/io/parquet.pyi +++ b/python/pylibcudf/pylibcudf/io/parquet.pyi @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION. +# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 from collections.abc import Mapping, Sequence diff --git a/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd b/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd index d0e23cedad79..0f5a531dea8a 100644 --- a/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd +++ b/python/pylibcudf/pylibcudf/libcudf/io/parquet.pxd @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: Copyright (c) 2020-2026, NVIDIA CORPORATION. +# SPDX-FileCopyrightText: Copyright (c) 2020-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 from libc.stdint cimport int32_t, int64_t, uint8_t from libcpp cimport bool diff --git a/python/pylibcudf/tests/io/test_parquet.py b/python/pylibcudf/tests/io/test_parquet.py index 45ead06214a3..a2ba752dbba1 100644 --- a/python/pylibcudf/tests/io/test_parquet.py +++ b/python/pylibcudf/tests/io/test_parquet.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION. +# SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 import io import os