[FLINK-39833][pipeline][starrocks] Escape special characters in column comment and default value when building DDL

Author: linguoxuan

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

@@ -191,11 +191,18 @@ private String buildColumnStmt(StarRocksColumn column) {
         builder.append(" ");
         builder.append(column.isNullable() ? "NULL" : "NOT NULL");
         if (column.getDefaultValue().isPresent()) {
-            builder.append(String.format(" DEFAULT \"%s\"", column.getDefaultValue().get()));
+            builder.append(
+                    String.format(
+                            " DEFAULT \"%s\"",
+                            StarRocksUtils.escapeSqlStringLiteral(column.getDefaultValue().get())));
         }
 
         if (column.getColumnComment().isPresent()) {
-            builder.append(String.format(" COMMENT \"%s\"", column.getColumnComment().get()));
+            builder.append(
+                    String.format(
+                            " COMMENT \"%s\"",
+                            StarRocksUtils.escapeSqlStringLiteral(
+                                    column.getColumnComment().get())));
         }
         return builder.toString();
     }

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

@@ -158,10 +158,13 @@ private void applyAddColumn(AddColumnEvent addColumnEvent) throws SchemaEvolveEx
                     new StarRocksColumn.Builder()
                             .setColumnName(column.getName())
                             .setOrdinalPosition(-1)
-                            .setColumnComment(column.getComment())
+                            .setColumnComment(
+                                    StarRocksUtils.escapeSqlStringLiteral(column.getComment()))
                             .setDefaultValue(
-                                    StarRocksUtils.convertInvalidTimestampDefaultValue(
-                                            column.getDefaultValueExpression(), column.getType()));
+                                    StarRocksUtils.escapeSqlStringLiteral(
+                                            StarRocksUtils.convertInvalidTimestampDefaultValue(
+                                                    column.getDefaultValueExpression(),
+                                                    column.getType())));
             toStarRocksDataType(column, false, builder);
             addColumns.add(builder.build());
         }

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

@@ -91,10 +91,12 @@ public static StarRocksTable toStarRocksTable(
                     new StarRocksColumn.Builder()
                             .setColumnName(column.getName())
                             .setOrdinalPosition(i)
-                            .setColumnComment(column.getComment())
+                            .setColumnComment(escapeSqlStringLiteral(column.getComment()))
                             .setDefaultValue(
-                                    convertInvalidTimestampDefaultValue(
-                                            column.getDefaultValueExpression(), column.getType()));
+                                    escapeSqlStringLiteral(
+                                            convertInvalidTimestampDefaultValue(
+                                                    column.getDefaultValueExpression(),
+                                                    column.getType())));
             toStarRocksDataType(column, i < primaryKeyCount, builder);
             starRocksColumns.add(builder.build());
         }
@@ -108,7 +110,7 @@ public static StarRocksTable toStarRocksTable(
                         .setTableKeys(schema.primaryKeys())
                         // use primary keys as distribution keys by default
                         .setDistributionKeys(schema.primaryKeys())
-                        .setComment(schema.comment());
+                        .setComment(escapeSqlStringLiteral(schema.comment()));
         if (tableCreateConfig.getNumBuckets().isPresent()) {
             tableBuilder.setNumBuckets(tableCreateConfig.getNumBuckets().get());
         }
@@ -492,4 +494,29 @@ public static String convertInvalidTimestampDefaultValue(
 
         return defaultValue;
     }
+
+    public static String escapeSqlStringLiteral(String value) {
+        if (value == null) {
+            return null;
+        }
+        StringBuilder sb = new StringBuilder(value.length() + 8);
+        for (int i = 0; i < value.length(); i++) {
+            char c = value.charAt(i);
+            switch (c) {
+                case '\\':
+                    sb.append("\\\\");
+                    break;
+                case '"':
+                    sb.append("\"\"");
+                    break;
+                case '\n':
+                case '\r':
+                    sb.append(' ');
+                    break;
+                default:
+                    sb.append(c);
+            }
+        }
+        return sb.toString();
+    }
 }