From f74b3caeea84d3e5eefd085dd04fb5e31751b22f Mon Sep 17 00:00:00 2001 From: Joob1n Date: Tue, 15 Sep 2026 15:32:37 +0800 Subject: [PATCH 1/2] fix(parquet): decode DECIMAL page statistics by physical type ColumnIndexFilter inferred the physical type of DECIMAL page statistics from the length of the encoded min/max: 4 bytes meant INT32 and 8 bytes INT64, both decoded with a native-endian memcpy, and anything else was treated as big-endian two's complement. Parquet allows DECIMAL as FIXED_LEN_BYTE_ARRAY for any precision, with length ceil((precision * log2(10) + 1) / 8). Precision 7-9 yields exactly 4 bytes and precision 17-18 exactly 8, so those columns took the INT32/INT64 branch and had their big-endian bounds read little-endian. Page min 1.00 (unscaled 100, stored as 00 00 00 64) read back as 1,677,721,600, so `col < 5.00` dropped a page that does match. Page-index filtering is pruning, not filtering, so nothing downstream restores the rows. Pass the SchemaDescriptor down from FileReaderWrapper::CalculateFilteredRowRanges and dispatch on the column's physical type, the way parquet_stats_extractor already does for row-group statistics. When no schema is available the comparison is skipped so that no page is pruned. Paimon's own writers use INT32 for precision <= 9 and INT64 for precision <= 18, so this only reaches users through files written elsewhere: Arrow with the default store_decimal_as_integer = false, Spark with spark.sql.parquet.writeLegacyFormat = true, or Hive. Claude-Session: https://claude.ai/code/session_01NnX2Q5dAwrmeWYmf9n1mCC --- .../format/parquet/column_index_filter.cpp | 194 +++++++++++------- .../format/parquet/column_index_filter.h | 51 +++-- .../parquet/column_index_filter_test.cpp | 164 +++++++++++++-- .../format/parquet/file_reader_wrapper.cpp | 6 +- 4 files changed, 302 insertions(+), 113 deletions(-) diff --git a/src/paimon/format/parquet/column_index_filter.cpp b/src/paimon/format/parquet/column_index_filter.cpp index 1e0a8f597..29abf1899 100644 --- a/src/paimon/format/parquet/column_index_filter.cpp +++ b/src/paimon/format/parquet/column_index_filter.cpp @@ -39,27 +39,29 @@ namespace paimon::parquet { Result ColumnIndexFilter::CalculateRowRanges( const std::shared_ptr& predicate, const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - const std::map& column_name_to_index, int64_t row_group_row_count) { + const std::map& column_name_to_index, int64_t row_group_row_count, + const ::parquet::SchemaDescriptor* schema) { if (!predicate || !rg_page_index_reader) { return RowRanges::CreateSingle(row_group_row_count); } return VisitPredicate(predicate, column_name_to_index, row_group_row_count, - rg_page_index_reader.get()); + rg_page_index_reader.get(), schema); } Result ColumnIndexFilter::VisitPredicate( const std::shared_ptr& predicate, const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader) { + ::parquet::RowGroupPageIndexReader* rg_page_index_reader, + const ::parquet::SchemaDescriptor* schema) { if (auto leaf_predicate = std::dynamic_pointer_cast(predicate)) { return VisitLeafPredicate(leaf_predicate, column_name_to_index, row_group_row_count, - rg_page_index_reader); + rg_page_index_reader, schema); } if (auto compound_predicate = std::dynamic_pointer_cast(predicate)) { return VisitCompoundPredicate(compound_predicate, column_name_to_index, row_group_row_count, - rg_page_index_reader); + rg_page_index_reader, schema); } return Status::Invalid("Unknown predicate type"); @@ -68,7 +70,8 @@ Result ColumnIndexFilter::VisitPredicate( Result ColumnIndexFilter::VisitLeafPredicate( const std::shared_ptr& leaf_predicate, const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader) { + ::parquet::RowGroupPageIndexReader* rg_page_index_reader, + const ::parquet::SchemaDescriptor* schema) { const std::string& field_name = leaf_predicate->FieldName(); auto it = column_name_to_index.find(field_name); if (it == column_name_to_index.end()) { @@ -92,6 +95,14 @@ Result ColumnIndexFilter::VisitLeafPredicate( const auto& literals = leaf_predicate->Literals(); FieldType field_type = leaf_predicate->GetFieldType(); + // The physical type decides how the encoded page statistics are laid out. It cannot + // be inferred from the length of the encoded bounds: a DECIMAL stored as a 4-byte + // FIXED_LEN_BYTE_ARRAY (precision 7-9) is indistinguishable from an INT32 one by + // length alone, yet is big-endian rather than little-endian. + ::parquet::Type::type physical_type = ::parquet::Type::UNDEFINED; + if (schema && column_index < schema->num_columns()) { + physical_type = schema->Column(column_index)->physical_type(); + } if (function_type != Function::Type::IS_NULL && function_type != Function::Type::IS_NOT_NULL && literals.empty()) { @@ -108,25 +119,31 @@ Result ColumnIndexFilter::VisitLeafPredicate( matching_pages = FilterPagesByIsNotNull(column_index_ptr); break; case Function::Type::EQUAL: - matching_pages = FilterPagesByEqual(column_index_ptr, literals[0], field_type); + matching_pages = + FilterPagesByEqual(column_index_ptr, literals[0], field_type, physical_type); break; case Function::Type::NOT_EQUAL: - matching_pages = FilterPagesByNotEqual(column_index_ptr, literals[0], field_type); + matching_pages = + FilterPagesByNotEqual(column_index_ptr, literals[0], field_type, physical_type); break; case Function::Type::LESS_THAN: - matching_pages = FilterPagesByLessThan(column_index_ptr, literals[0], field_type); + matching_pages = + FilterPagesByLessThan(column_index_ptr, literals[0], field_type, physical_type); break; case Function::Type::LESS_OR_EQUAL: - matching_pages = FilterPagesByLessOrEqual(column_index_ptr, literals[0], field_type); + matching_pages = + FilterPagesByLessOrEqual(column_index_ptr, literals[0], field_type, physical_type); break; case Function::Type::GREATER_THAN: - matching_pages = FilterPagesByGreaterThan(column_index_ptr, literals[0], field_type); + matching_pages = + FilterPagesByGreaterThan(column_index_ptr, literals[0], field_type, physical_type); break; case Function::Type::GREATER_OR_EQUAL: - matching_pages = FilterPagesByGreaterOrEqual(column_index_ptr, literals[0], field_type); + matching_pages = FilterPagesByGreaterOrEqual(column_index_ptr, literals[0], field_type, + physical_type); break; case Function::Type::IN: - matching_pages = FilterPagesByIn(column_index_ptr, literals, field_type); + matching_pages = FilterPagesByIn(column_index_ptr, literals, field_type, physical_type); break; case Function::Type::NOT_IN: matching_pages = FilterPagesByNotIn(column_index_ptr, literals); @@ -142,7 +159,8 @@ Result ColumnIndexFilter::VisitLeafPredicate( Result ColumnIndexFilter::VisitCompoundPredicate( const std::shared_ptr& compound_predicate, const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader) { + ::parquet::RowGroupPageIndexReader* rg_page_index_reader, + const ::parquet::SchemaDescriptor* schema) { const auto& children = compound_predicate->Children(); const auto& function = compound_predicate->GetFunction(); auto function_type = function.GetType(); @@ -154,7 +172,7 @@ Result ColumnIndexFilter::VisitCompoundPredicate( // Calculate row ranges for first child PAIMON_ASSIGN_OR_RAISE(RowRanges result, VisitPredicate(children[0], column_name_to_index, row_group_row_count, - rg_page_index_reader)); + rg_page_index_reader, schema)); if (function_type == Function::Type::AND) { // Short-circuit: if result is empty, no need to continue @@ -163,9 +181,10 @@ Result ColumnIndexFilter::VisitCompoundPredicate( } for (size_t i = 1; i < children.size(); ++i) { - PAIMON_ASSIGN_OR_RAISE(RowRanges child_ranges, - VisitPredicate(children[i], column_name_to_index, - row_group_row_count, rg_page_index_reader)); + PAIMON_ASSIGN_OR_RAISE( + RowRanges child_ranges, + VisitPredicate(children[i], column_name_to_index, row_group_row_count, + rg_page_index_reader, schema)); result = RowRanges::Intersection(result, child_ranges); @@ -181,9 +200,10 @@ Result ColumnIndexFilter::VisitCompoundPredicate( } for (size_t i = 1; i < children.size(); ++i) { - PAIMON_ASSIGN_OR_RAISE(RowRanges child_ranges, - VisitPredicate(children[i], column_name_to_index, - row_group_row_count, rg_page_index_reader)); + PAIMON_ASSIGN_OR_RAISE( + RowRanges child_ranges, + VisitPredicate(children[i], column_name_to_index, row_group_row_count, + rg_page_index_reader, schema)); result = RowRanges::Union(result, child_ranges); @@ -201,7 +221,7 @@ Result ColumnIndexFilter::VisitCompoundPredicate( std::vector ColumnIndexFilter::FilterPagesByEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type) { + FieldType field_type, ::parquet::Type::type physical_type) { std::vector matching_pages; if (literal.IsNull()) { @@ -219,7 +239,8 @@ std::vector ColumnIndexFilter::FilterPagesByEqual( continue; } - if (PageMightContainEqual(min_values[i], max_values[i], literal, field_type)) { + if (PageMightContainEqual(min_values[i], max_values[i], literal, field_type, + physical_type)) { matching_pages.push_back(i); } } @@ -229,7 +250,7 @@ std::vector ColumnIndexFilter::FilterPagesByEqual( std::vector ColumnIndexFilter::FilterPagesByNotEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type) { + FieldType field_type, ::parquet::Type::type physical_type) { std::vector matching_pages; if (literal.IsNull()) { @@ -251,8 +272,8 @@ std::vector ColumnIndexFilter::FilterPagesByNotEqual( // Try to exclude pages where min == max == literal (all non-null values equal literal). // NULL != literal is NULL (UNKNOWN) in SQL, so nulls don't produce true either. - auto cmp_min = CompareEncodedWithLiteral(min_values[i], literal, field_type); - auto cmp_max = CompareEncodedWithLiteral(max_values[i], literal, field_type); + auto cmp_min = CompareEncodedWithLiteral(min_values[i], literal, field_type, physical_type); + auto cmp_max = CompareEncodedWithLiteral(max_values[i], literal, field_type, physical_type); if (cmp_min.has_value() && cmp_max.has_value() && *cmp_min == 0 && *cmp_max == 0) { // min == max == literal: all non-null values equal literal, and nulls // don't satisfy != either. Skip this page entirely. @@ -267,7 +288,7 @@ std::vector ColumnIndexFilter::FilterPagesByNotEqual( std::vector ColumnIndexFilter::FilterPagesByLessThan( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type) { + FieldType field_type, ::parquet::Type::type physical_type) { std::vector matching_pages; const auto& null_pages = column_index->null_pages(); const auto& min_values = column_index->encoded_min_values(); @@ -278,7 +299,7 @@ std::vector ColumnIndexFilter::FilterPagesByLessThan( continue; } - if (PageMightContainLessThan(min_values[i], literal, field_type)) { + if (PageMightContainLessThan(min_values[i], literal, field_type, physical_type)) { matching_pages.push_back(i); } } @@ -288,7 +309,7 @@ std::vector ColumnIndexFilter::FilterPagesByLessThan( std::vector ColumnIndexFilter::FilterPagesByLessOrEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type) { + FieldType field_type, ::parquet::Type::type physical_type) { std::vector matching_pages; const auto& null_pages = column_index->null_pages(); const auto& min_values = column_index->encoded_min_values(); @@ -299,7 +320,7 @@ std::vector ColumnIndexFilter::FilterPagesByLessOrEqual( continue; } - if (PageMightContainLessOrEqual(min_values[i], literal, field_type)) { + if (PageMightContainLessOrEqual(min_values[i], literal, field_type, physical_type)) { matching_pages.push_back(i); } } @@ -309,7 +330,7 @@ std::vector ColumnIndexFilter::FilterPagesByLessOrEqual( std::vector ColumnIndexFilter::FilterPagesByGreaterThan( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type) { + FieldType field_type, ::parquet::Type::type physical_type) { std::vector matching_pages; const auto& null_pages = column_index->null_pages(); const auto& max_values = column_index->encoded_max_values(); @@ -320,7 +341,7 @@ std::vector ColumnIndexFilter::FilterPagesByGreaterThan( continue; } - if (PageMightContainGreaterThan(max_values[i], literal, field_type)) { + if (PageMightContainGreaterThan(max_values[i], literal, field_type, physical_type)) { matching_pages.push_back(i); } } @@ -330,7 +351,7 @@ std::vector ColumnIndexFilter::FilterPagesByGreaterThan( std::vector ColumnIndexFilter::FilterPagesByGreaterOrEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type) { + FieldType field_type, ::parquet::Type::type physical_type) { std::vector matching_pages; const auto& null_pages = column_index->null_pages(); const auto& max_values = column_index->encoded_max_values(); @@ -341,7 +362,7 @@ std::vector ColumnIndexFilter::FilterPagesByGreaterOrEqual( continue; } - if (PageMightContainGreaterOrEqual(max_values[i], literal, field_type)) { + if (PageMightContainGreaterOrEqual(max_values[i], literal, field_type, physical_type)) { matching_pages.push_back(i); } } @@ -390,7 +411,8 @@ std::vector ColumnIndexFilter::FilterPagesByIsNotNull( std::vector ColumnIndexFilter::FilterPagesByIn( const std::shared_ptr<::parquet::ColumnIndex>& column_index, - const std::vector& literals, FieldType field_type) { + const std::vector& literals, FieldType field_type, + ::parquet::Type::type physical_type) { std::vector matching_pages; const auto& null_pages = column_index->null_pages(); const auto& min_values = column_index->encoded_min_values(); @@ -426,7 +448,8 @@ std::vector ColumnIndexFilter::FilterPagesByIn( if (literal.IsNull()) { continue; } - if (PageMightContainEqual(min_values[i], max_values[i], literal, field_type)) { + if (PageMightContainEqual(min_values[i], max_values[i], literal, field_type, + physical_type)) { matching_pages.push_back(i); break; // Page matched, no need to check more literals } @@ -501,9 +524,9 @@ RowRanges ColumnIndexFilter::BuildRowRangesFromPageIndices( return ranges; } -std::optional ColumnIndexFilter::CompareEncodedWithLiteral(const std::string& encoded, - const Literal& literal, - FieldType field_type) { +std::optional ColumnIndexFilter::CompareEncodedWithLiteral( + const std::string& encoded, const Literal& literal, FieldType field_type, + ::parquet::Type::type physical_type) { if (literal.IsNull()) { return std::nullopt; } @@ -570,33 +593,54 @@ std::optional ColumnIndexFilter::CompareEncodedWithLiteral(const std::s return (cmp < 0) ? -1 : (cmp > 0) ? 1 : 0; } case FieldType::DECIMAL: { - // Parquet stores DECIMAL as INT32, INT64, or FIXED_LEN_BYTE_ARRAY depending - // on precision. All are stored as unscaled integer values. + // Parquet stores DECIMAL as INT32, INT64, FIXED_LEN_BYTE_ARRAY or BYTE_ARRAY. + // All carry the unscaled integer value, but INT32/INT64 are little-endian + // plain values while the byte arrays are big-endian two's complement, so the + // physical type from the file schema must decide. The encoded length cannot: + // FIXED_LEN_BYTE_ARRAY is legal for any precision with length + // ceil((precision * log2(10) + 1) / 8), which is 4 bytes for precision 7-9 + // and 8 bytes for precision 17-18. auto lit_decimal = literal.GetValue(); Decimal::int128_t lit_val = lit_decimal.Value(); Decimal::int128_t enc_val; - if (encoded.size() == sizeof(int32_t)) { - // INT32 physical type (precision <= 9) - int32_t raw; - std::memcpy(&raw, encoded.data(), sizeof(int32_t)); - enc_val = static_cast(raw); - } else if (encoded.size() == sizeof(int64_t)) { - // INT64 physical type (precision <= 18) - int64_t raw; - std::memcpy(&raw, encoded.data(), sizeof(int64_t)); - enc_val = static_cast(raw); - } else { - // FIXED_LEN_BYTE_ARRAY / BYTE_ARRAY: big-endian two's complement. - // Defer to Decimal::FromUnscaledBytes so endianness, padding, and - // sign extension stay consistent with parquet_stats_extractor. - if (encoded.empty()) { - return std::nullopt; + switch (physical_type) { + case ::parquet::Type::INT32: { + if (encoded.size() < sizeof(int32_t)) { + return std::nullopt; + } + int32_t raw; + std::memcpy(&raw, encoded.data(), sizeof(int32_t)); + enc_val = static_cast(raw); + break; } - Bytes bytes(encoded, GetDefaultPool().get()); - enc_val = - Decimal::FromUnscaledBytes(lit_decimal.Precision(), lit_decimal.Scale(), &bytes) - .Value(); + case ::parquet::Type::INT64: { + if (encoded.size() < sizeof(int64_t)) { + return std::nullopt; + } + int64_t raw; + std::memcpy(&raw, encoded.data(), sizeof(int64_t)); + enc_val = static_cast(raw); + break; + } + case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: + case ::parquet::Type::BYTE_ARRAY: { + // Big-endian two's complement. Defer to Decimal::FromUnscaledBytes so + // endianness, padding, and sign extension stay consistent with + // parquet_stats_extractor. + if (encoded.empty()) { + return std::nullopt; + } + Bytes bytes(encoded, GetDefaultPool().get()); + enc_val = Decimal::FromUnscaledBytes(lit_decimal.Precision(), + lit_decimal.Scale(), &bytes) + .Value(); + break; + } + default: + // Physical type unknown (no schema available) or not a valid decimal + // encoding: fall back to safe behavior (include page). + return std::nullopt; } return (enc_val < lit_val) ? -1 : (enc_val > lit_val) ? 1 : 0; @@ -611,13 +655,14 @@ std::optional ColumnIndexFilter::CompareEncodedWithLiteral(const std::s bool ColumnIndexFilter::PageMightContainEqual(const std::string& encoded_min, const std::string& encoded_max, - const Literal& literal, FieldType field_type) { + const Literal& literal, FieldType field_type, + ::parquet::Type::type physical_type) { if (literal.IsNull()) { return false; // Null is handled separately via null_pages } // Page might contain equal if min <= literal <= max - auto cmp_min = CompareEncodedWithLiteral(encoded_min, literal, field_type); + auto cmp_min = CompareEncodedWithLiteral(encoded_min, literal, field_type, physical_type); if (!cmp_min.has_value()) { return true; // Can't compare, assume match } @@ -625,7 +670,7 @@ bool ColumnIndexFilter::PageMightContainEqual(const std::string& encoded_min, return false; // min > literal } - auto cmp_max = CompareEncodedWithLiteral(encoded_max, literal, field_type); + auto cmp_max = CompareEncodedWithLiteral(encoded_max, literal, field_type, physical_type); if (!cmp_max.has_value()) { return true; } @@ -637,13 +682,14 @@ bool ColumnIndexFilter::PageMightContainEqual(const std::string& encoded_min, } bool ColumnIndexFilter::PageMightContainLessThan(const std::string& encoded_min, - const Literal& literal, FieldType field_type) { + const Literal& literal, FieldType field_type, + ::parquet::Type::type physical_type) { if (literal.IsNull()) { return false; } // Page might contain values < literal if min < literal - auto cmp_min = CompareEncodedWithLiteral(encoded_min, literal, field_type); + auto cmp_min = CompareEncodedWithLiteral(encoded_min, literal, field_type, physical_type); if (!cmp_min.has_value()) { return true; } @@ -651,13 +697,14 @@ bool ColumnIndexFilter::PageMightContainLessThan(const std::string& encoded_min, } bool ColumnIndexFilter::PageMightContainLessOrEqual(const std::string& encoded_min, - const Literal& literal, FieldType field_type) { + const Literal& literal, FieldType field_type, + ::parquet::Type::type physical_type) { if (literal.IsNull()) { return false; } // Page might contain values <= literal if min <= literal - auto cmp_min = CompareEncodedWithLiteral(encoded_min, literal, field_type); + auto cmp_min = CompareEncodedWithLiteral(encoded_min, literal, field_type, physical_type); if (!cmp_min.has_value()) { return true; } @@ -665,13 +712,14 @@ bool ColumnIndexFilter::PageMightContainLessOrEqual(const std::string& encoded_m } bool ColumnIndexFilter::PageMightContainGreaterThan(const std::string& encoded_max, - const Literal& literal, FieldType field_type) { + const Literal& literal, FieldType field_type, + ::parquet::Type::type physical_type) { if (literal.IsNull()) { return false; } // Page might contain values > literal if max > literal - auto cmp_max = CompareEncodedWithLiteral(encoded_max, literal, field_type); + auto cmp_max = CompareEncodedWithLiteral(encoded_max, literal, field_type, physical_type); if (!cmp_max.has_value()) { return true; } @@ -679,14 +727,14 @@ bool ColumnIndexFilter::PageMightContainGreaterThan(const std::string& encoded_m } bool ColumnIndexFilter::PageMightContainGreaterOrEqual(const std::string& encoded_max, - const Literal& literal, - FieldType field_type) { + const Literal& literal, FieldType field_type, + ::parquet::Type::type physical_type) { if (literal.IsNull()) { return false; } // Page might contain values >= literal if max >= literal - auto cmp_max = CompareEncodedWithLiteral(encoded_max, literal, field_type); + auto cmp_max = CompareEncodedWithLiteral(encoded_max, literal, field_type, physical_type); if (!cmp_max.has_value()) { return true; } diff --git a/src/paimon/format/parquet/column_index_filter.h b/src/paimon/format/parquet/column_index_filter.h index a30162416..555aae0ae 100644 --- a/src/paimon/format/parquet/column_index_filter.h +++ b/src/paimon/format/parquet/column_index_filter.h @@ -32,6 +32,7 @@ #include "paimon/predicate/predicate.h" #include "paimon/result.h" #include "parquet/page_index.h" +#include "parquet/schema.h" namespace paimon { class CompoundPredicate; @@ -60,60 +61,68 @@ class ColumnIndexFilter { /// @param rg_page_index_reader The page index reader of target row group for the file. /// @param column_name_to_index Map from column name to column index. /// @param row_group_row_count The number of rows in the row group. + /// @param schema The schema descriptor of the file, used to learn the physical type + /// each column is encoded with. May be null; page statistics whose decoding + /// depends on the physical type are then left uncompared so that no page is + /// pruned on them. /// @return RowRanges that may contain matching rows. static Result CalculateRowRanges( const std::shared_ptr& predicate, const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - const std::map& column_name_to_index, int64_t row_group_row_count); + const std::map& column_name_to_index, int64_t row_group_row_count, + const ::parquet::SchemaDescriptor* schema); private: /// Visit a predicate and calculate row ranges. static Result VisitPredicate( const std::shared_ptr& predicate, const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader); + ::parquet::RowGroupPageIndexReader* rg_page_index_reader, + const ::parquet::SchemaDescriptor* schema); /// Visit a leaf predicate and calculate row ranges. static Result VisitLeafPredicate( const std::shared_ptr& leaf_predicate, const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader); + ::parquet::RowGroupPageIndexReader* rg_page_index_reader, + const ::parquet::SchemaDescriptor* schema); /// Visit a compound predicate (AND/OR) and calculate row ranges. static Result VisitCompoundPredicate( const std::shared_ptr& compound_predicate, const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader); + ::parquet::RowGroupPageIndexReader* rg_page_index_reader, + const ::parquet::SchemaDescriptor* schema); /// Filter pages based on column index statistics for EQUAL predicate. static std::vector FilterPagesByEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Filter pages based on column index statistics for NOT_EQUAL predicate. static std::vector FilterPagesByNotEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Filter pages based on column index statistics for LESS_THAN predicate. static std::vector FilterPagesByLessThan( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Filter pages based on column index statistics for LESS_OR_EQUAL predicate. static std::vector FilterPagesByLessOrEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Filter pages based on column index statistics for GREATER_THAN predicate. static std::vector FilterPagesByGreaterThan( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Filter pages based on column index statistics for GREATER_OR_EQUAL predicate. static std::vector FilterPagesByGreaterOrEqual( const std::shared_ptr<::parquet::ColumnIndex>& column_index, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Filter pages based on column index statistics for IS_NULL predicate. static std::vector FilterPagesByIsNull( @@ -126,7 +135,8 @@ class ColumnIndexFilter { /// Filter pages based on column index statistics for IN predicate. static std::vector FilterPagesByIn( const std::shared_ptr<::parquet::ColumnIndex>& column_index, - const std::vector& literals, FieldType field_type); + const std::vector& literals, FieldType field_type, + ::parquet::Type::type physical_type); /// Filter pages based on column index statistics for NOT_IN predicate. static std::vector FilterPagesByNotIn( @@ -139,37 +149,44 @@ class ColumnIndexFilter { const std::shared_ptr<::parquet::OffsetIndex>& offset_index, int64_t row_group_row_count); /// Compare a parquet encoded value with a Literal. + /// @param physical_type The parquet physical type the column is encoded with, which + /// decides how the encoded statistics are decoded for types that Parquet can + /// store in more than one physical representation. /// @return -1 if encoded < literal, 0 if equal, 1 if encoded > literal. /// nullopt if comparison cannot be performed (unsupported type, etc.). static std::optional CompareEncodedWithLiteral(const std::string& encoded, const Literal& literal, - FieldType field_type); + FieldType field_type, + ::parquet::Type::type physical_type); /// Check if a page might contain a value equal to the literal. /// Condition: min <= literal <= max static bool PageMightContainEqual(const std::string& encoded_min, const std::string& encoded_max, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Check if a page might contain values less than the literal. /// Condition: min < literal static bool PageMightContainLessThan(const std::string& encoded_min, const Literal& literal, - FieldType field_type); + FieldType field_type, ::parquet::Type::type physical_type); /// Check if a page might contain values less than or equal to the literal. /// Condition: min <= literal static bool PageMightContainLessOrEqual(const std::string& encoded_min, const Literal& literal, - FieldType field_type); + FieldType field_type, + ::parquet::Type::type physical_type); /// Check if a page might contain values greater than the literal. /// Condition: max > literal static bool PageMightContainGreaterThan(const std::string& encoded_max, const Literal& literal, - FieldType field_type); + FieldType field_type, + ::parquet::Type::type physical_type); /// Check if a page might contain values greater than or equal to the literal. /// Condition: max >= literal static bool PageMightContainGreaterOrEqual(const std::string& encoded_max, - const Literal& literal, FieldType field_type); + const Literal& literal, FieldType field_type, + ::parquet::Type::type physical_type); }; } // namespace paimon::parquet diff --git a/src/paimon/format/parquet/column_index_filter_test.cpp b/src/paimon/format/parquet/column_index_filter_test.cpp index 94c65ef5f..0c7f64bab 100644 --- a/src/paimon/format/parquet/column_index_filter_test.cpp +++ b/src/paimon/format/parquet/column_index_filter_test.cpp @@ -36,6 +36,7 @@ #include "paimon/common/predicate/leaf_predicate_impl.h" #include "paimon/common/utils/arrow/arrow_input_stream_adapter.h" #include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/data/decimal.h" #include "paimon/defs.h" #include "paimon/format/parquet/parquet_format_defs.h" #include "paimon/format/parquet/parquet_format_writer.h" @@ -306,7 +307,8 @@ class ColumnIndexFilterTest : public ::testing::Test { Result Filter(const std::shared_ptr& predicate) { return ColumnIndexFilter::CalculateRowRanges(predicate, page_index_reader_->RowGroup(0), - column_name_to_index_, row_group_row_count_); + column_name_to_index_, row_group_row_count_, + parquet_reader_->metadata()->schema()); } std::shared_ptr arrow_pool_; @@ -552,19 +554,20 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { auto less_negative_zero = PredicateBuilder::LessThan( /*field_index=*/0, /*field_name=*/"value", field_type, field_type == FieldType::FLOAT ? Literal(-0.0f) : Literal(-0.0)); - ASSERT_OK_AND_ASSIGN(auto ranges, - ColumnIndexFilter::CalculateRowRanges( - less_negative_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows())); + ASSERT_OK_AND_ASSIGN( + auto ranges, + ColumnIndexFilter::CalculateRowRanges( + less_negative_zero, page_index_reader->RowGroup(0), {{"value", 0}}, + reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); ASSERT_TRUE(ranges.IsEmpty()) << "field type: " << static_cast(field_type); auto less_positive_zero = PredicateBuilder::LessThan( /*field_index=*/0, /*field_name=*/"value", field_type, field_type == FieldType::FLOAT ? Literal(0.0f) : Literal(0.0)); - ASSERT_OK_AND_ASSIGN(ranges, - ColumnIndexFilter::CalculateRowRanges( - less_positive_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows())); + ASSERT_OK_AND_ASSIGN( + ranges, ColumnIndexFilter::CalculateRowRanges( + less_positive_zero, page_index_reader->RowGroup(0), {{"value", 0}}, + reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); ASSERT_EQ(20, ranges.RowCount()); ASSERT_EQ(1, ranges.GetRanges().size()); ASSERT_EQ(0, ranges.GetRanges()[0].from); @@ -576,7 +579,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( greater_negative_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows())); + reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); ASSERT_EQ(30, ranges.RowCount()); auto not_equal_negative_zero = PredicateBuilder::NotEqual( @@ -585,16 +588,16 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( not_equal_negative_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows())); + reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); ASSERT_EQ(30, ranges.RowCount()); auto greater_finite = PredicateBuilder::GreaterThan( /*field_index=*/0, /*field_name=*/"value", field_type, field_type == FieldType::FLOAT ? Literal(2.0f) : Literal(2.0)); - ASSERT_OK_AND_ASSIGN(ranges, - ColumnIndexFilter::CalculateRowRanges( - greater_finite, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows())); + ASSERT_OK_AND_ASSIGN( + ranges, ColumnIndexFilter::CalculateRowRanges( + greater_finite, page_index_reader->RowGroup(0), {{"value", 0}}, + reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); ASSERT_TRUE(ranges.IsEmpty()); auto greater_between_pages = PredicateBuilder::GreaterThan( @@ -603,7 +606,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( greater_between_pages, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows())); + reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); ASSERT_EQ(10, ranges.RowCount()); ASSERT_EQ(1, ranges.GetRanges().size()); ASSERT_EQ(20, ranges.GetRanges()[0].from); @@ -612,14 +615,135 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { auto equal_finite = PredicateBuilder::Equal( /*field_index=*/0, /*field_name=*/"value", field_type, field_type == FieldType::FLOAT ? Literal(2.0f) : Literal(2.0)); - ASSERT_OK_AND_ASSIGN(ranges, - ColumnIndexFilter::CalculateRowRanges( - equal_finite, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows())); + ASSERT_OK_AND_ASSIGN( + ranges, ColumnIndexFilter::CalculateRowRanges( + equal_finite, page_index_reader->RowGroup(0), {{"value", 0}}, + reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); ASSERT_TRUE(ranges.IsEmpty()); } } +/// Arrow writes DECIMAL as FIXED_LEN_BYTE_ARRAY unless store_decimal_as_integer is set, +/// and those page bounds are big-endian two's complement. Precision 7-9 encodes into +/// exactly 4 bytes and precision 17-18 into exactly 8 bytes, so the encoded length alone +/// cannot tell them apart from INT32/INT64 decimals; decoding them little-endian turns +/// min 1.00 into 1,677,721,600 and prunes away pages that do match. +TEST_F(ColumnIndexFilterTest, DecimalFixedLenByteArrayPages) { + // Page 0: 1.00, page 1: 10.00, page 2: 100.00 (10 rows each). + constexpr int32_t kScale = 2; + const std::vector page_unscaled_values = {100, 1000, 10000}; + + for (int32_t precision : {9, 18}) { + auto decimal_type = arrow::decimal128(precision, kScale); + arrow::Decimal128Builder builder(decimal_type); + for (int64_t unscaled : page_unscaled_values) { + for (int32_t i = 0; i < 10; ++i) { + ASSERT_TRUE(builder.Append(arrow::Decimal128(unscaled)).ok()); + } + } + std::shared_ptr values = builder.Finish().ValueOrDie(); + auto field = arrow::field("value", decimal_type); + auto data = arrow::StructArray::Make({values}, {field}).ValueOrDie(); + + std::string file_name = + dir_->Str() + "/decimal_flba_p" + std::to_string(precision) + ".parquet"; + WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/30); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr in, fs_->Open(file_name)); + ASSERT_OK_AND_ASSIGN(int64_t length, in->Length()); + auto in_stream = std::make_shared(in, length, arrow_pool_); + auto reader = ::parquet::ParquetFileReader::Open(in_stream); + ASSERT_TRUE(reader); + const auto* schema = reader->metadata()->schema(); + // The fix is only exercised if the writer really chose the byte-array encoding + // with a length that collides with INT32 (precision 9) or INT64 (precision 18). + ASSERT_EQ(::parquet::Type::FIXED_LEN_BYTE_ARRAY, schema->Column(0)->physical_type()); + ASSERT_EQ(precision == 9 ? 4 : 8, schema->Column(0)->type_length()); + + auto page_index_reader = reader->GetPageIndexReader(); + ASSERT_TRUE(page_index_reader); + auto row_group_page_index = page_index_reader->RowGroup(0); + ASSERT_TRUE(row_group_page_index); + auto column_index = row_group_page_index->GetColumnIndex(0); + ASSERT_TRUE(column_index); + ASSERT_EQ(3, column_index->encoded_min_values().size()); + + int64_t row_count = reader->metadata()->RowGroup(0)->num_rows(); + auto filter = [&](const std::shared_ptr& predicate) { + return ColumnIndexFilter::CalculateRowRanges(predicate, page_index_reader->RowGroup(0), + {{"value", 0}}, row_count, schema); + }; + + // value < 5.00 → only page 0 (rows [0, 9]). + auto less_than = PredicateBuilder::LessThan( + /*field_index=*/0, /*field_name=*/"value", FieldType::DECIMAL, + Literal(Decimal(precision, kScale, 500))); + ASSERT_OK_AND_ASSIGN(auto ranges, filter(less_than)); + ASSERT_EQ(1, ranges.GetRanges().size()) << "precision: " << precision; + EXPECT_EQ(0, ranges.GetRanges()[0].from); + EXPECT_EQ(9, ranges.GetRanges()[0].to); + + // value > 50.00 → only page 2 (rows [20, 29]). + auto greater_than = PredicateBuilder::GreaterThan( + /*field_index=*/0, /*field_name=*/"value", FieldType::DECIMAL, + Literal(Decimal(precision, kScale, 5000))); + ASSERT_OK_AND_ASSIGN(ranges, filter(greater_than)); + ASSERT_EQ(1, ranges.GetRanges().size()) << "precision: " << precision; + EXPECT_EQ(20, ranges.GetRanges()[0].from); + EXPECT_EQ(29, ranges.GetRanges()[0].to); + + // value = 10.00 → only page 1 (rows [10, 19]). + auto equal = PredicateBuilder::Equal( + /*field_index=*/0, /*field_name=*/"value", FieldType::DECIMAL, + Literal(Decimal(precision, kScale, 1000))); + ASSERT_OK_AND_ASSIGN(ranges, filter(equal)); + ASSERT_EQ(1, ranges.GetRanges().size()) << "precision: " << precision; + EXPECT_EQ(10, ranges.GetRanges()[0].from); + EXPECT_EQ(19, ranges.GetRanges()[0].to); + + // value > 1000.00 → no page matches. + auto greater_than_all = PredicateBuilder::GreaterThan( + /*field_index=*/0, /*field_name=*/"value", FieldType::DECIMAL, + Literal(Decimal(precision, kScale, 100000))); + ASSERT_OK_AND_ASSIGN(ranges, filter(greater_than_all)); + EXPECT_TRUE(ranges.IsEmpty()) << "precision: " << precision; + } +} + +/// Without a schema descriptor the physical type of a DECIMAL column is unknown, so no +/// page may be pruned on its statistics. +TEST_F(ColumnIndexFilterTest, DecimalWithoutSchemaKeepsAllPages) { + auto decimal_type = arrow::decimal128(9, 2); + arrow::Decimal128Builder builder(decimal_type); + for (int64_t unscaled : {100, 1000, 10000}) { + for (int32_t i = 0; i < 10; ++i) { + ASSERT_TRUE(builder.Append(arrow::Decimal128(unscaled)).ok()); + } + } + std::shared_ptr values = builder.Finish().ValueOrDie(); + auto field = arrow::field("value", decimal_type); + auto data = arrow::StructArray::Make({values}, {field}).ValueOrDie(); + + std::string file_name = dir_->Str() + "/decimal_no_schema.parquet"; + WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/30); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr in, fs_->Open(file_name)); + ASSERT_OK_AND_ASSIGN(int64_t length, in->Length()); + auto in_stream = std::make_shared(in, length, arrow_pool_); + auto reader = ::parquet::ParquetFileReader::Open(in_stream); + ASSERT_TRUE(reader); + auto page_index_reader = reader->GetPageIndexReader(); + ASSERT_TRUE(page_index_reader); + + int64_t row_count = reader->metadata()->RowGroup(0)->num_rows(); + auto less_than = PredicateBuilder::LessThan( + /*field_index=*/0, /*field_name=*/"value", FieldType::DECIMAL, Literal(Decimal(9, 2, 500))); + ASSERT_OK_AND_ASSIGN(auto ranges, ColumnIndexFilter::CalculateRowRanges( + less_than, page_index_reader->RowGroup(0), {{"value", 0}}, + row_count, /*schema=*/nullptr)); + EXPECT_EQ(row_count, ranges.RowCount()); +} + /// Predicates referencing fields absent from the data file are stripped upstream /// by FieldMappingBuilder, so reaching ColumnIndexFilter with such a predicate is /// a contract violation and surfaces as an error. diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp b/src/paimon/format/parquet/file_reader_wrapper.cpp index f69b89d46..0cdab1bd8 100644 --- a/src/paimon/format/parquet/file_reader_wrapper.cpp +++ b/src/paimon/format/parquet/file_reader_wrapper.cpp @@ -631,9 +631,9 @@ Result FileReaderWrapper::CalculateFilteredRowRanges( {/*column_index=*/true, /*offset_index=*/true}); // Keep this restricted reader separate: projected payload columns still need // their offset indexes when the data reader is initialized later. - return ColumnIndexFilter::CalculateRowRanges(predicate, - page_index_reader->RowGroup(row_group_index), - column_name_to_index, row_count); + return ColumnIndexFilter::CalculateRowRanges( + predicate, page_index_reader->RowGroup(row_group_index), column_name_to_index, + row_count, meta_data->schema()); } PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("FileReaderWrapper::CalculateFilteredRowRanges") } From 263f44a4ac30c744d847cf23abb5bd4ae19f909a Mon Sep 17 00:00:00 2001 From: Joob1n Date: Tue, 15 Sep 2026 16:03:15 +0800 Subject: [PATCH 2/2] style(parquet): move the schema descriptor out of the trailing parameter slot Address review feedback: the trailing position is reserved for output parameters, so the const SchemaDescriptor* input now sits next to column_name_to_index instead of at the end of the list. Claude-Session: https://claude.ai/code/session_01NnX2Q5dAwrmeWYmf9n1mCC --- .../format/parquet/column_index_filter.cpp | 52 +++++++++---------- .../format/parquet/column_index_filter.h | 24 ++++----- .../parquet/column_index_filter_test.cpp | 31 ++++++----- .../format/parquet/file_reader_wrapper.cpp | 2 +- 4 files changed, 53 insertions(+), 56 deletions(-) diff --git a/src/paimon/format/parquet/column_index_filter.cpp b/src/paimon/format/parquet/column_index_filter.cpp index 29abf1899..74dc37fee 100644 --- a/src/paimon/format/parquet/column_index_filter.cpp +++ b/src/paimon/format/parquet/column_index_filter.cpp @@ -39,29 +39,29 @@ namespace paimon::parquet { Result ColumnIndexFilter::CalculateRowRanges( const std::shared_ptr& predicate, const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - const std::map& column_name_to_index, int64_t row_group_row_count, - const ::parquet::SchemaDescriptor* schema) { + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count) { if (!predicate || !rg_page_index_reader) { return RowRanges::CreateSingle(row_group_row_count); } - return VisitPredicate(predicate, column_name_to_index, row_group_row_count, - rg_page_index_reader.get(), schema); + return VisitPredicate(predicate, column_name_to_index, schema, row_group_row_count, + rg_page_index_reader.get()); } Result ColumnIndexFilter::VisitPredicate( const std::shared_ptr& predicate, - const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader, - const ::parquet::SchemaDescriptor* schema) { + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count, + ::parquet::RowGroupPageIndexReader* rg_page_index_reader) { if (auto leaf_predicate = std::dynamic_pointer_cast(predicate)) { - return VisitLeafPredicate(leaf_predicate, column_name_to_index, row_group_row_count, - rg_page_index_reader, schema); + return VisitLeafPredicate(leaf_predicate, column_name_to_index, schema, row_group_row_count, + rg_page_index_reader); } if (auto compound_predicate = std::dynamic_pointer_cast(predicate)) { - return VisitCompoundPredicate(compound_predicate, column_name_to_index, row_group_row_count, - rg_page_index_reader, schema); + return VisitCompoundPredicate(compound_predicate, column_name_to_index, schema, + row_group_row_count, rg_page_index_reader); } return Status::Invalid("Unknown predicate type"); @@ -69,9 +69,9 @@ Result ColumnIndexFilter::VisitPredicate( Result ColumnIndexFilter::VisitLeafPredicate( const std::shared_ptr& leaf_predicate, - const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader, - const ::parquet::SchemaDescriptor* schema) { + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count, + ::parquet::RowGroupPageIndexReader* rg_page_index_reader) { const std::string& field_name = leaf_predicate->FieldName(); auto it = column_name_to_index.find(field_name); if (it == column_name_to_index.end()) { @@ -158,9 +158,9 @@ Result ColumnIndexFilter::VisitLeafPredicate( Result ColumnIndexFilter::VisitCompoundPredicate( const std::shared_ptr& compound_predicate, - const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader, - const ::parquet::SchemaDescriptor* schema) { + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count, + ::parquet::RowGroupPageIndexReader* rg_page_index_reader) { const auto& children = compound_predicate->Children(); const auto& function = compound_predicate->GetFunction(); auto function_type = function.GetType(); @@ -171,8 +171,8 @@ Result ColumnIndexFilter::VisitCompoundPredicate( // Calculate row ranges for first child PAIMON_ASSIGN_OR_RAISE(RowRanges result, - VisitPredicate(children[0], column_name_to_index, row_group_row_count, - rg_page_index_reader, schema)); + VisitPredicate(children[0], column_name_to_index, schema, + row_group_row_count, rg_page_index_reader)); if (function_type == Function::Type::AND) { // Short-circuit: if result is empty, no need to continue @@ -181,10 +181,9 @@ Result ColumnIndexFilter::VisitCompoundPredicate( } for (size_t i = 1; i < children.size(); ++i) { - PAIMON_ASSIGN_OR_RAISE( - RowRanges child_ranges, - VisitPredicate(children[i], column_name_to_index, row_group_row_count, - rg_page_index_reader, schema)); + PAIMON_ASSIGN_OR_RAISE(RowRanges child_ranges, + VisitPredicate(children[i], column_name_to_index, schema, + row_group_row_count, rg_page_index_reader)); result = RowRanges::Intersection(result, child_ranges); @@ -200,10 +199,9 @@ Result ColumnIndexFilter::VisitCompoundPredicate( } for (size_t i = 1; i < children.size(); ++i) { - PAIMON_ASSIGN_OR_RAISE( - RowRanges child_ranges, - VisitPredicate(children[i], column_name_to_index, row_group_row_count, - rg_page_index_reader, schema)); + PAIMON_ASSIGN_OR_RAISE(RowRanges child_ranges, + VisitPredicate(children[i], column_name_to_index, schema, + row_group_row_count, rg_page_index_reader)); result = RowRanges::Union(result, child_ranges); diff --git a/src/paimon/format/parquet/column_index_filter.h b/src/paimon/format/parquet/column_index_filter.h index 555aae0ae..a0b7eecf3 100644 --- a/src/paimon/format/parquet/column_index_filter.h +++ b/src/paimon/format/parquet/column_index_filter.h @@ -60,39 +60,39 @@ class ColumnIndexFilter { /// @param predicate The predicate to evaluate. /// @param rg_page_index_reader The page index reader of target row group for the file. /// @param column_name_to_index Map from column name to column index. - /// @param row_group_row_count The number of rows in the row group. /// @param schema The schema descriptor of the file, used to learn the physical type /// each column is encoded with. May be null; page statistics whose decoding /// depends on the physical type are then left uncompared so that no page is /// pruned on them. + /// @param row_group_row_count The number of rows in the row group. /// @return RowRanges that may contain matching rows. static Result CalculateRowRanges( const std::shared_ptr& predicate, const std::shared_ptr<::parquet::RowGroupPageIndexReader>& rg_page_index_reader, - const std::map& column_name_to_index, int64_t row_group_row_count, - const ::parquet::SchemaDescriptor* schema); + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count); private: /// Visit a predicate and calculate row ranges. static Result VisitPredicate( const std::shared_ptr& predicate, - const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader, - const ::parquet::SchemaDescriptor* schema); + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count, + ::parquet::RowGroupPageIndexReader* rg_page_index_reader); /// Visit a leaf predicate and calculate row ranges. static Result VisitLeafPredicate( const std::shared_ptr& leaf_predicate, - const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader, - const ::parquet::SchemaDescriptor* schema); + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count, + ::parquet::RowGroupPageIndexReader* rg_page_index_reader); /// Visit a compound predicate (AND/OR) and calculate row ranges. static Result VisitCompoundPredicate( const std::shared_ptr& compound_predicate, - const std::map& column_name_to_index, int64_t row_group_row_count, - ::parquet::RowGroupPageIndexReader* rg_page_index_reader, - const ::parquet::SchemaDescriptor* schema); + const std::map& column_name_to_index, + const ::parquet::SchemaDescriptor* schema, int64_t row_group_row_count, + ::parquet::RowGroupPageIndexReader* rg_page_index_reader); /// Filter pages based on column index statistics for EQUAL predicate. static std::vector FilterPagesByEqual( diff --git a/src/paimon/format/parquet/column_index_filter_test.cpp b/src/paimon/format/parquet/column_index_filter_test.cpp index 0c7f64bab..f1b916bb7 100644 --- a/src/paimon/format/parquet/column_index_filter_test.cpp +++ b/src/paimon/format/parquet/column_index_filter_test.cpp @@ -306,9 +306,9 @@ class ColumnIndexFilterTest : public ::testing::Test { } Result Filter(const std::shared_ptr& predicate) { - return ColumnIndexFilter::CalculateRowRanges(predicate, page_index_reader_->RowGroup(0), - column_name_to_index_, row_group_row_count_, - parquet_reader_->metadata()->schema()); + return ColumnIndexFilter::CalculateRowRanges( + predicate, page_index_reader_->RowGroup(0), column_name_to_index_, + parquet_reader_->metadata()->schema(), row_group_row_count_); } std::shared_ptr arrow_pool_; @@ -554,11 +554,10 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { auto less_negative_zero = PredicateBuilder::LessThan( /*field_index=*/0, /*field_name=*/"value", field_type, field_type == FieldType::FLOAT ? Literal(-0.0f) : Literal(-0.0)); - ASSERT_OK_AND_ASSIGN( - auto ranges, - ColumnIndexFilter::CalculateRowRanges( - less_negative_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); + ASSERT_OK_AND_ASSIGN(auto ranges, ColumnIndexFilter::CalculateRowRanges( + less_negative_zero, page_index_reader->RowGroup(0), + {{"value", 0}}, reader->metadata()->schema(), + reader->metadata()->RowGroup(0)->num_rows())); ASSERT_TRUE(ranges.IsEmpty()) << "field type: " << static_cast(field_type); auto less_positive_zero = PredicateBuilder::LessThan( @@ -567,7 +566,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( less_positive_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); + reader->metadata()->schema(), reader->metadata()->RowGroup(0)->num_rows())); ASSERT_EQ(20, ranges.RowCount()); ASSERT_EQ(1, ranges.GetRanges().size()); ASSERT_EQ(0, ranges.GetRanges()[0].from); @@ -579,7 +578,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( greater_negative_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); + reader->metadata()->schema(), reader->metadata()->RowGroup(0)->num_rows())); ASSERT_EQ(30, ranges.RowCount()); auto not_equal_negative_zero = PredicateBuilder::NotEqual( @@ -588,7 +587,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( not_equal_negative_zero, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); + reader->metadata()->schema(), reader->metadata()->RowGroup(0)->num_rows())); ASSERT_EQ(30, ranges.RowCount()); auto greater_finite = PredicateBuilder::GreaterThan( @@ -597,7 +596,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( greater_finite, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); + reader->metadata()->schema(), reader->metadata()->RowGroup(0)->num_rows())); ASSERT_TRUE(ranges.IsEmpty()); auto greater_between_pages = PredicateBuilder::GreaterThan( @@ -606,7 +605,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( greater_between_pages, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); + reader->metadata()->schema(), reader->metadata()->RowGroup(0)->num_rows())); ASSERT_EQ(10, ranges.RowCount()); ASSERT_EQ(1, ranges.GetRanges().size()); ASSERT_EQ(20, ranges.GetRanges()[0].from); @@ -618,7 +617,7 @@ TEST_F(ColumnIndexFilterTest, SignedZeroUsesJavaOrderForFloatingPointPages) { ASSERT_OK_AND_ASSIGN( ranges, ColumnIndexFilter::CalculateRowRanges( equal_finite, page_index_reader->RowGroup(0), {{"value", 0}}, - reader->metadata()->RowGroup(0)->num_rows(), reader->metadata()->schema())); + reader->metadata()->schema(), reader->metadata()->RowGroup(0)->num_rows())); ASSERT_TRUE(ranges.IsEmpty()); } } @@ -671,7 +670,7 @@ TEST_F(ColumnIndexFilterTest, DecimalFixedLenByteArrayPages) { int64_t row_count = reader->metadata()->RowGroup(0)->num_rows(); auto filter = [&](const std::shared_ptr& predicate) { return ColumnIndexFilter::CalculateRowRanges(predicate, page_index_reader->RowGroup(0), - {{"value", 0}}, row_count, schema); + {{"value", 0}}, schema, row_count); }; // value < 5.00 → only page 0 (rows [0, 9]). @@ -740,7 +739,7 @@ TEST_F(ColumnIndexFilterTest, DecimalWithoutSchemaKeepsAllPages) { /*field_index=*/0, /*field_name=*/"value", FieldType::DECIMAL, Literal(Decimal(9, 2, 500))); ASSERT_OK_AND_ASSIGN(auto ranges, ColumnIndexFilter::CalculateRowRanges( less_than, page_index_reader->RowGroup(0), {{"value", 0}}, - row_count, /*schema=*/nullptr)); + /*schema=*/nullptr, row_count)); EXPECT_EQ(row_count, ranges.RowCount()); } diff --git a/src/paimon/format/parquet/file_reader_wrapper.cpp b/src/paimon/format/parquet/file_reader_wrapper.cpp index 0cdab1bd8..19c35310b 100644 --- a/src/paimon/format/parquet/file_reader_wrapper.cpp +++ b/src/paimon/format/parquet/file_reader_wrapper.cpp @@ -633,7 +633,7 @@ Result FileReaderWrapper::CalculateFilteredRowRanges( // their offset indexes when the data reader is initialized later. return ColumnIndexFilter::CalculateRowRanges( predicate, page_index_reader->RowGroup(row_group_index), column_name_to_index, - row_count, meta_data->schema()); + meta_data->schema(), row_count); } PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("FileReaderWrapper::CalculateFilteredRowRanges") }