diff --git a/src/paimon/format/parquet/column_index_filter.cpp b/src/paimon/format/parquet/column_index_filter.cpp index 1e0a8f597..74dc37fee 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, + 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, + 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, + 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, + 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); + return VisitCompoundPredicate(compound_predicate, column_name_to_index, schema, + row_group_row_count, rg_page_index_reader); } return Status::Invalid("Unknown predicate type"); @@ -67,7 +69,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, + 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); @@ -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); @@ -141,7 +158,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, + 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(); @@ -153,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)); + 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 @@ -164,7 +182,7 @@ 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, + VisitPredicate(children[i], column_name_to_index, schema, row_group_row_count, rg_page_index_reader)); result = RowRanges::Intersection(result, child_ranges); @@ -182,7 +200,7 @@ 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, + VisitPredicate(children[i], column_name_to_index, schema, row_group_row_count, rg_page_index_reader)); result = RowRanges::Union(result, child_ranges); @@ -201,7 +219,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 +237,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 +248,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 +270,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 +286,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 +297,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 +307,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 +318,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 +328,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 +339,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 +349,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 +360,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 +409,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 +446,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 +522,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 +591,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 +653,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 +668,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 +680,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 +695,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 +710,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 +725,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..a0b7eecf3 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; @@ -59,61 +60,69 @@ 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 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 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, + 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, + 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, + 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( 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..f1b916bb7 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" @@ -305,8 +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_); + 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_; @@ -552,19 +554,19 @@ 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()->schema(), + reader->metadata()->RowGroup(0)->num_rows())); 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()->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); @@ -576,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()->RowGroup(0)->num_rows())); ASSERT_EQ(30, ranges.RowCount()); auto not_equal_negative_zero = PredicateBuilder::NotEqual( @@ -585,16 +587,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()->schema(), reader->metadata()->RowGroup(0)->num_rows())); 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()->schema(), reader->metadata()->RowGroup(0)->num_rows())); ASSERT_TRUE(ranges.IsEmpty()); auto greater_between_pages = PredicateBuilder::GreaterThan( @@ -603,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()->RowGroup(0)->num_rows())); ASSERT_EQ(10, ranges.RowCount()); ASSERT_EQ(1, ranges.GetRanges().size()); ASSERT_EQ(20, ranges.GetRanges()[0].from); @@ -612,14 +614,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()->schema(), reader->metadata()->RowGroup(0)->num_rows())); 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}}, schema, row_count); + }; + + // 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}}, + /*schema=*/nullptr, row_count)); + 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..19c35310b 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, + meta_data->schema(), row_count); } PAIMON_PARQUET_CATCH_AND_RETURN_STATUS("FileReaderWrapper::CalculateFilteredRowRanges") }