diff --git a/integration-test/src/test/java/com/datastrato/gravitino/integration/test/spark/SparkIT.java b/integration-test/src/test/java/com/datastrato/gravitino/integration/test/spark/SparkIT.java
index 3eff71eae45..27f79040a8e 100644
--- a/integration-test/src/test/java/com/datastrato/gravitino/integration/test/spark/SparkIT.java
+++ b/integration-test/src/test/java/com/datastrato/gravitino/integration/test/spark/SparkIT.java
@@ -339,6 +339,29 @@ void testAlterTableRenameColumn() {
     checkTableColumns(tableName, renameColumns, getTableInfo(tableName));
   }
 
+  @Test
+  void testAlterTableUpdateColumnComment() {
+    String tableName = "test_update_column_comment";
+    dropTableIfExists(tableName);
+    List<SparkColumnInfo> simpleTableColumns = getSimpleTableColumn();
+    createSimpleTable(tableName);
+    checkTableColumns(tableName, simpleTableColumns, getTableInfo(tableName));
+
+    String oldColumnComment = "col1_comment";
+    String newColumnComment = "col1_new_comment";
+
+    sql(
+        String.format(
+            "ALTER TABLE %S ADD COLUMNS (col1 int comment '%S')", tableName, oldColumnComment));
+    sql(
+        String.format(
+            "ALTER TABLE %S CHANGE COLUMN col1 col1 int comment '%S'",
+            tableName, newColumnComment));
+    ArrayList<SparkColumnInfo> updateCommentColumns = new ArrayList<>(simpleTableColumns);
+    updateCommentColumns.add(SparkColumnInfo.of("col1", DataTypes.IntegerType, newColumnComment));
+    checkTableColumns(tableName, updateCommentColumns, getTableInfo(tableName));
+  }
+
   private void checkTableColumns(
       String tableName, List<SparkColumnInfo> columnInfos, SparkTableInfo tableInfo) {
     SparkTableInfoChecker.create()
diff --git a/spark-connector/src/main/java/com/datastrato/gravitino/spark/connector/catalog/GravitinoCatalog.java b/spark-connector/src/main/java/com/datastrato/gravitino/spark/connector/catalog/GravitinoCatalog.java
index 53b873f99a1..60453a7b5f7 100644
--- a/spark-connector/src/main/java/com/datastrato/gravitino/spark/connector/catalog/GravitinoCatalog.java
+++ b/spark-connector/src/main/java/com/datastrato/gravitino/spark/connector/catalog/GravitinoCatalog.java
@@ -382,6 +382,11 @@ static com.datastrato.gravitino.rel.TableChange transformTableChange(TableChange
       TableChange.RenameColumn renameColumn = (TableChange.RenameColumn) change;
       return com.datastrato.gravitino.rel.TableChange.renameColumn(
           renameColumn.fieldNames(), renameColumn.newName());
+    } else if (change instanceof TableChange.UpdateColumnComment) {
+      TableChange.UpdateColumnComment updateColumnComment =
+          (TableChange.UpdateColumnComment) change;
+      return com.datastrato.gravitino.rel.TableChange.updateColumnComment(
+          updateColumnComment.fieldNames(), updateColumnComment.newComment());
     } else {
       throw new UnsupportedOperationException(
           String.format("Unsupported table change %s", change.getClass().getName()));
diff --git a/spark-connector/src/test/java/com/datastrato/gravitino/spark/connector/catalog/TestTransformTableChange.java b/spark-connector/src/test/java/com/datastrato/gravitino/spark/connector/catalog/TestTransformTableChange.java
index c03558bd271..6041ad3248d 100644
--- a/spark-connector/src/test/java/com/datastrato/gravitino/spark/connector/catalog/TestTransformTableChange.java
+++ b/spark-connector/src/test/java/com/datastrato/gravitino/spark/connector/catalog/TestTransformTableChange.java
@@ -54,10 +54,29 @@ void testTransformRenameColumn() {
     com.datastrato.gravitino.rel.TableChange.RenameColumn gravitinoRenameColumn =
         (com.datastrato.gravitino.rel.TableChange.RenameColumn) gravitinoChange;
 
-    Assertions.assertEquals(oldFiledsName, gravitinoRenameColumn.getFieldName());
+    Assertions.assertArrayEquals(oldFiledsName, gravitinoRenameColumn.getFieldName());
     Assertions.assertEquals(newFiledName, gravitinoRenameColumn.getNewName());
   }
 
+  @Test
+  void testTransformUpdateColumnComment() {
+    String[] fieldNames = new String[] {"default_name"};
+    String newComment = "default_comment";
+
+    TableChange.UpdateColumnComment updateColumnComment =
+        (TableChange.UpdateColumnComment) TableChange.updateColumnComment(fieldNames, newComment);
+    com.datastrato.gravitino.rel.TableChange gravitinoChange =
+        GravitinoCatalog.transformTableChange(updateColumnComment);
+
+    Assertions.assertTrue(
+        gravitinoChange instanceof com.datastrato.gravitino.rel.TableChange.UpdateColumnComment);
+    com.datastrato.gravitino.rel.TableChange.UpdateColumnComment gravitinoUpdateColumnComment =
+        (com.datastrato.gravitino.rel.TableChange.UpdateColumnComment) gravitinoChange;
+
+    Assertions.assertArrayEquals(fieldNames, gravitinoUpdateColumnComment.getFieldName());
+    Assertions.assertEquals(newComment, gravitinoUpdateColumnComment.getNewComment());
+  }
+
   @Test
   void testTransformAddColumn() {
 
@@ -79,7 +98,8 @@ void testTransformAddColumn() {
     com.datastrato.gravitino.rel.TableChange.AddColumn gravitinoAddColumnFirst =
         (com.datastrato.gravitino.rel.TableChange.AddColumn) gravitinoChangeFirst;
 
-    Assertions.assertEquals(sparkAddColumnFirst.fieldNames(), gravitinoAddColumnFirst.fieldName());
+    Assertions.assertArrayEquals(
+        sparkAddColumnFirst.fieldNames(), gravitinoAddColumnFirst.fieldName());
     Assertions.assertTrue(
         "string".equalsIgnoreCase(gravitinoAddColumnFirst.getDataType().simpleString()));
     Assertions.assertTrue(
@@ -98,7 +118,8 @@ void testTransformAddColumn() {
     com.datastrato.gravitino.rel.TableChange.AddColumn gravitinoAddColumnAfter =
         (com.datastrato.gravitino.rel.TableChange.AddColumn) gravitinoChangeAfter;
 
-    Assertions.assertEquals(sparkAddColumnAfter.fieldNames(), gravitinoAddColumnAfter.fieldName());
+    Assertions.assertArrayEquals(
+        sparkAddColumnAfter.fieldNames(), gravitinoAddColumnAfter.fieldName());
     Assertions.assertTrue(
         "string".equalsIgnoreCase(gravitinoAddColumnAfter.getDataType().simpleString()));
     Assertions.assertTrue(
@@ -117,7 +138,7 @@ void testTransformAddColumn() {
     com.datastrato.gravitino.rel.TableChange.AddColumn gravitinoAddColumnDefault =
         (com.datastrato.gravitino.rel.TableChange.AddColumn) gravitinoChangeDefault;
 
-    Assertions.assertEquals(
+    Assertions.assertArrayEquals(
         sparkAddColumnDefault.fieldNames(), gravitinoAddColumnDefault.fieldName());
     Assertions.assertTrue(
         "string".equalsIgnoreCase(gravitinoAddColumnDefault.getDataType().simpleString()));
@@ -138,7 +159,7 @@ void testTransformDeleteColumn() {
     com.datastrato.gravitino.rel.TableChange.DeleteColumn gravitinoDeleteColumn =
         (com.datastrato.gravitino.rel.TableChange.DeleteColumn) gravitinoChange;
 
-    Assertions.assertEquals(sparkDeleteColumn.fieldNames(), gravitinoDeleteColumn.fieldName());
+    Assertions.assertArrayEquals(sparkDeleteColumn.fieldNames(), gravitinoDeleteColumn.fieldName());
     Assertions.assertEquals(sparkDeleteColumn.ifExists(), gravitinoDeleteColumn.getIfExists());
   }
 
@@ -155,7 +176,7 @@ void testTransformUpdateColumnType() {
     com.datastrato.gravitino.rel.TableChange.UpdateColumnType gravitinoUpdateColumnType =
         (com.datastrato.gravitino.rel.TableChange.UpdateColumnType) gravitinoChange;
 
-    Assertions.assertEquals(
+    Assertions.assertArrayEquals(
         sparkUpdateColumnType.fieldNames(), gravitinoUpdateColumnType.fieldName());
     Assertions.assertTrue(
         "string".equalsIgnoreCase(gravitinoUpdateColumnType.getNewDataType().simpleString()));