Added a fix for column rename in IcebergSchemaSync

xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergSchemaSync.java

@@ -85,12 +85,12 @@ private static Map<Integer, Supplier<UpdateSchema>> addUpdates(
       UpdateSchema updateSchema,
       String parentPath) {
     Map<Integer, Supplier<UpdateSchema>> updates = new HashMap<>();
-    Set<String> allColumnNames = new HashSet<>();
-    current.fields().stream().map(Types.NestedField::name).forEach(allColumnNames::add);
-    latest.fields().stream().map(Types.NestedField::name).forEach(allColumnNames::add);
-    for (String columnName : allColumnNames) {
-      Types.NestedField latestColumn = latest.field(columnName);
-      Types.NestedField currentColumn = current.field(columnName);
+    Set<Integer> allColumnFieldIds = new HashSet<>();
+    current.fields().stream().map(Types.NestedField::fieldId).forEach(allColumnFieldIds::add);
+    latest.fields().stream().map(Types.NestedField::fieldId).forEach(allColumnFieldIds::add);
+    for (int columnFieldId : allColumnFieldIds) {
+      Types.NestedField latestColumn = latest.field(columnFieldId);
+      Types.NestedField currentColumn = current.field(columnFieldId);
       if (currentColumn == null) {
         // add a new column
         if (latestColumn.isOptional()) {
@@ -110,7 +110,9 @@ private static Map<Integer, Supplier<UpdateSchema>> addUpdates(
         // drop the existing column, use fieldId 0 to perform deletes first
         updates.put(
             0,
-            () -> updateSchema.deleteColumn(constructFullyQualifiedName(columnName, parentPath)));
+            () ->
+                updateSchema.deleteColumn(
+                    constructFullyQualifiedName(currentColumn.name(), parentPath)));
       } else {
         updates.putAll(updateColumn(latestColumn, currentColumn, updateSchema, parentPath));
       }
@@ -134,6 +136,12 @@ private static Map<Integer, Supplier<UpdateSchema>> updateColumn(
                 updateSchema.updateColumn(
                     latestColumn.name(), latestColumn.type().asPrimitiveType()));
       }
+      // update the name of the column
+      if (!latestColumn.name().equals(currentColumn.name())) {
+        updates.put(
+            latestColumn.fieldId(),
+            () -> updateSchema.renameColumn(currentColumn.name(), latestColumn.name()));
+      }
       // update whether the column is required
       if (latestColumn.isOptional() != currentColumn.isOptional()) {
         if (latestColumn.isOptional()) {

xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSchemaSync.java

@@ -358,6 +358,33 @@ void testSyncWithProvidedIds() {
     }
   }
 
+  @Test
+  public void testUpdateColumnName() {
+    UpdateSchema mockUpdateSchema = Mockito.mock(UpdateSchema.class);
+    when(mockTransaction.updateSchema()).thenReturn(mockUpdateSchema);
+    schemaSync.sync(SCHEMA, updateColumnName(2), mockTransaction);
+
+    verify(mockUpdateSchema).renameColumn(SCHEMA.findColumnName(2), "updateColumnName");
+    verify(mockUpdateSchema).commit();
+  }
+
+  private Schema updateColumnName(int fieldId) {
+    List<Types.NestedField> fields = new ArrayList<>();
+    for (Types.NestedField existingField : SCHEMA.columns()) {
+      if (existingField.fieldId() == fieldId) {
+        fields.add(
+            Types.NestedField.of(
+                existingField.fieldId(),
+                existingField.isOptional(),
+                "updateColumnName",
+                existingField.type()));
+      } else {
+        fields.add(existingField);
+      }
+    }
+    return new Schema(fields);
+  }
+
   private Schema addColumnToDefault(Schema schema, Types.NestedField field, Integer parentId) {
     List<Types.NestedField> fields = new ArrayList<>();
     for (Types.NestedField existingField : schema.columns()) {