[fix](be) Disable Parquet binary min max pushdown

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;

be/test/format_v2/parquet/parquet_scan_test.cpp

@@ -222,6 +222,25 @@ std::shared_ptr<arrow::Array> build_int32_array(const std::vector<int32_t>& valu
     return finish_array(&builder);
 }
 
+std::shared_ptr<arrow::Array> build_string_array(const std::vector<std::string>& values) {
+    arrow::StringBuilder builder;
+    for (const auto& value : values) {
+        EXPECT_TRUE(builder.Append(value).ok());
+    }
+    return finish_array(&builder);
+}
+
+std::shared_ptr<arrow::Array> build_fixed_binary_array(const std::vector<std::string>& 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<const uint8_t*>(value.data())).ok());
+    }
+    return finish_array(&builder);
+}
+
 std::shared_ptr<arrow::Array> build_struct_array(const std::vector<int32_t>& ids,
                                                  const std::vector<std::string>& 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<TYPE_INT>(), 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<ErrorCode::NOT_IMPLEMENTED_ERROR>()) << status;
+    }
+}
+
 TEST_F(ParquetScanTest, AggregateRespectsStatisticsPrunedRowGroups) {
     write_int_pair_parquet_file(_file_path);
     auto reader = create_reader();