Skip to content

GH-50156: [C++][Parquet] Ignore min/max for unknown column order #50157

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pitrou merged 9 commits into from
Jun 16, 2026
Merged

GH-50156: [C++][Parquet] Ignore min/max for unknown column order #50157

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
34be8e8
GH-50156: [C++][Parquet] Ignore min/max for unknown column order
wgtmac Jun 11, 2026
fed1776
Potential fix for pull request finding
wgtmac Jun 11, 2026
a17f232
Apply suggestion from @wgtmac
wgtmac Jun 11, 2026
a92a70d
Apply suggestion from @wgtmac
wgtmac Jun 11, 2026
d42ce32
Apply suggestion from @wgtmac
wgtmac Jun 11, 2026
0ada0c0
add can_use_stats() and revert nullable column index
wgtmac Jun 12, 2026
b3360da
rename to can_use_min_max
wgtmac Jun 13, 2026
4a691c7
address comment
wgtmac Jun 15, 2026
7edb162
add new PageReader::Open with descr
wgtmac Jun 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion cpp/src/parquet/column_io_benchmark.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,8 @@ std::shared_ptr<Int64Reader> BuildReader(std::shared_ptr<Buffer>& buffer,
int64_t num_values, Compression::type codec,
ColumnDescriptor* schema) {
auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
std::unique_ptr<PageReader> page_reader = PageReader::Open(source, num_values, codec);
std::unique_ptr<PageReader> page_reader =
PageReader::Open(source, num_values, codec, ReaderProperties(), *schema);
return std::static_pointer_cast<Int64Reader>(
ColumnReader::Make(schema, std::move(page_reader)));
}
Expand Down
35 changes: 25 additions & 10 deletions cpp/src/parquet/column_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,10 +200,10 @@ namespace {

// Extracts encoded statistics from V1 and V2 data page headers
template <typename H>
EncodedStatistics ExtractStatsFromHeader(const H& header) {
EncodedStatistics ExtractStatsFromHeader(const H& header, StatisticsMinMaxField min_max) {
EncodedStatistics page_statistics;
if (header.__isset.statistics) {
page_statistics = FromThrift(header.statistics);
page_statistics = FromThrift(header.statistics, min_max);
}
return page_statistics;
}
Expand All @@ -225,13 +225,15 @@ class SerializedPageReader : public PageReader {
public:
SerializedPageReader(std::shared_ptr<ArrowInputStream> stream, int64_t total_num_values,
Compression::type codec, const ReaderProperties& properties,
const CryptoContext* crypto_ctx, bool always_compressed)
const CryptoContext* crypto_ctx, bool always_compressed,
StatisticsMinMaxField stats_min_max_field)
: properties_(properties),
stream_(std::move(stream)),
decompression_buffer_(AllocateBuffer(properties_.memory_pool(), 0)),
page_ordinal_(0),
seen_num_values_(0),
total_num_values_(total_num_values) {
total_num_values_(total_num_values),
stats_min_max_field_(stats_min_max_field) {
if (crypto_ctx != nullptr) {
crypto_ctx_ = *crypto_ctx;
InitDecryption();
Expand Down Expand Up @@ -302,6 +304,8 @@ class SerializedPageReader : public PageReader {
// Number of values in all the data pages
int64_t total_num_values_;

StatisticsMinMaxField stats_min_max_field_;

// data_page_aad_ and data_page_header_aad_ contain the AAD for data page and data page
// header in a single column respectively.
// While calculating AAD for different pages in a single column the pages AAD is
Expand Down Expand Up @@ -349,7 +353,7 @@ bool SerializedPageReader::ShouldSkipPage(EncodedStatistics* data_page_statistic
if (page_type == PageType::DATA_PAGE) {
const format::DataPageHeader& header = current_page_header_.data_page_header;
CheckNumValuesInHeader(header.num_values);
*data_page_statistics = ExtractStatsFromHeader(header);
*data_page_statistics = ExtractStatsFromHeader(header, stats_min_max_field_);
seen_num_values_ += header.num_values;
if (data_page_filter_) {
const EncodedStatistics* filter_statistics =
Expand All @@ -370,7 +374,7 @@ bool SerializedPageReader::ShouldSkipPage(EncodedStatistics* data_page_statistic
header.repetition_levels_byte_length < 0) {
throw ParquetException("Invalid page header (negative levels byte length)");
}
*data_page_statistics = ExtractStatsFromHeader(header);
*data_page_statistics = ExtractStatsFromHeader(header, stats_min_max_field_);
seen_num_values_ += header.num_values;
if (data_page_filter_) {
const EncodedStatistics* filter_statistics =
Expand Down Expand Up @@ -591,14 +595,25 @@ std::shared_ptr<Buffer> SerializedPageReader::DecompressIfNeeded(

} // namespace

std::unique_ptr<PageReader> PageReader::Open(
std::shared_ptr<ArrowInputStream> stream, int64_t total_num_values,
Compression::type codec, const ReaderProperties& properties,
const ColumnDescriptor& descr, bool always_compressed, const CryptoContext* ctx) {
const auto stats_min_max_field = GetStatisticsMinMaxField(descr);
return std::unique_ptr<PageReader>(
new SerializedPageReader(std::move(stream), total_num_values, codec, properties,
ctx, always_compressed, stats_min_max_field));
}

std::unique_ptr<PageReader> PageReader::Open(std::shared_ptr<ArrowInputStream> stream,
int64_t total_num_values,
Compression::type codec,
const ReaderProperties& properties,
bool always_compressed,
const CryptoContext* ctx) {
return std::unique_ptr<PageReader>(new SerializedPageReader(
std::move(stream), total_num_values, codec, properties, ctx, always_compressed));
std::move(stream), total_num_values, codec, properties, ctx, always_compressed,
StatisticsMinMaxField::kMinValueMaxValue));

@wgtmac wgtmac Jun 16, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This might be slightly a breaking change because StatisticsMinMaxField::kMinValueMaxValue only looks for min_value/max_value and does not try min/max anymore. But since all production code has been migrated to the new PageReeder::Open, it should be acceptable. @pitrou

@pitrou pitrou Jun 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed, also min_value/max_value have been introduced in 2017...

}

std::unique_ptr<PageReader> PageReader::Open(std::shared_ptr<ArrowInputStream> stream,
Expand All @@ -607,9 +622,9 @@ std::unique_ptr<PageReader> PageReader::Open(std::shared_ptr<ArrowInputStream> s
bool always_compressed,
::arrow::MemoryPool* pool,
const CryptoContext* ctx) {
return std::unique_ptr<PageReader>(
new SerializedPageReader(std::move(stream), total_num_values, codec,
ReaderProperties(pool), ctx, always_compressed));
return std::unique_ptr<PageReader>(new SerializedPageReader(
std::move(stream), total_num_values, codec, ReaderProperties(pool), ctx,
always_compressed, StatisticsMinMaxField::kMinValueMaxValue));
}

namespace {
Expand Down
10 changes: 10 additions & 0 deletions cpp/src/parquet/column_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,11 +117,21 @@ class PARQUET_EXPORT PageReader {
public:
virtual ~PageReader() = default;

static std::unique_ptr<PageReader> Open(std::shared_ptr<ArrowInputStream> stream,

@wgtmac wgtmac Jun 16, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've introduced a new Open for PageReader and deprecated two below.

int64_t total_num_values,
Compression::type codec,
const ReaderProperties& properties,
const ColumnDescriptor& descr,
bool always_compressed = false,
const CryptoContext* ctx = NULLPTR);

PARQUET_DEPRECATED("Deprecated in 25.0.0. Use the ColumnDescriptor overload instead.")
static std::unique_ptr<PageReader> Open(
std::shared_ptr<ArrowInputStream> stream, int64_t total_num_values,
Compression::type codec, bool always_compressed = false,
::arrow::MemoryPool* pool = ::arrow::default_memory_pool(),
const CryptoContext* ctx = NULLPTR);
PARQUET_DEPRECATED("Deprecated in 25.0.0. Use the ColumnDescriptor overload instead.")
static std::unique_ptr<PageReader> Open(std::shared_ptr<ArrowInputStream> stream,
int64_t total_num_values,
Compression::type codec,
Expand Down
9 changes: 5 additions & 4 deletions cpp/src/parquet/column_writer_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,8 +100,8 @@ class TestPrimitiveWriter : public PrimitiveTypedTest<TestType> {
auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
ReaderProperties readerProperties;
readerProperties.set_page_checksum_verification(page_checksum_verify);
std::unique_ptr<PageReader> page_reader =
PageReader::Open(std::move(source), num_rows, compression, readerProperties);
std::unique_ptr<PageReader> page_reader = PageReader::Open(
std::move(source), num_rows, compression, readerProperties, *this->descr_);
reader_ = std::static_pointer_cast<TypedColumnReader<TestType>>(
ColumnReader::Make(this->descr_, std::move(page_reader)));
}
Expand Down Expand Up @@ -2058,8 +2058,9 @@ TEST_F(TestValuesWriterInt32Type, AvoidCompressedInDataPageV2) {
ASSERT_OK_AND_ASSIGN(auto buffer, this->sink_->Finish());
auto source = std::make_shared<::arrow::io::BufferReader>(buffer);
ReaderProperties readerProperties;
std::unique_ptr<PageReader> page_reader = PageReader::Open(
std::move(source), total_num_values, compression, readerProperties);
std::unique_ptr<PageReader> page_reader =
PageReader::Open(std::move(source), total_num_values, compression,
readerProperties, *this->descr_);
auto data_page = std::static_pointer_cast<DataPageV2>(page_reader->NextPage());
ASSERT_TRUE(data_page != nullptr);
ASSERT_FALSE(data_page->is_compressed());
Expand Down
34 changes: 26 additions & 8 deletions cpp/src/parquet/file_deserialize_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,14 @@ namespace parquet {
using ::arrow::io::BufferReader;
using ::parquet::DataPageStats;

// Make a column descriptor with an undefined column order.
ColumnDescriptor MakeColumnDescriptor() {
auto node = schema::Int32("int_col", Repetition::REQUIRED);
static_cast<schema::PrimitiveNode*>(node.get())
->SetColumnOrder(ColumnOrder::undefined_);
return ColumnDescriptor(node, /*max_definition_level=*/0, /*max_repetition_level=*/0);
}

// Adds page statistics occupying a certain amount of bytes (for testing very
// large page headers)
template <typename H>
Expand Down Expand Up @@ -110,6 +118,8 @@ static std::vector<Compression::type> GetSupportedCodecTypes() {

class TestPageSerde : public ::testing::Test {
public:
TestPageSerde() : descr_(MakeColumnDescriptor()) {}

void SetUp() {
data_page_header_.encoding = format::Encoding::PLAIN;
data_page_header_.definition_level_encoding = format::Encoding::RLE;
Expand All @@ -124,7 +134,7 @@ class TestPageSerde : public ::testing::Test {
EndStream();

auto stream = std::make_shared<::arrow::io::BufferReader>(out_buffer_);
page_reader_ = PageReader::Open(stream, num_rows, codec, properties);
page_reader_ = PageReader::Open(stream, num_rows, codec, properties, descr_);
}

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

ColumnDescriptor descr_;
std::unique_ptr<PageReader> page_reader_;
format::PageHeader page_header_;
format::DataPageHeader data_page_header_;
Expand Down Expand Up @@ -466,7 +477,8 @@ TYPED_TEST(PageFilterTest, TestPageWithoutStatistics) {

auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
this->page_reader_ =
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
ReaderProperties(), this->descr_);

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

std::vector<EncodedStatistics> read_stats;
std::vector<int64_t> read_num_values;
Expand Down Expand Up @@ -526,7 +539,8 @@ TYPED_TEST(PageFilterTest, TestPageFilterCallback) {
{ // Skip all pages.
auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
this->page_reader_ =
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
ReaderProperties(), this->descr_);

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

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

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

auto stream = std::make_shared<::arrow::io::BufferReader>(this->out_buffer_);
this->page_reader_ =
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED);
PageReader::Open(stream, this->total_rows_, Compression::UNCOMPRESSED,
ReaderProperties(), this->descr_);

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

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

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

Expand Down Expand Up @@ -641,7 +658,8 @@ TEST_F(TestPageSerde, SkipsNonDataPages) {
EndStream();

auto stream = std::make_shared<::arrow::io::BufferReader>(out_buffer_);
page_reader_ = PageReader::Open(stream, /*num_rows=*/100, Compression::UNCOMPRESSED);
page_reader_ = PageReader::Open(stream, /*num_rows=*/100, Compression::UNCOMPRESSED,
ReaderProperties(), descr_);

// Only the two data pages are returned.
std::shared_ptr<Page> current_page = page_reader_->NextPage();
Expand Down
5 changes: 3 additions & 2 deletions cpp/src/parquet/file_reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,7 @@ class SerializedRowGroup : public RowGroupReader::Contents {
std::unique_ptr<PageReader> GetColumnPageReader(int i) override {
// Read column chunk from the file
auto col = row_group_metadata_->ColumnChunk(i);
const ColumnDescriptor* descr = row_group_metadata_->schema()->Column(i);

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

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

private:
Expand Down
78 changes: 41 additions & 37 deletions cpp/src/parquet/metadata.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,40 +90,48 @@ std::string ParquetVersionToString(ParquetVersion::type ver) {
return "UNKNOWN";
}

namespace {

template <typename DType>
static std::shared_ptr<Statistics> MakeTypedColumnStats(
const format::ColumnMetaData& metadata, const ColumnDescriptor* descr,
::arrow::MemoryPool* pool) {
std::optional<bool> min_exact =
metadata.statistics.__isset.is_min_value_exact
? std::optional<bool>(metadata.statistics.is_min_value_exact)
: std::nullopt;
std::optional<bool> max_exact =
metadata.statistics.__isset.is_max_value_exact
? std::optional<bool>(metadata.statistics.is_max_value_exact)
: std::nullopt;
// If ColumnOrder is defined, return max_value and min_value
if (descr->column_order().get_order() == ColumnOrder::TYPE_DEFINED_ORDER) {
return MakeStatistics<DType>(
descr, metadata.statistics.min_value, metadata.statistics.max_value,
metadata.num_values - metadata.statistics.null_count,
metadata.statistics.null_count, metadata.statistics.distinct_count,
metadata.statistics.__isset.max_value && metadata.statistics.__isset.min_value,
metadata.statistics.__isset.null_count,
metadata.statistics.__isset.distinct_count, min_exact, max_exact, pool);
}
// Default behavior
std::shared_ptr<Statistics> MakeTypedColumnStats(const format::ColumnMetaData& metadata,
const ColumnDescriptor* descr,
::arrow::MemoryPool* pool) {
const auto& statistics = metadata.statistics;
const std::string kEmpty = "";
const std::string* encoded_min = &kEmpty;
const std::string* encoded_max = &kEmpty;
bool has_min_max = false;
std::optional<bool> min_exact = std::nullopt;
std::optional<bool> max_exact = std::nullopt;

switch (GetStatisticsMinMaxField(*descr)) {
case StatisticsMinMaxField::kMinValueMaxValue:
encoded_min = &statistics.min_value;
encoded_max = &statistics.max_value;
has_min_max = statistics.__isset.max_value && statistics.__isset.min_value;
min_exact = statistics.__isset.is_min_value_exact
? std::optional<bool>(statistics.is_min_value_exact)
: std::nullopt;
max_exact = statistics.__isset.is_max_value_exact
? std::optional<bool>(statistics.is_max_value_exact)
: std::nullopt;
break;
case StatisticsMinMaxField::kLegacyMinMax:
encoded_min = &statistics.min;
encoded_max = &statistics.max;
has_min_max = statistics.__isset.max && statistics.__isset.min;
break;
case StatisticsMinMaxField::kInvalid:
break;
}

return MakeStatistics<DType>(
descr, metadata.statistics.min, metadata.statistics.max,
metadata.num_values - metadata.statistics.null_count,
metadata.statistics.null_count, metadata.statistics.distinct_count,
metadata.statistics.__isset.max && metadata.statistics.__isset.min,
metadata.statistics.__isset.null_count, metadata.statistics.__isset.distinct_count,
min_exact, max_exact, pool);
descr, *encoded_min, *encoded_max, metadata.num_values - statistics.null_count,
statistics.null_count, statistics.distinct_count, has_min_max,
statistics.__isset.null_count, statistics.__isset.distinct_count, min_exact,
max_exact, pool);
}

namespace {

std::shared_ptr<geospatial::GeoStatistics> MakeColumnGeometryStats(
const format::ColumnMetaData& metadata, const ColumnDescriptor* descr) {
if (metadata.__isset.geospatial_statistics) {
Expand Down Expand Up @@ -336,12 +344,8 @@ class ColumnChunkMetaData::ColumnChunkMetaDataImpl {
{
const std::lock_guard<std::mutex> guard(stats_mutex_);
if (possible_encoded_stats_ == nullptr) {
possible_encoded_stats_ =
std::make_shared<EncodedStatistics>(FromThrift(column_metadata_->statistics));
if (descr_->sort_order() == SortOrder::UNKNOWN) {
// If the column SortOrder is Unknown we can't trust max/min.
possible_encoded_stats_->ClearMinMax();
}
possible_encoded_stats_ = std::make_shared<EncodedStatistics>(
FromThrift(column_metadata_->statistics, GetStatisticsMinMaxField(*descr_)));
}
Comment thread
wgtmac marked this conversation as resolved.
}
return writer_version_->HasCorrectStatistics(type(), *possible_encoded_stats_,
Expand Down Expand Up @@ -1037,7 +1041,7 @@ class FileMetaData::FileMetaDataImpl {
if (column_order.__isset.TYPE_ORDER) {
column_orders.push_back(ColumnOrder::type_defined_);
} else {
column_orders.push_back(ColumnOrder::undefined_);
column_orders.push_back(ColumnOrder::unknown_);
}
}
} else {
Expand Down
Loading
Loading