[feature](iceberg) Support nested column schema change

be/test/format/table/iceberg/iceberg_reader_test.cpp

@@ -1284,13 +1284,18 @@ TEST_F(IcebergReaderTest, v1_materializes_missing_equality_delete_initial_defaul
         return field_ptr;
     };
 
+    static_assert(sizeof("0123456789abcdef0123456789abcdef") - 1 > StringView::kInlineSize);
+    const std::string binary_default = "0123456789abcdef0123456789abcdef";
+
     schema::external::TStructField root_field;
-    root_field.__set_fields({make_field("added_timestamp", 1, TPrimitiveType::DATETIMEV2,
-                                        "2024-01-01 00:00:00.123456", -1, 6, false),
-                             make_field("added_binary", 2, TPrimitiveType::VARBINARY,
-                                        "Ej5FZ+ibEtOkVkJmFBdAAA==", 16, -1, true),
-                             make_field("added_string_binary", 3, TPrimitiveType::STRING,
-                                        "AAEC/w==", -1, -1, true)});
+    root_field.__set_fields(
+            {make_field("added_timestamp", 1, TPrimitiveType::DATETIMEV2,
+                        "2024-01-01 00:00:00.123456", -1, 6, false),
+             make_field("added_binary", 2, TPrimitiveType::VARBINARY,
+                        "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=",
+                        static_cast<int32_t>(binary_default.size()), -1, true),
+             make_field("added_string_binary", 3, TPrimitiveType::STRING,
+                        "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=", -1, -1, true)});
     schema::external::TSchema current_schema;
     current_schema.__set_schema_id(-1);
     current_schema.__set_root_field(root_field);
@@ -1312,7 +1317,8 @@ TEST_F(IcebergReaderTest, v1_materializes_missing_equality_delete_initial_defaul
                                 &runtime_state, cache.get());
 
     const auto timestamp_type = make_nullable(std::make_shared<DataTypeDateTimeV2>(6));
-    const auto varbinary_type = make_nullable(std::make_shared<DataTypeVarbinary>(16));
+    const auto varbinary_type = make_nullable(
+            std::make_shared<DataTypeVarbinary>(static_cast<int32_t>(binary_default.size())));
     const auto string_type = make_nullable(std::make_shared<DataTypeString>());
     Block block;
     block.insert({timestamp_type->create_column(), timestamp_type, "added_timestamp"});
@@ -1335,10 +1341,8 @@ TEST_F(IcebergReaderTest, v1_materializes_missing_equality_delete_initial_defaul
     ASSERT_EQ(block.rows(), 3);
     EXPECT_EQ(timestamp_type->to_string(*block.get_by_position(0).column, 0),
               "2024-01-01 00:00:00.123456");
-    EXPECT_EQ(varbinary_type->to_string(*block.get_by_position(1).column, 0),
-              std::string("\x12\x3e\x45\x67\xe8\x9b\x12\xd3\xa4\x56\x42\x66\x14\x17\x40\x00", 16));
-    EXPECT_EQ(string_type->to_string(*block.get_by_position(2).column, 0),
-              std::string("\x00\x01\x02\xff", 4));
+    EXPECT_EQ(varbinary_type->to_string(*block.get_by_position(1).column, 0), binary_default);
+    EXPECT_EQ(string_type->to_string(*block.get_by_position(2).column, 0), binary_default);
     EXPECT_FALSE(is_column_const(*block.get_by_position(0).column));
     EXPECT_FALSE(is_column_const(*block.get_by_position(1).column));
     EXPECT_FALSE(is_column_const(*block.get_by_position(2).column));

be/test/format_v2/table/iceberg_reader_test.cpp

@@ -2449,8 +2449,8 @@ TEST(IcebergV2ReaderTest, IcebergEqualityDeleteMatchesVarbinaryInitialDefaultFor
     const auto file_path = (test_dir / "split.parquet").string();
     const auto delete_file_path = (test_dir / "equality-delete.parquet").string();
     write_single_int_parquet_file(file_path, "id", {1, 2, 3}, 0);
-    const std::string binary_default(
-            "\x12\x3e\x45\x67\xe8\x9b\x12\xd3\xa4\x56\x42\x66\x14\x17\x40\x00", 16);
+    static_assert(sizeof("0123456789abcdef0123456789abcdef") - 1 > StringView::kInlineSize);
+    const std::string binary_default = "0123456789abcdef0123456789abcdef";
     write_iceberg_binary_equality_delete_parquet_file(delete_file_path, 1, binary_default,
                                                       "added_binary");
 
@@ -2460,9 +2460,11 @@ TEST(IcebergV2ReaderTest, IcebergEqualityDeleteMatchesVarbinaryInitialDefaultFor
     scan_params.__set_current_schema_id(100);
     scan_params.__set_history_schema_info({external_schema(
             100, {external_schema_field("id", 0),
-                  external_schema_field("added_binary", 1, {}, "Ej5FZ+ibEtOkVkJmFBdAAA==",
-                                        external_primitive_type(TPrimitiveType::VARBINARY, 16),
-                                        true)})});
+                  external_schema_field(
+                          "added_binary", 1, {}, "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=",
+                          external_primitive_type(TPrimitiveType::VARBINARY,
+                                                  static_cast<int32_t>(binary_default.size())),
+                          true)})});
 
     RuntimeProfile profile("test_profile");
     RuntimeState state {TQueryOptions(), TQueryGlobals()};

fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java

@@ -142,6 +142,10 @@ public static Column generateBeforeValueColumn(Column column) {
     private boolean isKey;
     @SerializedName(value = "isAllowNull")
     private boolean isAllowNull;
+    // Runtime-only schema change intent; do not persist it as part of the table schema.
+    private transient boolean nullableSpecified;
+    // Runtime-only schema change intent; do not persist it as part of the table schema.
+    private transient boolean commentSpecified;
     @SerializedName(value = "isAutoInc")
     private boolean isAutoInc;
 
@@ -368,6 +372,8 @@ public Column(Column column) {
         this.isKey = column.isKey();
         this.isCompoundKey = column.isCompoundKey();
         this.isAllowNull = column.isAllowNull();
+        this.nullableSpecified = column.isNullableSpecified();
+        this.commentSpecified = column.isCommentSpecified();
         this.isAutoInc = column.isAutoInc();
         this.defaultValue = column.getDefaultValue();
         this.realDefaultValue = column.realDefaultValue;
@@ -583,6 +589,14 @@ public boolean isAllowNull() {
         return isAllowNull;
     }
 
+    public boolean isNullableSpecified() {
+        return nullableSpecified;
+    }
+
+    public boolean isCommentSpecified() {
+        return commentSpecified;
+    }
+
     public boolean isAutoInc() {
         return isAutoInc;
     }
@@ -595,6 +609,14 @@ public void setIsAllowNull(boolean isAllowNull) {
         this.isAllowNull = isAllowNull;
     }
 
+    public void setNullableSpecified(boolean nullableSpecified) {
+        this.nullableSpecified = nullableSpecified;
+    }
+
+    public void setCommentSpecified(boolean commentSpecified) {
+        this.commentSpecified = commentSpecified;
+    }
+
     public String getDefaultValue() {
         return this.defaultValue;
     }

fe/fe-catalog/src/main/java/org/apache/doris/catalog/ColumnType.java

@@ -322,11 +322,10 @@ private static void checkSupportSchemaChangeForComplexType(Type checkType, Type
                 existingNames.add(originalField.getName());
             }
 
-            // check new field name is not conflict with old field name
+            // check appended field names do not conflict with existing or earlier appended fields
             for (int i = originalFields.size(); i < otherStructType.getFields().size(); i++) {
-                // to check new field name is not conflict with old field name
                 String newFieldName = otherStructType.getFields().get(i).getName();
-                if (existingNames.contains(newFieldName)) {
+                if (!existingNames.add(newFieldName)) {
                     throw new DdlException("Added struct field '" + newFieldName + "' conflicts with existing field");
                 }
             }

fe/fe-catalog/src/main/java/org/apache/doris/catalog/info/ColumnPosition.java

@@ -18,6 +18,7 @@
 package org.apache.doris.catalog.info;
 
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.SqlUtils;
 
 import com.google.common.base.Strings;
 
@@ -57,7 +58,7 @@ public String toSql() {
         if (this == FIRST) {
             sb.append("FIRST");
         } else {
-            sb.append("AFTER `").append(lastCol).append("`");
+            sb.append("AFTER ").append(SqlUtils.getIdentSql(lastCol));
         }
         return sb.toString();
     }

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -415,20 +415,26 @@ private void processAlterTableForExternalTable(
                         table.getDbName(), table.getName(), renameTableOp.getNewTableName());
             } else if (alterOp instanceof AddColumnOp) {
                 AddColumnOp addColumn = (AddColumnOp) alterOp;
-                table.getCatalog().addColumn(table, addColumn.getColumn(), addColumn.getColPos());
+                table.getCatalog().addColumn(
+                        table, addColumn.getColumnPath(), addColumn.getColumn(), addColumn.getColPos());
             } else if (alterOp instanceof AddColumnsOp) {
                 AddColumnsOp addColumns = (AddColumnsOp) alterOp;
                 table.getCatalog().addColumns(table, addColumns.getColumns());
             } else if (alterOp instanceof DropColumnOp) {
                 DropColumnOp dropColumn = (DropColumnOp) alterOp;
-                table.getCatalog().dropColumn(table, dropColumn.getColName());
+                table.getCatalog().dropColumn(table, dropColumn.getColumnPath());
             } else if (alterOp instanceof RenameColumnOp) {
                 RenameColumnOp columnRename = (RenameColumnOp) alterOp;
                 table.getCatalog().renameColumn(
-                        table, columnRename.getColName(), columnRename.getNewColName());
+                        table, columnRename.getColumnPath(), columnRename.getNewColName());
             } else if (alterOp instanceof ModifyColumnOp) {
                 ModifyColumnOp modifyColumn = (ModifyColumnOp) alterOp;
-                table.getCatalog().modifyColumn(table, modifyColumn.getColumn(), modifyColumn.getColPos());
+                table.getCatalog().modifyColumn(
+                        table, modifyColumn.getColumnPath(), modifyColumn.getColumn(), modifyColumn.getColPos());
+            } else if (alterOp instanceof ModifyColumnCommentOp) {
+                ModifyColumnCommentOp modifyColumnComment = (ModifyColumnCommentOp) alterOp;
+                table.getCatalog().modifyColumnComment(
+                        table, modifyColumnComment.getColumnPath(), modifyColumnComment.getComment());
             } else if (alterOp instanceof ReorderColumnsOp) {
                 ReorderColumnsOp reorderColumns = (ReorderColumnsOp) alterOp;
                 table.getCatalog().reorderColumns(table, reorderColumns.getColumnsByPos());

fe/fe-core/src/main/java/org/apache/doris/analysis/ColumnPath.java

@@ -0,0 +1,92 @@
+// 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.
+
+package org.apache.doris.analysis;
+
+import org.apache.doris.common.util.SqlUtils;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Represents a column path used by schema change statements.
+ */
+public class ColumnPath {
+    private final ImmutableList<String> parts;
+
+    private ColumnPath(List<String> parts) {
+        Preconditions.checkArgument(parts != null && !parts.isEmpty(), "column path is empty");
+        for (String part : parts) {
+            Preconditions.checkArgument(part != null && !part.isEmpty(), "column path contains empty part");
+        }
+        this.parts = ImmutableList.copyOf(parts);
+    }
+
+    public static ColumnPath of(List<String> parts) {
+        return new ColumnPath(parts);
+    }
+
+    public static ColumnPath of(String name) {
+        return new ColumnPath(ImmutableList.of(name));
+    }
+
+    public static ColumnPath fromDotName(String name) {
+        return new ColumnPath(Arrays.asList(name.split("\\.")));
+    }
+
+    public List<String> getParts() {
+        return parts;
+    }
+
+    public boolean isNested() {
+        return parts.size() > 1;
+    }
+
+    public String getTopLevelName() {
+        return parts.get(0);
+    }
+
+    public String getLeafName() {
+        return parts.get(parts.size() - 1);
+    }
+
+    public ColumnPath getParentPath() {
+        Preconditions.checkState(isNested(), "top-level column path has no parent");
+        return new ColumnPath(parts.subList(0, parts.size() - 1));
+    }
+
+    public String getParentPathString() {
+        return getParentPath().getFullPath();
+    }
+
+    public String getFullPath() {
+        return String.join(".", parts);
+    }
+
+    public String toSql() {
+        return parts.stream().map(SqlUtils::getIdentSql).collect(Collectors.joining("."));
+    }
+
+    @Override
+    public String toString() {
+        return getFullPath();
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/common/FeNameFormat.java

@@ -107,11 +107,18 @@ public static void checkColumnName(String columnName) throws AnalysisException {
     }
 
     public static void checkColumnNameBypassHiddenColumn(String columnName) throws AnalysisException {
+        checkColumnNameBypassSystemColumnPrefix(columnName);
+        checkColumnNamePrefix(columnName, Column.SHADOW_NAME_PREFIX);
+    }
+
+    /**
+     * Check column name syntax without applying Doris top-level hidden/shadow column prefix rules.
+     */
+    public static void checkColumnNameBypassSystemColumnPrefix(String columnName) throws AnalysisException {
         if (Strings.isNullOrEmpty(columnName) || !columnName.matches(getColumnNameRegex())) {
             ErrorReport.reportAnalysisException(ErrorCode.ERR_WRONG_COLUMN_NAME,
                     columnName, getColumnNameRegex());
         }
-        checkColumnNamePrefix(columnName, Column.SHADOW_NAME_PREFIX);
     }
 
     private static void checkColumnNamePrefix(String columnName, String prefix) throws AnalysisException {

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

@@ -17,6 +17,7 @@
 
 package org.apache.doris.datasource;
 
+import org.apache.doris.analysis.ColumnPath;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.DatabaseIf;
 import org.apache.doris.catalog.TableIf;
@@ -249,6 +250,15 @@ default void addColumn(TableIf table, Column column, ColumnPosition columnPositi
         throw new UserException("Not support add column operation");
     }
 
+    default void addColumn(TableIf table, ColumnPath columnPath, Column column, ColumnPosition columnPosition)
+            throws UserException {
+        if (!columnPath.isNested()) {
+            addColumn(table, column, columnPosition);
+            return;
+        }
+        throw new UserException("Not support nested add column operation");
+    }
+
     default void addColumns(TableIf table, List<Column> columns) throws UserException {
         throw new UserException("Not support add columns operation");
     }
@@ -257,14 +267,47 @@ default void dropColumn(TableIf table, String name) throws UserException {
         throw new UserException("Not support drop column operation");
     }
 
+    default void dropColumn(TableIf table, ColumnPath columnPath) throws UserException {
+        if (!columnPath.isNested()) {
+            dropColumn(table, columnPath.getTopLevelName());
+            return;
+        }
+        throw new UserException("Not support nested drop column operation");
+    }
+
     default void renameColumn(TableIf table, String oldName, String newName) throws UserException {
         throw new UserException("Not support rename column operation");
     }
 
+    default void renameColumn(TableIf table, ColumnPath columnPath, String newName) throws UserException {
+        if (!columnPath.isNested()) {
+            renameColumn(table, columnPath.getTopLevelName(), newName);
+            return;
+        }
+        throw new UserException("Not support nested rename column operation");
+    }
+
     default void modifyColumn(TableIf table, Column column, ColumnPosition columnPosition) throws UserException {
         throw new UserException("Not support update column operation");
     }
 
+    default void modifyColumn(TableIf table, ColumnPath columnPath, Column column, ColumnPosition columnPosition)
+            throws UserException {
+        if (!columnPath.isNested()) {
+            modifyColumn(table, column, columnPosition);
+            return;
+        }
+        throw new UserException("Not support nested modify column operation");
+    }
+
+    default void modifyColumnComment(TableIf table, String name, String comment) throws UserException {
+        modifyColumnComment(table, ColumnPath.of(name), comment);
+    }
+
+    default void modifyColumnComment(TableIf table, ColumnPath columnPath, String comment) throws UserException {
+        throw new UserException("Not support modify column comment operation");
+    }
+
     default void reorderColumns(TableIf table, List<String> newOrder) throws UserException {
         throw new UserException("Not support reorder columns operation");
     }