diff --git a/xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergSchemaSync.java b/xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergSchemaSync.java
index 800938cb4..aef645be1 100644
--- a/xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergSchemaSync.java
+++ b/xtable-core/src/main/java/org/apache/xtable/iceberg/IcebergSchemaSync.java
@@ -74,12 +74,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()) {
@@ -99,7 +99,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));
       }
@@ -123,6 +125,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()) {
diff --git a/xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSchemaSync.java b/xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSchemaSync.java
index 98254591e..c240dc5f8 100644
--- a/xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSchemaSync.java
+++ b/xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSchemaSync.java
@@ -335,6 +335,33 @@ void testAddMapFieldComment() {
     verify(mockUpdateSchema).commit();
   }
 
+  @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()) {