diff --git a/docs/content/flink/sql-alter.md b/docs/content/flink/sql-alter.md
index 0bdb84847cda..fe96ec413796 100644
--- a/docs/content/flink/sql-alter.md
+++ b/docs/content/flink/sql-alter.md
@@ -36,6 +36,32 @@ ALTER TABLE my_table SET (
 );
 ```
 
+## Removing Table Properties
+
+The following SQL removes `write-buffer-size` table property.
+
+```sql
+ALTER TABLE my_table RESET ('write-buffer-size');
+```
+
+##  Changing/Adding Table Comment
+
+The following SQL changes comment of table `my_table` to `table comment`.
+
+```sql
+ALTER TABLE my_table SET (
+    'comment' = 'table comment'
+    );
+```
+
+## Removing Table Comment
+
+The following SQL removes table comment.
+
+```sql
+ALTER TABLE my_table RESET ('comment');
+```
+
 ## Rename Table Name
 
 The following SQL rename the table name to new name.
@@ -48,14 +74,6 @@ ALTER TABLE my_table RENAME TO my_table_new;
 If you use object storage, such as S3 or OSS, please use this syntax carefully, because the renaming of object storage is not atomic, and only partial files may be moved in case of failure.
 {{< /hint >}}
 
-## Removing Table Properties
-
-The following SQL removes `write-buffer-size` table property.
-
-```sql
-ALTER TABLE my_table RESET ('write-buffer-size');
-```
-
 ## Adding New Columns
 
 The following SQL adds two columns `c1` and `c2` to table `my_table`.
diff --git a/docs/content/spark/sql-alter.md b/docs/content/spark/sql-alter.md
index d14f6bb3315c..b0d264651606 100644
--- a/docs/content/spark/sql-alter.md
+++ b/docs/content/spark/sql-alter.md
@@ -36,6 +36,32 @@ ALTER TABLE my_table SET TBLPROPERTIES (
 );
 ```
 
+## Removing Table Properties
+
+The following SQL removes `write-buffer-size` table property.
+
+```sql
+ALTER TABLE my_table UNSET TBLPROPERTIES ('write-buffer-size');
+```
+
+##  Changing/Adding Table Comment
+
+The following SQL changes comment of table `my_table` to `table comment`.
+
+```sql
+ALTER TABLE my_table SET TBLPROPERTIES (
+    'comment' = 'table comment'
+    );
+```
+
+## Removing Table Comment
+
+The following SQL removes table comment.
+
+```sql
+ALTER TABLE my_table UNSET TBLPROPERTIES ('comment');
+```
+
 ## Rename Table Name
 
 The following SQL rename the table name to new name.
@@ -58,14 +84,6 @@ ALTER TABLE catalog.database.test1 RENAME to catalog.database.test2;
 If you use object storage, such as S3 or OSS, please use this syntax carefully, because the renaming of object storage is not atomic, and only partial files may be moved in case of failure.
 {{< /hint >}}
 
-## Removing Table Properties
-
-The following SQL removes `write-buffer-size` table property.
-
-```sql
-ALTER TABLE my_table UNSET TBLPROPERTIES ('write-buffer-size');
-```
-
 ## Adding New Columns
 
 The following SQL adds two columns `c1` and `c2` to table `my_table`.
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 96e85d20a009..9057b1c3d47b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -45,6 +45,7 @@ public interface Catalog extends AutoCloseable {
 
     String SYSTEM_TABLE_SPLITTER = "$";
     String SYSTEM_DATABASE_NAME = "sys";
+    String COMMENT_PROP = "comment";
 
     /** Warehouse root path containing all database directories in this catalog. */
     String warehouse();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
index fc74b73a4920..98efd479f20d 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -455,11 +455,20 @@ private List<SchemaChange> toSchemaChange(
 
             SchemaManager.checkAlterTablePath(key);
 
-            schemaChanges.add(SchemaChange.setOption(key, value));
+            if (Catalog.COMMENT_PROP.equals(key)) {
+                schemaChanges.add(SchemaChange.updateComment(value));
+            } else {
+                schemaChanges.add(SchemaChange.setOption(key, value));
+            }
             return schemaChanges;
         } else if (change instanceof ResetOption) {
             ResetOption resetOption = (ResetOption) change;
-            schemaChanges.add(SchemaChange.removeOption(resetOption.getKey()));
+            String key = resetOption.getKey();
+            if (Catalog.COMMENT_PROP.equals(key)) {
+                schemaChanges.add(SchemaChange.updateComment(null));
+            } else {
+                schemaChanges.add(SchemaChange.removeOption(resetOption.getKey()));
+            }
             return schemaChanges;
         } else if (change instanceof TableChange.ModifyColumn) {
             // let non-physical column handle by option
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
index 9db3775b327f..fc5a3dbe0545 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SchemaChangeITCase.java
@@ -879,6 +879,26 @@ public void testSetAndResetImmutableOptions() throws Exception {
                 .hasMessage("Change 'sequence.field' is not supported yet.");
     }
 
+    @Test
+    public void testAlterTableComment() throws Exception {
+        sql("CREATE TABLE T (a STRING, b STRING, c STRING)");
+
+        // add table comment
+        sql("ALTER TABLE T SET ('comment'='t comment')");
+        String comment = table("T").getComment();
+        assertThat(comment).isEqualTo("t comment");
+
+        // update table comment
+        sql("ALTER TABLE T SET ('comment'='t comment v2')");
+        comment = table("T").getComment();
+        assertThat(comment).isEqualTo("t comment v2");
+
+        // remove table comment
+        sql("ALTER TABLE T RESET ('comment')");
+        comment = table("T").getComment();
+        assertThat(comment).isEmpty();
+    }
+
     @Test
     public void testAlterTableSchema() {
         sql("CREATE TABLE T (a STRING, b STRING COMMENT 'from column b')");
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index cb651f22da8e..c7c947ce5ec2 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -102,7 +102,6 @@ public class HiveCatalog extends AbstractCatalog {
     private static final Logger LOG = LoggerFactory.getLogger(HiveCatalog.class);
 
     // Reserved properties
-    public static final String COMMENT_PROP = "comment";
     public static final String TABLE_TYPE_PROP = "table_type";
     public static final String PAIMON_TABLE_TYPE_VALUE = "paimon";
 
diff --git a/paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/PaimonMetaHook.java b/paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/PaimonMetaHook.java
index d82b36d8762a..5cc826b554bf 100644
--- a/paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/PaimonMetaHook.java
+++ b/paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/PaimonMetaHook.java
@@ -50,6 +50,7 @@
 
 import static org.apache.hadoop.hive.metastore.Warehouse.getDnsPath;
 import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
+import static org.apache.paimon.catalog.Catalog.COMMENT_PROP;
 import static org.apache.paimon.hive.HiveTypeUtils.toPaimonType;
 
 /**
@@ -60,7 +61,6 @@ public class PaimonMetaHook implements HiveMetaHook {
 
     private static final Logger LOG = LoggerFactory.getLogger(PaimonMetaHook.class);
 
-    private static final String COMMENT = "comment";
     private final Configuration conf;
 
     // paimon table existed before create hive table
@@ -110,7 +110,7 @@ public void preCreateTable(Table table) throws MetaException {
         // create paimon table
         List<FieldSchema> cols = table.getSd().getCols();
         Schema.Builder schemaBuilder =
-                Schema.newBuilder().comment(table.getParameters().get(COMMENT));
+                Schema.newBuilder().comment(table.getParameters().get(COMMENT_PROP));
         cols.iterator()
                 .forEachRemaining(
                         fieldSchema ->