[feature](iceberg) Support Iceberg V3 default values

Backport #65851 to branch-4.1.

Preserve missing Iceberg default columns across TopN row-id fetch batches and keep the branch-4.1 table reader layout.

Author: hubgeter

be/src/exec/scan/access_path_parser.cpp

@@ -90,10 +90,10 @@ void inherit_schema_metadata(format::ColumnDefinition* column,
     // The presence bit is part of the mapping contract: an explicit empty mapping must remain
     // authoritative after access-path pruning instead of enabling current-name fallback.
     column->has_name_mapping = schema_column->has_name_mapping;
-    // Initial defaults describe the logical value of fields absent from older files. Nested
-    // access-path pruning must retain them just like it retains rename metadata.
     column->initial_default_value = schema_column->initial_default_value;
     column->initial_default_value_is_base64 = schema_column->initial_default_value_is_base64;
+    column->is_optional = schema_column->is_optional;
+    column->default_expr = schema_column->default_expr;
 }
 
 const format::ColumnDefinition* find_schema_child_by_path(
@@ -152,8 +152,7 @@ int32_t schema_field_id_or(const format::ColumnDefinition* schema_column, int32_
 
 std::string schema_field_name_or(const format::ColumnDefinition* schema_column,
                                  std::string fallback) {
-    return schema_column == nullptr || schema_column->name.empty() ? std::move(fallback)
-                                                                   : schema_column->name;
+    return schema_column == nullptr || schema_column->name.empty() ? fallback : schema_column->name;
 }
 
 struct AccessPathNode {
@@ -253,7 +252,7 @@ Status build_all_nested_children_from_schema(format::ColumnDefinition* column,
     case TYPE_ARRAY: {
         const auto& array_type = assert_cast<const DataTypeArray&>(*nested_type);
         const auto* element_schema = schema_column != nullptr && !schema_column->children.empty()
-                                             ? &schema_column->children[0]
+                                             ? schema_column->children.data()
                                              : nullptr;
         auto* child = find_or_add_child(column, schema_field_id_or(element_schema, 0), "element",
                                         array_type.get_nested_type());
@@ -264,7 +263,7 @@ Status build_all_nested_children_from_schema(format::ColumnDefinition* column,
     case TYPE_MAP: {
         const auto& map_type = assert_cast<const DataTypeMap&>(*nested_type);
         const auto* key_schema = schema_column != nullptr && !schema_column->children.empty()
-                                         ? &schema_column->children[0]
+                                         ? schema_column->children.data()
                                          : nullptr;
         const auto* value_schema = schema_column != nullptr && schema_column->children.size() > 1
                                            ? &schema_column->children[1]
@@ -362,15 +361,7 @@ Status build_map_children_from_access_node(format::ColumnDefinition* column,
             merge_access_path_node(&key_node, child_node);
             continue;
         }
-        if (child_path == "VALUES") {
-            need_key = true;
-            key_node.project_all = true;
-            key_node.children.clear();
-            need_value = true;
-            merge_access_path_node(&value_node, child_node);
-            continue;
-        }
-        if (child_path == "*") {
+        if (child_path == "VALUES" || child_path == "*") {
             need_key = true;
             key_node.project_all = true;
             key_node.children.clear();
@@ -406,7 +397,7 @@ Status build_map_children_from_access_node(format::ColumnDefinition* column,
     }
 
     const auto* key_schema = schema_column != nullptr && !schema_column->children.empty()
-                                     ? &schema_column->children[0]
+                                     ? schema_column->children.data()
                                      : nullptr;
     const auto* value_schema = schema_column != nullptr && schema_column->children.size() > 1
                                        ? &schema_column->children[1]
@@ -455,7 +446,7 @@ Status build_nested_children_from_access_node(format::ColumnDefinition* column,
         }
         const auto& array_type = assert_cast<const DataTypeArray&>(*nested_type);
         const auto* element_schema = schema_column != nullptr && !schema_column->children.empty()
-                                             ? &schema_column->children[0]
+                                             ? schema_column->children.data()
                                              : nullptr;
         auto* child = find_or_add_child(column, schema_field_id_or(element_schema, 0), "element",
                                         array_type.get_nested_type());

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

@@ -17,6 +17,7 @@
 
 #include "exec/sink/writer/iceberg/viceberg_table_writer.h"
 
+#include "common/exception.h"
 #include "core/block/block.h"
 #include "core/block/column_with_type_and_name.h"
 #include "core/block/materialize_block.h"
@@ -127,7 +128,14 @@ VIcebergTableWriter::_to_iceberg_partition_columns() {
         id_to_column_idx[_schema->columns()[i].field_id()] = i;
     }
     for (const auto& partition_field : _partition_spec->fields()) {
-        int column_idx = id_to_column_idx[partition_field.source_id()];
+        auto column_idx_it = id_to_column_idx.find(partition_field.source_id());
+        if (column_idx_it == id_to_column_idx.end()) {
+            throw Exception(
+                    ErrorCode::INTERNAL_ERROR,
+                    "Iceberg partition field {} references source field {} outside writer schema",
+                    partition_field.field_id(), partition_field.source_id());
+        }
+        int column_idx = column_idx_it->second;
         std::unique_ptr<PartitionColumnTransform> partition_column_transform =
                 PartitionColumnTransforms::create(
                         partition_field, _vec_output_expr_ctxs[column_idx]->root()->data_type());

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

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <gen_cpp/DataSinks_types.h>
+#include <gtest/gtest_prod.h>
 
 #include "common/atomic_shared_ptr.h"
 #include "core/block/block.h"
@@ -81,6 +82,8 @@ class VIcebergTableWriter final : public AsyncResultWriter {
     std::shared_ptr<IPartitionWriterBase> current_writer() const { return _current_writer.load(); }
 
 private:
+    FRIEND_TEST(VIcebergTableWriterTest, RejectMissingPartitionSource);
+
     // The currently active partition writer (may be VIcebergPartitionWriter or VIcebergSortWriter).
     // Updated during write() to track which writer received the most recent data.
     // Wrapped in atomic_shared_ptr because revoke_memory / get_revocable_mem_size run on

be/src/format/arrow/arrow_block_convertor.cpp

@@ -39,6 +39,7 @@
 #include <utility>
 #include <vector>
 
+#include "common/cast_set.h"
 #include "common/status.h"
 #include "core/block/column_with_type_and_name.h"
 #include "core/column/column.h"
@@ -84,7 +85,9 @@ int hex_value(char c) {
     return -1;
 }
 
-Status parse_uuid_to_bytes(StringRef uuid, std::array<uint8_t, 16>* bytes) {
+} // namespace
+
+Status parse_iceberg_uuid_to_bytes(StringRef uuid, std::array<uint8_t, 16>* bytes) {
     if (uuid.size == 16) {
         std::memcpy(bytes->data(), uuid.data, bytes->size());
         return Status::OK();
@@ -126,6 +129,8 @@ Status parse_uuid_to_bytes(StringRef uuid, std::array<uint8_t, 16>* bytes) {
     return Status::OK();
 }
 
+namespace {
+
 Status write_iceberg_uuid_string_column_to_arrow(const IColumn& column, const DataTypePtr& type,
                                                  arrow::ArrayBuilder* array_builder, int64_t start,
                                                  int64_t end) {
@@ -153,13 +158,15 @@ Status write_iceberg_uuid_string_column_to_arrow(const IColumn& column, const Da
     }
 
     const auto& string_column = assert_cast<const ColumnString&>(*data_column);
-    for (size_t row = start; row < end; ++row) {
+    const auto begin_row = cast_set<size_t>(start);
+    const auto end_row = cast_set<size_t>(end);
+    for (size_t row = begin_row; row < end_row; ++row) {
         if (null_map != nullptr && (*null_map)[row]) {
             RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column, builder));
             continue;
         }
         std::array<uint8_t, 16> bytes;
-        RETURN_IF_ERROR(parse_uuid_to_bytes(string_column.get_data_at(row), &bytes));
+        RETURN_IF_ERROR(parse_iceberg_uuid_to_bytes(string_column.get_data_at(row), &bytes));
         RETURN_IF_ERROR(checkArrowStatus(builder.Append(bytes.data()), column, builder));
     }
     return Status::OK();

be/src/format/arrow/arrow_block_convertor.h

@@ -19,13 +19,15 @@
 
 #include <cctz/time_zone.h>
 
+#include <array>
 #include <cstdint>
 #include <memory>
 
 #include "common/status.h"
 #include "core/block/block.h"
 #include "core/column/column.h"
 #include "core/data_type/data_type.h"
+#include "core/string_ref.h"
 
 // This file will convert Doris Block to/from Arrow's RecordBatch
 // Block is used by Doris query engine to exchange data between
@@ -41,6 +43,8 @@ class Schema;
 
 namespace doris {
 
+Status parse_iceberg_uuid_to_bytes(StringRef uuid, std::array<uint8_t, 16>* bytes);
+
 class FromBlockToRecordBatchConverter {
 public:
     FromBlockToRecordBatchConverter(const Block& block,

be/src/format/orc/vorc_reader.cpp

@@ -81,6 +81,7 @@
 #include "exprs/vin_predicate.h"
 #include "exprs/vruntimefilter_wrapper.h"
 #include "format/orc/orc_file_reader.h"
+#include "format/table/iceberg_default_value.h"
 #include "format/table/iceberg_reader.h"
 #include "format/table/partition_column_filler.h"
 #include "format/table/transactional_hive_common.h"
@@ -559,6 +560,7 @@ Status OrcReader::init_reader(
 
     RETURN_IF_ERROR(_create_file_reader());
     RETURN_IF_ERROR(_init_read_columns());
+    _nested_initial_default_values.clear();
     return Status::OK();
 }
 
@@ -2290,7 +2292,8 @@ Status OrcReader::_fill_doris_data_column(const std::string& col_name,
         const auto* orc_struct = dynamic_cast<const orc::StructVectorBatch*>(cvb);
         auto& doris_struct = static_cast<ColumnStruct&>(*data_column);
         std::map<int, int> read_fields;
-        std::set<int> missing_fields;
+        std::set<int> schema_missing_fields;
+        std::set<int> projected_out_fields;
         const auto* doris_struct_type =
                 assert_cast<const DataTypeStruct*>(remove_nullable(data_type).get());
 
@@ -2305,7 +2308,7 @@ Status OrcReader::_fill_doris_data_column(const std::string& col_name,
         for (int i = 0; i < doris_struct.tuple_size(); ++i) {
             const auto& table_column_name = doris_struct_type->get_name_by_position(i);
             if (!root_node->children_column_exists(table_column_name)) {
-                missing_fields.insert(i);
+                schema_missing_fields.insert(i);
                 continue;
             }
             const auto& file_column_name = root_node->children_file_column_name(table_column_name);
@@ -2321,27 +2324,47 @@ Status OrcReader::_fill_doris_data_column(const std::string& col_name,
                            << "], table_column: " << table_column_name
                            << ", file_column: " << file_column_name_lower;
             } else {
-                missing_fields.insert(i);
-                VLOG_DEBUG << "[OrcReader] Missing field: doris_field[" << i
+                projected_out_fields.insert(i);
+                VLOG_DEBUG << "[OrcReader] Projected-out field: doris_field[" << i
                            << "], table_column: " << table_column_name
                            << ", file_column: " << file_column_name_lower
-                           << " (not found in ORC file)";
+                           << " (not found in projected ORC type)";
             }
         }
 
-        for (int missing_field : missing_fields) {
-            ColumnPtr& doris_field = doris_struct.get_column_ptr(missing_field);
-            if (!doris_field->is_nullable()) {
-                return Status::InternalError(
-                        "Child field of '{}' is not nullable, but is missing in orc file",
-                        col_name);
-            }
+        // The selected ORC type can omit physical struct children that were not requested. They
+        // still exist in the file schema, so they must not be treated as Iceberg schema-evolution
+        // misses. Append placeholders only to keep every ColumnStruct child at the same size; the
+        // projected-out values are never exposed to the query.
+        for (int projected_out_field : projected_out_fields) {
+            ColumnPtr& doris_field = doris_struct.get_column_ptr(projected_out_field);
             auto mutable_field = IColumn::mutate(std::move(doris_field));
-            reinterpret_cast<ColumnNullable*>(mutable_field.get())
-                    ->insert_many_defaults(num_values);
+            mutable_field->insert_many_defaults(num_values);
             doris_field = std::move(mutable_field);
         }
 
+        for (int missing_field : schema_missing_fields) {
+            ColumnPtr& doris_field = doris_struct.get_column_ptr(missing_field);
+            const auto& doris_name = doris_struct_type->get_name_by_position(missing_field);
+            const auto& doris_type = doris_struct_type->get_element(missing_field);
+            const auto* iceberg_field = root_node->get_missing_column_field(doris_name);
+            if (iceberg_field != nullptr) {
+                RETURN_IF_ERROR(iceberg::append_initial_default(
+                        *iceberg_field, doris_type, num_values, &_nested_initial_default_values,
+                        &doris_field));
+            } else {
+                if (!doris_field->is_nullable()) {
+                    return Status::InternalError(
+                            "Child field of '{}' is not nullable, but is missing in orc file",
+                            col_name);
+                }
+                auto mutable_field = IColumn::mutate(std::move(doris_field));
+                reinterpret_cast<ColumnNullable*>(mutable_field.get())
+                        ->insert_many_defaults(num_values);
+                doris_field = std::move(mutable_field);
+            }
+        }
+
         for (auto read_field : read_fields) {
             orc::ColumnVectorBatch* orc_field = orc_struct->fields[read_field.second];
             const orc::Type* orc_type = orc_column_type->getSubtype(read_field.second);
@@ -2353,7 +2376,7 @@ Status OrcReader::_fill_doris_data_column(const std::string& col_name,
                     field_name, doris_field, doris_type,
                     root_node->get_children_node(
                             doris_struct_type->get_name_by_position(read_field.first)),
-                    orc_type, orc_field, num_values));
+                    orc_type, orc_field, num_values, orc_struct));
         }
         return Status::OK();
     }
@@ -2368,7 +2391,8 @@ template <bool is_filter>
 Status OrcReader::_orc_column_to_doris_column(
         const std::string& col_name, ColumnPtr& doris_column, const DataTypePtr& data_type,
         std::shared_ptr<TableSchemaChangeHelper::Node> root_node, const orc::Type* orc_column_type,
-        const orc::ColumnVectorBatch* cvb, size_t num_values) {
+        const orc::ColumnVectorBatch* cvb, size_t num_values,
+        const orc::ColumnVectorBatch* parent_cvb) {
     DataTypePtr resolved_type;
     ColumnPtr resolved_column;
     MutableColumnPtr data_column;
@@ -2422,8 +2446,18 @@ Status OrcReader::_orc_column_to_doris_column(
             fill_orc_null_map(nullable_column, cvb, num_values);
         } else {
             if (cvb->hasNulls) {
-                return Status::InternalError("Not nullable column {} has null values in orc file",
-                                             col_name);
+                if (parent_cvb == nullptr || !parent_cvb->hasNulls) {
+                    return Status::InternalError(
+                            "Not nullable column {} has null values in orc file", col_name);
+                }
+                DORIS_CHECK_GE(parent_cvb->capacity, num_values);
+                DORIS_CHECK_GE(cvb->capacity, num_values);
+                for (size_t i = 0; i < num_values; ++i) {
+                    if (!cvb->notNull[i] && parent_cvb->notNull[i]) {
+                        return Status::InternalError(
+                                "Not nullable column {} has null values in orc file", col_name);
+                    }
+                }
             }
             data_column = std::move(mutable_resolved_column);
         }

be/src/format/orc/vorc_reader.h

@@ -195,6 +195,11 @@ class OrcReader : public GenericReader {
     Status get_parsed_schema(std::vector<std::string>* col_names,
                              std::vector<DataTypePtr>* col_types) override;
 
+    const orc::Type* get_file_root_type() const {
+        DORIS_CHECK(_reader != nullptr);
+        return &_reader->getType();
+    }
+
     void set_position_delete_rowids(const std::vector<int64_t>* delete_rows) {
         _position_delete_ordered_rowids = delete_rows;
     }
@@ -369,7 +374,8 @@ class OrcReader : public GenericReader {
                                        const DataTypePtr& data_type,
                                        std::shared_ptr<TableSchemaChangeHelper::Node> root_node,
                                        const orc::Type* orc_column_type,
-                                       const orc::ColumnVectorBatch* cvb, size_t num_values);
+                                       const orc::ColumnVectorBatch* cvb, size_t num_values,
+                                       const orc::ColumnVectorBatch* parent_cvb = nullptr);
 
     template <PrimitiveType PType, typename OrcColumnType>
     Status _decode_flat_column(const std::string& col_name, const MutableColumnPtr& data_column,
@@ -772,6 +778,9 @@ class OrcReader : public GenericReader {
     // Through this node, you can find the file column based on the table column.
     std::shared_ptr<TableSchemaChangeHelper::Node> _table_info_node_ptr =
             TableSchemaChangeHelper::ConstNode::get_instance();
+    // Hold the resolved type with the one-row value so equivalent complex types reconstructed for
+    // later Blocks reuse the same Iceberg field-ID entry outside the batch conversion path.
+    std::unordered_map<int32_t, std::pair<DataTypePtr, ColumnPtr>> _nested_initial_default_values;
 
     std::set<uint64_t> _column_ids;
     std::set<uint64_t> _filter_column_ids;