[fix](iceberg) Preserve nested field case in created schemas

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java

@@ -531,10 +531,14 @@ public void renameColumn(String dbName, String tableName, String oldName, String
         public void modifyColumn(String dbName, String tableName, IcebergColumnChange column,
                 boolean commentSpecified, ConnectorColumnPosition position) {
             withTable(dbName, tableName, table -> {
-                Types.NestedField current = table.schema().findField(column.getName());
+                Schema schema = table.schema();
+                Types.NestedField current = IcebergNestedColumnEvolution.findTopLevelField(
+                        schema, column.getName());
                 if (current == null) {
                     throw new DorisConnectorException("Column " + column.getName() + " does not exist");
                 }
+                // Iceberg update paths are case-sensitive, so stage every change with the persisted spelling.
+                String currentName = current.name();
                 // Iceberg can widen required -> optional but never optional -> required (existing data may hold
                 // nulls), so a NOT NULL request on an already-nullable column fails loud — legacy parity
                 // (IcebergMetadataOps.validateForModifyColumn / validateForModifyComplexColumn).
@@ -555,24 +559,25 @@ public void modifyColumn(String dbName, String tableName, IcebergColumnChange co
                         throw new DorisConnectorException("Modify column type from complex to primitive is not"
                                 + " supported: " + column.getName());
                     }
-                    updateSchema.updateColumn(column.getName(), newType.asPrimitiveType(), targetComment);
+                    updateSchema.updateColumn(currentName, newType.asPrimitiveType(), targetComment);
                 } else {
                     // A complex (STRUCT/ARRAY/MAP) modify diffs the new type against the current one field-by-field
                     // (IcebergComplexTypeDiff); the top-level column doc is updated separately, as in legacy.
                     if (current.type().isPrimitiveType()) {
                         throw new DorisConnectorException("Modify column type from non-complex to complex is not"
                                 + " supported: " + column.getName());
                     }
-                    IcebergComplexTypeDiff.apply(updateSchema, column.getName(), current.type(), newType,
+                    IcebergComplexTypeDiff.apply(updateSchema, currentName, current.type(), newType,
                             column.getSourceType());
                     if (!Objects.equals(current.doc(), targetComment)) {
-                        updateSchema.updateColumnDoc(column.getName(), targetComment);
+                        updateSchema.updateColumnDoc(currentName, targetComment);
                     }
                 }
                 if (column.isNullable()) {
-                    updateSchema.makeColumnOptional(column.getName());
+                    updateSchema.makeColumnOptional(currentName);
                 }
-                applyPosition(updateSchema, position, column.getName());
+                IcebergNestedColumnEvolution.applyTopLevelPosition(
+                        updateSchema, position, currentName, schema, "modify");
                 updateSchema.commit();
                 return null;
             });

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergComplexTypeDiff.java

@@ -172,10 +172,11 @@ private static void applyStructChange(UpdateSchema updateSchema, String path,
             Types.NestedField oldField = oldFields.get(i);
             Types.NestedField newField = newFields.get(i);
             String fieldPath = path + "." + oldField.name();
-            existingNames.add(oldField.name());
+            existingNames.add(lowercaseName(oldField.name()));
 
-            // Legacy ColumnType rule: existing fields are matched by position and may not be renamed.
-            if (!oldField.name().equals(newField.name())) {
+            // Iceberg defines case-insensitive identity with ROOT-lowercase keys. Java equalsIgnoreCase is
+            // broader for some Unicode characters and could otherwise route an update to the wrong field.
+            if (!lowercaseName(oldField.name()).equals(lowercaseName(newField.name()))) {
                 throw new DorisConnectorException("Cannot rename struct field from '" + oldField.name()
                         + "' to '" + newField.name() + "'");
             }
@@ -216,7 +217,7 @@ private static void applyStructChange(UpdateSchema updateSchema, String path,
         // Append the new fields (legacy parity: must be nullable and not clash with an existing name).
         for (int i = oldFields.size(); i < newFields.size(); i++) {
             Types.NestedField newField = newFields.get(i);
-            if (existingNames.contains(newField.name())) {
+            if (!existingNames.add(lowercaseName(newField.name()))) {
                 throw new DorisConnectorException("Added struct field '" + newField.name()
                         + "' conflicts with existing field");
             }
@@ -227,6 +228,10 @@ private static void applyStructChange(UpdateSchema updateSchema, String path,
         }
     }
 
+    private static String lowercaseName(String name) {
+        return name.toLowerCase(Locale.ROOT);
+    }
+
     private static void applyListChange(UpdateSchema updateSchema, String path,
             Types.ListType oldList, Types.ListType newList, ConnectorType newConn) {
         String elementPath = path + "." + oldList.field(oldList.elementId()).name();

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java

@@ -1303,7 +1303,7 @@ public void modifyColumn(ConnectorSession session, ConnectorTableHandle handle,
             // generic "Unsupported type for Iceberg: SMALLINT" here. Restore the legacy parity message ("Cannot
             // change int to smallint in nested types") by validating the requested nested type against the
             // CURRENT type — legacy validated in Doris type space, where the narrow target still exists.
-            throw upgradeNestedModifyError(iceHandle, column, buildError);
+            throw upgradeNestedModifyError(iceHandle, ConnectorColumnPath.of(column.getName()), column, buildError);
         }
         // Carry the neutral source type so a complex-type diff can read each STRUCT field's commentSpecified.
         IcebergColumnChange change = new IcebergColumnChange(column.getName(), icebergType,
@@ -1328,15 +1328,15 @@ public void modifyColumn(ConnectorSession session, ConnectorTableHandle handle,
      * against the CURRENT column type. Best-effort: a scalar modify, a load failure, or no offending nested leaf
      * keeps the original build error — so no other modify path changes.
      */
-    private DorisConnectorException upgradeNestedModifyError(IcebergTableHandle handle, ConnectorColumn column,
-            DorisConnectorException buildError) {
+    private DorisConnectorException upgradeNestedModifyError(IcebergTableHandle handle, ConnectorColumnPath path,
+            ConnectorColumn column, DorisConnectorException buildError) {
         if (!isComplexType(column.getType())) {
             return buildError;
         }
         try {
             Types.NestedField current = executeAuthenticated(() ->
                     catalogOps.withTable(handle.getDbName(), handle.getTableName(),
-                            table -> table.schema().findField(column.getName())));
+                            table -> IcebergNestedColumnEvolution.findFieldForErrorUpgrade(table.schema(), path)));
             if (current != null && !current.type().isPrimitiveType()) {
                 IcebergComplexTypeDiff.validateNestedModifyRepresentable(current.type(), column.getType());
             }
@@ -1487,7 +1487,8 @@ public void modifyNestedColumn(ConnectorSession session, ConnectorTableHandle ha
         try {
             icebergType = IcebergSchemaBuilder.buildColumnType(column.getType());
         } catch (DorisConnectorException buildError) {
-            throw upgradeNestedModifyError(iceHandle, column, buildError);
+            // Preserve the complete target identity so error parity cannot bind a same-named top-level field.
+            throw upgradeNestedModifyError(iceHandle, path, column, buildError);
         }
         // Carry the neutral source type so the nested complex-type diff can read each STRUCT field's
         // commentSpecified (an omitted COMMENT on a sub-field must keep its current doc, not clear it).

fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergNestedColumnEvolution.java

@@ -492,6 +492,19 @@ private static ResolvedColumnPath resolveColumnPath(Schema schema, ConnectorColu
         return new ResolvedColumnPath(ConnectorColumnPath.of(canonicalParts), currentType, currentField);
     }
 
+    static NestedField findTopLevelField(Schema schema, String columnName) {
+        return schema.asStruct().caseInsensitiveField(columnName);
+    }
+
+    static NestedField findFieldForErrorUpgrade(Schema schema, ConnectorColumnPath columnPath) {
+        try {
+            return resolveColumnPath(schema, columnPath, "modify").getField();
+        } catch (DorisConnectorException ignored) {
+            // Error-message upgrading is best-effort and must not replace the original type-build failure.
+            return null;
+        }
+    }
+
     /**
      * Resolves {@code columnPath}'s parent (which must be a struct) and its leaf within that struct
      * (case-insensitive). Used by nested DROP / RENAME, which target an existing struct field.
@@ -584,6 +597,13 @@ private static void applyPosition(UpdateSchema updateSchema, ConnectorColumnPosi
         }
     }
 
+    static void applyTopLevelPosition(UpdateSchema updateSchema, ConnectorColumnPosition position,
+            String columnName, Schema schema, String operation) {
+        if (position != null) {
+            applyPosition(updateSchema, position, ConnectorColumnPath.of(columnName), schema, operation);
+        }
+    }
+
     private static String getPositionReferencePath(Schema schema, ConnectorColumnPath columnPath,
             ConnectorColumnPosition position, String operation) {
         if (position == null || position.isFirst()) {

fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java

@@ -367,6 +367,32 @@ private static ConnectorType structType(List<String> names, List<ConnectorType>
         return ConnectorType.structOf(names, types, nullable, comments);
     }
 
+    @Test
+    public void testModifyColumnsCanonicalizeMixedCaseRootAndPosition() {
+        createMixedCaseTable();
+        ops.modifyColumn("db1", "mixed", change("id", Types.LongType.get(), "identifier", true), true, null);
+
+        ConnectorType requestedType = structType(
+                Arrays.asList("metric"), Arrays.asList(ConnectorType.of("BIGINT")),
+                Arrays.asList(true), Arrays.asList((String) null));
+
+        ops.modifyColumn("db1", "mixed",
+                new IcebergColumnChange("info", IcebergSchemaBuilder.buildColumnType(requestedType),
+                        "updated", null, true, requestedType),
+                true, ConnectorColumnPosition.after("id"));
+
+        Schema schema = reload("mixed");
+        Assertions.assertEquals(Arrays.asList("Id", "Info", "Label"), schema.columns().stream()
+                .map(Types.NestedField::name).collect(Collectors.toList()));
+        Assertions.assertEquals(Type.TypeID.LONG, schema.findField("Id").type().typeId());
+        Assertions.assertEquals("identifier", schema.findField("Id").doc());
+        Types.NestedField info = schema.findField("Info");
+        Assertions.assertEquals("updated", info.doc());
+        Types.NestedField metric = info.type().asStructType().fields().get(0);
+        Assertions.assertEquals("Metric", metric.name());
+        Assertions.assertEquals(Type.TypeID.LONG, metric.type().typeId());
+    }
+
     @Test
     public void testModifyStructAddsNullableField() {
         createTable("s_add", new ConnectorColumn("st",
@@ -397,6 +423,124 @@ public void testModifyStructWidensFieldTypeAndComment() {
         Assertions.assertEquals("new", a.doc());
     }
 
+    @Test
+    public void testModifyStructMatchesExistingFieldCaseInsensitively() {
+        createTable("s_case", new ConnectorColumn("st",
+                structType(Arrays.asList("CaseSensitive"), Arrays.asList(ConnectorType.of("INT")),
+                        Arrays.asList(true), Arrays.asList((String) null)), "", true, null, false));
+
+        modifyComplex("s_case", "st",
+                structType(Arrays.asList("casesensitive"), Arrays.asList(ConnectorType.of("BIGINT")),
+                        Arrays.asList(true), Arrays.asList((String) null)), true);
+
+        Types.NestedField field = reload("s_case").findField("st").type().asStructType().fields().get(0);
+        Assertions.assertEquals("CaseSensitive", field.name());
+        Assertions.assertEquals(Type.TypeID.LONG, field.type().typeId());
+    }
+
+    @Test
+    public void testModifyStructCommentMatchesExistingFieldCaseInsensitively() {
+        createTable("s_case_doc", new ConnectorColumn("st",
+                structType(Arrays.asList("CaseSensitive"), Arrays.asList(ConnectorType.of("INT")),
+                        Arrays.asList(true), Arrays.asList("old")), "", true, null, false));
+
+        modifyComplex("s_case_doc", "st",
+                structType(Arrays.asList("casesensitive"), Arrays.asList(ConnectorType.of("INT")),
+                        Arrays.asList(true), Arrays.asList("new")), true);
+
+        Types.NestedField field = reload("s_case_doc").findField("st").type().asStructType().fields().get(0);
+        Assertions.assertEquals("CaseSensitive", field.name());
+        Assertions.assertEquals("new", field.doc());
+    }
+
+    @Test
+    public void testModifyStructNullabilityMatchesExistingFieldCaseInsensitively() {
+        createTable("s_case_null", new ConnectorColumn("st",
+                structType(Arrays.asList("CaseSensitive"), Arrays.asList(ConnectorType.of("INT")),
+                        Arrays.asList(false), Arrays.asList((String) null)), "", true, null, false));
+
+        modifyComplex("s_case_null", "st",
+                structType(Arrays.asList("casesensitive"), Arrays.asList(ConnectorType.of("INT")),
+                        Arrays.asList(true), Arrays.asList((String) null)), true);
+
+        Types.NestedField field = reload("s_case_null").findField("st").type().asStructType().fields().get(0);
+        Assertions.assertEquals("CaseSensitive", field.name());
+        Assertions.assertTrue(field.isOptional());
+    }
+
+    @Test
+    public void testModifyStructUnderArrayMatchesExistingFieldCaseInsensitively() {
+        ConnectorType oldStruct = structType(
+                Arrays.asList("CaseSensitive"), Arrays.asList(ConnectorType.of("INT")),
+                Arrays.asList(true), Arrays.asList((String) null));
+        createTable("a_case", new ConnectorColumn(
+                "arr", ConnectorType.arrayOf(oldStruct), "", true, null, false));
+        ConnectorType newStruct = structType(
+                Arrays.asList("casesensitive"), Arrays.asList(ConnectorType.of("BIGINT")),
+                Arrays.asList(true), Arrays.asList((String) null));
+
+        modifyComplex("a_case", "arr", ConnectorType.arrayOf(newStruct), true);
+
+        Types.NestedField field = reload("a_case").findField("arr").type().asListType()
+                .elementType().asStructType().fields().get(0);
+        Assertions.assertEquals("CaseSensitive", field.name());
+        Assertions.assertEquals(Type.TypeID.LONG, field.type().typeId());
+    }
+
+    @Test
+    public void testModifyStructUnderMapMatchesExistingFieldCaseInsensitively() {
+        ConnectorType oldStruct = structType(
+                Arrays.asList("CaseSensitive"), Arrays.asList(ConnectorType.of("INT")),
+                Arrays.asList(true), Arrays.asList((String) null));
+        createTable("m_case", new ConnectorColumn("m",
+                ConnectorType.mapOf(ConnectorType.of("STRING"), oldStruct), "", true, null, false));
+        ConnectorType newStruct = structType(
+                Arrays.asList("casesensitive"), Arrays.asList(ConnectorType.of("BIGINT")),
+                Arrays.asList(true), Arrays.asList((String) null));
+
+        modifyComplex("m_case", "m",
+                ConnectorType.mapOf(ConnectorType.of("STRING"), newStruct), true);
+
+        Types.NestedField field = reload("m_case").findField("m").type().asMapType()
+                .valueType().asStructType().fields().get(0);
+        Assertions.assertEquals("CaseSensitive", field.name());
+        Assertions.assertEquals(Type.TypeID.LONG, field.type().typeId());
+    }
+
+    @Test
+    public void testModifyStructRejectsCaseInsensitiveAppendedFieldCollision() {
+        createTable("s_case_collision", new ConnectorColumn("st",
+                structType(Arrays.asList("CaseSensitive"), Arrays.asList(ConnectorType.of("INT")),
+                        Arrays.asList(true), Arrays.asList((String) null)), "", true, null, false));
+
+        DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class,
+                () -> modifyComplex("s_case_collision", "st",
+                        structType(Arrays.asList("casesensitive", "CASESENSITIVE"),
+                                Arrays.asList(ConnectorType.of("INT"), ConnectorType.of("STRING")),
+                                Arrays.asList(true, true), Arrays.asList(null, null)), true));
+
+        Assertions.assertTrue(ex.getMessage().contains("conflicts with existing field"), ex.getMessage());
+    }
+
+    @Test
+    public void testModifyStructUsesIcebergLowercaseIdentity() {
+        createTable("s_unicode_identity", new ConnectorColumn("st",
+                structType(Arrays.asList("Σ", "ς"),
+                        Arrays.asList(ConnectorType.of("INT"), ConnectorType.of("INT")),
+                        Arrays.asList(true, true), Arrays.asList(null, null)), "", true, null, false));
+
+        DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class,
+                () -> modifyComplex("s_unicode_identity", "st",
+                        structType(Arrays.asList("ς", "Σ"),
+                                Arrays.asList(ConnectorType.of("BIGINT"), ConnectorType.of("INT")),
+                                Arrays.asList(true, true), Arrays.asList(null, null)), true));
+
+        Assertions.assertTrue(ex.getMessage().contains("Cannot rename struct field"), ex.getMessage());
+        Types.StructType fields = reload("s_unicode_identity").findField("st").type().asStructType();
+        Assertions.assertEquals(Type.TypeID.INTEGER, fields.field("Σ").type().typeId());
+        Assertions.assertEquals(Type.TypeID.INTEGER, fields.field("ς").type().typeId());
+    }
+
     @Test
     public void testModifyStructFieldWidensNotNullToNullable() {
         createTable("s_null", new ConnectorColumn("st",