[FLINK-38839][runtime] Support semicolon delimiter for table-options

Author: tchivs

docs/content.zh/docs/core-concept/transform.md

@@ -311,7 +311,7 @@ transform:
     table-options: comment=web order
     description: auto creating table options example
 ```
-小技巧：table-options 的格式是 `key1=value1,key2=value2`。
+小技巧：table-options 的格式是 `key1=value1,key2=value2`；如果 value 中包含逗号，使用分号分隔：`key1=value1;key2=value2`。
 
 ## Classification mapping
 多个转换规则可以定义为分类映射。
@@ -466,4 +466,4 @@ pipeline:
 |---------------|--------|-------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | openai.model  | STRING | required          | Name of model to be called, for example: "text-embedding-3-small", Available options are "text-embedding-3-small", "text-embedding-3-large", "text-embedding-ada-002". |
 | openai.host   | STRING | required          | Host of the Model server to be connected, for example: `http://langchain4j.dev/demo/openai/v1`.                                                                        |
-| openai.apikey | STRING | required          | Api Key for verification of the Model server, for example, "demo".                                                                                                     |
+| openai.apikey | STRING | required          | Api Key for verification of the Model server, for example, "demo".                                                                                                     |

docs/content/docs/core-concept/transform.md

@@ -315,6 +315,7 @@ transform:
     description: auto creating table options example
 ```
 Tips: The format of table-options is `key1=value1,key2=value2`.
+If option values contain commas, use semicolon delimiter like `key1=value1;key2=value2`.
 
 ## Classification mapping
 Multiple transform rules can be defined to classify input data rows and apply different processing.
@@ -471,4 +472,4 @@ The following built-in models are provided:
 |---------------|--------|-------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | openai.model  | STRING | required          | Name of model to be called, for example: "text-embedding-3-small", Available options are "text-embedding-3-small", "text-embedding-3-large", "text-embedding-ada-002". |
 | openai.host   | STRING | required          | Host of the Model server to be connected, for example: `http://langchain4j.dev/demo/openai/v1`.                                                                        |
-| openai.apikey | STRING | required          | Api Key for verification of the Model server, for example, "demo".                                                                                                     |
+| openai.apikey | STRING | required          | Api Key for verification of the Model server, for example, "demo".                                                                                                     |

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransform.java

@@ -58,13 +58,17 @@ public SchemaMetadataTransform(
             partitionKeys = Arrays.asList(partitionKeyArr);
         }
         if (!StringUtils.isNullOrWhitespaceOnly(tableOptionString)) {
-            for (String tableOption : tableOptionString.split(",")) {
-                String[] kv = tableOption.split("=");
+            // Use semicolon as delimiter to support multi-value options like
+            // sequence.field=jjsj,gxsj.
+            // Keep comma delimiter for backward compatibility.
+            String delimiter = tableOptionString.contains(";") ? ";" : ",";
+            for (String tableOption : tableOptionString.split(delimiter)) {
+                String[] kv = tableOption.split("=", 2);
                 if (kv.length != 2) {
                     throw new IllegalArgumentException(
                             "table option format error: "
                                     + tableOptionString
-                                    + ", it should be like `key1=value1,key2=value2`.");
+                                    + ", it should be like `key1=value1,key2=value2` or `key1=value1;key2=value2`.");
                 }
                 options.put(kv[0].trim(), kv[1].trim());
             }

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/SchemaMetadataTransformTest.java

@@ -0,0 +1,68 @@
+/*
+ * 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.runtime.operators.transform;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Testcases for {@link SchemaMetadataTransform}. */
+class SchemaMetadataTransformTest {
+
+    @Test
+    void testTableOptionsWithCommaDelimiter() {
+        SchemaMetadataTransform transform =
+                new SchemaMetadataTransform(null, null, "key1=value1,key2=value2");
+        assertThat(transform.getOptions())
+                .containsEntry("key1", "value1")
+                .containsEntry("key2", "value2");
+    }
+
+    @Test
+    void testTableOptionsWithSemicolonDelimiter() {
+        SchemaMetadataTransform transform =
+                new SchemaMetadataTransform(null, null, "key1=value1;key2=value2");
+        assertThat(transform.getOptions())
+                .containsEntry("key1", "value1")
+                .containsEntry("key2", "value2");
+    }
+
+    @Test
+    void testTableOptionsWithCommaInValue() {
+        SchemaMetadataTransform transform =
+                new SchemaMetadataTransform(
+                        null,
+                        null,
+                        "sequence.field=gxsj,jjsj;"
+                                + "file-index.range-bitmap.columns=jjsj;"
+                                + "file-index.bloom-filter.columns=jjdbh");
+        assertThat(transform.getOptions())
+                .containsEntry("sequence.field", "gxsj,jjsj")
+                .containsEntry("file-index.range-bitmap.columns", "jjsj")
+                .containsEntry("file-index.bloom-filter.columns", "jjdbh");
+    }
+
+    @Test
+    void testTableOptionsSplitByFirstEqualSign() {
+        SchemaMetadataTransform transform =
+                new SchemaMetadataTransform(null, null, "key1=value=1;key2=value2");
+        assertThat(transform.getOptions())
+                .containsEntry("key1", "value=1")
+                .containsEntry("key2", "value2");
+    }
+}