This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 3366987b fix(parquet): decode DECIMAL page statistics by physical type
(#346)
3366987b is described below
commit 3366987bc90ebb6f03254ac91a2ccd6a9a668b96
Author: Yunpei Zhou <[email protected]>
AuthorDate: Tue Sep 15 17:14:20 2026 +0800
fix(parquet): decode DECIMAL page statistics by physical type (#346)
---
src/paimon/format/parquet/column_index_filter.cpp | 188 +++++++++++++--------
src/paimon/format/parquet/column_index_filter.h | 51 ++++--
.../format/parquet/column_index_filter_test.cpp | 165 +++++++++++++++---
src/paimon/format/parquet/file_reader_wrapper.cpp | 6 +-
4 files changed, 298 insertions(+), 112 deletions(-)
diff --git a/src/paimon/format/parquet/column_index_filter.cpp
b/src/paimon/format/parquet/column_index_filter.cpp
index 1e0a8f59..74dc37fe 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<RowRanges> ColumnIndexFilter::CalculateRowRanges(
const std::shared_ptr<Predicate>& predicate,
const std::shared_ptr<::parquet::RowGroupPageIndexReader>&
rg_page_index_reader,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count) {
+ const std::map<std::string, int32_t>& 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<RowRanges> ColumnIndexFilter::VisitPredicate(
const std::shared_ptr<Predicate>& predicate,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count,
+ const std::map<std::string, int32_t>& 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<LeafPredicate>(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<CompoundPredicate>(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<RowRanges> ColumnIndexFilter::VisitPredicate(
Result<RowRanges> ColumnIndexFilter::VisitLeafPredicate(
const std::shared_ptr<LeafPredicate>& leaf_predicate,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count,
+ const std::map<std::string, int32_t>& 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<RowRanges> 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<RowRanges> 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<RowRanges> ColumnIndexFilter::VisitLeafPredicate(
Result<RowRanges> ColumnIndexFilter::VisitCompoundPredicate(
const std::shared_ptr<CompoundPredicate>& compound_predicate,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count,
+ const std::map<std::string, int32_t>& 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<RowRanges> 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<RowRanges> 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<RowRanges> 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<RowRanges> ColumnIndexFilter::VisitCompoundPredicate(
std::vector<int32_t> 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<int32_t> matching_pages;
if (literal.IsNull()) {
@@ -219,7 +237,8 @@ std::vector<int32_t> 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<int32_t> ColumnIndexFilter::FilterPagesByEqual(
std::vector<int32_t> 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<int32_t> matching_pages;
if (literal.IsNull()) {
@@ -251,8 +270,8 @@ std::vector<int32_t>
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<int32_t>
ColumnIndexFilter::FilterPagesByNotEqual(
std::vector<int32_t> 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<int32_t> 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<int32_t>
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<int32_t>
ColumnIndexFilter::FilterPagesByLessThan(
std::vector<int32_t> 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<int32_t> 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<int32_t>
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<int32_t>
ColumnIndexFilter::FilterPagesByLessOrEqual(
std::vector<int32_t> 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<int32_t> 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<int32_t>
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<int32_t>
ColumnIndexFilter::FilterPagesByGreaterThan(
std::vector<int32_t> 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<int32_t> 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<int32_t>
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<int32_t>
ColumnIndexFilter::FilterPagesByIsNotNull(
std::vector<int32_t> ColumnIndexFilter::FilterPagesByIn(
const std::shared_ptr<::parquet::ColumnIndex>& column_index,
- const std::vector<Literal>& literals, FieldType field_type) {
+ const std::vector<Literal>& literals, FieldType field_type,
+ ::parquet::Type::type physical_type) {
std::vector<int32_t> 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<int32_t> 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<int32_t> ColumnIndexFilter::CompareEncodedWithLiteral(const
std::string& encoded,
- const
Literal& literal,
- FieldType
field_type) {
+std::optional<int32_t> 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<int32_t>
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>();
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<Decimal::int128_t>(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<Decimal::int128_t>(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<Decimal::int128_t>(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<Decimal::int128_t>(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<int32_t>
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 a3016241..a0b7eecf 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<RowRanges> CalculateRowRanges(
const std::shared_ptr<Predicate>& predicate,
const std::shared_ptr<::parquet::RowGroupPageIndexReader>&
rg_page_index_reader,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count);
+ const std::map<std::string, int32_t>& column_name_to_index,
+ const ::parquet::SchemaDescriptor* schema, int64_t
row_group_row_count);
private:
/// Visit a predicate and calculate row ranges.
static Result<RowRanges> VisitPredicate(
const std::shared_ptr<Predicate>& predicate,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count,
+ const std::map<std::string, int32_t>& 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<RowRanges> VisitLeafPredicate(
const std::shared_ptr<LeafPredicate>& leaf_predicate,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count,
+ const std::map<std::string, int32_t>& 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<RowRanges> VisitCompoundPredicate(
const std::shared_ptr<CompoundPredicate>& compound_predicate,
- const std::map<std::string, int32_t>& column_name_to_index, int64_t
row_group_row_count,
+ const std::map<std::string, int32_t>& 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<int32_t> 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<int32_t> 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<int32_t> 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<int32_t> 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<int32_t> 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<int32_t> 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<int32_t> FilterPagesByIsNull(
@@ -126,7 +135,8 @@ class ColumnIndexFilter {
/// Filter pages based on column index statistics for IN predicate.
static std::vector<int32_t> FilterPagesByIn(
const std::shared_ptr<::parquet::ColumnIndex>& column_index,
- const std::vector<Literal>& literals, FieldType field_type);
+ const std::vector<Literal>& literals, FieldType field_type,
+ ::parquet::Type::type physical_type);
/// Filter pages based on column index statistics for NOT_IN predicate.
static std::vector<int32_t> 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<int32_t> 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 94c65ef5..f1b916bb 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<RowRanges> Filter(const std::shared_ptr<Predicate>& 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::MemoryPool> 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<int32_t>(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<int64_t> 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<arrow::Array> 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<InputStream> in,
fs_->Open(file_name));
+ ASSERT_OK_AND_ASSIGN(int64_t length, in->Length());
+ auto in_stream = std::make_shared<ArrowInputStreamAdapter>(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>& 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<arrow::Array> 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<InputStream> in,
fs_->Open(file_name));
+ ASSERT_OK_AND_ASSIGN(int64_t length, in->Length());
+ auto in_stream = std::make_shared<ArrowInputStreamAdapter>(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 f69b89d4..19c35310 100644
--- a/src/paimon/format/parquet/file_reader_wrapper.cpp
+++ b/src/paimon/format/parquet/file_reader_wrapper.cpp
@@ -631,9 +631,9 @@ Result<RowRanges>
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")
}
