[fix](iceberg) Honor disabled write metrics

be/src/exec/sink/viceberg_merge_sink.cpp

@@ -254,6 +254,9 @@ Status VIcebergMergeSink::_build_inner_sinks() {
     if (merge_sink.__isset.broker_addresses) {
         table_sink.__set_broker_addresses(merge_sink.broker_addresses);
     }
+    if (merge_sink.__isset.collect_column_stats) {
+        table_sink.__set_collect_column_stats(merge_sink.collect_column_stats);
+    }
     _table_sink.__set_type(TDataSinkType::ICEBERG_TABLE_SINK);
     _table_sink.__set_iceberg_table_sink(table_sink);
 

be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp

@@ -47,7 +47,11 @@ VIcebergPartitionWriter::VIcebergPartitionWriter(
           _file_name_index(file_name_index),
           _file_format_type(file_format_type),
           _compress_type(compress_type),
-          _hadoop_conf(hadoop_conf) {}
+          _hadoop_conf(hadoop_conf) {
+    if (t_sink.iceberg_table_sink.__isset.collect_column_stats) {
+        _collect_column_stats = t_sink.iceberg_table_sink.collect_column_stats;
+    }
+}
 
 Status VIcebergPartitionWriter::open(RuntimeState* state, RuntimeProfile* profile,
                                      const RowDescriptor* row_desc) {
@@ -156,6 +160,10 @@ Status VIcebergPartitionWriter::_build_iceberg_commit_data(TIcebergCommitData* c
     commit_data->__set_file_size(_file_format_transformer->written_len());
     commit_data->__set_file_content(TFileContent::DATA);
     commit_data->__set_partition_values(_partition_values);
+    // ORC collection reopens the file, so honor the FE policy before any footer work.
+    if (!_collect_column_stats) {
+        return Status::OK();
+    }
     if (_file_format_type == TFileFormatType::FORMAT_PARQUET) {
         TIcebergColumnStats column_stats;
         RETURN_IF_ERROR(static_cast<VParquetTransformer*>(_file_format_transformer.get())

be/src/exec/sink/writer/iceberg/viceberg_partition_writer.h

@@ -67,6 +67,8 @@ class VIcebergPartitionWriter : public IPartitionWriterBase {
     inline size_t written_len() const override { return _file_format_transformer->written_len(); }
 
 private:
+    friend class VIcebergPartitionWriterTest;
+
     std::string _get_target_file_name();
 
     Status _build_iceberg_commit_data(TIcebergCommitData* commit_data);
@@ -91,6 +93,7 @@ class VIcebergPartitionWriter : public IPartitionWriterBase {
     TFileFormatType::type _file_format_type;
     TFileCompressType::type _compress_type;
     const std::map<std::string, std::string>& _hadoop_conf;
+    bool _collect_column_stats = true;
 
     std::shared_ptr<io::FileSystem> _fs = nullptr;
 

be/src/format/transformer/vorc_transformer.cpp

@@ -385,12 +385,19 @@ Status VOrcTransformer::collect_file_statistics_after_close(TIcebergColumnStats*
 
         const iceberg::StructType& root_struct = _iceberg_schema->root_struct();
         const auto& nested_fields = root_struct.fields();
+        const orc::Type& orc_root_type = reader->getType();
         for (uint32_t i = 0; i < nested_fields.size(); i++) {
-            uint32_t orc_col_id = i + 1; // skip root struct
-            if (orc_col_id >= file_stats->getNumberOfColumns()) {
+            if (i >= orc_root_type.getSubtypeCount()) {
+                continue;
+            }
+            // ORC IDs are depth-first, so top-level fields after a complex field are not i + 1.
+            const uint64_t raw_orc_col_id = orc_root_type.getSubtype(i)->getColumnId();
+            if (raw_orc_col_id >= file_stats->getNumberOfColumns()) {
                 continue;
             }
 
+            // The uint32_t column-count check above makes narrowing to the ORC API width safe.
+            const uint32_t orc_col_id = static_cast<uint32_t>(raw_orc_col_id);
             const orc::ColumnStatistics* col_stats = file_stats->getColumnStatistics(orc_col_id);
             if (col_stats == nullptr) {
                 continue;

be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp

@@ -0,0 +1,107 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <optional>
+
+#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
+
+namespace doris {
+
+namespace {
+
+class FakeFileFormatTransformer final : public VFileFormatTransformer {
+public:
+    explicit FakeFileFormatTransformer(const VExprContextSPtrs& output_exprs)
+            : VFileFormatTransformer(nullptr, output_exprs, false) {}
+
+    Status open() override { return Status::OK(); }
+    Status write(const Block&) override { return Status::OK(); }
+    Status close() override { return Status::OK(); }
+    int64_t written_len() override { return 64; }
+};
+
+TDataSink make_table_sink(std::optional<bool> collect_column_stats) {
+    TIcebergTableSink iceberg_sink;
+    if (collect_column_stats.has_value()) {
+        iceberg_sink.__set_collect_column_stats(*collect_column_stats);
+    }
+    TDataSink sink;
+    sink.__set_type(TDataSinkType::ICEBERG_TABLE_SINK);
+    sink.__set_iceberg_table_sink(iceberg_sink);
+    return sink;
+}
+
+} // namespace
+
+class VIcebergPartitionWriterTest : public testing::Test {
+protected:
+    static std::unique_ptr<VIcebergPartitionWriter> make_writer(
+            const TDataSink& sink, const VExprContextSPtrs& output_exprs,
+            const iceberg::Schema& schema, const std::string* schema_json,
+            const std::map<std::string, std::string>& hadoop_conf) {
+        IPartitionWriterBase::WriteInfo write_info;
+        write_info.file_type = TFileType::FILE_LOCAL;
+        return std::make_unique<VIcebergPartitionWriter>(
+                sink, std::vector<std::string> {}, output_exprs, schema, schema_json,
+                std::vector<std::string> {}, std::move(write_info), "data", 0,
+                TFileFormatType::FORMAT_ORC, TFileCompressType::ZLIB, hadoop_conf);
+    }
+
+    static void install_fake_transformer(VIcebergPartitionWriter* writer,
+                                         const VExprContextSPtrs& output_exprs) {
+        writer->_file_format_transformer =
+                std::make_unique<FakeFileFormatTransformer>(output_exprs);
+    }
+
+    static Status build_commit_data(VIcebergPartitionWriter* writer,
+                                    TIcebergCommitData* commit_data) {
+        return writer->_build_iceberg_commit_data(commit_data);
+    }
+
+    static bool collect_column_stats(const VIcebergPartitionWriter& writer) {
+        return writer._collect_column_stats;
+    }
+};
+
+TEST_F(VIcebergPartitionWriterTest, OrcSkipsFooterCollectionWhenMetricsAreDisabled) {
+    VExprContextSPtrs output_exprs;
+    iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+    std::string schema_json;
+    std::map<std::string, std::string> hadoop_conf;
+    auto writer =
+            make_writer(make_table_sink(false), output_exprs, schema, &schema_json, hadoop_conf);
+    install_fake_transformer(writer.get(), output_exprs);
+
+    TIcebergCommitData commit_data;
+    ASSERT_TRUE(build_commit_data(writer.get(), &commit_data).ok());
+    EXPECT_FALSE(commit_data.__isset.column_stats);
+}
+
+TEST_F(VIcebergPartitionWriterTest, MissingPolicyKeepsCollectionEnabledForRollingUpgrade) {
+    VExprContextSPtrs output_exprs;
+    iceberg::Schema schema(std::vector<iceberg::NestedField> {});
+    std::string schema_json;
+    std::map<std::string, std::string> hadoop_conf;
+    auto writer = make_writer(make_table_sink(std::nullopt), output_exprs, schema, &schema_json,
+                              hadoop_conf);
+
+    EXPECT_TRUE(collect_column_stats(*writer));
+}
+
+} // namespace doris

be/test/format/transformer/vorc_transformer_test.cpp

@@ -0,0 +1,107 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "format/transformer/vorc_transformer.h"
+
+#include <gtest/gtest.h>
+
+#include "core/block/block.h"
+#include "core/column/column_string.h"
+#include "core/column/column_struct.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_struct.h"
+#include "format/table/iceberg/schema_parser.h"
+#include "io/fs/local_file_system.h"
+#include "runtime/runtime_state.h"
+#include "testutil/mock/mock_slot_ref.h"
+#include "util/uid_util.h"
+
+namespace doris {
+
+class VOrcTransformerTest : public testing::Test {
+protected:
+    void SetUp() override {
+        _file_path = "./vorc_transformer_" + UniqueId::gen_uid().to_string() + ".orc";
+        _fs = io::global_local_filesystem();
+    }
+
+    void TearDown() override { static_cast<void>(_fs->delete_file(_file_path)); }
+
+    std::string _file_path;
+    std::shared_ptr<io::FileSystem> _fs;
+};
+
+TEST_F(VOrcTransformerTest, CollectsBoundsForTopLevelFieldAfterStruct) {
+    auto int_type = std::make_shared<DataTypeInt32>();
+    auto struct_type = std::make_shared<DataTypeStruct>(DataTypes {int_type}, Strings {"a"});
+    auto string_type = std::make_shared<DataTypeString>();
+    VExprContextSPtrs output_exprs =
+            MockSlotRef::create_mock_contexts(DataTypes {struct_type, string_type});
+
+    const std::string schema_json = R"({
+        "type": "struct",
+        "fields": [
+            {
+                "id": 1,
+                "name": "s",
+                "required": true,
+                "type": {
+                    "type": "struct",
+                    "fields": [
+                        {"id": 2, "name": "a", "required": true, "type": "int"}
+                    ]
+                }
+            },
+            {"id": 3, "name": "b", "required": true, "type": "string"}
+        ]
+    })";
+    std::unique_ptr<iceberg::Schema> schema = iceberg::SchemaParser::from_json(schema_json);
+
+    io::FileWriterPtr file_writer;
+    ASSERT_TRUE(_fs->create_file(_file_path, &file_writer).ok());
+    RuntimeState state;
+    VOrcTransformer transformer(&state, file_writer.get(), output_exprs, "", {"s", "b"}, false,
+                                TFileCompressType::PLAIN, schema.get(), _fs);
+    ASSERT_TRUE(transformer.open().ok());
+
+    auto nested_column = ColumnInt32::create();
+    nested_column->insert_value(-1);
+    Columns struct_columns;
+    struct_columns.emplace_back(std::move(nested_column));
+    auto struct_column = ColumnStruct::create(std::move(struct_columns));
+    auto string_column = ColumnString::create();
+    string_column->insert_data("hello", 5);
+
+    Block block;
+    block.insert(ColumnWithTypeAndName(std::move(struct_column), struct_type, "s"));
+    block.insert(ColumnWithTypeAndName(std::move(string_column), string_type, "b"));
+    ASSERT_TRUE(transformer.write(block).ok());
+    ASSERT_TRUE(transformer.close().ok());
+
+    TIcebergColumnStats stats;
+    ASSERT_TRUE(transformer.collect_file_statistics_after_close(&stats).ok());
+    ASSERT_TRUE(stats.__isset.lower_bounds);
+    ASSERT_TRUE(stats.__isset.upper_bounds);
+    ASSERT_EQ(1, stats.lower_bounds.count(3));
+    ASSERT_EQ(1, stats.upper_bounds.count(3));
+    EXPECT_EQ("hello", stats.lower_bounds.at(3));
+    EXPECT_EQ("hello", stats.upper_bounds.at(3));
+}
+
+} // namespace doris

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java

@@ -79,14 +79,17 @@
 import com.google.gson.reflect.TypeToken;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.ManifestFile;
 import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.MetadataTableType;
 import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.MetricsModes;
+import org.apache.iceberg.MetricsUtil;
 import org.apache.iceberg.PartitionData;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
@@ -1961,10 +1964,25 @@ public static Schema appendRowLineageFieldsForV3(Schema schema) {
                 MetadataColumns.ROW_ID, MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER));
     }
 
+    public static boolean shouldCollectColumnStats(Table table, Schema writerSchema) {
+        MetricsConfig metricsConfig = MetricsConfig.forTable(table);
+        if (getFileFormat(table) == FileFormat.ORC) {
+            // Match the footer collectors: ORC reports top-level collection counts, while Parquet reports leaf fields.
+            return writerSchema.columns().stream()
+                    .anyMatch(field -> MetricsUtil.metricsMode(writerSchema, metricsConfig, field.fieldId())
+                            != MetricsModes.None.get());
+        }
+        return TypeUtil.indexById(writerSchema.asStruct()).values().stream()
+                .filter(field -> field.type().isPrimitiveType())
+                .anyMatch(field -> MetricsUtil.metricsMode(writerSchema, metricsConfig, field.fieldId())
+                        != MetricsModes.None.get());
+    }
+
     public static int getFormatVersion(Table table) {
         int formatVersion = 2; // default format version : 2
-        if (table instanceof BaseTable) {
-            formatVersion = ((BaseTable) table).operations().current().formatVersion();
+        if (table instanceof HasTableOperations) {
+            // TransactionTable exposes the real format version through operations, not table properties.
+            formatVersion = ((HasTableOperations) table).operations().current().formatVersion();
         } else if (table != null && table.properties() != null) {
             String version = table.properties().get(TableProperties.FORMAT_VERSION);
             if (version != null) {