Skip to content

Commit 55958e0

Browse files
wgtmacCopilot
wgtmac
and
Copilot
authored
GH-50156: [C++][Parquet] Ignore min/max for unknown column order (#50157)
### Rationale for this change Parquet column order defines how min/max statistics should be interpreted. If a reader sees an unsupported ColumnOrder, it cannot safely use chunk statistics or page index min/max values for that column. ### What changes are included in this PR? - Added an internal `ColumnOrder::UNKNOWN` state for unsupported thrift column order. - Kept missing ColumnOrder as `UNDEFINED`, preserving legacy min/max behavior. - Ignored chunk-level min/max statistics when column order or sort order is unknown. - Exposed `ColumnDescriptor::can_use_min_max()` to indicate whether min/max fields can be trusted. ### Are these changes tested? Added regression tests for unsupported column order, missing column order, and page index guards. ### Are there any user-facing changes? No. * GitHub Issue: #50156 Lead-authored-by: Gang Wu <ustcwg@gmail.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Antoine Pitrou <antoine@python.org>
1 parent 4b64398 commit 55958e0

14 files changed

Lines changed: 303 additions & 71 deletions

‎cpp/src/parquet/column_io_benchmark.cc‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,8 @@ std::shared_ptr<Int64Reader> BuildReader(std::shared_ptr<Buffer>& buffer,
161161
int64_t num_values, Compression::type codec,
162162
ColumnDescriptor* schema) {
163163
auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
164-
std::unique_ptr<PageReader> page_reader = PageReader::Open(source, num_values, codec);
164+
std::unique_ptr<PageReader> page_reader =
165+
PageReader::Open(source, num_values, codec, ReaderProperties(), *schema);
165166
return std::static_pointer_cast<Int64Reader>(
166167
ColumnReader::Make(schema, std::move(page_reader)));
167168
}

‎cpp/src/parquet/column_reader.cc‎

Lines changed: 25 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -200,10 +200,10 @@ namespace {
200200

201201
// Extracts encoded statistics from V1 and V2 data page headers
202202
template <typename H>
203-
EncodedStatistics ExtractStatsFromHeader(const H& header) {
203+
EncodedStatistics ExtractStatsFromHeader(const H& header, StatisticsMinMaxField min_max) {
204204
EncodedStatistics page_statistics;
205205
if (header.__isset.statistics) {
206-
page_statistics = FromThrift(header.statistics);
206+
page_statistics = FromThrift(header.statistics, min_max);
207207
}
208208
return page_statistics;
209209
}
@@ -225,13 +225,15 @@ class SerializedPageReader : public PageReader {
225225
public:
226226
SerializedPageReader(std::shared_ptr<ArrowInputStream> stream, int64_t total_num_values,
227227
Compression::type codec, const ReaderProperties& properties,
228-
const CryptoContext* crypto_ctx, bool always_compressed)
228+
const CryptoContext* crypto_ctx, bool always_compressed,
229+
StatisticsMinMaxField stats_min_max_field)
229230
: properties_(properties),
230231
stream_(std::move(stream)),
231232
decompression_buffer_(AllocateBuffer(properties_.memory_pool(), 0)),
232233
page_ordinal_(0),
233234
seen_num_values_(0),
234-
total_num_values_(total_num_values) {
235+
total_num_values_(total_num_values),
236+
stats_min_max_field_(stats_min_max_field) {
235237
if (crypto_ctx != nullptr) {
236238
crypto_ctx_ = *crypto_ctx;
237239
InitDecryption();
@@ -302,6 +304,8 @@ class SerializedPageReader : public PageReader {
302304
// Number of values in all the data pages
303305
int64_t total_num_values_;
304306

307+
StatisticsMinMaxField stats_min_max_field_;
308+
305309
// data_page_aad_ and data_page_header_aad_ contain the AAD for data page and data page
306310
// header in a single column respectively.
307311
// While calculating AAD for different pages in a single column the pages AAD is
@@ -349,7 +353,7 @@ bool SerializedPageReader::ShouldSkipPage(EncodedStatistics* data_page_statistic
349353
if (page_type == PageType::DATA_PAGE) {
350354
const format::DataPageHeader& header = current_page_header_.data_page_header;
351355
CheckNumValuesInHeader(header.num_values);
352-
*data_page_statistics = ExtractStatsFromHeader(header);
356+
*data_page_statistics = ExtractStatsFromHeader(header, stats_min_max_field_);
353357
seen_num_values_ += header.num_values;
354358
if (data_page_filter_) {
355359
const EncodedStatistics* filter_statistics =
@@ -370,7 +374,7 @@ bool SerializedPageReader::ShouldSkipPage(EncodedStatistics* data_page_statistic
370374
header.repetition_levels_byte_length < 0) {
371375
throw ParquetException("Invalid page header (negative levels byte length)");
372376
}
373-
*data_page_statistics = ExtractStatsFromHeader(header);
377+
*data_page_statistics = ExtractStatsFromHeader(header, stats_min_max_field_);
374378
seen_num_values_ += header.num_values;
375379
if (data_page_filter_) {
376380
const EncodedStatistics* filter_statistics =
@@ -591,14 +595,25 @@ std::shared_ptr<Buffer> SerializedPageReader::DecompressIfNeeded(
591595

592596
} // namespace
593597

598+
std::unique_ptr<PageReader> PageReader::Open(
599+
std::shared_ptr<ArrowInputStream> stream, int64_t total_num_values,
600+
Compression::type codec, const ReaderProperties& properties,
601+
const ColumnDescriptor& descr, bool always_compressed, const CryptoContext* ctx) {
602+
const auto stats_min_max_field = GetStatisticsMinMaxField(descr);
603+
return std::unique_ptr<PageReader>(
604+
new SerializedPageReader(std::move(stream), total_num_values, codec, properties,
605+
ctx, always_compressed, stats_min_max_field));
606+
}
607+
594608
std::unique_ptr<PageReader> PageReader::Open(std::shared_ptr<ArrowInputStream> stream,
595609
int64_t total_num_values,
596610
Compression::type codec,
597611
const ReaderProperties& properties,
598612
bool always_compressed,
599613
const CryptoContext* ctx) {
600614
return std::unique_ptr<PageReader>(new SerializedPageReader(
601-
std::move(stream), total_num_values, codec, properties, ctx, always_compressed));
615+
std::move(stream), total_num_values, codec, properties, ctx, always_compressed,
616+
StatisticsMinMaxField::kMinValueMaxValue));
602617
}
603618

604619
std::unique_ptr<PageReader> PageReader::Open(std::shared_ptr<ArrowInputStream> stream,
@@ -607,9 +622,9 @@ std::unique_ptr<PageReader> PageReader::Open(std::shared_ptr<ArrowInputStream> s
607622
bool always_compressed,
608623
::arrow::MemoryPool* pool,
609624
const CryptoContext* ctx) {
610-
return std::unique_ptr<PageReader>(
611-
new SerializedPageReader(std::move(stream), total_num_values, codec,
612-
ReaderProperties(pool), ctx, always_compressed));
625+
return std::unique_ptr<PageReader>(new SerializedPageReader(
626+
std::move(stream), total_num_values, codec, ReaderProperties(pool), ctx,
627+
always_compressed, StatisticsMinMaxField::kMinValueMaxValue));
613628
}
614629

615630
namespace {

‎cpp/src/parquet/column_reader.h‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,11 +117,21 @@ class PARQUET_EXPORT PageReader {
117117
public:
118118
virtual ~PageReader() = default;
119119

120+
static std::unique_ptr<PageReader> Open(std::shared_ptr<ArrowInputStream> stream,
121+
int64_t total_num_values,
122+
Compression::type codec,
123+
const ReaderProperties& properties,
124+
const ColumnDescriptor& descr,
125+
bool always_compressed = false,
126+
const CryptoContext* ctx = NULLPTR);
127+
128+
PARQUET_DEPRECATED("Deprecated in 25.0.0. Use the ColumnDescriptor overload instead.")
120129
static std::unique_ptr<PageReader> Open(
121130
std::shared_ptr<ArrowInputStream> stream, int64_t total_num_values,
122131
Compression::type codec, bool always_compressed = false,
123132
::arrow::MemoryPool* pool = ::arrow::default_memory_pool(),
124133
const CryptoContext* ctx = NULLPTR);
134+
PARQUET_DEPRECATED("Deprecated in 25.0.0. Use the ColumnDescriptor overload instead.")
125135
static std::unique_ptr<PageReader> Open(std::shared_ptr<ArrowInputStream> stream,
126136
int64_t total_num_values,
127137
Compression::type codec,

‎cpp/src/parquet/column_writer_test.cc‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -100,8 +100,8 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
100100
auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
101101
ReaderProperties readerProperties;
102102
readerProperties.set_page_checksum_verification(page_checksum_verify);
103-
std::unique_ptr<PageReader> page_reader =
104-
PageReader::Open(std::move(source), num_rows, compression, readerProperties);
103+
std::unique_ptr<PageReader> page_reader = PageReader::Open(
104+
std::move(source), num_rows, compression, readerProperties, *this->descr_);
105105
reader_ = std::static_pointer_cast<TypedColumnReader<TestType>>(
106106
ColumnReader::Make(this->descr_, std::move(page_reader)));
107107
}
@@ -2058,8 +2058,9 @@ TEST_F(TestValuesWriterInt32Type, AvoidCompressedInDataPageV2) {
20582058
ASSERT_OK_AND_ASSIGN(auto buffer, this->sink_->Finish());
20592059
auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
20602060
ReaderProperties readerProperties;
2061-
std::unique_ptr<PageReader> page_reader = PageReader::Open(
2062-
std::move(source), total_num_values, compression, readerProperties);
2061+
std::unique_ptr<PageReader> page_reader =
2062+
PageReader::Open(std::move(source), total_num_values, compression,
2063+
readerProperties, *this->descr_);
20632064
auto data_page = std::static_pointer_cast<DataPageV2>(page_reader->NextPage());
20642065
ASSERT_TRUE(data_page != nullptr);
20652066
ASSERT_FALSE(data_page->is_compressed());

‎cpp/src/parquet/file_deserialize_test.cc‎

Lines changed: 26 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,14 @@ namespace parquet {
4646
using ::arrow::io::BufferReader;
4747
using ::parquet::DataPageStats;
4848

49+
// Make a column descriptor with an undefined column order.
50+
ColumnDescriptor MakeColumnDescriptor() {
51+
auto node = schema::Int32("int_col", Repetition::REQUIRED);
52+
static_cast<schema::PrimitiveNode*>(node.get())
53+
->SetColumnOrder(ColumnOrder::undefined_);
54+
return ColumnDescriptor(node, /*max_definition_level=*/0, /*max_repetition_level=*/0);
55+
}
56+
4957
// Adds page statistics occupying a certain amount of bytes (for testing very
5058
// large page headers)
5159
template <typename H>
@@ -110,6 +118,8 @@ static std::vector<Compression::type> GetSupportedCodecTypes() {
110118

111119
class TestPageSerde : public ::testing::Test {
112120
public:
121+
TestPageSerde() : descr_(MakeColumnDescriptor()) {}
122+
113123
void SetUp() {
114124
data_page_header_.encoding = format::Encoding::PLAIN;
115125
data_page_header_.definition_level_encoding = format::Encoding::RLE;
@@ -124,7 +134,7 @@ class TestPageSerde : public ::testing::Test {
124134
EndStream();
125135

126136
auto stream = std::make_shared<::arrow::io::BufferReader>(out_buffer_);
127-
page_reader_ = PageReader::Open(stream, num_rows, codec, properties);
137+
page_reader_ = PageReader::Open(stream, num_rows, codec, properties, descr_);
128138
}
129139

130140
void WriteDataPageHeader(int max_serialized_len = 1024, int32_t uncompressed_size = 0,
@@ -204,6 +214,7 @@ class TestPageSerde : public ::testing::Test {
204214
std::shared_ptr<::arrow::io::BufferOutputStream> out_stream_;
205215
std::shared_ptr<Buffer> out_buffer_;
206216

217+
ColumnDescriptor descr_;
207218
std::unique_ptr<PageReader> page_reader_;
208219
format::PageHeader page_header_;
209220
format::DataPageHeader data_page_header_;
@@ -466,7 +477,8 @@ TYPED_TEST(PageFilterTest, TestPageWithoutStatistics) {
466477

467478
auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
468479
this->page_reader_ =
469-
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
480+
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
481+
ReaderProperties(), this->descr_);
470482

471483
int num_pages = 0;
472484
bool is_stats_null = false;
@@ -492,7 +504,8 @@ TYPED_TEST(PageFilterTest, TestPageFilterCallback) {
492504
// are right.
493505
auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
494506
this->page_reader_ =
495-
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
507+
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
508+
ReaderProperties(), this->descr_);
496509

497510
std::vector<EncodedStatistics> read_stats;
498511
std::vector<int64_t> read_num_values;
@@ -526,7 +539,8 @@ TYPED_TEST(PageFilterTest, TestPageFilterCallback) {
526539
{ // Skip all pages.
527540
auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
528541
this->page_reader_ =
529-
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
542+
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
543+
ReaderProperties(), this->descr_);
530544

531545
auto skip_all_pages = [](const DataPageStats& stats) -> bool { return true; };
532546

@@ -538,7 +552,8 @@ TYPED_TEST(PageFilterTest, TestPageFilterCallback) {
538552
{ // Skip every other page.
539553
auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
540554
this->page_reader_ =
541-
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
555+
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
556+
ReaderProperties(), this->descr_);
542557

543558
// Skip pages with even number of values.
544559
auto skip_even_pages = [](const DataPageStats& stats) -> bool {
@@ -569,7 +584,8 @@ TYPED_TEST(PageFilterTest, TestChangingPageFilter) {
569584

570585
auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
571586
this->page_reader_ =
572-
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
587+
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
588+
ReaderProperties(), this->descr_);
573589

574590
// This callback will always return false.
575591
auto read_all_pages = [](const DataPageStats& stats) -> bool { return false; };
@@ -604,7 +620,8 @@ TEST_F(TestPageSerde, DoesNotFilterDictionaryPages) {
604620

605621
// Try to read it back while asking for all data pages to be skipped.
606622
auto stream = std::make_shared<::arrow::io::BufferReader>(out_buffer_);
607-
page_reader_ = PageReader::Open(stream, /*num_rows=*/100, Compression::UNCOMPRESSED);
623+
page_reader_ = PageReader::Open(stream, /*num_rows=*/100, Compression::UNCOMPRESSED,
624+
ReaderProperties(), descr_);
608625

609626
auto skip_all_pages = [](const DataPageStats& stats) -> bool { return true; };
610627

@@ -641,7 +658,8 @@ TEST_F(TestPageSerde, SkipsNonDataPages) {
641658
EndStream();
642659

643660
auto stream = std::make_shared<::arrow::io::BufferReader>(out_buffer_);
644-
page_reader_ = PageReader::Open(stream, /*num_rows=*/100, Compression::UNCOMPRESSED);
661+
page_reader_ = PageReader::Open(stream, /*num_rows=*/100, Compression::UNCOMPRESSED,
662+
ReaderProperties(), descr_);
645663

646664
// Only the two data pages are returned.
647665
std::shared_ptr<Page> current_page = page_reader_->NextPage();

‎cpp/src/parquet/file_reader.cc‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,7 @@ class SerializedRowGroup : public RowGroupReader::Contents {
239239
std::unique_ptr<PageReader> GetColumnPageReader(int i) override {
240240
// Read column chunk from the file
241241
auto col = row_group_metadata_->ColumnChunk(i);
242+
const ColumnDescriptor* descr = row_group_metadata_->schema()->Column(i);
242243

243244
::arrow::io::ReadRange col_range =
244245
ComputeColumnChunkRange(file_metadata_, source_size_, row_group_ordinal_, i);
@@ -263,7 +264,7 @@ class SerializedRowGroup : public RowGroupReader::Contents {
263264
// Column is encrypted only if crypto_metadata exists.
264265
if (!crypto_metadata) {
265266
return PageReader::Open(stream, col->num_values(), col->compression(), properties_,
266-
always_compressed);
267+
*descr, always_compressed);
267268
}
268269

269270
// The column is encrypted
@@ -286,7 +287,7 @@ class SerializedRowGroup : public RowGroupReader::Contents {
286287
std::move(meta_decryptor_factory),
287288
std::move(data_decryptor_factory)};
288289
return PageReader::Open(stream, col->num_values(), col->compression(), properties_,
289-
always_compressed, &ctx);
290+
*descr, always_compressed, &ctx);
290291
}
291292

292293
private:

‎cpp/src/parquet/metadata.cc‎

Lines changed: 41 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -90,40 +90,48 @@ std::string ParquetVersionToString(ParquetVersion::type ver) {
9090
return "UNKNOWN";
9191
}
9292

93+
namespace {
94+
9395
template <typename DType>
94-
static std::shared_ptr<Statistics> MakeTypedColumnStats(
95-
const format::ColumnMetaData& metadata, const ColumnDescriptor* descr,
96-
::arrow::MemoryPool* pool) {
97-
std::optional<bool> min_exact =
98-
metadata.statistics.__isset.is_min_value_exact
99-
? std::optional<bool>(metadata.statistics.is_min_value_exact)
100-
: std::nullopt;
101-
std::optional<bool> max_exact =
102-
metadata.statistics.__isset.is_max_value_exact
103-
? std::optional<bool>(metadata.statistics.is_max_value_exact)
104-
: std::nullopt;
105-
// If ColumnOrder is defined, return max_value and min_value
106-
if (descr->column_order().get_order() == ColumnOrder::TYPE_DEFINED_ORDER) {
107-
return MakeStatistics<DType>(
108-
descr, metadata.statistics.min_value, metadata.statistics.max_value,
109-
metadata.num_values - metadata.statistics.null_count,
110-
metadata.statistics.null_count, metadata.statistics.distinct_count,
111-
metadata.statistics.__isset.max_value && metadata.statistics.__isset.min_value,
112-
metadata.statistics.__isset.null_count,
113-
metadata.statistics.__isset.distinct_count, min_exact, max_exact, pool);
114-
}
115-
// Default behavior
96+
std::shared_ptr<Statistics> MakeTypedColumnStats(const format::ColumnMetaData& metadata,
97+
const ColumnDescriptor* descr,
98+
::arrow::MemoryPool* pool) {
99+
const auto& statistics = metadata.statistics;
100+
const std::string kEmpty = "";
101+
const std::string* encoded_min = &kEmpty;
102+
const std::string* encoded_max = &kEmpty;
103+
bool has_min_max = false;
104+
std::optional<bool> min_exact = std::nullopt;
105+
std::optional<bool> max_exact = std::nullopt;
106+
107+
switch (GetStatisticsMinMaxField(*descr)) {
108+
case StatisticsMinMaxField::kMinValueMaxValue:
109+
encoded_min = &statistics.min_value;
110+
encoded_max = &statistics.max_value;
111+
has_min_max = statistics.__isset.max_value && statistics.__isset.min_value;
112+
min_exact = statistics.__isset.is_min_value_exact
113+
? std::optional<bool>(statistics.is_min_value_exact)
114+
: std::nullopt;
115+
max_exact = statistics.__isset.is_max_value_exact
116+
? std::optional<bool>(statistics.is_max_value_exact)
117+
: std::nullopt;
118+
break;
119+
case StatisticsMinMaxField::kLegacyMinMax:
120+
encoded_min = &statistics.min;
121+
encoded_max = &statistics.max;
122+
has_min_max = statistics.__isset.max && statistics.__isset.min;
123+
break;
124+
case StatisticsMinMaxField::kInvalid:
125+
break;
126+
}
127+
116128
return MakeStatistics<DType>(
117-
descr, metadata.statistics.min, metadata.statistics.max,
118-
metadata.num_values - metadata.statistics.null_count,
119-
metadata.statistics.null_count, metadata.statistics.distinct_count,
120-
metadata.statistics.__isset.max && metadata.statistics.__isset.min,
121-
metadata.statistics.__isset.null_count, metadata.statistics.__isset.distinct_count,
122-
min_exact, max_exact, pool);
129+
descr, *encoded_min, *encoded_max, metadata.num_values - statistics.null_count,
130+
statistics.null_count, statistics.distinct_count, has_min_max,
131+
statistics.__isset.null_count, statistics.__isset.distinct_count, min_exact,
132+
max_exact, pool);
123133
}
124134

125-
namespace {
126-
127135
std::shared_ptr<geospatial::GeoStatistics> MakeColumnGeometryStats(
128136
const format::ColumnMetaData& metadata, const ColumnDescriptor* descr) {
129137
if (metadata.__isset.geospatial_statistics) {
@@ -336,12 +344,8 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
336344
{
337345
const std::lock_guard<std::mutex> guard(stats_mutex_);
338346
if (possible_encoded_stats_ == nullptr) {
339-
possible_encoded_stats_ =
340-
std::make_shared<EncodedStatistics>(FromThrift(column_metadata_->statistics));
341-
if (descr_->sort_order() == SortOrder::UNKNOWN) {
342-
// If the column SortOrder is Unknown we can't trust max/min.
343-
possible_encoded_stats_->ClearMinMax();
344-
}
347+
possible_encoded_stats_ = std::make_shared<EncodedStatistics>(
348+
FromThrift(column_metadata_->statistics, GetStatisticsMinMaxField(*descr_)));
345349
}
346350
}
347351
return writer_version_->HasCorrectStatistics(type(), *possible_encoded_stats_,
@@ -1037,7 +1041,7 @@ class FileMetaData::FileMetaDataImpl {
10371041
if (column_order.__isset.TYPE_ORDER) {
10381042
column_orders.push_back(ColumnOrder::type_defined_);
10391043
} else {
1040-
column_orders.push_back(ColumnOrder::undefined_);
1044+
column_orders.push_back(ColumnOrder::unknown_);
10411045
}
10421046
}
10431047
} else {

0 commit comments

Comments
 (0)