From 942c3750789cf0326991ef374d33778fe85a67eb Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 12 Jul 2026 13:36:16 +0800 Subject: [PATCH] [fix](be) Disable Parquet binary min max pushdown ### What problem does this PR solve? Issue Number: N/A Related PR: N/A Problem Summary: FileScannerV2 returned Parquet footer min/max statistics as exact aggregate values. For BYTE_ARRAY and FIXED_LEN_BYTE_ARRAY columns, Parquet permits truncated lower and upper bounds, and Arrow 17 does not expose the exactness flags. MIN/MAX pushdown could therefore return a synthetic bound that does not occur in the data. Reject aggregate pushdown for these physical types so the upper reader falls back to a normal scan, while retaining exact numeric pushdown and safe statistics pruning. ### Release note Fix incorrect Parquet MIN/MAX aggregate results for truncated binary statistics. ### Check List (For Author) - Test: Unit Test - ParquetScanTest targeted binary rejection and numeric success tests - Behavior changed: Yes, Parquet binary MIN/MAX aggregate pushdown now falls back to scanning. - Does this need documentation: No --- be/src/format_v2/parquet/parquet_reader.cpp | 17 +++++++ .../format_v2/parquet/parquet_scan_test.cpp | 46 +++++++++++++++++++ 2 files changed, 63 insertions(+) diff --git a/be/src/format_v2/parquet/parquet_reader.cpp b/be/src/format_v2/parquet/parquet_reader.cpp index 82ad3f78891c80..01ad3333818b27 100644 --- a/be/src/format_v2/parquet/parquet_reader.cpp +++ b/be/src/format_v2/parquet/parquet_reader.cpp @@ -287,6 +287,22 @@ static Status find_projected_minmax_leaf(const ParquetColumnSchema& column_schem child_projection.local_id(), column_schema.name); } +static Status validate_minmax_aggregate_statistics(const ParquetColumnSchema& column_schema) { + DORIS_CHECK(column_schema.descriptor != nullptr); + switch (column_schema.descriptor->physical_type()) { + case ::parquet::Type::BYTE_ARRAY: + case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: + // Arrow 17 does not expose Parquet's min/max exactness flags. Binary statistics may be + // truncated bounds rather than values present in the file, so they are safe for pruning + // but cannot be returned as exact aggregate results. + return Status::NotSupported( + "Parquet MIN/MAX aggregate pushdown requires exact statistics for column {}", + column_schema.name); + default: + return Status::OK(); + } +} + void ParquetReader::_fill_column_definition(const ParquetColumnSchema& column_schema, format::ColumnDefinition* field) const { if (column_schema.parquet_field_id >= 0) { @@ -671,6 +687,7 @@ Status ParquetReader::get_aggregate_result(const format::FileAggregateRequest& r RETURN_IF_ERROR(find_projected_minmax_leaf( *column_schema, request.columns[request_column_idx].projection, &leaf_schema)); DORIS_CHECK(leaf_schema != nullptr); + RETURN_IF_ERROR(validate_minmax_aggregate_statistics(*leaf_schema)); auto& aggregate_column = result->columns[request_column_idx]; aggregate_column.projection = request.columns[request_column_idx].projection; diff --git a/be/test/format_v2/parquet/parquet_scan_test.cpp b/be/test/format_v2/parquet/parquet_scan_test.cpp index edd51c4520307a..d73e08235c8b4e 100644 --- a/be/test/format_v2/parquet/parquet_scan_test.cpp +++ b/be/test/format_v2/parquet/parquet_scan_test.cpp @@ -222,6 +222,25 @@ std::shared_ptr build_int32_array(const std::vector& valu return finish_array(&builder); } +std::shared_ptr build_string_array(const std::vector& values) { + arrow::StringBuilder builder; + for (const auto& value : values) { + EXPECT_TRUE(builder.Append(value).ok()); + } + return finish_array(&builder); +} + +std::shared_ptr build_fixed_binary_array(const std::vector& values, + int byte_width) { + auto type = arrow::fixed_size_binary(byte_width); + arrow::FixedSizeBinaryBuilder builder(type, arrow::default_memory_pool()); + for (const auto& value : values) { + EXPECT_EQ(value.size(), byte_width); + EXPECT_TRUE(builder.Append(reinterpret_cast(value.data())).ok()); + } + return finish_array(&builder); +} + std::shared_ptr build_struct_array(const std::vector& ids, const std::vector& names) { auto struct_type = arrow::struct_({arrow::field("id", arrow::int32(), false), @@ -295,6 +314,16 @@ void write_int_pair_parquet_file(const std::string& file_path, int64_t row_group write_table(file_path, table, row_group_size, false, false, enable_statistics); } +void write_binary_minmax_parquet_file(const std::string& file_path) { + auto schema = arrow::schema({ + arrow::field("text", arrow::utf8(), false), + arrow::field("fixed", arrow::fixed_size_binary(4), false), + }); + auto table = arrow::Table::Make(schema, {build_string_array({"alpha", "omega"}), + build_fixed_binary_array({"aaaa", "zzzz"}, 4)}); + write_table(file_path, table, 2); +} + void write_struct_parquet_file(const std::string& file_path) { auto struct_type = arrow::struct_({arrow::field("id", arrow::int32(), false), arrow::field("name", arrow::utf8(), false)}); @@ -692,6 +721,23 @@ TEST_F(ParquetScanTest, AggregateCountAndMinMaxUseAllSelectedRowGroups) { EXPECT_EQ(minmax_result.columns[1].max_value.get(), 60); } +TEST_F(ParquetScanTest, AggregateMinMaxRejectsInexactBinaryStatistics) { + write_binary_minmax_parquet_file(_file_path); + auto reader = create_reader(); + RuntimeState state {TQueryOptions(), TQueryGlobals()}; + ASSERT_TRUE(reader->init(&state).ok()); + open_all_row_groups(reader.get()); + + for (int32_t column_id = 0; column_id < 2; ++column_id) { + format::FileAggregateRequest request; + request.agg_type = TPushAggOp::MINMAX; + request.columns.push_back({.projection = field_projection(column_id)}); + format::FileAggregateResult result; + const auto status = reader->get_aggregate_result(request, &result); + EXPECT_TRUE(status.is()) << status; + } +} + TEST_F(ParquetScanTest, AggregateRespectsStatisticsPrunedRowGroups) { write_int_pair_parquet_file(_file_path); auto reader = create_reader();