support modify nullable

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

@@ -112,15 +112,7 @@ public void alterColumnType(String databaseName, String tableName, StarRocksColu
         Preconditions.checkArgument(
                 !StringUtils.isNullOrWhitespaceOnly(column.getColumnName()),
                 "column name cannot be null or empty.");
-        String alterSql =
-                buildAlterColumnTypeSql(
-                        databaseName,
-                        tableName,
-                        column.getColumnName(),
-                        getFullColumnType(
-                                column.getDataType(),
-                                column.getColumnSize(),
-                                column.getDecimalDigits()));
+        String alterSql = buildAlterColumnTypeSql(databaseName, tableName, buildColumnStmt(column));
         try {
             long startTimeMillis = System.currentTimeMillis();
             executeUpdateStatement(alterSql);
@@ -161,10 +153,9 @@ private String buildRenameColumnSql(
     }
 
     private String buildAlterColumnTypeSql(
-            String databaseName, String tableName, String columnName, String dataType) {
+            String databaseName, String tableName, String columnStmt) {
         return String.format(
-                "ALTER TABLE `%s`.`%s` MODIFY COLUMN %s %s",
-                databaseName, tableName, columnName, dataType);
+                "ALTER TABLE `%s`.`%s` MODIFY COLUMN %s", databaseName, tableName, columnStmt);
     }
 
     private void executeUpdateStatement(String sql) throws StarRocksCatalogException {
@@ -189,6 +180,26 @@ private void checkTableArgument(String databaseName, String 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();

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

@@ -162,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();