This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 15cb21833 [FLINK-39228][starrocks] Escape quoted default/comment
literals in DDL (#4308)
15cb21833 is described below
commit 15cb21833c4df9873ce3a0e0d18beba30fb00ad4
Author: Mingliang Zhu <[email protected]>
AuthorDate: Wed Sep 23 22:48:47 2026 +0800
[FLINK-39228][starrocks] Escape quoted default/comment literals in DDL
(#4308)
Co-authored-by: lvyanquan <[email protected]>
Co-authored-by: Codex <[email protected]>
---
.../starrocks/sink/StarRocksMetadataApplier.java | 9 ++++--
.../connectors/starrocks/sink/StarRocksUtils.java | 9 ++++--
.../sink/utils/StarRocksEscapingUtils.java | 34 ++++++++++++++++++++++
.../sink/StarRocksMetadataApplierITCase.java | 25 ++++++++++++----
.../sink/StarRocksMetadataApplierTest.java | 8 ++++-
.../starrocks/sink/StarRocksUtilsTest.java | 24 +++++++++++++++
6 files changed, 96 insertions(+), 13 deletions(-)
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
index 7f0208441..6d1a27f39 100644
---
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
@@ -47,6 +47,7 @@ import java.util.Map;
import java.util.Set;
import static
org.apache.flink.cdc.connectors.starrocks.sink.StarRocksUtils.toStarRocksDataType;
+import static
org.apache.flink.cdc.connectors.starrocks.sink.utils.StarRocksEscapingUtils.escape;
/** A {@code MetadataApplier} that applies metadata changes to StarRocks. */
public class StarRocksMetadataApplier implements MetadataApplier {
@@ -158,10 +159,12 @@ public class StarRocksMetadataApplier implements
MetadataApplier {
new StarRocksColumn.Builder()
.setColumnName(column.getName())
.setOrdinalPosition(-1)
- .setColumnComment(column.getComment())
+ .setColumnComment(escape(column.getComment()))
.setDefaultValue(
-
StarRocksUtils.convertInvalidTimestampDefaultValue(
-
column.getDefaultValueExpression(), column.getType()));
+ escape(
+
StarRocksUtils.convertInvalidTimestampDefaultValue(
+
column.getDefaultValueExpression(),
+ column.getType())));
toStarRocksDataType(column, false, builder,
tableCreateConfig.getUnicodeCharMaxBytes());
addColumns.add(builder.build());
}
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/StarRocksUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtils.java
index d2a50d4ab..120f198d6 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtils.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtils.java
@@ -53,6 +53,7 @@ import java.util.List;
import static org.apache.flink.cdc.common.types.DataTypeChecks.getPrecision;
import static org.apache.flink.cdc.common.types.DataTypeChecks.getScale;
+import static
org.apache.flink.cdc.connectors.starrocks.sink.utils.StarRocksEscapingUtils.escape;
/** Utilities for conversion from source table to StarRocks table. */
public class StarRocksUtils {
@@ -91,10 +92,12 @@ public class StarRocksUtils {
new StarRocksColumn.Builder()
.setColumnName(column.getName())
.setOrdinalPosition(i)
- .setColumnComment(column.getComment())
+ .setColumnComment(escape(column.getComment()))
.setDefaultValue(
- convertInvalidTimestampDefaultValue(
-
column.getDefaultValueExpression(), column.getType()));
+ escape(
+
convertInvalidTimestampDefaultValue(
+
column.getDefaultValueExpression(),
+ column.getType())));
toStarRocksDataType(
column,
i < primaryKeyCount,
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/utils/StarRocksEscapingUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksEscapingUtils.java
new file mode 100644
index 000000000..fc4369a43
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/utils/StarRocksEscapingUtils.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.starrocks.sink.utils;
+
+import org.apache.flink.cdc.common.annotation.Internal;
+
+/** Utilities for escaping values embedded in double-quoted StarRocks SQL
literals. */
+@Internal
+public final class StarRocksEscapingUtils {
+
+ private StarRocksEscapingUtils() {}
+
+ public static String escape(String value) {
+ if (value == null) {
+ return null;
+ }
+ return value.replace("\\", "\\\\").replace("\"", "\\\"");
+ }
+}
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/StarRocksMetadataApplierITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java
index d9164d60c..bdbaaceea 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksMetadataApplierITCase.java
@@ -101,7 +101,12 @@ class StarRocksMetadataApplierITCase extends
StarRocksSinkTestBase {
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),
+ "\"name\"",
+ "\"name\\\\default\""))
.primaryKey("id")
.build();
@@ -123,9 +128,16 @@ class StarRocksMetadataApplierITCase extends
StarRocksSinkTestBase {
Collections.singletonList(
new AddColumnEvent.ColumnWithPosition(
new PhysicalColumn(
- "extra_decimal",
- DataTypes.DECIMAL(17, 0),
- null)))));
+ "extra_decimal",
DataTypes.DECIMAL(17, 0), null)))),
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new AddColumnEvent.ColumnWithPosition(
+ new PhysicalColumn(
+ "extra_string",
+ DataTypes.VARCHAR(17),
+ "\"extra_string\"",
+ "\"extra\\\\string\"")))));
}
private List<Event> generateDropColumnEvents(TableId tableId) {
@@ -314,10 +326,11 @@ class StarRocksMetadataApplierITCase extends
StarRocksSinkTestBase {
Arrays.asList(
"id | int | NO | true | null",
"number | double | YES | false | null",
- "name | varchar(51) | YES | false | null",
+ "name | varchar(51) | YES | false |
\"name\\\\default\"",
"extra_date | date | YES | false | null",
"extra_bool | boolean | YES | false | null",
- "extra_decimal | decimal(17,0) | YES | false | null");
+ "extra_decimal | decimal(17,0) | YES | false | null",
+ "extra_string | varchar(51) | YES | false |
\"extra\\\\string\"");
assertEqualsInOrder(expected, actual);
}
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
index 2da34d6cf..2f41df669 100644
---
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
@@ -140,7 +140,11 @@ class StarRocksMetadataApplierTest {
new AddColumnEvent.ColumnWithPosition(
Column.physicalColumn("col2", new
DecimalType(20, 5))),
new AddColumnEvent.ColumnWithPosition(
- Column.physicalColumn("col3", new
SmallIntType()))));
+ Column.physicalColumn(
+ "col3",
+ new SmallIntType(),
+ "comment \"quoted\" \\ path",
+ "default \"quoted\" \\
path"))));
metadataApplier.applySchemaChange(addColumnEvent);
StarRocksTable actualTable =
@@ -170,6 +174,8 @@ class StarRocksMetadataApplierTest {
.setOrdinalPosition(2)
.setDataType("smallint")
.setNullable(true)
+ .setColumnComment("comment \\\"quoted\\\" \\\\ path")
+ .setDefaultValue("default \\\"quoted\\\" \\\\ path")
.build());
StarRocksTable expectTable =
new StarRocksTable.Builder()
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/StarRocksUtilsTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtilsTest.java
index e1caaba50..827456635 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtilsTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksUtilsTest.java
@@ -240,6 +240,30 @@ class StarRocksUtilsTest {
assertThat(table.getComment()).hasValue("test table comment");
}
+ @Test
+ void testToStarRocksTableEscapesColumnDefaultValueAndComment() {
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn(
+ "escaped",
+ DataTypes.VARCHAR(100),
+ "comment \"quoted\" \\ path",
+ "default \"quoted\" \\ path")
+ .primaryKey("id")
+ .build();
+
+ StarRocksTable table =
+ StarRocksUtils.toStarRocksTable(
+ TableId.tableId("db", "table"),
+ schema,
+ new TableCreateConfig(null, Collections.emptyMap()));
+
+ StarRocksColumn column = table.getColumn("escaped");
+ assertThat(column.getColumnComment()).hasValue("comment \\\"quoted\\\"
\\\\ path");
+ assertThat(column.getDefaultValue()).hasValue("default \\\"quoted\\\"
\\\\ path");
+ }
+
//
--------------------------------------------------------------------------------------------
// Tests for convertInvalidTimestampDefaultValue
//
--------------------------------------------------------------------------------------------
