fix

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

@@ -28,6 +28,7 @@
 
 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 {
@@ -113,7 +114,13 @@ public void alterColumnType(String databaseName, String tableName, StarRocksColu
                 "column name cannot be null or empty.");
         String alterSql =
                 buildAlterColumnTypeSql(
-                        databaseName, tableName, column.getColumnName(), column.getDataType());
+                        databaseName,
+                        tableName,
+                        column.getColumnName(),
+                        getFullColumnType(
+                                column.getDataType(),
+                                column.getColumnSize(),
+                                column.getDecimalDigits()));
         try {
             long startTimeMillis = System.currentTimeMillis();
             executeUpdateStatement(alterSql);
@@ -181,4 +188,24 @@ private void checkTableArgument(String databaseName, String tableName) {
                 !StringUtils.isNullOrWhitespaceOnly(tableName),
                 "Table name cannot be null or empty.");
     }
+
+    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;
+        }
+    }
 }