[FLINK-38834][pipline-connector/starrocks] Support alter column type DDL for starrocks sink connector (#4198)

Author: zml1206

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksEnrichedCatalog.java

@@ -22,11 +22,13 @@
 
 import com.starrocks.connector.flink.catalog.StarRocksCatalog;
 import com.starrocks.connector.flink.catalog.StarRocksCatalogException;
+import com.starrocks.connector.flink.catalog.StarRocksColumn;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.util.Optional;
 
 /** An enriched {@code StarRocksCatalog} with more schema evolution abilities. */
 public class StarRocksEnrichedCatalog extends StarRocksCatalog {
@@ -104,6 +106,37 @@ public void renameColumn(
         }
     }
 
+    public void alterColumnType(String databaseName, String tableName, StarRocksColumn column)
+            throws StarRocksCatalogException {
+        checkTableArgument(databaseName, tableName);
+        Preconditions.checkArgument(
+                !StringUtils.isNullOrWhitespaceOnly(column.getColumnName()),
+                "column name cannot be null or empty.");
+        String alterSql = buildAlterColumnTypeSql(databaseName, tableName, buildColumnStmt(column));
+        try {
+            long startTimeMillis = System.currentTimeMillis();
+            executeUpdateStatement(alterSql);
+            LOG.info(
+                    "Success to alter table {}.{} modify column type, duration: {}ms, sql: {}",
+                    databaseName,
+                    tableName,
+                    System.currentTimeMillis() - startTimeMillis,
+                    alterSql);
+        } catch (Exception e) {
+            LOG.error(
+                    "Failed to alter table {}.{} modify column type, sql: {}",
+                    databaseName,
+                    tableName,
+                    alterSql,
+                    e);
+            throw new StarRocksCatalogException(
+                    String.format(
+                            "Failed to alter table %s.%s modify column type",
+                            databaseName, tableName),
+                    e);
+        }
+    }
+
     private String buildTruncateTableSql(String databaseName, String tableName) {
         return String.format("TRUNCATE TABLE `%s`.`%s`;", databaseName, tableName);
     }
@@ -119,6 +152,12 @@ private String buildRenameColumnSql(
                 databaseName, tableName, oldColumnName, newColumnName);
     }
 
+    private String buildAlterColumnTypeSql(
+            String databaseName, String tableName, String columnStmt) {
+        return String.format(
+                "ALTER TABLE `%s`.`%s` MODIFY COLUMN %s", databaseName, tableName, columnStmt);
+    }
+
     private void executeUpdateStatement(String sql) throws StarRocksCatalogException {
         try {
             Method m =
@@ -140,4 +179,44 @@ private void checkTableArgument(String databaseName, String tableName) {
                 !StringUtils.isNullOrWhitespaceOnly(tableName),
                 "Table name cannot be null or empty.");
     }
+
+    private String buildColumnStmt(StarRocksColumn column) {
+        StringBuilder builder = new StringBuilder();
+        builder.append("`");
+        builder.append(column.getColumnName());
+        builder.append("` ");
+        builder.append(
+                getFullColumnType(
+                        column.getDataType(), column.getColumnSize(), column.getDecimalDigits()));
+        builder.append(" ");
+        builder.append(column.isNullable() ? "NULL" : "NOT NULL");
+        if (column.getDefaultValue().isPresent()) {
+            builder.append(String.format(" DEFAULT \"%s\"", column.getDefaultValue().get()));
+        }
+
+        if (column.getColumnComment().isPresent()) {
+            builder.append(String.format(" COMMENT \"%s\"", column.getColumnComment().get()));
+        }
+        return builder.toString();
+    }
+
+    private String getFullColumnType(
+            String type, Optional<Integer> columnSize, Optional<Integer> decimalDigits) {
+        String dataType = type.toUpperCase();
+        switch (dataType) {
+            case "DECIMAL":
+                Preconditions.checkArgument(
+                        columnSize.isPresent(), "DECIMAL type must have column size");
+                Preconditions.checkArgument(
+                        decimalDigits.isPresent(), "DECIMAL type must have decimal digits");
+                return String.format("DECIMAL(%d, %s)", columnSize.get(), decimalDigits.get());
+            case "CHAR":
+            case "VARCHAR":
+                Preconditions.checkArgument(
+                        columnSize.isPresent(), type + " type must have column size");
+                return String.format("%s(%d)", dataType, columnSize.get());
+            default:
+                return dataType;
+        }
+    }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplier.java

@@ -29,9 +29,9 @@
 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;
 
 import org.apache.flink.shaded.guava31.com.google.common.collect.Sets;
 
@@ -91,6 +91,7 @@ public Set<SchemaChangeEventType> getSupportedSchemaEvolutionTypes() {
                 SchemaChangeEventType.ADD_COLUMN,
                 SchemaChangeEventType.DROP_COLUMN,
                 SchemaChangeEventType.RENAME_COLUMN,
+                SchemaChangeEventType.ALTER_COLUMN_TYPE,
                 SchemaChangeEventType.DROP_TABLE,
                 SchemaChangeEventType.TRUNCATE_TABLE);
     }
@@ -325,15 +326,21 @@ private void applyRenameColumn(RenameColumnEvent renameColumnEvent)
         }
     }
 
-    private void applyAlterColumnType(AlterColumnTypeEvent alterColumnTypeEvent)
-            throws SchemaEvolveException {
-        // TODO There are limitations for data type conversions. We should know the data types
-        // before and after changing so that we can make a validation. But the event only contains
-        // data
-        // types after changing. One way is that the framework delivers the old schema. We can
-        // support
-        // the alter after a discussion.
-        throw new UnsupportedSchemaChangeEventException(alterColumnTypeEvent);
+    private void applyAlterColumnType(AlterColumnTypeEvent event) throws SchemaEvolveException {
+        try {
+            TableId tableId = event.tableId();
+            Map<String, DataType> typeMapping = event.getTypeMapping();
+
+            for (Map.Entry<String, DataType> entry : typeMapping.entrySet()) {
+                StarRocksColumn.Builder builder =
+                        new StarRocksColumn.Builder().setColumnName(entry.getKey());
+                toStarRocksDataType(entry.getValue(), false, builder);
+                catalog.alterColumnType(
+                        tableId.getSchemaName(), tableId.getTableName(), builder.build());
+            }
+        } catch (Exception e) {
+            throw new SchemaEvolveException(event, "fail to apply alter column type event", e);
+        }
     }
 
     private void applyTruncateTable(TruncateTableEvent truncateTableEvent) {

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtils.java

@@ -113,10 +113,15 @@ public static StarRocksTable toStarRocksTable(
 
     /** Convert CDC data type to StarRocks data type. */
     public static void toStarRocksDataType(
-            Column cdcColumn, boolean isPrimaryKeys, StarRocksColumn.Builder builder) {
+            DataType cdcDataType, boolean isPrimaryKeys, StarRocksColumn.Builder builder) {
         CdcDataTypeTransformer dataTypeTransformer =
                 new CdcDataTypeTransformer(isPrimaryKeys, builder);
-        cdcColumn.getType().accept(dataTypeTransformer);
+        cdcDataType.accept(dataTypeTransformer);
+    }
+
+    public static void toStarRocksDataType(
+            Column cdcColumn, boolean isPrimaryKeys, StarRocksColumn.Builder builder) {
+        toStarRocksDataType(cdcColumn.getType(), isPrimaryKeys, builder);
     }
 
     /** Format DATE type data. */

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java

@@ -55,7 +55,6 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
@@ -163,7 +162,7 @@ private List<Event> generateAlterColumnTypeEvents(TableId tableId) {
                 Schema.newBuilder()
                         .column(new PhysicalColumn("id", DataTypes.INT().notNull(), null))
                         .column(new PhysicalColumn("number", DataTypes.DOUBLE(), null))
-                        .column(new PhysicalColumn("name", DataTypes.VARCHAR(17), null))
+                        .column(new PhysicalColumn("name", DataTypes.VARCHAR(17).notNull(), null))
                         .primaryKey("id")
                         .build();
 
@@ -327,15 +326,14 @@ void testStarRocksRenameColumn() throws Exception {
     }
 
     @Test
-    @Disabled("Alter column type is not supported currently.")
     void testStarRocksAlterColumnType() throws Exception {
         TableId tableId =
                 TableId.tableId(
                         StarRocksContainer.STARROCKS_DATABASE_NAME,
                         StarRocksContainer.STARROCKS_TABLE_NAME);
 
         runJobWithEvents(generateAlterColumnTypeEvents(tableId));
-
+        waitAlterDone(tableId, 60000L);
         List<String> actual = inspectTableSchema(tableId);
 
         List<String> expected =
@@ -348,7 +346,6 @@ void testStarRocksAlterColumnType() throws Exception {
     }
 
     @Test
-    @Disabled("Alter column type is not supported currently.")
     void testStarRocksNarrowingAlterColumnType() throws Exception {
         Assertions.assertThatThrownBy(
                 () -> {

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

@@ -39,6 +39,7 @@
 import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
 import org.testcontainers.lifecycle.Startables;
 
+import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.time.Duration;
@@ -227,6 +228,33 @@ public List<String> fetchTableContent(TableId tableId, int columnCount) throws S
         return results;
     }
 
+    // Starrocks alter column is asynchronous and does not support Light mode.
+    public void waitAlterDone(TableId tableId, long timeout)
+            throws SQLException, InterruptedException {
+        Connection conn = STARROCKS_CONTAINER.createConnection("");
+        conn.createStatement().execute(String.format("USE `%s`", tableId.getSchemaName()));
+        long t0 = System.currentTimeMillis();
+        while (System.currentTimeMillis() - t0 < timeout) {
+            ResultSet rs =
+                    conn.createStatement()
+                            .executeQuery(
+                                    String.format(
+                                            "SHOW ALTER TABLE COLUMN WHERE TableName = '%s' ORDER BY CreateTime DESC LIMIT 1",
+                                            tableId.getTableName()));
+            if (rs.next()) {
+                String state = rs.getString("State");
+                if ("FINISHED".equals(state)) {
+                    return;
+                }
+                if ("CANCELLED".equals(state)) {
+                    throw new RuntimeException("Alter failed: " + rs.getString("Msg"));
+                }
+            }
+            Thread.sleep(1000L);
+        }
+        throw new RuntimeException("Alter job timeout");
+    }
+
     public static <T> void assertEqualsInAnyOrder(List<T> expected, List<T> actual) {
         Assertions.assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
     }