diff --git a/cpp/include/cudf/io/parquet_metadata.hpp b/cpp/include/cudf/io/parquet_metadata.hpp index c6c0e6f4e376..23a1356f402b 100644 --- a/cpp/include/cudf/io/parquet_metadata.hpp +++ b/cpp/include/cudf/io/parquet_metadata.hpp @@ -10,12 +10,18 @@ #pragma once +#include #include #include #include +#include +#include #include +#include +#include #include +#include #include #include @@ -294,6 +300,49 @@ parquet_metadata read_parquet_metadata(source_info const& src_info); std::vector read_parquet_footers( std::span const> sources); +/** + * @brief Min/max bounds decoded from parquet column-chunk statistics. + * + * ``file_indices`` and ``row_group_indices`` identify the column chunks represented by each row of + * every table in ``bounds``. Each table in ``bounds`` corresponds positionally to a requested + * column and contains exactly two columns: decoded minimum values followed by decoded maximum + * values. + */ +struct column_chunk_bounds_result { + /// File index for each row in every bounds table + std::unique_ptr file_indices; + /// File-local row-group index for each row in every bounds table + std::unique_ptr row_group_indices; + /// One two-column table per requested column, where column 0 is min and column 1 is max + std::vector> bounds; +}; + +/** + * @brief Decode parquet column-chunk min/max statistics for selected leaf columns. + * + * Missing min/max statistics are represented as nulls in the corresponding output column. Parquet + * min/max exactness flags are not interpreted by this function. The requested column names are + * resolved against each file's schema. + * + * @ingroup io_readers + * + * @param parquet_metadatas Parquet file metadata, one per source + * @param column_names Dotted leaf-column paths to decode statistics for + * @param stream CUDA stream used for device memory operations + * @param mr Device memory resource to use for device memory allocation + * @return Decoded min/max bounds and row-group identifiers + * + * @throw std::invalid_argument If a requested leaf-column path is missing or ambiguous. + * @throw std::invalid_argument If a requested column has unsupported or compound statistics dtype. + * @throw std::invalid_argument If a requested column has mismatching statistics dtype across + * sources. + */ +column_chunk_bounds_result column_chunk_bounds( + std::vector parquet_metadatas, + std::span column_names, + rmm::cuda_stream_view stream = cudf::get_default_stream(), + rmm::device_async_resource_ref mr = cudf::get_current_device_resource_ref()); + /** @} */ // end of group } // namespace io } // namespace CUDF_EXPORT cudf diff --git a/cpp/src/io/parquet/predicate_pushdown.cpp b/cpp/src/io/parquet/predicate_pushdown.cpp index 15124db4b2c8..f64d8ff730a8 100644 --- a/cpp/src/io/parquet/predicate_pushdown.cpp +++ b/cpp/src/io/parquet/predicate_pushdown.cpp @@ -5,17 +5,16 @@ #include "expression_transform_helpers.hpp" #include "reader_impl_helpers.hpp" +#include "row_group_stats_helpers.hpp" #include "stats_filter_helpers.hpp" -#include "timestamp_utils.cuh" #include #include #include -#include +#include #include #include #include -#include #include #include @@ -28,104 +27,6 @@ namespace cudf::io::parquet::detail { -namespace { - -/** - * @brief Converts column chunk statistics to 2 device columns - min, max values. - * - * Each column's number of rows equals the total number of row groups. - * - */ -struct row_group_stats_caster : public stats_caster_base { - size_type total_row_groups; - std::vector const& per_file_metadata; - host_span const> row_group_indices; - bool has_is_null_operator; - - // Creates device columns from column statistics (min, max) - template - std:: - tuple, std::unique_ptr, std::optional>> - operator()(host_span per_source_schema_indices, - cudf::data_type dtype, - cuda::stream_ref stream, - rmm::device_async_resource_ref mr) const - { - // List, Struct, Dictionary types are not supported - if constexpr (cudf::is_compound() && !std::is_same_v) { - CUDF_FAIL("Compound types do not have statistics"); - } else { - host_column min(total_row_groups, stream); - host_column max(total_row_groups, stream); - std::optional> is_null; - if (has_is_null_operator) { is_null = host_column(total_row_groups, stream); } - - size_type stats_idx = 0; - for (size_t src_idx = 0; src_idx < row_group_indices.size(); ++src_idx) { - auto const mapped_schema_idx = per_source_schema_indices[src_idx]; - // Compute timestamp scale factor for precision conversion from the mapped source schema. - auto const ts_scale = [&] { - if constexpr (cudf::is_timestamp()) { - auto const& schema = per_file_metadata[src_idx].schema[mapped_schema_idx]; - return calc_timestamp_scale(schema.logical_type, static_cast(T::period::den)); - } - return 0; - }(); - - for (auto const rg_idx : row_group_indices[src_idx]) { - auto const& row_group = per_file_metadata[src_idx].row_groups[rg_idx]; - auto col = std::find_if(row_group.columns.begin(), - row_group.columns.end(), - [mapped_schema_idx](ColumnChunk const& col) { - return col.schema_idx == mapped_schema_idx; - }); - if (col != std::end(row_group.columns)) { - auto const& colchunk = *col; - // To support deprecated min, max fields. - auto const& min_value = colchunk.meta_data.statistics.min_value.has_value() - ? colchunk.meta_data.statistics.min_value - : colchunk.meta_data.statistics.min; - auto const& max_value = colchunk.meta_data.statistics.max_value.has_value() - ? colchunk.meta_data.statistics.max_value - : colchunk.meta_data.statistics.max; - // translate binary data to Type then to - min.set_index(stats_idx, min_value, colchunk.meta_data.type, ts_scale); - max.set_index(stats_idx, max_value, colchunk.meta_data.type, ts_scale); - // Check the nullability of this column chunk - if (has_is_null_operator) { - if (colchunk.meta_data.statistics.null_count.has_value()) { - auto const& null_count = colchunk.meta_data.statistics.null_count.value(); - if (null_count == 0) { - is_null->val[stats_idx] = false; - } else if (null_count < colchunk.meta_data.num_values) { - is_null->set_index(stats_idx, std::nullopt, {}); - } else if (null_count == colchunk.meta_data.num_values) { - is_null->val[stats_idx] = true; - } else { - CUDF_FAIL("Invalid null count"); - } - } - } - } else { - // Marking it null, if column present in row group - min.set_index(stats_idx, std::nullopt, {}); - max.set_index(stats_idx, std::nullopt, {}); - if (has_is_null_operator) { is_null->set_index(stats_idx, std::nullopt, {}); } - } - stats_idx++; - } - }; - return {min.to_device(dtype, stream, mr), - max.to_device(dtype, stream, mr), - has_is_null_operator ? std::make_optional(is_null->to_device( - data_type{cudf::type_id::BOOL8}, stream, mr)) - : std::nullopt}; - } - } -}; - -} // namespace - bool aggregate_reader_metadata::any_row_group_stats_available( host_span const> input_row_group_indices, host_span filter_column_schemas) const @@ -288,7 +189,8 @@ aggregate_reader_metadata::filter_row_groups( // Span of row groups to apply bloom filtering on. auto const bloom_filter_input_row_groups = stats_filtered_row_groups.has_value() - ? host_span const>(stats_filtered_row_groups.value()) + ? host_span const>{stats_filtered_row_groups.value().data(), + stats_filtered_row_groups.value().size()} : input_row_group_indices; // Collect equality literals for each input table column for bloom filtering diff --git a/cpp/src/io/parquet/reader_impl_helpers.cpp b/cpp/src/io/parquet/reader_impl_helpers.cpp index 804d7b2eabd1..3e428bb1dd16 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.cpp +++ b/cpp/src/io/parquet/reader_impl_helpers.cpp @@ -12,13 +12,21 @@ #include "ipc/Message_generated.h" #include "ipc/Schema_generated.h" #include "parquet_common.hpp" +#include "row_group_stats_helpers.hpp" +#include #include #include #include +#include #include +#include #include #include +#include +#include +#include +#include #include #include @@ -32,6 +40,7 @@ #include #include #include +#include #include #include @@ -79,6 +88,56 @@ namespace flatbuf = cudf::io::parquet::flatbuf; namespace { +[[nodiscard]] std::unique_ptr make_size_type_column( + cudf::detail::host_vector const& values, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) +{ + auto data = cudf::detail::make_device_uvector_async( + host_span{values.data(), values.size()}, stream, mr); + stream.synchronize(); + return std::make_unique(data_type{type_id::INT32}, + static_cast(values.size()), + data.release(), + rmm::device_buffer{0, stream, mr}, + 0); +} + +[[nodiscard]] int find_leaf_schema_index(std::span schema_tree, + std::string_view column_name) +{ + auto found = std::optional{}; + for (auto idx = 1; std::cmp_less(idx, schema_tree.size()); ++idx) { + if (not schema_tree[idx].children_idx.empty()) { continue; } + if (column_path_from_index(schema_tree, idx) != column_name) { continue; } + CUDF_EXPECTS(not found.has_value(), + std::string{"Ambiguous parquet leaf column path: "} + std::string{column_name}, + std::invalid_argument); + found = idx; + } + CUDF_EXPECTS(found.has_value(), + std::string{"Parquet leaf column path not found: "} + std::string{column_name}, + std::invalid_argument); + return found.value(); +} + +[[nodiscard]] data_type statistics_dtype(SchemaElement const& schema) +{ + auto const dtype = to_data_type(to_type_id(schema, + false, // strings_to_categorical + type_id::EMPTY, // timestamp_type_id + type_id::EMPTY), + schema); + CUDF_EXPECTS(dtype.id() != type_id::EMPTY, + std::string{"Unsupported parquet statistics dtype for column: "} + schema.name, + std::invalid_argument); + CUDF_EXPECTS( + not cudf::is_compound(dtype) or dtype.id() == type_id::STRING, + std::string{"Compound parquet statistics are not supported for column: "} + schema.name, + std::invalid_argument); + return dtype; +} + /** * @brief Computes the total number of row groups in input span of row group indices */ @@ -1366,6 +1425,86 @@ aggregate_reader_metadata::get_column_chunk_metadata() const return column_chunk_metadata; } +column_chunk_bounds_result aggregate_reader_metadata::column_chunk_bounds( + host_span column_names, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const +{ + CUDF_EXPECTS(column_names.empty() or not per_file_metadata.empty(), + "Cannot decode parquet column-chunk bounds without source metadata", + std::invalid_argument); + + auto const total_row_groups = get_num_row_groups(); + + auto input_row_group_indices = std::vector>(per_file_metadata.size()); + auto file_indices = cudf::detail::make_empty_host_vector(total_row_groups, stream); + auto row_group_indices = + cudf::detail::make_empty_host_vector(total_row_groups, stream); + + for (auto src_idx = size_type{0}; std::cmp_less(src_idx, per_file_metadata.size()); ++src_idx) { + auto const num_source_row_groups = + static_cast(per_file_metadata[src_idx].row_groups.size()); + auto& source_row_group_indices = input_row_group_indices[src_idx]; + source_row_group_indices.resize(num_source_row_groups); + std::iota(source_row_group_indices.begin(), source_row_group_indices.end(), size_type{0}); + + std::fill_n(std::back_inserter(file_indices), num_source_row_groups, src_idx); + std::copy(source_row_group_indices.begin(), + source_row_group_indices.end(), + std::back_inserter(row_group_indices)); + } + + auto result = column_chunk_bounds_result{make_size_type_column(file_indices, stream, mr), + make_size_type_column(row_group_indices, stream, mr), + {}}; + result.bounds.reserve(column_names.size()); + + row_group_stats_caster const stats_col{ + .total_row_groups = total_row_groups, + .per_file_metadata = per_file_metadata, + .row_group_indices = host_span const>{input_row_group_indices.data(), + input_row_group_indices.size()}, + .has_is_null_operator = false}; + + for (auto const& column_name : column_names) { + auto per_source_schema_indices = std::vector(per_file_metadata.size()); + auto dtype = data_type{type_id::EMPTY}; + + for (auto src_idx = size_type{0}; std::cmp_less(src_idx, per_file_metadata.size()); ++src_idx) { + auto const& schema_tree = get_schema_tree(src_idx); + auto const schema_idx = find_leaf_schema_index( + std::span{schema_tree.data(), schema_tree.size()}, column_name); + auto const source_dtype = statistics_dtype(schema_tree[schema_idx]); + + if (src_idx == 0) { + dtype = source_dtype; + } else { + CUDF_EXPECTS( + source_dtype == dtype, + std::string{"Mismatching parquet statistics dtype across sources for column: "} + + column_name, + std::invalid_argument); + } + per_source_schema_indices[src_idx] = schema_idx; + } + + auto [min_col, max_col, _] = cudf::type_dispatcher( + dtype, + stats_col, + host_span{per_source_schema_indices.data(), per_source_schema_indices.size()}, + dtype, + stream, + mr); + std::vector> columns; + columns.reserve(2); + columns.push_back(std::move(min_col)); + columns.push_back(std::move(max_col)); + result.bounds.push_back(std::make_unique(std::move(columns))); + } + + return result; +} + bool aggregate_reader_metadata::is_schema_index_mapped(int schema_idx, int src_idx) const { // Check if schema_idx or src_idx is invalid @@ -1649,7 +1788,8 @@ aggregate_reader_metadata::select_row_groups( }); // Set the current span of row group indices to the vector of all row group indices - current_row_group_indices = host_span const>(all_row_group_indices); + current_row_group_indices = host_span const>{ + all_row_group_indices.data(), all_row_group_indices.size()}; } // Otherwise, set the current span of row group indices to the specified input row group indices else { @@ -1678,7 +1818,8 @@ aggregate_reader_metadata::select_row_groups( apply_row_bounds_filter(current_row_group_indices, rows_to_skip, rows_to_read); // Update the current span of row group indices - current_row_group_indices = host_span const>(trimmed_row_group_indices); + current_row_group_indices = host_span const>{ + trimmed_row_group_indices.data(), trimmed_row_group_indices.size()}; } // Flag to check if the row groups will be filtered using byte bounds @@ -1697,7 +1838,8 @@ aggregate_reader_metadata::select_row_groups( apply_byte_bounds_filter(current_row_group_indices, skip_bytes_opt, byte_count_opt); // Update the current span of row group indices - current_row_group_indices = host_span const>(trimmed_row_group_indices); + current_row_group_indices = host_span const>{ + trimmed_row_group_indices.data(), trimmed_row_group_indices.size()}; } // Compute number of input row groups after row or byte bounds trimming @@ -1725,8 +1867,9 @@ aggregate_reader_metadata::select_row_groups( // rows to skip relative to the first surviving row group if (filtered_row_group_indices.has_value()) { // Update the current span of row group indices + auto const& filtered_indices = filtered_row_group_indices.value(); current_row_group_indices = - host_span const>(filtered_row_group_indices.value()); + host_span const>{filtered_indices.data(), filtered_indices.size()}; // Only need to update the rows to skip relative to the first surviving row group // if row bounds were previously applied @@ -2278,3 +2421,24 @@ std::vector aggregate_reader_metadata::get_parquet_types( } } // namespace cudf::io::parquet::detail + +namespace cudf::io { + +column_chunk_bounds_result column_chunk_bounds(std::vector parquet_metadatas, + std::span column_names, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) +{ + CUDF_FUNC_RANGE(); + + auto metadata = parquet::detail::aggregate_reader_metadata{ + std::move(parquet_metadatas), + false, // use_arrow_schema + false // has_cols_from_mismatched_srcs + }; + + return metadata.column_chunk_bounds( + host_span{column_names.data(), column_names.size()}, stream, mr); +} + +} // namespace cudf::io diff --git a/cpp/src/io/parquet/reader_impl_helpers.hpp b/cpp/src/io/parquet/reader_impl_helpers.hpp index d5b53e262bfd..a7dde8cd8d03 100644 --- a/cpp/src/io/parquet/reader_impl_helpers.hpp +++ b/cpp/src/io/parquet/reader_impl_helpers.hpp @@ -516,6 +516,19 @@ class aggregate_reader_metadata { [[nodiscard]] std::unordered_map> get_column_chunk_metadata() const; + /** + * @brief Decodes min/max statistics for selected column chunks. + * + * @param column_names Dotted leaf-column paths to decode statistics for + * @param stream CUDA stream used for device memory operations + * @param mr Device memory resource to use for device memory allocation + * @return Decoded min/max bounds and row-group identifiers + */ + [[nodiscard]] column_chunk_bounds_result column_chunk_bounds( + host_span column_names, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const; + /** * @brief Get total number of rows across all files * diff --git a/cpp/src/io/parquet/row_group_stats_helpers.hpp b/cpp/src/io/parquet/row_group_stats_helpers.hpp new file mode 100644 index 000000000000..89cbda42d8b5 --- /dev/null +++ b/cpp/src/io/parquet/row_group_stats_helpers.hpp @@ -0,0 +1,147 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include "reader_impl_helpers.hpp" +#include "stats_filter_helpers.hpp" +#include "timestamp_utils.cuh" + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace cudf::io::parquet::detail { + +/** + * @brief Converts row-group column chunk statistics to device columns. + * + * Each output column has one row for every selected row group. + */ +struct row_group_stats_caster : public stats_caster_base { + using result_type = std:: + tuple, std::unique_ptr, std::optional>>; + + size_type total_row_groups; + std::vector const& per_file_metadata; + host_span const> row_group_indices; + bool has_is_null_operator; + + template + result_type operator()(host_span per_source_schema_indices, + cudf::data_type dtype, + rmm::cuda_stream_view stream, + rmm::device_async_resource_ref mr) const + { + CUDF_EXPECTS(row_group_indices.size() == per_file_metadata.size(), + "Row-group indices must match parquet metadata sources", + std::invalid_argument); + CUDF_EXPECTS(per_source_schema_indices.size() == per_file_metadata.size(), + "Per-source schema indices must match parquet metadata sources", + std::invalid_argument); + auto const computed_total_row_groups = + std::accumulate(row_group_indices.begin(), + row_group_indices.end(), + size_type{0}, + [](auto count, auto const& source_row_group_indices) { + return count + static_cast(source_row_group_indices.size()); + }); + CUDF_EXPECTS(total_row_groups == computed_total_row_groups, + "Total row groups must match selected row-group indices", + std::invalid_argument); + + if constexpr (cudf::is_compound() and not std::is_same_v) { + CUDF_FAIL("Compound types do not have statistics"); + } else { + host_column min(total_row_groups, stream); + host_column max(total_row_groups, stream); + std::optional> is_null; + if (has_is_null_operator) { is_null = host_column(total_row_groups, stream); } + + size_type stats_idx = 0; + for (size_t src_idx = 0; src_idx < row_group_indices.size(); ++src_idx) { + auto const mapped_schema_idx = per_source_schema_indices[src_idx]; + auto const& source_metadata = per_file_metadata[src_idx]; + CUDF_EXPECTS(mapped_schema_idx >= 0 and + static_cast(mapped_schema_idx) < source_metadata.schema.size(), + "Mapped schema index is out of bounds", + std::invalid_argument); + // Compute timestamp scale factor for precision conversion from the mapped source schema. + auto const ts_scale = [&] { + if constexpr (cudf::is_timestamp()) { + auto const& schema = source_metadata.schema[mapped_schema_idx]; + return calc_timestamp_scale(schema.logical_type, static_cast(T::period::den)); + } + return 0; + }(); + + for (auto const rg_idx : row_group_indices[src_idx]) { + CUDF_EXPECTS( + rg_idx >= 0 and static_cast(rg_idx) < source_metadata.row_groups.size(), + "Row-group index is out of bounds", + std::invalid_argument); + auto const& row_group = source_metadata.row_groups[rg_idx]; + auto col = std::find_if(row_group.columns.begin(), + row_group.columns.end(), + [mapped_schema_idx](ColumnChunk const& col) { + return col.schema_idx == mapped_schema_idx; + }); + if (col != std::end(row_group.columns)) { + auto const& colchunk = *col; + // To support deprecated min, max fields. + auto const& min_value = colchunk.meta_data.statistics.min_value.has_value() + ? colchunk.meta_data.statistics.min_value + : colchunk.meta_data.statistics.min; + auto const& max_value = colchunk.meta_data.statistics.max_value.has_value() + ? colchunk.meta_data.statistics.max_value + : colchunk.meta_data.statistics.max; + // translate binary data to Type then to + min.set_index(stats_idx, min_value, colchunk.meta_data.type, ts_scale); + max.set_index(stats_idx, max_value, colchunk.meta_data.type, ts_scale); + // Check the nullability of this column chunk + if (has_is_null_operator) { + if (colchunk.meta_data.statistics.null_count.has_value()) { + auto const& null_count = colchunk.meta_data.statistics.null_count.value(); + if (null_count == 0) { + is_null->val[stats_idx] = false; + } else if (null_count < colchunk.meta_data.num_values) { + is_null->set_index(stats_idx, std::nullopt, {}); + } else if (null_count == colchunk.meta_data.num_values) { + is_null->val[stats_idx] = true; + } else { + CUDF_FAIL("Invalid null count"); + } + } + } + } else { + // Mark it null if the column chunk is absent from this row group. + min.set_index(stats_idx, std::nullopt, {}); + max.set_index(stats_idx, std::nullopt, {}); + if (has_is_null_operator) { is_null->set_index(stats_idx, std::nullopt, {}); } + } + stats_idx++; + } + }; + return {min.to_device(dtype, stream, mr), + max.to_device(dtype, stream, mr), + has_is_null_operator ? std::make_optional(is_null->to_device( + data_type{cudf::type_id::BOOL8}, stream, mr)) + : std::nullopt}; + } + } +}; + +} // namespace cudf::io::parquet::detail diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index 08e48dd213eb..b35e62f701f3 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -3646,6 +3646,57 @@ TEST_F(ParquetMetadataReaderTest, PreMaterializedMetadata) test_parquet_metadata(3); } +TEST_F(ParquetMetadataReaderTest, ColumnChunkBounds) +{ + auto values = column_wrapper{1, 2, 3, 4}; + auto input = table_view{{values}}; + + cudf::io::table_input_metadata metadata(input); + metadata.column_metadata[0].set_name("value"); + + auto filepath = temp_env->get_temp_filepath("ColumnChunkBounds.parquet"); + cudf::io::parquet_writer_options const out_opts = + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{filepath}, input) + .metadata(std::move(metadata)) + .row_group_size_rows(2) + .max_page_size_rows(2) + .max_page_fragment_size(2) + .stats_level(cudf::io::statistics_freq::STATISTICS_ROWGROUP); + cudf::io::write_parquet(out_opts); + + auto datasources = cudf::io::make_datasources(cudf::io::source_info{filepath}); + auto metadatas = cudf::io::read_parquet_footers(datasources); + auto column_names = std::vector{"value"}; + auto bounds = cudf::io::column_chunk_bounds( + std::move(metadatas), + cudf::host_span{column_names.data(), column_names.size()}, + cudf::get_default_stream(), + cudf::get_current_device_resource_ref()); + + ASSERT_EQ(bounds.bounds.size(), 1); + + auto expected_file_indices = column_wrapper{0, 0}; + auto expected_rg_indices = column_wrapper{0, 1}; + CUDF_TEST_EXPECT_COLUMNS_EQUAL(expected_file_indices, bounds.file_indices->view()); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(expected_rg_indices, bounds.row_group_indices->view()); + + auto const all_valid = cudf::test::iterators::no_nulls(); + auto expected_min = column_wrapper({1, 3}, all_valid); + auto expected_max = column_wrapper({2, 4}, all_valid); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(expected_min, bounds.bounds.front()->view().column(0)); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(expected_max, bounds.bounds.front()->view().column(1)); +} + +TEST_F(ParquetMetadataReaderTest, ColumnChunkBoundsEmptyMetadata) +{ + auto column_names = std::vector{"value"}; + auto metadatas = std::vector{}; + EXPECT_THROW(cudf::io::column_chunk_bounds( + std::move(metadatas), + cudf::host_span{column_names.data(), column_names.size()}), + std::invalid_argument); +} + TEST_F(ParquetMetadataReaderTest, Nested) { auto const num_rows = 1200; diff --git a/python/pylibcudf/pylibcudf/io/parquet_metadata.pxd b/python/pylibcudf/pylibcudf/io/parquet_metadata.pxd index fedbca1800f2..6ffc20cbfad3 100644 --- a/python/pylibcudf/pylibcudf/io/parquet_metadata.pxd +++ b/python/pylibcudf/pylibcudf/io/parquet_metadata.pxd @@ -10,6 +10,7 @@ from pylibcudf.libcudf.io.parquet_schema cimport ( FileMetaData as cpp_FileMetaData, RowGroup as cpp_RowGroup, SortingColumn as cpp_SortingColumn, + Statistics as cpp_Statistics, ) from pylibcudf.libcudf.io.parquet_metadata cimport( parquet_metadata, @@ -17,6 +18,7 @@ from pylibcudf.libcudf.io.parquet_metadata cimport( parquet_column_schema, ) from pylibcudf.types cimport DataType +from rmm.pylibrmm.memory_resource cimport DeviceMemoryResource cdef class ParquetColumnSchema: cdef parquet_column_schema column_schema @@ -78,6 +80,12 @@ cdef class SortingColumn: @staticmethod cdef SortingColumn from_cpp(cpp_SortingColumn sorting_column) +cdef class ColumnChunkStatistics: + cdef cpp_Statistics c_obj + + @staticmethod + cdef ColumnChunkStatistics from_cpp(cpp_Statistics statistics) + cdef class ColumnChunk: cdef cpp_ColumnChunk c_obj @@ -98,3 +106,9 @@ cdef class RowGroup: cpdef ParquetMetadata read_parquet_metadata(SourceInfo src_info) cpdef list read_parquet_footers(SourceInfo src_info) +cpdef tuple column_chunk_bounds( + object file_metadatas, + object columns, + object stream=*, + DeviceMemoryResource mr=*, +) diff --git a/python/pylibcudf/pylibcudf/io/parquet_metadata.pyi b/python/pylibcudf/pylibcudf/io/parquet_metadata.pyi index 6aa9efb19713..4b3bb0a909a1 100644 --- a/python/pylibcudf/pylibcudf/io/parquet_metadata.pyi +++ b/python/pylibcudf/pylibcudf/io/parquet_metadata.pyi @@ -1,8 +1,14 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +from collections.abc import Sequence + +from pylibcudf.column import Column from pylibcudf.io.types import SourceInfo +from pylibcudf.table import Table from pylibcudf.types import DataType +from pylibcudf.utils import CudaStreamLike +from rmm.pylibrmm.memory_resource import DeviceMemoryResource try: from collections.abc import Buffer @@ -12,12 +18,14 @@ except ImportError: __all__ = [ "ColumnChunk", "ColumnChunkMetaData", + "ColumnChunkStatistics", "FileMetaData", "ParquetColumnSchema", "ParquetMetadata", "ParquetSchema", "RowGroup", "SortingColumn", + "column_chunk_bounds", "read_parquet_footers", "read_parquet_metadata", ] @@ -83,6 +91,22 @@ class ColumnChunk: @property def meta_data(self) -> ColumnChunkMetaData: ... +class ColumnChunkStatistics: + @property + def has_min_max(self) -> bool: ... + @property + def min_encoded(self) -> bytes | None: ... + @property + def max_encoded(self) -> bytes | None: ... + @property + def null_count(self) -> int | None: ... + @property + def distinct_count(self) -> int | None: ... + @property + def is_min_value_exact(self) -> bool | None: ... + @property + def is_max_value_exact(self) -> bool | None: ... + class ColumnChunkMetaData: @property def path_in_schema(self) -> list[str]: ... @@ -92,6 +116,8 @@ class ColumnChunkMetaData: def total_uncompressed_size(self) -> int: ... @property def total_compressed_size(self) -> int: ... + @property + def statistics(self) -> ColumnChunkStatistics: ... class RowGroup: @property @@ -111,3 +137,9 @@ class RowGroup: def read_parquet_metadata(src_info: SourceInfo) -> ParquetMetadata: ... def read_parquet_footers(src_info: SourceInfo) -> list[FileMetaData]: ... +def column_chunk_bounds( + file_metadatas: Sequence[FileMetaData], + columns: Sequence[str], + stream: CudaStreamLike | None = None, + mr: DeviceMemoryResource | None = None, +) -> tuple[Column, Column, tuple[Table, ...]]: ... diff --git a/python/pylibcudf/pylibcudf/io/parquet_metadata.pyx b/python/pylibcudf/pylibcudf/io/parquet_metadata.pyx index e6015786173b..407b0e85af3d 100644 --- a/python/pylibcudf/pylibcudf/io/parquet_metadata.pyx +++ b/python/pylibcudf/pylibcudf/io/parquet_metadata.pyx @@ -3,11 +3,16 @@ from cython.operator cimport dereference from libc.stdint cimport uint8_t +from cuda.bindings.cyruntime cimport cudaStream_t +from cpython.bytes cimport PyBytes_FromStringAndSize from libcpp.memory cimport make_unique, unique_ptr +from libcpp.optional cimport optional +from libcpp.span cimport span as std_span from libcpp.string cimport string from libcpp.utility cimport move from libcpp.vector cimport vector +from pylibcudf.column cimport Column from pylibcudf.io.types cimport SourceInfo from pylibcudf.libcudf.io.datasource cimport datasource, make_datasources from pylibcudf.libcudf.io.hybrid_scan cimport ( @@ -22,9 +27,15 @@ from pylibcudf.libcudf.io.parquet_schema cimport ( FileMetaData as cpp_FileMetaData, RowGroup as cpp_RowGroup, SortingColumn as cpp_SortingColumn, + Statistics as cpp_Statistics, ) +from pylibcudf.libcudf.table.table cimport table as cpp_table from pylibcudf.libcudf.utilities.span cimport host_span +from pylibcudf.table cimport Table from pylibcudf.types cimport DataType +from pylibcudf.utils cimport _get_memory_resource, _get_stream +from rmm.pylibrmm.memory_resource cimport DeviceMemoryResource +from rmm.pylibrmm.stream cimport Stream from typing import TYPE_CHECKING @@ -32,17 +43,20 @@ if TYPE_CHECKING: from typing_extensions import Buffer ctypedef const unique_ptr[datasource] const_unique_ptr_datasource +ctypedef const string const_string __all__ = [ "ColumnChunk", "ColumnChunkMetaData", + "ColumnChunkStatistics", "FileMetaData", "ParquetColumnSchema", "ParquetMetadata", "ParquetSchema", "RowGroup", "SortingColumn", + "column_chunk_bounds", "read_parquet_footers", "read_parquet_metadata", ] @@ -306,6 +320,83 @@ cdef class SortingColumn: return self.c_obj.nulls_first +cdef object _optional_bytes(optional[vector[uint8_t]] value): + cdef vector[uint8_t]* buffer + if not value.has_value(): + return None + buffer = &value.value() + if buffer.size() == 0: + return b"" + return PyBytes_FromStringAndSize( + buffer.data(), buffer.size() + ) + + +cdef class ColumnChunkStatistics: + """Column chunk statistics.""" + + def __init__(self): + raise ValueError("ColumnChunkStatistics cannot be constructed directly") + + @staticmethod + cdef ColumnChunkStatistics from_cpp(cpp_Statistics statistics): + cdef ColumnChunkStatistics result = ColumnChunkStatistics.__new__( + ColumnChunkStatistics + ) + result.c_obj = statistics + return result + + @property + def has_min_max(self) -> bool: + """Whether this column chunk has encoded minimum and maximum values.""" + return ( + (self.c_obj.min_value.has_value() or self.c_obj.min.has_value()) + and (self.c_obj.max_value.has_value() or self.c_obj.max.has_value()) + ) + + @property + def min_encoded(self) -> bytes | None: + """Encoded minimum value, preferring ``min_value`` over deprecated ``min``.""" + if self.c_obj.min_value.has_value(): + return _optional_bytes(self.c_obj.min_value) + return _optional_bytes(self.c_obj.min) + + @property + def max_encoded(self) -> bytes | None: + """Encoded maximum value, preferring ``max_value`` over deprecated ``max``.""" + if self.c_obj.max_value.has_value(): + return _optional_bytes(self.c_obj.max_value) + return _optional_bytes(self.c_obj.max) + + @property + def null_count(self) -> int | None: + """Number of null values in the column chunk.""" + if not self.c_obj.null_count.has_value(): + return None + return self.c_obj.null_count.value() + + @property + def distinct_count(self) -> int | None: + """Number of distinct values in the column chunk.""" + if not self.c_obj.distinct_count.has_value(): + return None + return self.c_obj.distinct_count.value() + + @property + def is_min_value_exact(self) -> bool | None: + """Whether ``min_value`` is the exact column-chunk minimum.""" + if not self.c_obj.is_min_value_exact.has_value(): + return None + return self.c_obj.is_min_value_exact.value() + + @property + def is_max_value_exact(self) -> bool | None: + """Whether ``max_value`` is the exact column-chunk maximum.""" + if not self.c_obj.is_max_value_exact.has_value(): + return None + return self.c_obj.is_max_value_exact.value() + + cdef class ColumnChunk: """Metadata for a row group's column chunk.""" @@ -394,6 +485,11 @@ cdef class ColumnChunkMetaData: """Total compressed page bytes for this chunk.""" return self.c_obj.total_compressed_size + @property + def statistics(self) -> ColumnChunkStatistics: + """Column chunk statistics.""" + return ColumnChunkStatistics.from_cpp(self.c_obj.statistics) + cdef class RowGroup: """Parquet row group metadata.""" @@ -691,3 +787,74 @@ cpdef list read_parquet_footers(SourceInfo src_info): # GIL held only for Python object allocation + list build return [FileMetaData.from_libcudf(move(owned[i])) for i in range(n)] + + +cpdef tuple column_chunk_bounds( + object file_metadatas, + object columns, + object stream=None, + DeviceMemoryResource mr=None, +): + """ + Decode parquet column-chunk min/max statistics for selected columns. + + Missing min/max statistics are returned as nulls. Parquet min/max + exactness flags are not interpreted by this function. + + Parameters + ---------- + file_metadatas : Sequence[FileMetaData] + Parquet footer metadata objects, one per source. + columns : Sequence[str] + Dotted leaf-column paths to decode statistics for. + stream : CudaStreamLike, optional + CUDA stream used for device memory operations. + mr : DeviceMemoryResource, optional + Device memory resource used for device memory allocation. + + Returns + ------- + tuple[Column, Column, tuple[Table, ...]] + File indices, file-local row-group indices, and one two-column + ``(min, max)`` table per requested column. + """ + cdef vector[cpp_FileMetaData] c_metadatas + cdef vector[string] c_columns + cdef cpp_parquet_metadata.column_chunk_bounds_result c_result + cdef object metadata_obj + cdef object column_name + cdef vector[unique_ptr[cpp_table]].size_type i + cdef Stream _stream = _get_stream(stream) + cdef cudaStream_t _cs = _stream.view().value() + cdef list bounds = [] + mr = _get_memory_resource(mr) + + for metadata_obj in file_metadatas: + if not isinstance(metadata_obj, FileMetaData): + raise TypeError("file_metadatas must contain only FileMetaData objects") + c_metadatas.push_back(dereference((metadata_obj).c_obj)) + + if isinstance(columns, str): + raise TypeError("columns must be a sequence of strings") + + for column_name in columns: + if not isinstance(column_name, str): + raise TypeError("columns must contain only strings") + c_columns.push_back(column_name.encode()) + + with nogil: + c_result = cpp_parquet_metadata.column_chunk_bounds( + move(c_metadatas), + std_span[const_string](c_columns.data(), c_columns.size()), + _cs, + mr.get_mr(), + ) + + for i in range(c_result.bounds.size()): + bounds.append(Table.from_libcudf(move(c_result.bounds[i]), _stream, mr)) + + return ( + Column.from_libcudf(move(c_result.file_indices), _stream, mr), + Column.from_libcudf(move(c_result.row_group_indices), _stream, mr), + tuple(bounds), + ) diff --git a/python/pylibcudf/pylibcudf/libcudf/io/parquet_metadata.pxd b/python/pylibcudf/pylibcudf/libcudf/io/parquet_metadata.pxd index 3a2b41de2774..1c27d581a93c 100644 --- a/python/pylibcudf/pylibcudf/libcudf/io/parquet_metadata.pxd +++ b/python/pylibcudf/pylibcudf/libcudf/io/parquet_metadata.pxd @@ -1,18 +1,24 @@ -# 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 from libcpp.memory cimport unique_ptr +from libcpp.span cimport span as std_span from libcpp.string cimport string from libcpp.unordered_map cimport unordered_map from libcpp.vector cimport vector from pylibcudf.exception_handler cimport libcudf_exception_handler +from pylibcudf.libcudf.column.column cimport column from pylibcudf.libcudf.io.datasource cimport datasource from pylibcudf.libcudf.io.parquet_schema cimport FileMetaData +from pylibcudf.libcudf.table.table cimport table from pylibcudf.libcudf.types cimport data_type, size_type from pylibcudf.libcudf.io.types cimport source_info from pylibcudf.libcudf.utilities.span cimport host_span +from cuda.bindings.cyruntime cimport cudaStream_t +from rmm.librmm.memory_resource cimport device_async_resource_ref ctypedef const unique_ptr[datasource] const_unique_ptr_datasource +ctypedef const string const_string cdef extern from "cudf/io/parquet_metadata.hpp" namespace "cudf::io" nogil: @@ -40,6 +46,11 @@ cdef extern from "cudf/io/parquet_metadata.hpp" namespace "cudf::io" nogil: unordered_map[string, vector[int64_t]] \ columnchunk_metadata() except +libcudf_exception_handler + cdef cppclass column_chunk_bounds_result: + unique_ptr[column] file_indices + unique_ptr[column] row_group_indices + vector[unique_ptr[table]] bounds + cdef parquet_metadata read_parquet_metadata( source_info src_info ) except +libcudf_exception_handler @@ -47,3 +58,10 @@ cdef extern from "cudf/io/parquet_metadata.hpp" namespace "cudf::io" nogil: cdef vector[FileMetaData] read_parquet_footers( host_span[const_unique_ptr_datasource] sources ) except +libcudf_exception_handler + + cdef column_chunk_bounds_result column_chunk_bounds( + vector[FileMetaData] parquet_metadatas, + std_span[const_string] column_names, + cudaStream_t stream, + device_async_resource_ref mr, + ) except +libcudf_exception_handler diff --git a/python/pylibcudf/pylibcudf/libcudf/io/parquet_schema.pxd b/python/pylibcudf/pylibcudf/libcudf/io/parquet_schema.pxd index 823f2d192111..c12a32351265 100644 --- a/python/pylibcudf/pylibcudf/libcudf/io/parquet_schema.pxd +++ b/python/pylibcudf/pylibcudf/libcudf/io/parquet_schema.pxd @@ -1,7 +1,8 @@ -# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION. +# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -from libc.stdint cimport int16_t, int32_t, int64_t +from libc.stdint cimport int16_t, int32_t, int64_t, uint8_t +from libcpp cimport bool from libcpp.optional cimport optional from libcpp.string cimport string from libcpp.vector cimport vector @@ -9,6 +10,16 @@ from pylibcudf.exception_handler cimport libcudf_exception_handler cdef extern from "cudf/io/parquet_schema.hpp" namespace "cudf::io::parquet" nogil: + cdef cppclass Statistics: + optional[vector[uint8_t]] max + optional[vector[uint8_t]] min + optional[int64_t] null_count + optional[int64_t] distinct_count + optional[vector[uint8_t]] max_value + optional[vector[uint8_t]] min_value + optional[bool] is_max_value_exact + optional[bool] is_min_value_exact + cdef cppclass SortingColumn: int32_t column_idx bint descending @@ -22,6 +33,7 @@ cdef extern from "cudf/io/parquet_schema.hpp" namespace "cudf::io::parquet" nogi int64_t data_page_offset int64_t index_page_offset int64_t dictionary_page_offset + Statistics statistics cdef cppclass ColumnChunk: string file_path diff --git a/python/pylibcudf/tests/io/test_parquet.py b/python/pylibcudf/tests/io/test_parquet.py index be56621c2bc9..64e0526aa7b7 100644 --- a/python/pylibcudf/tests/io/test_parquet.py +++ b/python/pylibcudf/tests/io/test_parquet.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 import io import os +import struct import pyarrow as pa import pyarrow.compute as pc @@ -494,6 +495,203 @@ def test_file_metadata_row_groups_and_column_chunks() -> None: assert meta_data.path_in_schema[-1] == pa_col_chunk.path_in_schema +def test_file_metadata_columnchunk_statistics() -> None: + table = pa.table( + { + "a": pa.array([1, 2, 3, None], type=pa.int64()), + "s": pa.array(["aa", "bb", "cc", None]), + } + ) + sink = io.BytesIO() + write_table(table, sink, row_group_size=2) + sink.seek(0) + + source_info = plc.io.SourceInfo([sink]) + file_metadata = plc.io.parquet_metadata.read_parquet_footers(source_info)[ + 0 + ] + + expected = [ + (1, 2, 0, b"aa", b"bb"), + (3, 3, 1, b"cc", b"cc"), + ] + for row_group, ( + min_a, + max_a, + null_count, + min_s, + max_s, + ) in zip(file_metadata.row_groups, expected, strict=True): + stats_by_name = { + column.meta_data.path_in_schema[-1]: column.meta_data.statistics + for column in row_group.columns + } + + a_stats = stats_by_name["a"] + assert a_stats.has_min_max + assert a_stats.min_encoded == struct.pack(" None: + table = pa.table({"a": pa.array([1, 2], type=pa.int64())}) + sink = io.BytesIO() + write_table(table, sink, write_statistics=False) + sink.seek(0) + + source_info = plc.io.SourceInfo([sink]) + file_metadata = plc.io.parquet_metadata.read_parquet_footers(source_info)[ + 0 + ] + statistics = file_metadata.row_groups[0].columns[0].meta_data.statistics + + assert not statistics.has_min_max + assert statistics.min_encoded is None + assert statistics.max_encoded is None + assert statistics.is_min_value_exact is None + assert statistics.is_max_value_exact is None + + +def test_column_chunk_bounds(tmp_path) -> None: + table_0 = pa.table( + { + "a": pa.array([1, 2, 3, 4], type=pa.int64()), + "s": ["b", "a", "d", "c"], + "ts": pa.array([0, 1, 2, 3], type=pa.timestamp("us")), + "n": pa.array([None, 2, None, None], type=pa.int64()), + } + ) + table_1 = pa.table( + { + "a": pa.array([10, 20, 30, 40], type=pa.int64()), + "s": ["z", "y", "x", "w"], + "ts": pa.array([10, 20, 30, 40], type=pa.timestamp("us")), + "n": pa.array([5, None, None, 8], type=pa.int64()), + } + ) + path_0 = tmp_path / "part-0.parquet" + path_1 = tmp_path / "part-1.parquet" + write_table(table_0, path_0, row_group_size=2) + write_table(table_1, path_1, row_group_size=2) + + file_metadatas = plc.io.parquet_metadata.read_parquet_footers( + plc.io.SourceInfo([path_0, path_1]) + ) + + file_indices, row_group_indices, bounds = ( + plc.io.parquet_metadata.column_chunk_bounds( + file_metadatas, + columns=["a", "s", "ts", "n"], + ) + ) + + assert file_indices.to_pylist() == [0, 0, 1, 1] + assert row_group_indices.to_pylist() == [0, 1, 0, 1] + assert len(bounds) == 4 + + a_min, a_max = bounds[0].columns() + assert a_min.to_pylist() == [1, 3, 10, 30] + assert a_max.to_pylist() == [2, 4, 20, 40] + + s_min, s_max = bounds[1].columns() + assert s_min.to_pylist() == ["a", "c", "y", "w"] + assert s_max.to_pylist() == ["b", "d", "z", "x"] + + ts_min, ts_max = bounds[2].columns() + assert ts_min.to_arrow().equals( + pa.array([0, 2, 10, 30], type=pa.timestamp("us")) + ) + assert ts_max.to_arrow().equals( + pa.array([1, 3, 20, 40], type=pa.timestamp("us")) + ) + + n_min, n_max = bounds[3].columns() + assert n_min.to_pylist() == [2, None, 5, 8] + assert n_max.to_pylist() == [2, None, 5, 8] + + +def test_column_chunk_bounds_without_minmax(tmp_path) -> None: + table = pa.table({"a": pa.array([1, 2], type=pa.int64())}) + path = tmp_path / "no-stats.parquet" + write_table(table, path, write_statistics=False) + + file_metadatas = plc.io.parquet_metadata.read_parquet_footers( + plc.io.SourceInfo([path]) + ) + + file_indices, row_group_indices, (bounds,) = ( + plc.io.parquet_metadata.column_chunk_bounds( + file_metadatas, + columns=["a"], + ) + ) + + assert file_indices.to_pylist() == [0] + assert row_group_indices.to_pylist() == [0] + min_col, max_col = bounds.columns() + assert min_col.to_pylist() == [None] + assert max_col.to_pylist() == [None] + + +def test_column_chunk_bounds_invalid_inputs(tmp_path) -> None: + table = pa.table({"a": pa.array([1, 2], type=pa.int64())}) + path = tmp_path / "input.parquet" + write_table(table, path) + + file_metadatas = plc.io.parquet_metadata.read_parquet_footers( + plc.io.SourceInfo([path]) + ) + + with pytest.raises( + ValueError, match="Parquet leaf column path not found: missing" + ): + plc.io.parquet_metadata.column_chunk_bounds( + file_metadatas, + columns=["missing"], + ) + + with pytest.raises(TypeError, match="columns must contain only strings"): + plc.io.parquet_metadata.column_chunk_bounds( + file_metadatas, + columns=[1], + ) + + with pytest.raises( + TypeError, match="columns must be a sequence of strings" + ): + plc.io.parquet_metadata.column_chunk_bounds( + file_metadatas, + columns="a", + ) + + with pytest.raises( + TypeError, + match="file_metadatas must contain only FileMetaData objects", + ): + plc.io.parquet_metadata.column_chunk_bounds( + [object()], + columns=["a"], + ) + + with pytest.raises( + ValueError, + match="without source metadata", + ): + plc.io.parquet_metadata.column_chunk_bounds( + [], + columns=["a"], + ) + + def test_file_metadata_wrappers_not_directly_constructible() -> None: with pytest.raises( ValueError, match="SortingColumn cannot be constructed directly" @@ -507,6 +705,11 @@ def test_file_metadata_wrappers_not_directly_constructible() -> None: ValueError, match="ColumnChunkMetaData cannot be constructed directly" ): plc.io.parquet_metadata.ColumnChunkMetaData() + with pytest.raises( + ValueError, + match="ColumnChunkStatistics cannot be constructed directly", + ): + plc.io.parquet_metadata.ColumnChunkStatistics() with pytest.raises( ValueError, match="RowGroup cannot be constructed directly" ):