[flink] support flink sql alter table comment

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`.

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`.

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();

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

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')");

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";
 

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 ->