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 d1ae9f1da [FLINK-40239][runtime] Support additional Flink SQL regular
expression functions in YAML Transform (#4491)
d1ae9f1da is described below
commit d1ae9f1da84f7b5e06de73458bd939e14a0e1a95
Author: haruki <[email protected]>
AuthorDate: Wed Aug 5 16:00:00 2026 +0800
[FLINK-40239][runtime] Support additional Flink SQL regular expression
functions in YAML Transform (#4491)
---
docs/content.zh/docs/core-concept/transform.md | 5 ++
docs/content/docs/core-concept/transform.md | 5 ++
.../runtime/functions/impl/StringFunctions.java | 83 ++++++++++++++++++++++
.../parser/metadata/TransformSqlOperatorTable.java | 54 ++++++++++++++
.../parser/metadata/TransformSqlReturnTypes.java | 14 ++++
.../functions/impl/StringFunctionsTest.java | 61 ++++++++++++++++
.../transform/PostTransformOperatorTest.java | 53 ++++++++++++++
.../cdc/runtime/parser/TransformParserTest.java | 33 +++++++++
8 files changed, 308 insertions(+)
diff --git a/docs/content.zh/docs/core-concept/transform.md
b/docs/content.zh/docs/core-concept/transform.md
index dc405f777..a563c040b 100644
--- a/docs/content.zh/docs/core-concept/transform.md
+++ b/docs/content.zh/docs/core-concept/transform.md
@@ -178,6 +178,11 @@ Flink CDC 使用 [Calcite](https://calcite.apache.org/)
来解析表达式并且
| LOWER(string) | lower(string)
| 返回小写形式的字符串。
|
| TRIM(string1) | trim('BOTH',string1)
| 返回去除两端空格的字符串。
|
| REGEXP_REPLACE(string1, string2, string3) | regexpReplace(string1,
string2, string3) | 返回将 STRING1 中所有匹配正则表达式 STRING2 的子串替换为 STRING3
后的字符串。例如，'foobar'.regexpReplace('oo\|ar', '') 返回 "fb"。 |
+| REGEXP_EXTRACT(string, regex[, extractIndex]) | regexpExtract(string,
regex[, extractIndex]) | 返回正则表达式组 extractIndex 捕获的子串。extractIndex 默认为 0，0
表示完整匹配。输入为 NULL、无匹配、正则表达式非法或组索引非法时返回 NULL。 |
+| REGEXP_EXTRACT_ALL(string, regex[, extractIndex]) | regexpExtractAll(string,
regex[, extractIndex]) | 返回组 extractIndex 捕获的所有子串组成的
ARRAY<STRING>。extractIndex 默认为 1，0 表示完整匹配。无匹配时返回空数组，输入为
NULL、正则表达式非法或组索引非法时返回 NULL。 |
+| REGEXP_COUNT(string, regex) | regexpCount(string,
regex) | 返回正则表达式匹配的非重叠子串数量。无匹配时返回 0，输入为 NULL 或正则表达式非法时返回 NULL。 |
+| REGEXP_INSTR(string, regex) | regexpInstr(string,
regex) | 返回第一个匹配正则表达式的子串的 1-based 位置。无匹配时返回 0，输入为 NULL
或正则表达式非法时返回 NULL。 |
+| REGEXP_SUBSTR(string, regex) | regexpSubstr(string,
regex) | 返回第一个匹配正则表达式的子串。输入为 NULL、无匹配或正则表达式非法时返回 NULL。 |
| SUBSTR(string, integer1[, integer2]) |
substr(string,integer1,integer2) | 返回 STRING 从位置 integer1 开始、长度为
integer2（默认到末尾）的子串。
|
| SUBSTRING(string FROM integer1 [ FOR integer2 ]) |
substring(string,integer1,integer2) | 返回 STRING 从位置 integer1 开始、长度为
integer2（默认到末尾）的子串。
|
| CONCAT(string1, string2,…) | concat(string1,
string2,…) | 返回连接 string1、string2、… 后的字符串。例如，CONCAT('AA', 'BB',
'CC') 返回 'AABBCC'。 |
diff --git a/docs/content/docs/core-concept/transform.md
b/docs/content/docs/core-concept/transform.md
index eb7629057..e0a8451dd 100644
--- a/docs/content/docs/core-concept/transform.md
+++ b/docs/content/docs/core-concept/transform.md
@@ -179,6 +179,11 @@ Logical functions follow SQL three-valued logic for
nullable BOOLEAN values. `AN
| LOWER(string) | lower(string)
| Returns string in lowercase.
|
| TRIM(string1) | trim('BOTH',string1)
| Returns a string that removes whitespaces at both sides.
|
| REGEXP_REPLACE(string1, string2, string3) | regexpReplace(string1,
string2, string3) | Returns a string from STRING1 with all the substrings that
match a regular expression STRING2 consecutively being replaced with STRING3.
E.g., 'foobar'.regexpReplace('oo\|ar', '') returns "fb". |
+| REGEXP_EXTRACT(string, regex[, extractIndex]) | regexpExtract(string,
regex[, extractIndex]) | Returns the substring captured by the regular
expression group extractIndex. extractIndex defaults to 0, where 0 means the
whole match. Returns NULL for NULL input, no match, invalid regex, or invalid
group index. |
+| REGEXP_EXTRACT_ALL(string, regex[, extractIndex]) | regexpExtractAll(string,
regex[, extractIndex]) | Returns ARRAY<STRING> with all substrings
captured by group extractIndex. extractIndex defaults to 1, and 0 means the
whole match. Returns an empty array if there is no match, and NULL for NULL
input, invalid regex, or invalid group index. |
+| REGEXP_COUNT(string, regex) | regexpCount(string,
regex) | Returns the number of non-overlapping substrings that
match regex. Returns 0 if there is no match, and NULL for NULL input or invalid
regex. |
+| REGEXP_INSTR(string, regex) | regexpInstr(string,
regex) | Returns the 1-based position of the first substring
that matches regex. Returns 0 if there is no match, and NULL for NULL input or
invalid regex. |
+| REGEXP_SUBSTR(string, regex) | regexpSubstr(string,
regex) | Returns the first substring that matches regex. Returns
NULL for NULL input, no match, or invalid regex. |
| SUBSTR(string, integer1[, integer2]) |
substr(string,integer1,integer2) | Returns a substring of STRING
starting from position integer1 with length integer2 (to the end by default).
|
| SUBSTRING(string FROM integer1 [ FOR integer2 ]) |
substring(string,integer1,integer2) | Returns a substring of STRING
starting from position integer1 with length integer2 (to the end by default).
|
| CONCAT(string1, string2,…) | concat(string1,
string2,…) | Returns a string that concatenates string1, string2,
…. E.g., CONCAT('AA', 'BB', 'CC') returns 'AABBCC'.
|
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
index 7e29f63e7..fca030674 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
@@ -19,19 +19,25 @@ package org.apache.flink.cdc.runtime.functions.impl;
import org.apache.flink.cdc.common.types.variant.BinaryVariantInternalBuilder;
import org.apache.flink.cdc.common.types.variant.Variant;
+import org.apache.flink.cdc.common.utils.ThreadLocalCache;
import org.apache.calcite.runtime.SqlFunctions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
/** String built-in functions. */
public class StringFunctions {
private static final Logger LOG =
LoggerFactory.getLogger(StringFunctions.class);
+ private static final ThreadLocalCache<String, Pattern>
REGEXP_PATTERN_CACHE =
+ ThreadLocalCache.of(Pattern::compile);
public static int charLength(String str) {
return str.length();
@@ -62,6 +68,72 @@ public class StringFunctions {
}
}
+ /** Returns a string extracted with a specified regular expression and
capture group index. */
+ public static String regexpExtract(String str, String regex) {
+ return regexpExtract(str, regex, 0);
+ }
+
+ public static String regexpExtract(String str, String regex, Number
extractIndex) {
+ if (extractIndex == null || extractIndex.longValue() < 0) {
+ return null;
+ }
+ Matcher matcher = getRegexpMatcher(str, regex);
+ if (matcher == null || matcher.groupCount() <
extractIndex.longValue()) {
+ return null;
+ }
+ return matcher.find() ? matcher.group(extractIndex.intValue()) : null;
+ }
+
+ /** Returns all strings extracted with a specified regular expression. */
+ public static List<String> regexpExtractAll(String str, String regex) {
+ return regexpExtractAll(str, regex, 1);
+ }
+
+ public static List<String> regexpExtractAll(String str, String regex,
Number extractIndex) {
+ if (extractIndex == null || extractIndex.longValue() < 0) {
+ return null;
+ }
+ Matcher matcher = getRegexpMatcher(str, regex);
+ if (matcher == null || matcher.groupCount() <
extractIndex.longValue()) {
+ return null;
+ }
+
+ List<String> result = new ArrayList<>();
+ while (matcher.find()) {
+ result.add(matcher.group(extractIndex.intValue()));
+ }
+ return result;
+ }
+
+ /** Returns the number of non-overlapping matches of a regular expression.
*/
+ public static Integer regexpCount(String str, String regex) {
+ Matcher matcher = getRegexpMatcher(str, regex);
+ if (matcher == null) {
+ return null;
+ }
+
+ int count = 0;
+ while (matcher.find()) {
+ count++;
+ }
+ return count;
+ }
+
+ /** Returns the 1-based position of the first regular expression match. */
+ public static Integer regexpInstr(String str, String regex) {
+ Matcher matcher = getRegexpMatcher(str, regex);
+ if (matcher == null) {
+ return null;
+ }
+ return matcher.find() ? matcher.start() + 1 : 0;
+ }
+
+ /** Returns the first substring that matches a regular expression. */
+ public static String regexpSubstr(String str, String regex) {
+ Matcher matcher = getRegexpMatcher(str, regex);
+ return matcher != null && matcher.find() ? matcher.group(0) : null;
+ }
+
public static String concat(String... str) {
return String.join("", str);
}
@@ -216,4 +288,15 @@ public class StringFunctions {
String.format("Failed to parse json string: %s", jsonStr),
e);
}
}
+
+ private static Matcher getRegexpMatcher(String str, String regex) {
+ if (str == null || regex == null) {
+ return null;
+ }
+ try {
+ return REGEXP_PATTERN_CACHE.get(regex).matcher(str);
+ } catch (PatternSyntaxException e) {
+ return null;
+ }
+ }
}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
index 7a94e0586..a278799af 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
@@ -185,6 +185,60 @@ public class TransformSqlOperatorTable extends
ReflectiveSqlOperatorTable {
OperandTypes.family(
SqlTypeFamily.STRING, SqlTypeFamily.STRING,
SqlTypeFamily.STRING),
SqlFunctionCategory.STRING);
+ public static final SqlFunction REGEXP_EXTRACT =
+ new SqlFunction(
+ "REGEXP_EXTRACT",
+ SqlKind.OTHER_FUNCTION,
+ TransformSqlReturnTypes.VARCHAR_FORCE_NULLABLE,
+ null,
+ OperandTypes.or(
+ OperandTypes.family(SqlTypeFamily.CHARACTER,
SqlTypeFamily.CHARACTER),
+ OperandTypes.family(
+ SqlTypeFamily.CHARACTER,
+ SqlTypeFamily.CHARACTER,
+ SqlTypeFamily.INTEGER)),
+ SqlFunctionCategory.STRING);
+ public static final SqlFunction REGEXP_EXTRACT_ALL =
+ new SqlFunction(
+ "REGEXP_EXTRACT_ALL",
+ SqlKind.OTHER_FUNCTION,
+ TransformSqlReturnTypes.VARCHAR_ARRAY_FORCE_NULLABLE,
+ null,
+ OperandTypes.or(
+ OperandTypes.family(SqlTypeFamily.CHARACTER,
SqlTypeFamily.CHARACTER),
+ OperandTypes.family(
+ SqlTypeFamily.CHARACTER,
+ SqlTypeFamily.CHARACTER,
+ SqlTypeFamily.INTEGER)),
+ SqlFunctionCategory.STRING);
+ public static final SqlFunction REGEXP_COUNT =
+ new SqlFunction(
+ "REGEXP_COUNT",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.cascade(
+ ReturnTypes.explicit(SqlTypeName.INTEGER),
+ SqlTypeTransforms.FORCE_NULLABLE),
+ null,
+ OperandTypes.family(SqlTypeFamily.CHARACTER,
SqlTypeFamily.CHARACTER),
+ SqlFunctionCategory.STRING);
+ public static final SqlFunction REGEXP_INSTR =
+ new SqlFunction(
+ "REGEXP_INSTR",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.cascade(
+ ReturnTypes.explicit(SqlTypeName.INTEGER),
+ SqlTypeTransforms.FORCE_NULLABLE),
+ null,
+ OperandTypes.family(SqlTypeFamily.CHARACTER,
SqlTypeFamily.CHARACTER),
+ SqlFunctionCategory.STRING);
+ public static final SqlFunction REGEXP_SUBSTR =
+ new SqlFunction(
+ "REGEXP_SUBSTR",
+ SqlKind.OTHER_FUNCTION,
+ TransformSqlReturnTypes.VARCHAR_FORCE_NULLABLE,
+ null,
+ OperandTypes.family(SqlTypeFamily.CHARACTER,
SqlTypeFamily.CHARACTER),
+ SqlFunctionCategory.STRING);
public static final SqlFunction SUBSTR =
new SqlFunction(
"SUBSTR",
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java
index fb647fcd6..cd4681089 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java
@@ -111,6 +111,20 @@ public class TransformSqlReturnTypes {
ReturnTypes.cascade(
ReturnTypes.explicit(SqlTypeName.VARCHAR),
SqlTypeTransforms.FORCE_NULLABLE);
+ public static final SqlReturnTypeInference VARCHAR_ARRAY_FORCE_NULLABLE =
+ opBinding -> {
+ RelDataType elementType =
+ opBinding
+ .getTypeFactory()
+ .createTypeWithNullability(
+ opBinding
+ .getTypeFactory()
+
.createSqlType(SqlTypeName.VARCHAR),
+ true);
+ RelDataType arrayType =
opBinding.getTypeFactory().createArrayType(elementType, -1);
+ return
opBinding.getTypeFactory().createTypeWithNullability(arrayType, true);
+ };
+
public static final SqlReturnTypeInference VARCHAR_NOT_NULL =
ReturnTypes.cascade(
ReturnTypes.explicit(SqlTypeName.VARCHAR),
SqlTypeTransforms.TO_NOT_NULLABLE);
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
index 420b6713f..4e55db85b 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
@@ -44,6 +44,12 @@ class StringFunctionsTest {
assertThat(StringFunctions.like("Alice", "A%", null)).isNull();
}
+ @Test
+ void testRegexpReplaceKeepsLiteralReplacement() {
+ assertThat(StringFunctions.regexpReplace("a", "a",
"$1")).isEqualTo("$1");
+ assertThat(StringFunctions.regexpReplace("a", "a",
"\\")).isEqualTo("\\");
+ }
+
@Test
void testSimilarTo() {
assertThat(StringFunctions.similarTo("Alice", "(A|B)%")).isTrue();
@@ -57,4 +63,59 @@ class StringFunctionsTest {
assertThat(StringFunctions.similarTo("Alice", null)).isNull();
assertThat(StringFunctions.notSimilarTo(null, "(A|B)%")).isNull();
}
+
+ @Test
+ void testRegexpFunctionsNullArguments() {
+ assertThat(StringFunctions.regexpExtract(null, "a")).isNull();
+ assertThat(StringFunctions.regexpExtract("a", null)).isNull();
+ assertThat(StringFunctions.regexpExtractAll(null, "a")).isNull();
+ assertThat(StringFunctions.regexpExtractAll("a", null)).isNull();
+ assertThat(StringFunctions.regexpExtractAll("a", "a", null)).isNull();
+ assertThat(StringFunctions.regexpCount(null, "a")).isNull();
+ assertThat(StringFunctions.regexpCount("a", null)).isNull();
+ assertThat(StringFunctions.regexpInstr(null, "a")).isNull();
+ assertThat(StringFunctions.regexpInstr("a", null)).isNull();
+ assertThat(StringFunctions.regexpSubstr(null, "a")).isNull();
+ assertThat(StringFunctions.regexpSubstr("a", null)).isNull();
+ }
+
+ @Test
+ void testRegexpFunctions() {
+ assertThat(StringFunctions.regexpExtract("foothebar", "foo(.*?)(bar)",
2)).isEqualTo("bar");
+ assertThat(StringFunctions.regexpExtract("foothebar", "foo(.*?)(bar)"))
+ .isEqualTo("foothebar");
+ assertThat(StringFunctions.regexpExtract("foothebar", "foo(.*?)(bar)",
3)).isNull();
+ assertThat(StringFunctions.regexpExtract("foobar", "(foo)|(bar)",
2)).isNull();
+ assertThat(StringFunctions.regexpExtract("abcd", "z", 0)).isNull();
+ assertThat(StringFunctions.regexpExtract("abcd", "(", 0)).isNull();
+ assertThat(StringFunctions.regexpExtract("abcd", "a", -1)).isNull();
+
+ assertThat(StringFunctions.regexpExtractAll("100-200, 300-400",
"(\\d+)-(\\d+)"))
+ .containsExactly("100", "300");
+ assertThat(StringFunctions.regexpExtractAll("100-200, 300-400",
"(\\d+)-(\\d+)", 0))
+ .containsExactly("100-200", "300-400");
+ assertThat(StringFunctions.regexpExtractAll("100-200, 300-400",
"(\\d+)-(\\d+)", 2))
+ .containsExactly("200", "400");
+ assertThat(StringFunctions.regexpExtractAll("abcdeabde",
"(abcdeabde)|([a-z]*)", 2))
+ .containsExactly(null, "");
+ assertThat(StringFunctions.regexpExtractAll("100-200", "[a-z]",
0)).isEmpty();
+ assertThat(StringFunctions.regexpExtractAll("abcdeabde",
"abcdeabde")).isNull();
+ assertThat(StringFunctions.regexpExtractAll("abcdeabde",
"(abcdeabde)", 2)).isNull();
+ assertThat(StringFunctions.regexpExtractAll("abcdeabde", "(",
0)).isNull();
+ assertThat(StringFunctions.regexpExtractAll("abcdeabde",
"(abcdeabde)", -1)).isNull();
+
+ assertThat(StringFunctions.regexpCount("abc123xyz456",
"\\d")).isEqualTo(6);
+ assertThat(StringFunctions.regexpCount("abcd", "z")).isZero();
+ assertThat(StringFunctions.regexpCount("abcd", "(")).isNull();
+
+ assertThat(StringFunctions.regexpInstr("hello world! Hello everyone!",
"Hello"))
+ .isEqualTo(14);
+ assertThat(StringFunctions.regexpInstr("abcd", "z")).isZero();
+ assertThat(StringFunctions.regexpInstr("abcd", "(")).isNull();
+
+ assertThat(StringFunctions.regexpSubstr("100-200, 300-400",
"(\\d+)-(\\d+)"))
+ .isEqualTo("100-200");
+ assertThat(StringFunctions.regexpSubstr("abcd", "z")).isNull();
+ assertThat(StringFunctions.regexpSubstr("abcd", "(")).isNull();
+ }
}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
index 57563b280..fa2277a99 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
@@ -19,6 +19,7 @@ package org.apache.flink.cdc.runtime.operators.transform;
import org.apache.flink.cdc.common.data.DateData;
import org.apache.flink.cdc.common.data.DecimalData;
+import org.apache.flink.cdc.common.data.GenericArrayData;
import org.apache.flink.cdc.common.data.LocalZonedTimestampData;
import org.apache.flink.cdc.common.data.TimeData;
import org.apache.flink.cdc.common.data.TimestampData;
@@ -468,6 +469,58 @@ class PostTransformOperatorTest {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testRegexpExtractAllProjection() throws Exception {
+ TableId tableId = TableId.tableId("my_company", "my_branch",
"regexp_table");
+ Schema inputSchema =
+ Schema.newBuilder().physicalColumn("text_value",
DataTypes.STRING()).build();
+ Schema outputSchema =
+ Schema.newBuilder()
+ .physicalColumn("regexp_values",
DataTypes.ARRAY(DataTypes.STRING()))
+ .build();
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ tableId.identifier(),
+ "REGEXP_EXTRACT_ALL(text_value,
'([0-9]+)-([0-9]+)') AS regexp_values",
+ null)
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event> harness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+
+ harness.open();
+ harness.getOperator()
+ .processElement(new StreamRecord<>(new
CreateTableEvent(tableId, inputSchema)));
+ Assertions.assertThat(harness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(new CreateTableEvent(tableId,
outputSchema)));
+
+ BinaryRecordDataGenerator inputGenerator =
+ new BinaryRecordDataGenerator((RowType)
inputSchema.toRowDataType());
+ BinaryRecordDataGenerator outputGenerator =
+ new BinaryRecordDataGenerator((RowType)
outputSchema.toRowDataType());
+ DataChangeEvent inputEvent =
+ DataChangeEvent.insertEvent(
+ tableId,
+ inputGenerator.generate(
+ new Object[]
{BinaryStringData.fromString("100-200, 300-400")}));
+ DataChangeEvent expectedEvent =
+ DataChangeEvent.insertEvent(
+ tableId,
+ outputGenerator.generate(
+ new Object[] {
+ new GenericArrayData(
+ new Object[] {
+
BinaryStringData.fromString("100"),
+
BinaryStringData.fromString("300")
+ })
+ }));
+
+ harness.getOperator().processElement(new StreamRecord<>(inputEvent));
+ Assertions.assertThat(harness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(expectedEvent));
+ harness.close();
+ }
+
@Test
void testDataChangeEventTransformProjectionDataTypeConvert() throws
Exception {
PostTransformOperator transform =
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
index 9943ed069..519fe2669 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
@@ -196,6 +196,27 @@ class TransformParserTest {
testFilterExpression("trim(id)", "trim(\"BOTH\", \" \", id)");
testFilterExpression(
"REGEXP_REPLACE(id, '[a-zA-Z]', '')", "regexpReplace(id,
\"[a-zA-Z]\", \"\")");
+ testFilterExpression(
+ "REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 2)",
+ "regexpExtract(\"foothebar\", \"foo(.*?)(bar)\", 2)");
+ testFilterExpression(
+ "REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)')",
+ "regexpExtract(\"foothebar\", \"foo(.*?)(bar)\")");
+ testFilterExpression(
+ "REGEXP_EXTRACT_ALL('100-200, 300-400', '([0-9]+)-([0-9]+)',
2)",
+ "regexpExtractAll(\"100-200, 300-400\", \"([0-9]+)-([0-9]+)\",
2)");
+ testFilterExpression(
+ "REGEXP_EXTRACT_ALL('100-200, 300-400', '([0-9]+)-([0-9]+)')",
+ "regexpExtractAll(\"100-200, 300-400\",
\"([0-9]+)-([0-9]+)\")");
+ testFilterExpression(
+ "REGEXP_COUNT('abc123xyz456', '[0-9]')",
+ "regexpCount(\"abc123xyz456\", \"[0-9]\")");
+ testFilterExpression(
+ "REGEXP_INSTR('abc123xyz456', '[0-9]')",
+ "regexpInstr(\"abc123xyz456\", \"[0-9]\")");
+ testFilterExpression(
+ "REGEXP_SUBSTR('100-200, 300-400', '([0-9]+)-([0-9]+)')",
+ "regexpSubstr(\"100-200, 300-400\", \"([0-9]+)-([0-9]+)\")");
testFilterExpression("upper(id)", "upper(id)");
testFilterExpression("lower(id)", "lower(id)");
testFilterExpression("concat(a,b)", "concat(a, b)");
@@ -570,6 +591,18 @@ class TransformParserTest {
"ProjectionColumn{column=`bmi` DOUBLE,
expression='`TB`.`weight` / (`TB`.`height` * `TB`.`height`)',
scriptExpression='$0 / $1 * $1', originalColumnNames=[weight, height, height],
columnNameMap={weight=$0, height=$1}}");
Assertions.assertThat(result).hasToString("[" + String.join(", ",
expected) + "]");
+ List<ProjectionColumn> regexpResult =
+ TransformParser.generateProjectionColumns(
+ "REGEXP_EXTRACT_ALL(name, '([0-9]+)-([0-9]+)') AS
regexp_values",
+ testColumns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]);
+ Assertions.assertThat(regexpResult).hasSize(1);
+ Assertions.assertThat(regexpResult.get(0).getDataType())
+ .isEqualTo(DataTypes.ARRAY(DataTypes.STRING()));
+ Assertions.assertThat(regexpResult.get(0).getScriptExpression())
+ .isEqualTo("regexpExtractAll($0, \"([0-9]+)-([0-9]+)\")");
+
List<ProjectionColumn> metadataResult =
TransformParser.generateProjectionColumns(
"*, __namespace_name__, __schema_name__,
__table_name__, __data_event_type__ AS op_type",
