[FLINK-38829][pipline-connector/starrocks] Support rename column DDL for starrocks sink connector (#4197)

zml1206 · web-flow · commit 2293651e992f · 2025-12-23T19:42:25.000+08:00
diff --git a/docs/content.zh/docs/get-started/quickstart/mysql-to-starrocks.md b/docs/content.zh/docs/get-started/quickstart/mysql-to-starrocks.md
@@ -67,7 +67,7 @@ under the License.
    version: '2.1'
    services:
      StarRocks:
-       image: starrocks/allin1-ubuntu:3.2.6
+       image: starrocks/allin1-ubuntu:3.5.10
        ports:
          - "8080:8080"
          - "9030:9030"
diff --git a/docs/content/docs/get-started/quickstart/mysql-to-starrocks.md b/docs/content/docs/get-started/quickstart/mysql-to-starrocks.md
@@ -70,7 +70,7 @@ Create a `docker-compose.yml` file using the content provided below:
    version: '2.1'
    services:
       StarRocks:
-         image: starrocks/allin1-ubuntu:3.2.6
+         image: starrocks/allin1-ubuntu:3.5.10
          ports:
             - "8080:8080"
             - "9030:9030"
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java
@@ -38,12 +38,7 @@ public StarRocksEnrichedCatalog(String jdbcUrl, String username, String password
 
     public void truncateTable(String databaseName, String tableName)
             throws StarRocksCatalogException {
-        Preconditions.checkArgument(
-                !StringUtils.isNullOrWhitespaceOnly(databaseName),
-                "Database name cannot be null or empty.");
-        Preconditions.checkArgument(
-                !StringUtils.isNullOrWhitespaceOnly(tableName),
-                "Table name cannot be null or empty.");
+        checkTableArgument(databaseName, tableName);
         String alterSql = this.buildTruncateTableSql(databaseName, tableName);
         try {
             // TRUNCATE TABLE is not regarded as a column-based schema change for StarRocks, so
@@ -62,12 +57,7 @@ public void truncateTable(String databaseName, String tableName)
     }
 
     public void dropTable(String databaseName, String tableName) throws StarRocksCatalogException {
-        Preconditions.checkArgument(
-                !StringUtils.isNullOrWhitespaceOnly(databaseName),
-                "Database name cannot be null or empty.");
-        Preconditions.checkArgument(
-                !StringUtils.isNullOrWhitespaceOnly(tableName),
-                "Table name cannot be null or empty.");
+        checkTableArgument(databaseName, tableName);
         String alterSql = this.buildDropTableSql(databaseName, tableName);
         try {
             // like TRUNCATE TABLE, DROP TABLE isn't a column-affecting operation and `executeAlter`
@@ -84,6 +74,36 @@ public void dropTable(String databaseName, String tableName) throws StarRocksCat
         }
     }
 
+    public void renameColumn(
+            String databaseName, String tableName, String oldColumnName, String newColumnName)
+            throws StarRocksCatalogException {
+        checkTableArgument(databaseName, tableName);
+        Preconditions.checkArgument(
+                !StringUtils.isNullOrWhitespaceOnly(oldColumnName),
+                "old column name cannot be null or empty.");
+        Preconditions.checkArgument(
+                !StringUtils.isNullOrWhitespaceOnly(newColumnName),
+                "new column name cannot be null or empty.");
+        String alterSql =
+                this.buildRenameColumnSql(databaseName, tableName, oldColumnName, newColumnName);
+        try {
+            executeUpdateStatement(alterSql);
+        } catch (Exception e) {
+            LOG.error(
+                    "Failed to alter table `{}`.`{}` rename column {} to {}. SQL executed: {}",
+                    databaseName,
+                    tableName,
+                    oldColumnName,
+                    newColumnName,
+                    alterSql);
+            throw new StarRocksCatalogException(
+                    String.format(
+                            "Failed to alter table `%s`.`%s` rename column %s to %s.",
+                            databaseName, tableName, oldColumnName, newColumnName),
+                    e);
+        }
+    }
+
     private String buildTruncateTableSql(String databaseName, String tableName) {
         return String.format("TRUNCATE TABLE `%s`.`%s`;", databaseName, tableName);
     }
@@ -92,6 +112,13 @@ private String buildDropTableSql(String databaseName, String tableName) {
         return String.format("DROP TABLE `%s`.`%s`;", databaseName, tableName);
     }
 
+    private String buildRenameColumnSql(
+            String databaseName, String tableName, String oldColumnName, String newColumnName) {
+        return String.format(
+                "ALTER TABLE `%s`.`%s` RENAME COLUMN %s TO %s;",
+                databaseName, tableName, oldColumnName, newColumnName);
+    }
+
     private void executeUpdateStatement(String sql) throws StarRocksCatalogException {
         try {
             Method m =
@@ -104,4 +131,13 @@ private void executeUpdateStatement(String sql) throws StarRocksCatalogException
             throw new RuntimeException(e);
         }
     }
+
+    private void checkTableArgument(String databaseName, String tableName) {
+        Preconditions.checkArgument(
+                !StringUtils.isNullOrWhitespaceOnly(databaseName),
+                "Database name cannot be null or empty.");
+        Preconditions.checkArgument(
+                !StringUtils.isNullOrWhitespaceOnly(tableName),
+                "Table name cannot be null or empty.");
+    }
 }
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java
@@ -43,6 +43,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import static org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksDataType;
@@ -89,6 +90,7 @@ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
                 SchemaChangeEventType.CREATE_TABLE,
                 SchemaChangeEventType.ADD_COLUMN,
                 SchemaChangeEventType.DROP_COLUMN,
+                SchemaChangeEventType.RENAME_COLUMN,
                 SchemaChangeEventType.DROP_TABLE,
                 SchemaChangeEventType.TRUNCATE_TABLE);
     }
@@ -307,9 +309,20 @@ private void applyDropColumn(DropColumnEvent dropColumnEvent) throws SchemaEvolv
 
     private void applyRenameColumn(RenameColumnEvent renameColumnEvent)
             throws SchemaEvolveException {
-        // TODO StarRocks plans to support column rename since 3.3 which has not been released.
-        // Support it later.
-        throw new UnsupportedSchemaChangeEventException(renameColumnEvent);
+        try {
+            TableId tableId = renameColumnEvent.tableId();
+            Map<String, String> nameMapping = renameColumnEvent.getNameMapping();
+            for (Map.Entry<String, String> entry : nameMapping.entrySet()) {
+                catalog.renameColumn(
+                        tableId.getSchemaName(),
+                        tableId.getTableName(),
+                        entry.getKey(),
+                        entry.getValue());
+            }
+        } catch (Exception e) {
+            throw new SchemaEvolveException(
+                    renameColumnEvent, "fail to apply rename column event", e);
+        }
     }
 
     private void applyAlterColumnType(AlterColumnTypeEvent alterColumnTypeEvent)
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java
@@ -307,7 +307,6 @@ void testStarRocksDropColumn() throws Exception {
     }
 
     @Test
-    @Disabled("Rename column is not supported currently.")
     void testStarRocksRenameColumn() throws Exception {
         TableId tableId =
                 TableId.tableId(
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksContainer.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksContainer.java
@@ -33,7 +33,7 @@
 @Testcontainers
 public class StarRocksContainer extends JdbcDatabaseContainer<StarRocksContainer> {
 
-    private static final String DOCKER_IMAGE_NAME = "starrocks/allin1-ubuntu:3.2.6";
+    private static final String DOCKER_IMAGE_NAME = "starrocks/allin1-ubuntu:3.5.10";
 
     // exposed ports
     public static final int FE_HTTP_SERVICE_PORT = 8080;
diff --git a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksSinkTestBase.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksSinkTestBase.java
@@ -83,7 +83,7 @@ public static void startContainers() {
         long startWaitingTimestamp = System.currentTimeMillis();
 
         new LogMessageWaitStrategy()
-                .withRegEx(".*Enjoy the journal to StarRocks blazing-fast lake-house engine!.*\\s")
+                .withRegEx(".*Enjoy the journey to StarRocks blazing-fast lake-house engine!.*\\s")
                 .withTimes(1)
                 .withStartupTimeout(
                         Duration.of(DEFAULT_STARTUP_TIMEOUT_SECONDS, ChronoUnit.SECONDS))
diff --git a/tools/cdcup/src/sink/star_rocks.rb b/tools/cdcup/src/sink/star_rocks.rb
@@ -24,7 +24,7 @@ def connector_name
 
     def prepend_to_docker_compose_yaml(docker_compose_yaml)
       docker_compose_yaml['services']['starrocks'] = {
-        'image' => 'starrocks/allin1-ubuntu:3.2.6',
+        'image' => 'starrocks/allin1-ubuntu:3.5.10',
         'hostname' => 'starrocks',
         'ports' => %w[8080 9030],
         'volumes' => ["#{CDC_DATA_VOLUME}:/data"]

Original file line number	Diff line number	Diff line change
`@@ -307,7 +307,6 @@ void testStarRocksDropColumn() throws Exception {`
`307`	`307`	`}`
`308`	`308`
`309`	`309`	`@Test`
`310`		`- @Disabled("Rename column is not supported currently.")`
`311`	`310`	`void testStarRocksRenameColumn() throws Exception {`
`312`	`311`	`TableId tableId =`
`313`	`312`	`TableId.tableId(`