[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(
@@ -103,15 +110,20 @@ const format::ColumnDefinition* find_schema_child_by_path(
                 });
         return child_it == schema_column->children.end() ? nullptr : &*child_it;
     }
-    const auto child_it = std::ranges::find_if(schema_column->children, [&](const auto& child) {
-        if (to_lower(child.name) == to_lower(child_path)) {
-            return true;
-        }
+    // Iceberg can reuse a historical name for a newly added sibling. Current names therefore
+    // have precedence across the entire struct; an earlier alias must not steal that access path.
+    const auto exact_it = std::ranges::find_if(schema_column->children, [&](const auto& child) {
+        return to_lower(child.name) == to_lower(child_path);
+    });
+    if (exact_it != schema_column->children.end()) {
+        return &*exact_it;
+    }
+    const auto alias_it = std::ranges::find_if(schema_column->children, [&](const auto& child) {
         return std::ranges::any_of(child.name_mapping, [&](const std::string& alias) {
             return to_lower(alias) == to_lower(child_path);
         });
     });
-    return child_it == schema_column->children.end() ? nullptr : &*child_it;
+    return alias_it == schema_column->children.end() ? nullptr : &*alias_it;
 }
 
 int32_t schema_field_id(const format::ColumnDefinition* schema_column) {

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_position_delete_sys_table_reader.cpp

@@ -37,6 +37,7 @@
 #include "format/orc/vorc_reader.h"
 #include "format/parquet/schema_desc.h"
 #include "format/parquet/vparquet_reader.h"
+#include "format/table/iceberg_scan_semantics.h"
 #include "format/table/parquet_utils.h"
 #include "format/table/table_schema_change_helper.h"
 #include "runtime/runtime_state.h"
@@ -99,15 +100,7 @@ const ColumnInt64* get_int64_column(const Block& block, const std::string& name)
     return check_and_get_column<ColumnInt64>(block.get_by_position(pos).column.get());
 }
 
-const schema::external::TField* get_field_ptr(const schema::external::TFieldPtr& field_ptr) {
-    if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) {
-        return nullptr;
-    }
-    return field_ptr.field_ptr.get();
-}
-
-const schema::external::TField* find_current_schema_field(const TFileScanRangeParams* params,
-                                                          const std::string& name) {
+const schema::external::TSchema* find_current_schema(const TFileScanRangeParams* params) {
     if (params == nullptr || !params->__isset.history_schema_info ||
         params->history_schema_info.empty()) {
         return nullptr;
@@ -121,16 +114,7 @@ const schema::external::TField* find_current_schema_field(const TFileScanRangePa
             }
         }
     }
-    if (!schema->__isset.root_field || !schema->root_field.__isset.fields) {
-        return nullptr;
-    }
-    for (const auto& field_ptr : schema->root_field.fields) {
-        const auto* field = get_field_ptr(field_ptr);
-        if (field != nullptr && field->__isset.name && field->name == name) {
-            return field;
-        }
-    }
-    return nullptr;
+    return schema;
 }
 
 template <typename ReadColumns>
@@ -248,13 +232,24 @@ Status IcebergPositionDeleteSysTableReader::_init_position_delete_reader() {
                                              &_state->timezone_obj(), _io_ctx, _state, _meta_cache);
 
         const FieldDescriptor* schema = nullptr;
-        int row_index = -1;
+        std::shared_ptr<TableSchemaChangeHelper::Node> mapped_file_schema;
         if (row_requested) {
             RETURN_IF_ERROR(parquet_reader->get_file_metadata_schema(&schema));
             DORIS_CHECK(schema != nullptr);
-            row_index = schema->get_column_index(kRowColumn);
+            const auto* table_schema = find_current_schema(_range_params);
+            if (table_schema == nullptr || !table_schema->__isset.root_field) {
+                return Status::InternalError(
+                        "Iceberg position delete system table row schema is missing");
+            }
+            // Position-delete mapping mode is file-wide: file_path/pos IDs must prevent an
+            // ID-less physical row from being rebound by name in either reader generation.
+            RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::
+                                    by_parquet_field_id_with_name_mapping(
+                                            table_schema->root_field, *schema, mapped_file_schema,
+                                            supports_iceberg_scan_semantics_v1(_range_params)));
         }
-        const bool read_row = row_requested && row_index >= 0;
+        const bool read_row =
+                row_requested && mapped_file_schema->children_column_exists(kRowColumn);
         _init_read_columns(read_row);
         std::vector<std::string> read_column_names;
         read_column_names.reserve(_read_columns.size());
@@ -270,18 +265,10 @@ Status IcebergPositionDeleteSysTableReader::_init_position_delete_reader() {
         pctx.range = &_range;
         pctx.filter_groups = false;
         if (read_row) {
-            const auto* table_row_field = find_current_schema_field(_range_params, kRowColumn);
-            if (table_row_field == nullptr) {
-                return Status::InternalError(
-                        "Iceberg position delete system table row schema is missing");
-            }
-            const auto* file_row_field = schema->get_column(static_cast<size_t>(row_index));
-            std::shared_ptr<TableSchemaChangeHelper::Node> row_node;
-            RETURN_IF_ERROR(TableSchemaChangeHelper::BuildTableInfoUtil::
-                                    by_parquet_field_id_with_name_mapping(
-                                            *table_row_field, *file_row_field, row_node));
             auto root_node = create_position_delete_root_node(_read_columns);
-            root_node->add_children(kRowColumn, file_row_field->name, row_node);
+            root_node->add_children(kRowColumn,
+                                    mapped_file_schema->children_file_column_name(kRowColumn),
+                                    mapped_file_schema->get_children_node(kRowColumn));
             pctx.table_info_node = std::move(root_node);
         }
         RETURN_IF_ERROR(static_cast<GenericReader*>(parquet_reader.get())->init_reader(&pctx));
@@ -294,19 +281,25 @@ Status IcebergPositionDeleteSysTableReader::_init_position_delete_reader() {
                 OrcReader::create_unique(_profile, _state, *_range_params, _range, _batch_size,
                                          _state->timezone(), _io_ctx, _meta_cache);
 
-        const orc::Type* row_type = nullptr;
+        std::shared_ptr<TableSchemaChangeHelper::Node> mapped_file_schema;
         if (row_requested) {
             const orc::Type* root_type = nullptr;
             RETURN_IF_ERROR(orc_reader->get_file_type(&root_type));
             DORIS_CHECK(root_type != nullptr);
-            for (uint64_t i = 0; i < root_type->getSubtypeCount(); ++i) {
-                if (root_type->getFieldName(i) == kRowColumn) {
-                    row_type = root_type->getSubtype(i);
-                    break;
-                }
+            const auto* table_schema = find_current_schema(_range_params);
+            if (table_schema == nullptr || !table_schema->__isset.root_field) {
+                return Status::InternalError(
+                        "Iceberg position delete system table row schema is missing");
             }
+            // Resolve row against the complete delete-file type so top-level IDs keep ORC in ID
+            // projection throughout the nested row subtree.
+            RETURN_IF_ERROR(
+                    TableSchemaChangeHelper::BuildTableInfoUtil::by_orc_field_id_with_name_mapping(
+                            table_schema->root_field, root_type, kIcebergOrcAttribute,
+                            mapped_file_schema, supports_iceberg_scan_semantics_v1(_range_params)));
         }
-        const bool read_row = row_requested && row_type != nullptr;
+        const bool read_row =
+                row_requested && mapped_file_schema->children_column_exists(kRowColumn);
         _init_read_columns(read_row);
         std::vector<std::string> read_column_names;
         read_column_names.reserve(_read_columns.size());
@@ -321,17 +314,10 @@ Status IcebergPositionDeleteSysTableReader::_init_position_delete_reader() {
         octx.params = _range_params;
         octx.range = &_range;
         if (read_row) {
-            const auto* table_row_field = find_current_schema_field(_range_params, kRowColumn);
-            if (table_row_field == nullptr) {
-                return Status::InternalError(
-                        "Iceberg position delete system table row schema is missing");
-            }
-            std::shared_ptr<TableSchemaChangeHelper::Node> row_node;
-            RETURN_IF_ERROR(
-                    TableSchemaChangeHelper::BuildTableInfoUtil::by_orc_field_id_with_name_mapping(
-                            *table_row_field, row_type, kIcebergOrcAttribute, row_node));
             auto root_node = create_position_delete_root_node(_read_columns);
-            root_node->add_children(kRowColumn, kRowColumn, row_node);
+            root_node->add_children(kRowColumn,
+                                    mapped_file_schema->children_file_column_name(kRowColumn),
+                                    mapped_file_schema->get_children_node(kRowColumn));
             octx.table_info_node = std::move(root_node);
         }
         RETURN_IF_ERROR(static_cast<GenericReader*>(orc_reader.get())->init_reader(&octx));