Skip to content

[FLINK-40239][runtime] Support additional Flink SQL regular expression functions in YAML Transform #4491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
haruki-830 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[FLINK-40239][runtime] Support additional Flink SQL regular expression functions in YAML Transform #4491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions docs/content.zh/docs/core-concept/transform.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'。 |
Expand Down
5 changes: 5 additions & 0 deletions docs/content/docs/core-concept/transform.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'. |
Expand Down
80 changes: 80 additions & 0 deletions ...dc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,12 @@
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 {
Expand Down Expand Up @@ -62,6 +65,72 @@ public static String regexpReplace(String str, String regex, String replacement)
}
}

/** 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);

@lvyanquan lvyanquan Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

getRegexpMatcher() calls Pattern.compile(regex) for every record. Since regexes in YAML transforms are usually constant, this adds unnecessary CPU overhead to the per-record hot path. Please use a bounded ThreadLocalCache<String, Pattern> while preserving the current invalid-regex behavior.

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);
}
Expand Down Expand Up @@ -216,4 +285,15 @@ public static Variant parseJson(String jsonStr, boolean allowDuplicateKeys) {
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 Pattern.compile(regex).matcher(str);
} catch (PatternSyntaxException e) {
return null;
}
}
}
54 changes: 54 additions & 0 deletions ...src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,60 @@ public void lookupOperatorOverloads(
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",
Expand Down
14 changes: 14 additions & 0 deletions ...e/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,20 @@ public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
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);
Expand Down
61 changes: 61 additions & 0 deletions ...untime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/StringFunctionsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,12 @@ void testLikeEscapeNullReturnsUnknown() {
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();
Expand All @@ -57,4 +63,59 @@ void testSimilarToNullReturnsUnknown() {
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();
}
}
Loading
Loading