[FLINK-38829][pipline-connector/starrocks] Support rename column DDL for starrocks sink connector

zml1206 · zml1206 · commit 0240447a9d5f · 2025-12-23T09:13:07.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;
@@ -307,9 +308,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/StarRocksMetadataApplierTest.java b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierTest.java
@@ -18,10 +18,7 @@
 package org.apache.flink.cdc.connectors.starrocks.sink;
 
 import org.apache.flink.cdc.common.configuration.Configuration;
-import org.apache.flink.cdc.common.event.AddColumnEvent;
-import org.apache.flink.cdc.common.event.CreateTableEvent;
-import org.apache.flink.cdc.common.event.DropColumnEvent;
-import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.event.*;
 import org.apache.flink.cdc.common.schema.Column;
 import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.common.types.BooleanType;
@@ -38,10 +35,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 
 import static org.apache.flink.cdc.connectors.starrocks.sink.StarRocksDataSinkOptions.TABLE_CREATE_NUM_BUCKETS;
 import static org.apache.flink.cdc.connectors.starrocks.sink.StarRocksDataSinkOptions.TABLE_SCHEMA_CHANGE_TIMEOUT;
@@ -226,4 +220,63 @@ void testDropColumn() throws Exception {
                         .build();
         Assertions.assertThat(actualTable).isEqualTo(expectTable);
     }
+
+    @Test
+    void testRenameColumn() throws Exception {
+        TableId tableId = TableId.parse("test.tbl4");
+        Schema schema =
+                Schema.newBuilder()
+                        .physicalColumn("col1", new IntType())
+                        .physicalColumn("col2", new BooleanType())
+                        .physicalColumn("col3", new TimestampType())
+                        .primaryKey("col1")
+                        .build();
+        CreateTableEvent createTableEvent = new CreateTableEvent(tableId, schema);
+        metadataApplier.applySchemaChange(createTableEvent);
+
+        Map<String, String> nameMapping = new HashMap<>();
+        nameMapping.put("col2", "newCol2");
+        nameMapping.put("col3", "newCol3");
+        RenameColumnEvent renameColumnEvent = new RenameColumnEvent(tableId, nameMapping);
+        metadataApplier.applySchemaChange(renameColumnEvent);
+
+        StarRocksTable actualTable =
+                catalog.getTable(tableId.getSchemaName(), tableId.getTableName()).orElse(null);
+        Assertions.assertThat(actualTable).isNotNull();
+
+        List<StarRocksColumn> columns = new ArrayList<>();
+        columns.add(
+                new StarRocksColumn.Builder()
+                        .setColumnName("col1")
+                        .setOrdinalPosition(0)
+                        .setDataType("int")
+                        .setNullable(true)
+                        .build());
+        columns.add(
+                new StarRocksColumn.Builder()
+                        .setColumnName("newCol2")
+                        .setOrdinalPosition(1)
+                        .setDataType("boolean")
+                        .setNullable(true)
+                        .build());
+        columns.add(
+                new StarRocksColumn.Builder()
+                        .setColumnName("newCol3")
+                        .setOrdinalPosition(2)
+                        .setDataType("datetime")
+                        .setNullable(true)
+                        .build());
+        StarRocksTable expectTable =
+                new StarRocksTable.Builder()
+                        .setDatabaseName(tableId.getSchemaName())
+                        .setTableName(tableId.getTableName())
+                        .setTableType(StarRocksTable.TableType.PRIMARY_KEY)
+                        .setColumns(columns)
+                        .setTableKeys(schema.primaryKeys())
+                        .setDistributionKeys(schema.primaryKeys())
+                        .setNumBuckets(10)
+                        .setTableProperties(Collections.singletonMap("replication_num", "5"))
+                        .build();
+        Assertions.assertThat(actualTable).isEqualTo(expectTable);
+    }
 }
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/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"]
