[fix](iceberg) Honor partial name mappings for legacy files

be/src/exec/scan/access_path_parser.cpp

@@ -87,6 +87,13 @@ void inherit_schema_metadata(format::ColumnDefinition* column,
         return;
     }
     column->name_mapping = schema_column->name_mapping;
+    // 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;
 }
 
 const format::ColumnDefinition* find_schema_child_by_path(

be/src/format/orc/vorc_reader.cpp

@@ -34,6 +34,7 @@
 #include "exprs/vexpr.h"
 #include "exprs/vslot_ref.h"
 #include "exprs/vtopn_pred.h"
+#include "util/url_coding.h"
 
 // IWYU pragma: no_include <bits/chrono.h>
 #include <chrono> // IWYU pragma: keep
@@ -105,6 +106,35 @@
 #include "util/unaligned.h"
 
 namespace doris {
+static Status build_orc_initial_default_column(
+        const std::optional<TableSchemaChangeHelper::InitialDefaultValue>& metadata,
+        const DataTypePtr& type, size_t rows, ColumnPtr* column) {
+    DORIS_CHECK(column != nullptr);
+    if (!metadata.has_value()) {
+        *column = nullptr;
+        return Status::OK();
+    }
+    const auto nested_type = remove_nullable(type);
+    Field value;
+    if (metadata->is_base64 || nested_type->get_primitive_type() == TYPE_VARBINARY) {
+        std::string decoded;
+        if (!base64_decode(metadata->value, &decoded)) {
+            return Status::InvalidArgument("Invalid Base64 Iceberg nested initial default");
+        }
+        value = nested_type->get_primitive_type() == TYPE_VARBINARY
+                        ? Field::create_field<TYPE_VARBINARY>(StringView(decoded))
+                        : Field::create_field<TYPE_STRING>(decoded);
+        // StringView borrows decoded for payloads longer than 12 bytes. Build the owning column
+        // before the decode buffer leaves scope (UUID/FIXED defaults routinely exceed that size).
+        *column = type->create_column_const(rows, value)->convert_to_full_column_if_const();
+        return Status::OK();
+    } else {
+        RETURN_IF_ERROR(nested_type->get_serde()->from_fe_string(metadata->value, value));
+    }
+    *column = type->create_column_const(rows, value)->convert_to_full_column_if_const();
+    return Status::OK();
+}
+
 class RuntimeState;
 namespace io {
 struct IOContext;
@@ -2145,15 +2175,28 @@ Status OrcReader::_fill_doris_data_column(const std::string& col_name,
 
         for (int missing_field : missing_fields) {
             ColumnPtr& doris_field = doris_struct.get_column_ptr(missing_field);
-            if (!doris_field->is_nullable()) {
+            const auto& table_column_name = doris_struct_type->get_name_by_position(missing_field);
+            const auto& doris_type = doris_struct_type->get_element(missing_field);
+            ColumnPtr initial_default;
+            RETURN_IF_ERROR(build_orc_initial_default_column(
+                    root_node->children_initial_default_value(table_column_name), doris_type,
+                    num_values, &initial_default));
+            if (initial_default.get() != nullptr) {
+                // ORC projection may synthesize a missing nested field, but its Iceberg initial
+                // default remains the logical value for every row in the older file.
+                auto mutable_field = IColumn::mutate(std::move(doris_field));
+                mutable_field->insert_range_from(*initial_default, 0, num_values);
+                doris_field = std::move(mutable_field);
+            } else if (!doris_field->is_nullable()) {
                 return Status::InternalError(
                         "Child field of '{}' is not nullable, but is missing in orc file",
                         col_name);
+            } else {
+                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);
             }
-            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) {

be/src/format/parquet/vparquet_column_reader.cpp

@@ -40,8 +40,38 @@
 #include "format/parquet/vparquet_column_chunk_reader.h"
 #include "io/fs/tracing_file_reader.h"
 #include "runtime/runtime_profile.h"
+#include "util/url_coding.h"
 
 namespace doris {
+static Status build_initial_default_column(
+        const std::optional<TableSchemaChangeHelper::InitialDefaultValue>& metadata,
+        const DataTypePtr& type, size_t rows, ColumnPtr* column) {
+    DORIS_CHECK(column != nullptr);
+    if (!metadata.has_value()) {
+        *column = nullptr;
+        return Status::OK();
+    }
+    const auto nested_type = remove_nullable(type);
+    Field value;
+    if (metadata->is_base64 || nested_type->get_primitive_type() == TYPE_VARBINARY) {
+        std::string decoded;
+        if (!base64_decode(metadata->value, &decoded)) {
+            return Status::InvalidArgument("Invalid Base64 Iceberg nested initial default");
+        }
+        value = nested_type->get_primitive_type() == TYPE_VARBINARY
+                        ? Field::create_field<TYPE_VARBINARY>(StringView(decoded))
+                        : Field::create_field<TYPE_STRING>(decoded);
+        // StringView borrows decoded for payloads longer than 12 bytes. Build the owning column
+        // before the decode buffer leaves scope (UUID/FIXED defaults routinely exceed that size).
+        *column = type->create_column_const(rows, value)->convert_to_full_column_if_const();
+        return Status::OK();
+    } else {
+        RETURN_IF_ERROR(nested_type->get_serde()->from_fe_string(metadata->value, value));
+    }
+    *column = type->create_column_const(rows, value)->convert_to_full_column_if_const();
+    return Status::OK();
+}
+
 static void fill_struct_null_map(FieldSchema* field, NullMap& null_map,
                                  const std::vector<level_t>& rep_levels,
                                  const std::vector<level_t>& def_levels) {
@@ -990,11 +1020,21 @@ Status StructColumnReader::read_column_data(
     for (auto idx : missing_column_idxs) {
         auto& doris_field = doris_struct.get_column_ptr(idx);
         auto& doris_type = doris_struct_type->get_element(idx);
-        DCHECK(doris_type->is_nullable());
-        doris_field = IColumn::mutate(std::move(doris_field));
-        auto mutable_column = doris_field->assert_mutable();
-        auto* nullable_column = static_cast<ColumnNullable*>(mutable_column.get());
-        nullable_column->insert_many_defaults(missing_column_sz);
+        auto mutable_field = IColumn::mutate(std::move(doris_field));
+        ColumnPtr initial_default;
+        RETURN_IF_ERROR(build_initial_default_column(
+                root_node->children_initial_default_value(doris_struct_type->get_element_name(idx)),
+                doris_type, missing_column_sz, &initial_default));
+        if (initial_default.get() != nullptr) {
+            // Iceberg initial defaults are logical row values, including for nested fields absent
+            // from the physical file; append them instead of the type's generic NULL/default.
+            mutable_field->insert_range_from(*initial_default, 0, missing_column_sz);
+        } else {
+            DCHECK(doris_type->is_nullable());
+            static_cast<ColumnNullable*>(mutable_field.get())
+                    ->insert_many_defaults(missing_column_sz);
+        }
+        doris_field = std::move(mutable_field);
     }
 
     if (null_map_ptr != nullptr) {

be/src/format/table/iceberg_reader_mixin.h

@@ -157,6 +157,30 @@ class IcebergReaderMixin : public BaseReader, public TableSchemaChangeHelper {
 protected:
     // ---- Hook implementations ----
 
+    Status on_fill_missing_columns(Block* block, size_t rows,
+                                   const std::vector<std::string>& cols) override {
+        std::vector<std::string> generic_columns;
+        for (const auto& name : cols) {
+            const auto* field = _find_current_schema_field(name);
+            if (field == nullptr || !field->__isset.initial_default_value) {
+                generic_columns.push_back(name);
+                continue;
+            }
+            if (!this->col_name_to_block_idx_ref()->contains(name)) {
+                return Status::InternalError("Missing column: {} not found in block {}", name,
+                                             block->dump_structure());
+            }
+            const auto position = this->col_name_to_block_idx_ref()->at(name);
+            ColumnPtr value;
+            RETURN_IF_ERROR(_build_owned_initial_default_column(
+                    *field, block->get_by_position(position).type, rows, &value));
+            // Iceberg metadata is authoritative for pre-add rows; a generic FE expression may be
+            // the Base64 carrier rather than the logical BINARY/FIXED/UUID value.
+            block->get_by_position(position).column = value->convert_to_full_column_if_const();
+        }
+        return BaseReader::on_fill_missing_columns(block, rows, generic_columns);
+    }
+
     // Called before reading a block: expand block for equality delete columns + detect row_id
     Status on_before_read_block(Block* block) override {
         RETURN_IF_ERROR(_expand_block_if_need(block));
@@ -270,6 +294,11 @@ class IcebergReaderMixin : public BaseReader, public TableSchemaChangeHelper {
 
     Status _expand_block_if_need(Block* block);
     Status _shrink_block_if_need(Block* block);
+    const schema::external::TStructField* _current_schema_root() const;
+    const schema::external::TField* _find_current_schema_field(const std::string& name) const;
+    Status _build_owned_initial_default_column(const schema::external::TField& field,
+                                               const DataTypePtr& type, size_t rows,
+                                               ColumnPtr* column) const;
     Status _register_missing_equality_delete_column(int32_t field_id, const std::string& name,
                                                     const DataTypePtr& delete_key_type);
     Status _materialize_missing_equality_delete_column(Block* block, const std::string& name,
@@ -664,6 +693,76 @@ Status IcebergReaderMixin<BaseReader>::_expand_block_if_need(Block* block) {
     return Status::OK();
 }
 
+template <typename BaseReader>
+const schema::external::TStructField* IcebergReaderMixin<BaseReader>::_current_schema_root() const {
+    const auto& scan_params = this->get_scan_params();
+    if (!scan_params.__isset.history_schema_info || scan_params.history_schema_info.empty()) {
+        return nullptr;
+    }
+    const schema::external::TSchema* current_schema = &scan_params.history_schema_info.front();
+    if (scan_params.__isset.current_schema_id) {
+        const auto schema_it = std::ranges::find_if(
+                scan_params.history_schema_info, [&](const schema::external::TSchema& schema) {
+                    return schema.__isset.schema_id &&
+                           schema.schema_id == scan_params.current_schema_id;
+                });
+        if (schema_it == scan_params.history_schema_info.end()) {
+            return nullptr;
+        }
+        current_schema = &*schema_it;
+    }
+    return current_schema->__isset.root_field ? &current_schema->root_field : nullptr;
+}
+
+template <typename BaseReader>
+const schema::external::TField* IcebergReaderMixin<BaseReader>::_find_current_schema_field(
+        const std::string& name) const {
+    const auto* root = _current_schema_root();
+    if (root == nullptr) {
+        return nullptr;
+    }
+    const auto field = std::ranges::find_if(root->fields, [&](const auto& field_ptr) {
+        return field_ptr.__isset.field_ptr && field_ptr.field_ptr != nullptr &&
+               field_ptr.field_ptr->__isset.name && field_ptr.field_ptr->name == name;
+    });
+    return field == root->fields.end() ? nullptr : field->field_ptr.get();
+}
+
+template <typename BaseReader>
+Status IcebergReaderMixin<BaseReader>::_build_owned_initial_default_column(
+        const schema::external::TField& field, const DataTypePtr& type, size_t rows,
+        ColumnPtr* column) const {
+    DORIS_CHECK(type != nullptr);
+    DORIS_CHECK(column != nullptr);
+    DORIS_CHECK(field.__isset.initial_default_value);
+    const auto nested_type = remove_nullable(type);
+    const bool is_base64 =
+            (field.__isset.initial_default_value_is_base64 &&
+             field.initial_default_value_is_base64) ||
+            (field.__isset.type && thrift_to_type(field.type.type) == TYPE_VARBINARY);
+    Field value;
+    if (is_base64) {
+        std::string decoded;
+        if (!base64_decode(field.initial_default_value, &decoded)) {
+            return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field {}",
+                                           field.name);
+        }
+        if (nested_type->get_primitive_type() == TYPE_VARBINARY) {
+            value = Field::create_field<TYPE_VARBINARY>(StringView(decoded));
+        } else {
+            DORIS_CHECK(is_string_type(nested_type->get_primitive_type()));
+            value = Field::create_field<TYPE_STRING>(decoded);
+        }
+        // Variable-width Fields borrow decoded. Materialize before it leaves scope so every V1
+        // missing-column and equality-delete boundary keeps UUID/FIXED payloads alive.
+        *column = type->create_column_const(rows, value);
+        return Status::OK();
+    }
+    RETURN_IF_ERROR(nested_type->get_serde()->from_fe_string(field.initial_default_value, value));
+    *column = type->create_column_const(rows, value);
+    return Status::OK();
+}
+
 template <typename BaseReader>
 Status IcebergReaderMixin<BaseReader>::_register_missing_equality_delete_column(
         int32_t field_id, const std::string& name, const DataTypePtr& delete_key_type) {
@@ -703,33 +802,15 @@ Status IcebergReaderMixin<BaseReader>::_register_missing_equality_delete_column(
                 "Missing Iceberg schema metadata for equality-delete field id {}", field_id);
     }
 
-    Field value;
+    ColumnPtr value_column;
     if (table_field->__isset.initial_default_value) {
-        const auto nested_type = remove_nullable(delete_key_type);
-        const bool default_is_base64 = (table_field->__isset.initial_default_value_is_base64 &&
-                                        table_field->initial_default_value_is_base64) ||
-                                       (table_field->__isset.type &&
-                                        thrift_to_type(table_field->type.type) == TYPE_VARBINARY);
-        if (default_is_base64) {
-            std::string decoded_default;
-            if (!base64_decode(table_field->initial_default_value, &decoded_default)) {
-                return Status::InvalidArgument(
-                        "Invalid Base64 Iceberg initial default for field {}", table_field->name);
-            }
-            if (nested_type->get_primitive_type() == TYPE_VARBINARY) {
-                value = Field::create_field<TYPE_VARBINARY>(StringView(decoded_default));
-            } else {
-                DORIS_CHECK(is_string_type(nested_type->get_primitive_type()));
-                value = Field::create_field<TYPE_STRING>(decoded_default);
-            }
-        } else {
-            RETURN_IF_ERROR(nested_type->get_serde()->from_fe_string(
-                    table_field->initial_default_value, value));
-        }
+        RETURN_IF_ERROR(_build_owned_initial_default_column(*table_field, delete_key_type, 1,
+                                                            &value_column));
+    } else {
+        value_column = delete_key_type->create_column_const(1, Field());
     }
-    const bool inserted = _missing_equality_delete_values
-                                  .emplace(name, delete_key_type->create_column_const(1, value))
-                                  .second;
+    const bool inserted =
+            _missing_equality_delete_values.emplace(name, std::move(value_column)).second;
     DORIS_CHECK(inserted);
     this->register_synthesized_column_handler(
             name, [this, name](Block* block, size_t rows) -> Status {