apache · haruki-830 · Jul 16, 2026
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/starrocks.md b/docs/content.zh/docs/connectors/pipeline-connectors/starrocks.md
@@ -247,7 +247,7 @@ pipeline:
     自动设置分桶数量</a>。对于 StarRocks 2.5 之前的版本必须设置，否则无法自动创建表。
 
 * 对于表结构变更同步
-  * 支持创建/删除/清空表，增加/删除/重命名列，修改列类型
+  * 支持创建/删除/清空表，增加/删除/重命名列，修改列类型；StarRocks 3.1 及之后版本支持修改表注释
   * 新增列只能添加到最后一列
   * 如果使用 StarRocks 3.2 及之后版本，并且通过连接器来自动建表, 可以通过配置 `table.create.properties.fast_schema_evolution` 为 `true`
     来加速 StarRocks 执行变更。

diff --git a/docs/content/docs/connectors/pipeline-connectors/starrocks.md b/docs/content/docs/connectors/pipeline-connectors/starrocks.md
@@ -255,7 +255,7 @@ pipeline:
     otherwise you must set the option.
 
 * For schema change synchronization
-  * supports create/drop/truncate table, add/drop/rename columns and alter column types
+  * supports create/drop/truncate table, add/drop/rename columns, alter column types, and alter table comments for StarRocks 3.1 or later
   * the new column will always be added to the last position
   * if your StarRocks version is 3.2 or later, and using the connector to create table automatically,
     you can set `table.create.properties.fast_schema_evolution` to `true` to speed up the schema change.

diff --git a/...rc/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java b/...rc/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java
@@ -28,16 +28,32 @@
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /** An enriched {@code StarRocksCatalog} with more schema evolution abilities. */
 public class StarRocksEnrichedCatalog extends StarRocksCatalog {
+    private static final Logger LOG = LoggerFactory.getLogger(StarRocksEnrichedCatalog.class);
+    private static final Pattern VERSION_PATTERN = Pattern.compile("(\\d+)\\.(\\d+)(?:\\.(\\d+))?");
+
+    private final String jdbcUrl;
+    private final String username;
+    private final String password;
+    private Boolean supportsAlterTableComment;
+
     public StarRocksEnrichedCatalog(String jdbcUrl, String username, String password) {
         super(jdbcUrl, username, password);
+        this.jdbcUrl = jdbcUrl;
+        this.username = username;
+        this.password = password;
     }
 
-    private static final Logger LOG = LoggerFactory.getLogger(StarRocksEnrichedCatalog.class);
-
     public void truncateTable(String databaseName, String tableName)
             throws StarRocksCatalogException {
         checkTableArgument(databaseName, tableName);
@@ -137,6 +153,41 @@ public void alterColumnType(String databaseName, String tableName, StarRocksColu
         }
     }
 
+    public boolean supportsAlterTableComment() throws StarRocksCatalogException {
+        if (supportsAlterTableComment == null) {
+            supportsAlterTableComment =
+                    isAlterTableCommentSupportedVersion(fetchStarRocksVersion());
+        }
+        return supportsAlterTableComment;
+    }
+
+    public void alterTableComment(String databaseName, String tableName, String comment)
+            throws StarRocksCatalogException {
+        checkTableArgument(databaseName, tableName);
+        Preconditions.checkArgument(comment != null, "Table comment cannot be null.");
+        String alterSql = buildAlterTableCommentSql(databaseName, tableName, comment);
+        try {
+            long startTimeMillis = System.currentTimeMillis();
+            executeUpdateStatement(alterSql);
+            LOG.info(
+                    "Success to alter table {}.{} comment, duration: {}ms, sql: {}",
+                    databaseName,
+                    tableName,
+                    System.currentTimeMillis() - startTimeMillis,
+                    alterSql);
+        } catch (Exception e) {
+            LOG.error(
+                    "Failed to alter table {}.{} comment, sql: {}",
+                    databaseName,
+                    tableName,
+                    alterSql,
+                    e);
+            throw new StarRocksCatalogException(
+                    String.format("Failed to alter table %s.%s comment", databaseName, tableName),
+                    e);
+        }
+    }
+
     private String buildTruncateTableSql(String databaseName, String tableName) {
         return String.format("TRUNCATE TABLE `%s`.`%s`;", databaseName, tableName);
     }
@@ -158,6 +209,13 @@ private String buildAlterColumnTypeSql(
                 "ALTER TABLE `%s`.`%s` MODIFY COLUMN %s", databaseName, tableName, columnStmt);
     }
 
+    private String buildAlterTableCommentSql(
+            String databaseName, String tableName, String comment) {
+        return String.format(
+                "ALTER TABLE `%s`.`%s` COMMENT = \"%s\";",
+                databaseName, tableName, escapeSqlStringLiteral(comment));
+    }
+
     private void executeUpdateStatement(String sql) throws StarRocksCatalogException {
         try {
             Method m =
@@ -171,6 +229,52 @@ private void executeUpdateStatement(String sql) throws StarRocksCatalogException
         }
     }
 
+    private String fetchStarRocksVersion() throws StarRocksCatalogException {
+        try {
+            return executeScalar("SELECT current_version()");
+        } catch (Exception e) {
+            LOG.debug("Failed to get StarRocks version by current_version().", e);
+        }
+
+        try {
+            // version() returns the MySQL-compatible server version, not the StarRocks version.
+            return executeScalar("SELECT @@version_comment");
+        } catch (Exception e) {
+            throw new StarRocksCatalogException("Failed to get StarRocks version.", e);
+        }
+    }
+
+    private String executeScalar(String sql) throws SQLException {
+        try (Connection connection = DriverManager.getConnection(jdbcUrl, username, password);
+                Statement statement = connection.createStatement();
+                ResultSet resultSet = statement.executeQuery(sql)) {
+            if (resultSet.next()) {
+                return resultSet.getString(1);
+            }
+        }
+        throw new SQLException("No result returned for SQL: " + sql);
+    }
+
+    static boolean isAlterTableCommentSupportedVersion(String version) {
+        if (StringUtils.isNullOrWhitespaceOnly(version)) {
+            return false;
+        }
+        Matcher matcher = VERSION_PATTERN.matcher(version);
+        if (!matcher.find()) {
+            return false;
+        }
+        int major = Integer.parseInt(matcher.group(1));
+        int minor = Integer.parseInt(matcher.group(2));
+        return major > 3 || (major == 3 && minor >= 1);
+    }
+
+    static String escapeSqlStringLiteral(String str) {
+        return str.replace("\\", "\\\\")
+                .replace("\"", "\\\"")
+                .replace("\n", "\\n")
+                .replace("\r", "\\r");
+    }
+
     private void checkTableArgument(String databaseName, String tableName) {
         Preconditions.checkArgument(
                 !StringUtils.isNullOrWhitespaceOnly(databaseName),

diff --git a/...rc/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java b/...rc/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java
@@ -19,6 +19,7 @@
 
 import org.apache.flink.cdc.common.event.AddColumnEvent;
 import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.AlterTableCommentEvent;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
 import org.apache.flink.cdc.common.event.DropColumnEvent;
 import org.apache.flink.cdc.common.event.DropTableEvent;
@@ -29,6 +30,7 @@
 import org.apache.flink.cdc.common.event.TruncateTableEvent;
 import org.apache.flink.cdc.common.event.visitor.SchemaChangeEventVisitor;
 import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
+import org.apache.flink.cdc.common.exceptions.UnsupportedSchemaChangeEventException;
 import org.apache.flink.cdc.common.schema.Column;
 import org.apache.flink.cdc.common.sink.MetadataApplier;
 import org.apache.flink.cdc.common.types.DataType;
@@ -93,7 +95,8 @@ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
                 SchemaChangeEventType.RENAME_COLUMN,
                 SchemaChangeEventType.ALTER_COLUMN_TYPE,
                 SchemaChangeEventType.DROP_TABLE,
-                SchemaChangeEventType.TRUNCATE_TABLE);
+                SchemaChangeEventType.TRUNCATE_TABLE,
+                SchemaChangeEventType.ALTER_TABLE_COMMENT);
     }
 
     @Override
@@ -113,14 +116,7 @@ public void applySchemaChange(SchemaChangeEvent schemaChangeEvent)
                 this::applyDropTable,
                 this::applyRenameColumn,
                 this::applyTruncateTable,
-                alterTableCommentEvent -> {
-                    // TODO Currently, table comments cannot be modified.
-                    // See
-                    // https://docs.starrocks.io/docs/sql-reference/sql-statements/table_bucket_part_index/ALTER_TABLE/#alter-table-comment-from-v31
-                    LOG.warn(
-                            "AlterTableCommentEvent is not supported by StarRocks connector yet. Event: {}",
-                            alterTableCommentEvent);
-                });
+                this::applyAlterTableComment);
     }
 
     private void applyCreateTable(CreateTableEvent createTableEvent) throws SchemaEvolveException {
@@ -334,6 +330,22 @@ private void applyAlterColumnType(AlterColumnTypeEvent event) throws SchemaEvolv
         }
     }
 
+    private void applyAlterTableComment(AlterTableCommentEvent event) throws SchemaEvolveException {
+        try {
+            if (!catalog.supportsAlterTableComment()) {
+                throw new UnsupportedSchemaChangeEventException(
+                        event, "Alter table comment requires StarRocks 3.1 or later.");
+            }
+            TableId tableId = event.tableId();
+            catalog.alterTableComment(
+                    tableId.getSchemaName(), tableId.getTableName(), event.getComment());
+        } catch (UnsupportedSchemaChangeEventException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new SchemaEvolveException(event, "fail to apply alter table comment event", e);
+        }
+    }
+
     private void applyTruncateTable(TruncateTableEvent truncateTableEvent) {
         try {
             catalog.truncateTable(

diff --git a/...ks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/MockStarRocksCatalog.java b/...ks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/MockStarRocksCatalog.java
@@ -34,9 +34,12 @@ public class MockStarRocksCatalog extends StarRocksEnrichedCatalog {
     /** database name -> table name -> table. */
     private final Map<String, Map<String, StarRocksTable>> tables;
 
+    private boolean supportsAlterTableComment;
+
     public MockStarRocksCatalog() {
         super("jdbc:mysql://127.0.0.1:9030", "root", "");
         this.tables = new HashMap<>();
+        this.supportsAlterTableComment = true;
     }
 
     @Override
@@ -96,6 +99,33 @@ public void createTable(StarRocksTable table, boolean ignoreIfExists)
         }
     }
 
+    public void setSupportsAlterTableComment(boolean supportsAlterTableComment) {
+        this.supportsAlterTableComment = supportsAlterTableComment;
+    }
+
+    @Override
+    public boolean supportsAlterTableComment() {
+        return supportsAlterTableComment;
+    }
+
+    @Override
+    public void alterTableComment(String databaseName, String tableName, String comment)
+            throws StarRocksCatalogException {
+        Map<String, StarRocksTable> dbTables = tables.get(databaseName);
+        if (dbTables == null) {
+            throw new StarRocksCatalogException(
+                    String.format("database %s does not exist", databaseName));
+        }
+
+        StarRocksTable oldTable = dbTables.get(tableName);
+        if (oldTable == null) {
+            throw new StarRocksCatalogException(
+                    String.format("table %s.%s does not exist", databaseName, tableName));
+        }
+
+        dbTables.put(tableName, copyTableWithComment(oldTable, comment));
+    }
+
     @Override
     public void alterAddColumns(
             String databaseName,
@@ -195,4 +225,18 @@ public void alterDropColumns(
                         .build();
         dbTables.put(tableName, newTable);
     }
+
+    private StarRocksTable copyTableWithComment(StarRocksTable oldTable, String comment) {
+        return new StarRocksTable.Builder()
+                .setDatabaseName(oldTable.getDatabaseName())
+                .setTableName(oldTable.getTableName())
+                .setTableType(oldTable.getTableType())
+                .setColumns(oldTable.getColumns())
+                .setTableKeys(oldTable.getTableKeys().orElse(null))
+                .setDistributionKeys(oldTable.getDistributionKeys().orElse(null))
+                .setNumBuckets(oldTable.getNumBuckets().orElse(null))
+                .setComment(comment)
+                .setTableProperties(oldTable.getProperties())
+                .build();
+    }
 }
diff --git a/...t/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java b/...t/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java
@@ -22,6 +22,7 @@
 import org.apache.flink.cdc.common.data.binary.BinaryStringData;
 import org.apache.flink.cdc.common.event.AddColumnEvent;
 import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.AlterTableCommentEvent;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
 import org.apache.flink.cdc.common.event.DataChangeEvent;
 import org.apache.flink.cdc.common.event.DropColumnEvent;
@@ -380,6 +381,29 @@ void testStarRocksAlterColumnType() throws Exception {
         assertEqualsInOrder(expected, actual);
     }
 
+    @Test
+    void testStarRocksAlterTableComment() throws Exception {
+        TableId tableId =
+                TableId.tableId(
+                        StarRocksContainer.STARROCKS_DATABASE_NAME,
+                        StarRocksContainer.STARROCKS_TABLE_NAME);
+
+        Schema schema =
+                Schema.newBuilder()
+                        .column(new PhysicalColumn("id", DataTypes.INT().notNull(), null))
+                        .column(new PhysicalColumn("name", DataTypes.STRING(), null))
+                        .primaryKey("id")
+                        .comment("old table comment")
+                        .build();
+
+        runJobWithEvents(
+                Arrays.asList(
+                        new CreateTableEvent(tableId, schema),
+                        new AlterTableCommentEvent(tableId, "new table comment")));
+
+        Assertions.assertThat(inspectTableComment(tableId)).isEqualTo("new table comment");
+    }
+
     @Test
     void testStarRocksNarrowingAlterColumnType() throws Exception {
         Assertions.assertThatThrownBy(