You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2024/08/21 14:11:20 UTC
(flink) branch master updated (9399e669e5a -> e8c1d1b9d41)
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 9399e669e5a [hotfix][API/DataStream] Log the right syncDurationMillis time
new 697b4b26010 [FLINK-35931][table-planner] Use a separate test class for the regexp family of functions
new 7fa4ebad057 [FLINK-35931][table] Add the built-in function REGEXP_EXTRACT_ALL
new 8189bba1953 [FLINK-35932][table] Add the built-in function REGEXP_COUNT
new cbbb1cd178a [FLINK-35962][table] Add the built-in function REGEXP_INSTR
new e8c1d1b9d41 [FLINK-35963][table] Add the built-in function REGEXP_SUBSTR
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/data/sql_functions.yml | 49 ++-
docs/data/sql_functions_zh.yml | 50 ++-
.../docs/reference/pyflink.table/expressions.rst | 4 +
flink-python/pyflink/table/expression.py | 50 +++
.../pyflink/table/tests/test_expression.py | 13 +-
.../flink/table/api/internal/BaseExpressions.java | 70 ++++
.../functions/BuiltInFunctionDefinitions.java | 68 ++++
.../planner/functions/RegexpFunctionsITCase.java | 390 +++++++++++++++++++++
.../planner/functions/StringFunctionsITCase.java | 25 +-
.../table/runtime/functions/SqlFunctionUtils.java | 19 +
...ncodeFunction.java => RegexpCountFunction.java} | 30 +-
...Function.java => RegexpExtractAllFunction.java} | 58 +--
...UnhexFunction.java => RegexpInstrFunction.java} | 20 +-
...TrimFunction.java => RegexpSubstrFunction.java} | 30 +-
14 files changed, 769 insertions(+), 107 deletions(-)
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
copy flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/{UrlEncodeFunction.java => RegexpCountFunction.java} (61%)
copy flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/{SplitFunction.java => RegexpExtractAllFunction.java} (50%)
copy flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/{UnhexFunction.java => RegexpInstrFunction.java} (65%)
copy flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/{BTrimFunction.java => RegexpSubstrFunction.java} (63%)
(flink) 04/05: [FLINK-35962][table] Add the built-in function REGEXP_INSTR
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit cbbb1cd178a5b53200ee793f77bac73a0d49ff14
Author: dylanhz <dy...@163.com>
AuthorDate: Fri Aug 16 14:02:40 2024 +0800
[FLINK-35962][table] Add the built-in function REGEXP_INSTR
---
docs/data/sql_functions.yml | 10 +++
docs/data/sql_functions_zh.yml | 10 +++
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 11 ++++
.../pyflink/table/tests/test_expression.py | 1 +
.../flink/table/api/internal/BaseExpressions.java | 14 ++++
.../functions/BuiltInFunctionDefinitions.java | 15 +++++
.../planner/functions/RegexpFunctionsITCase.java | 76 +++++++++++++++++++++-
.../functions/scalar/RegexpInstrFunction.java | 55 ++++++++++++++++
9 files changed, 192 insertions(+), 1 deletion(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 29b0c27ebcd..e6f768ca4c4 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -366,6 +366,16 @@ string:
`str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>, extractIndex <TINYINT | SMALLINT | INTEGER | BIGINT>`
Returns an `ARRAY<STRING>` representation of all the matched substrings. `NULL` if any of the arguments are `NULL` or invalid.
+ - sql: REGEXP_INSTR(str, regex)
+ table: str.regexpInstr(regex)
+ description: |
+ Returns the position of the first substring in str that matches regex.
+
+ Result indexes begin at 1, 0 if there is no match.
+
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
+
+ Returns an `INTEGER` representation of the first matched substring index. `NULL` if any of the arguments are `NULL` or regex is invalid.
- sql: TRANSLATE(expr, fromStr, toStr)
table: expr.translate(fromStr, toStr)
description: |
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index 76607687f17..e50040608c8 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -433,6 +433,16 @@ string:
`str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>, extractIndex <TINYINT | SMALLINT | INTEGER | BIGINT>`
返回一个 `ARRAY<STRING>`，表示所有匹配的子串。如果任何参数为 `NULL`或非法，则返回 `NULL`。
+ - sql: REGEXP_INSTR(str, regex)
+ table: str.regexpInstr(regex)
+ description: |
+ 返回 str 中第一个匹配 regex 的子字符串的索引。
+
+ 结果索引从 1 开始，如果匹配失败则返回 0。
+
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
+
+ 返回一个 `INTEGER` 表示 str 中第一个匹配 regex 的子字符串索引。如果任何参数为 `NULL` 或 regex 非法，则返回 `NULL`。
- sql: TRANSLATE(expr, fromStr, toStr)
table: expr.translate(fromStr, toStr)
description: |
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
index 8698e884e1a..23796fbaca3 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -182,6 +182,7 @@ string functions
Expression.regexp_replace
Expression.regexp_extract
Expression.regexp_extract_all
+ Expression.regexp_instr
Expression.from_base64
Expression.to_base64
Expression.ascii
diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
index 5975a79502b..6eabd528e02 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1254,6 +1254,17 @@ class Expression(Generic[T]):
else:
return _ternary_op("regexpExtractAll")(self, regex, extract_index)
+ def regexp_instr(self, regex) -> 'Expression':
+ """
+ Returns the position of the first substring in str that matches regex.
+ Result indexes begin at 1, 0 if there is no match.
+ null if any of the arguments are null or regex is invalid.
+
+ :param regex: A STRING expression with a matching pattern.
+ :return: An INTEGER representation of the first matched substring index.
+ """
+ return _binary_op("regexpInstr")(self, regex)
+
@property
def from_base64(self) -> 'Expression[str]':
"""
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index fb94a02ec7d..3aa101b685e 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -180,6 +180,7 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
self.assertEqual('REGEXP_EXTRACT_ALL(a, b)', str(expr1.regexp_extract_all(expr2)))
self.assertEqual('REGEXP_EXTRACT_ALL(a, b, 3)', str(expr1.regexp_extract_all(expr2, 3)))
self.assertEqual("regexpReplace(a, b, 'abc')", str(expr1.regexp_replace(expr2, 'abc')))
+ self.assertEqual("REGEXP_INSTR(a, b)", str(expr1.regexp_instr(expr2)))
# temporal functions
self.assertEqual('cast(a, DATE)', str(expr1.to_date))
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 5092937e7df..8c2f9f1a073 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -163,6 +163,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_COUNT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_EXTRACT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_EXTRACT_ALL;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_INSTR;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_REPLACE;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REPEAT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REPLACE;
@@ -1208,6 +1209,19 @@ public abstract class BaseExpressions<InType, OutType> {
unresolvedCall(REGEXP_EXTRACT_ALL, toExpr(), objectToExpression(regex)));
}
+ /**
+ * Returns the position of the first substring in {@code str} that matches {@code regex}. <br>
+ * Result indexes begin at 1, 0 if there is no match. <br>
+ *
+ * @param regex A STRING expression with a matching pattern.
+ * @return An INTEGER representation of the first matched substring index. <br>
+ * null if any of the arguments are null or {@code regex} is invalid.
+ */
+ public OutType regexpInstr(InType regex) {
+ return toApiSpecificExpression(
+ unresolvedCall(REGEXP_INSTR, toExpr(), objectToExpression(regex)));
+ }
+
/**
* Returns a string by quotes a string as a JSON value and wrapping it with double quote
* characters.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index e726051a4c4..19aabcc6134 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -1164,6 +1164,21 @@ public final class BuiltInFunctionDefinitions {
"org.apache.flink.table.runtime.functions.scalar.RegexpExtractAllFunction")
.build();
+ public static final BuiltInFunctionDefinition REGEXP_INSTR =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("REGEXP_INSTR")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Arrays.asList("str", "regex"),
+ Arrays.asList(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.CHARACTER_STRING))))
+ .outputTypeStrategy(explicit(DataTypes.INT()))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.RegexpInstrFunction")
+ .build();
+
public static final BuiltInFunctionDefinition JSON_QUOTE =
BuiltInFunctionDefinition.newBuilder()
.name("JSON_QUOTE")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
index 3809f98637f..24ec1479a30 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
@@ -35,7 +35,8 @@ class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
return Stream.of(
regexpCountTestCases(),
regexpExtractTestCases(),
- regexpExtractAllTestCases())
+ regexpExtractAllTestCases(),
+ regexpInstrTestCases())
.flatMap(s -> s);
}
@@ -238,4 +239,77 @@ class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
+ "REGEXP_EXTRACT_ALL(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)\n"
+ "REGEXP_EXTRACT_ALL(str <CHARACTER_STRING>, regex <CHARACTER_STRING>, extractIndex <INTEGER_NUMERIC>)"));
}
+
+ private Stream<TestSetSpec> regexpInstrTestCases() {
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.REGEXP_INSTR)
+ .onFieldsWithData(null, "abcdeabde", "100-200, 300-400")
+ .andDataTypes(DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING())
+ // null input
+ .testResult(
+ $("f0").regexpInstr($("f1")),
+ "REGEXP_INSTR(f0, f1)",
+ null,
+ DataTypes.INT())
+ .testResult(
+ $("f1").regexpInstr($("f0")),
+ "REGEXP_INSTR(f1, f0)",
+ null,
+ DataTypes.INT())
+ // invalid regexp
+ .testResult(
+ $("f1").regexpInstr("("),
+ "REGEXP_INSTR(f1, '(')",
+ null,
+ DataTypes.INT())
+ // not found
+ .testResult(
+ $("f2").regexpInstr("[a-z]"),
+ "REGEXP_INSTR(f2, '[a-z]')",
+ 0,
+ DataTypes.INT())
+ // border chars
+ .testResult(
+ lit("Helloworld! Hello everyone!").regexpInstr("\\bHello\\b"),
+ "REGEXP_INSTR('Helloworld! Hello everyone!', '\\bHello\\b')",
+ 13,
+ DataTypes.INT())
+ .testResult(
+ lit("Helloworld! Hello everyone!").regexpInstr("\\bHello\\b"),
+ "REGEXP_INSTR('Helloworld! Hello everyone!', '\\bHello\\b')",
+ 14,
+ DataTypes.INT())
+ // normal cases
+ .testResult(
+ lit("hello world! Hello everyone!").regexpInstr("Hello"),
+ "REGEXP_INSTR('hello world! Hello everyone!', 'Hello')",
+ 14,
+ DataTypes.INT())
+ .testResult(
+ lit("a.b.c.d").regexpInstr("\\."),
+ "REGEXP_INSTR('a.b.c.d', '\\.')",
+ 2,
+ DataTypes.INT())
+ .testResult(
+ lit("abc123xyz456").regexpInstr("\\d"),
+ "REGEXP_INSTR('abc123xyz456', '\\d')",
+ 4,
+ DataTypes.INT())
+ .testResult(
+ $("f2").regexpInstr("(\\d+)-(\\d+)"),
+ "REGEXP_INSTR(f2, '(\\d+)-(\\d+)')",
+ 1,
+ DataTypes.INT()),
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.REGEXP_INSTR, "Validation Error")
+ .onFieldsWithData(1024)
+ .andDataTypes(DataTypes.INT())
+ .testTableApiValidationError(
+ $("f0").regexpInstr("1024"),
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_INSTR(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)")
+ .testSqlValidationError(
+ "REGEXP_INSTR(f0, '1024')",
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_INSTR(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)"));
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java
new file mode 100644
index 00000000000..2a233c042b1
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java
@@ -0,0 +1,55 @@
+/*
+ * 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.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
+
+import javax.annotation.Nullable;
+
+import java.util.regex.Matcher;
+import java.util.regex.PatternSyntaxException;
+
+import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.REGEXP_PATTERN_CACHE;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_INSTR}. */
+@Internal
+public class RegexpInstrFunction extends BuiltInScalarFunction {
+
+ public RegexpInstrFunction(SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.REGEXP_INSTR, context);
+ }
+
+ public @Nullable Integer eval(@Nullable StringData str, @Nullable StringData regex) {
+ if (str == null || regex == null) {
+ return null;
+ }
+
+ Matcher matcher;
+ try {
+ matcher = REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
+ } catch (PatternSyntaxException e) {
+ return null;
+ }
+
+ return matcher.find() ? matcher.start() + 1 : 0;
+ }
+}
(flink) 05/05: [FLINK-35963][table] Add the built-in function REGEXP_SUBSTR
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e8c1d1b9d411e17130a57851c0bbaecb2e816048
Author: dylanhz <dy...@163.com>
AuthorDate: Fri Aug 16 14:19:57 2024 +0800
[FLINK-35963][table] Add the built-in function REGEXP_SUBSTR
---
docs/data/sql_functions.yml | 8 +++
docs/data/sql_functions_zh.yml | 8 +++
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 10 +++
.../pyflink/table/tests/test_expression.py | 1 +
.../flink/table/api/internal/BaseExpressions.java | 14 ++++
.../functions/BuiltInFunctionDefinitions.java | 15 +++++
.../planner/functions/RegexpFunctionsITCase.java | 77 +++++++++++++++++++++-
.../table/runtime/functions/SqlFunctionUtils.java | 21 +++++-
.../functions/scalar/RegexpCountFunction.java | 14 +---
.../functions/scalar/RegexpExtractAllFunction.java | 16 ++---
.../functions/scalar/RegexpInstrFunction.java | 14 +---
...nstrFunction.java => RegexpSubstrFunction.java} | 30 ++++-----
13 files changed, 176 insertions(+), 53 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index e6f768ca4c4..08f0b860238 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -376,6 +376,14 @@ string:
`str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
Returns an `INTEGER` representation of the first matched substring index. `NULL` if any of the arguments are `NULL` or regex is invalid.
+ - sql: REGEXP_SUBSTR(str, regex)
+ table: str.regexpSubstr(regex)
+ description: |
+ Returns the first substring in str that matches regex.
+
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
+
+ Returns an `STRING` representation of the first matched substring. `NULL` if any of the arguments are `NULL` or regex if invalid or pattern is not found.
- sql: TRANSLATE(expr, fromStr, toStr)
table: expr.translate(fromStr, toStr)
description: |
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index e50040608c8..c7c332ed007 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -443,6 +443,14 @@ string:
`str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
返回一个 `INTEGER` 表示 str 中第一个匹配 regex 的子字符串索引。如果任何参数为 `NULL` 或 regex 非法，则返回 `NULL`。
+ - sql: REGEXP_SUBSTR(str, regex)
+ table: str.regexpSubStr(regex)
+ description: |
+ 返回 str 中第一个匹配 regex 的子字符串。
+
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
+
+ 返回一个 `STRING` 表示 str 中第一个匹配 regex 的子字符串。如果任何参数为 `NULL` 或 regex 非法或匹配失败，则返回 `NULL`。
- sql: TRANSLATE(expr, fromStr, toStr)
table: expr.translate(fromStr, toStr)
description: |
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
index 23796fbaca3..3f494f18438 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -183,6 +183,7 @@ string functions
Expression.regexp_extract
Expression.regexp_extract_all
Expression.regexp_instr
+ Expression.regexp_substr
Expression.from_base64
Expression.to_base64
Expression.ascii
diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
index 6eabd528e02..54806af9498 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1265,6 +1265,16 @@ class Expression(Generic[T]):
"""
return _binary_op("regexpInstr")(self, regex)
+ def regexp_substr(self, regex) -> 'Expression':
+ """
+ Returns the first substring in str that matches regex.
+ null if any of the arguments are null or regex is invalid or pattern is not found.
+
+ :param regex: A STRING expression with a matching pattern.
+ :return: A STRING representation of the first matched substring.
+ """
+ return _binary_op("regexpSubstr")(self, regex)
+
@property
def from_base64(self) -> 'Expression[str]':
"""
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index 3aa101b685e..9ae5e22dc5e 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -181,6 +181,7 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
self.assertEqual('REGEXP_EXTRACT_ALL(a, b, 3)', str(expr1.regexp_extract_all(expr2, 3)))
self.assertEqual("regexpReplace(a, b, 'abc')", str(expr1.regexp_replace(expr2, 'abc')))
self.assertEqual("REGEXP_INSTR(a, b)", str(expr1.regexp_instr(expr2)))
+ self.assertEqual("REGEXP_SUBSTR(a, b)", str(expr1.regexp_substr(expr2)))
# temporal functions
self.assertEqual('cast(a, DATE)', str(expr1.to_date))
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 8c2f9f1a073..3a5e5397378 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -165,6 +165,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_EXTRACT_ALL;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_INSTR;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_REPLACE;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_SUBSTR;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REPEAT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REPLACE;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REVERSE;
@@ -1222,6 +1223,19 @@ public abstract class BaseExpressions<InType, OutType> {
unresolvedCall(REGEXP_INSTR, toExpr(), objectToExpression(regex)));
}
+ /**
+ * Returns the first substring in {@code str} that matches {@code regex}.
+ *
+ * @param regex A STRING expression with a matching pattern.
+ * @return A STRING representation of the first matched substring. <br>
+ * null if any of the arguments are null or {@code regex} is invalid or pattern is not
+ * found.
+ */
+ public OutType regexpSubstr(InType regex) {
+ return toApiSpecificExpression(
+ unresolvedCall(REGEXP_SUBSTR, toExpr(), objectToExpression(regex)));
+ }
+
/**
* Returns a string by quotes a string as a JSON value and wrapping it with double quote
* characters.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 19aabcc6134..be68d1dd7f0 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -1179,6 +1179,21 @@ public final class BuiltInFunctionDefinitions {
"org.apache.flink.table.runtime.functions.scalar.RegexpInstrFunction")
.build();
+ public static final BuiltInFunctionDefinition REGEXP_SUBSTR =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("REGEXP_SUBSTR")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Arrays.asList("str", "regex"),
+ Arrays.asList(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.CHARACTER_STRING))))
+ .outputTypeStrategy(explicit(DataTypes.STRING()))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.RegexpSubstrFunction")
+ .build();
+
public static final BuiltInFunctionDefinition JSON_QUOTE =
BuiltInFunctionDefinition.newBuilder()
.name("JSON_QUOTE")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
index 24ec1479a30..c5a63826c7e 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
@@ -36,7 +36,8 @@ class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
regexpCountTestCases(),
regexpExtractTestCases(),
regexpExtractAllTestCases(),
- regexpInstrTestCases())
+ regexpInstrTestCases(),
+ regexpSubstrTestCases())
.flatMap(s -> s);
}
@@ -312,4 +313,78 @@ class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
"Invalid input arguments. Expected signatures are:\n"
+ "REGEXP_INSTR(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)"));
}
+
+ private Stream<TestSetSpec> regexpSubstrTestCases() {
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.REGEXP_SUBSTR)
+ .onFieldsWithData(null, "abcdeabde", "100-200, 300-400")
+ .andDataTypes(DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING())
+ // null input
+ .testResult(
+ $("f0").regexpSubstr($("f1")),
+ "REGEXP_SUBSTR(f0, f1)",
+ null,
+ DataTypes.STRING())
+ .testResult(
+ $("f1").regexpSubstr($("f0")),
+ "REGEXP_SUBSTR(f1, f0)",
+ null,
+ DataTypes.STRING())
+ // invalid regexp
+ .testResult(
+ $("f1").regexpSubstr("("),
+ "REGEXP_SUBSTR(f1, '(')",
+ null,
+ DataTypes.STRING())
+ // not found
+ .testResult(
+ $("f2").regexpSubstr("[a-z]"),
+ "REGEXP_SUBSTR(f2, '[a-z]')",
+ null,
+ DataTypes.STRING())
+ // border chars
+ .testResult(
+ lit("Helloworld! Hello everyone!").regexpSubstr("\\bHello\\b"),
+ "REGEXP_SUBSTR('Helloworld! Hello everyone!', '\\bHello\\b')",
+ "Hello",
+ DataTypes.STRING())
+ .testResult(
+ $("f2").regexpSubstr("(\\d+)-(\\d+)$"),
+ "REGEXP_SUBSTR(f2, '(\\d+)-(\\d+)$')",
+ "300-400",
+ DataTypes.STRING())
+ // normal cases
+ .testResult(
+ lit("hello world! Hello everyone!").regexpSubstr("Hello"),
+ "REGEXP_SUBSTR('hello world! Hello everyone!', 'Hello')",
+ "Hello",
+ DataTypes.STRING())
+ .testResult(
+ lit("a.b.c.d").regexpSubstr("\\."),
+ "REGEXP_SUBSTR('a.b.c.d', '\\.')",
+ ".",
+ DataTypes.STRING())
+ .testResult(
+ lit("abc123xyz456").regexpSubstr("\\d"),
+ "REGEXP_SUBSTR('abc123xyz456', '\\d')",
+ "1",
+ DataTypes.STRING())
+ .testResult(
+ $("f2").regexpSubstr("(\\d+)-(\\d+)"),
+ "REGEXP_SUBSTR(f2, '(\\d+)-(\\d+)')",
+ "100-200",
+ DataTypes.STRING()),
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.REGEXP_SUBSTR, "Validation Error")
+ .onFieldsWithData(1024)
+ .andDataTypes(DataTypes.INT())
+ .testTableApiValidationError(
+ $("f0").regexpSubstr("1024"),
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_SUBSTR(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)")
+ .testSqlValidationError(
+ "REGEXP_SUBSTR(f0, '1024')",
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_SUBSTR(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)"));
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
index d51dd9f1cb9..0fdd5213613 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.functions;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.DecimalDataUtils;
+import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.binary.BinaryStringData;
import org.apache.flink.table.data.binary.BinaryStringDataUtil;
import org.apache.flink.table.utils.EncodingUtils;
@@ -30,6 +31,8 @@ import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.math.BigInteger;
@@ -46,6 +49,7 @@ import java.util.UUID;
import java.util.regex.MatchResult;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
import static org.apache.flink.table.data.DecimalDataUtils.castFrom;
import static org.apache.flink.table.data.DecimalDataUtils.castToIntegral;
@@ -64,7 +68,7 @@ public class SqlFunctionUtils {
private static final Logger LOG = LoggerFactory.getLogger(SqlFunctionUtils.class);
- public static final ThreadLocalCache<String, Pattern> REGEXP_PATTERN_CACHE =
+ private static final ThreadLocalCache<String, Pattern> REGEXP_PATTERN_CACHE =
new ThreadLocalCache<String, Pattern>() {
@Override
public Pattern getNewInstance(String regex) {
@@ -472,6 +476,21 @@ public class SqlFunctionUtils {
return regexpExtract(str, regex, 0);
}
+ /**
+ * Returns a Matcher object that represents the result of matching given StringData against a
+ * specified regular expression pattern.
+ */
+ public static Matcher getRegexpMatcher(@Nullable StringData str, @Nullable StringData regex) {
+ if (str == null || regex == null) {
+ return null;
+ }
+ try {
+ return REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
+ } catch (PatternSyntaxException e) {
+ return null;
+ }
+ }
+
/**
* Parse string as key-value string and return the value matches key name. example:
* keyvalue('k1=v1;k2=v2', ';', '=', 'k2') = 'v2' keyvalue('k1:v1,k2:v2', ',', ':', 'k3') = NULL
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpCountFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpCountFunction.java
index 48f9cdb731e..378dda4b1c9 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpCountFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpCountFunction.java
@@ -26,9 +26,8 @@ import org.apache.flink.table.functions.SpecializedFunction;
import javax.annotation.Nullable;
import java.util.regex.Matcher;
-import java.util.regex.PatternSyntaxException;
-import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.REGEXP_PATTERN_CACHE;
+import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.getRegexpMatcher;
/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_COUNT}. */
@Internal
@@ -39,14 +38,8 @@ public class RegexpCountFunction extends BuiltInScalarFunction {
}
public @Nullable Integer eval(@Nullable StringData str, @Nullable StringData regex) {
- if (str == null || regex == null) {
- return null;
- }
-
- Matcher matcher;
- try {
- matcher = REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
- } catch (PatternSyntaxException e) {
+ Matcher matcher = getRegexpMatcher(str, regex);
+ if (matcher == null) {
return null;
}
@@ -54,7 +47,6 @@ public class RegexpCountFunction extends BuiltInScalarFunction {
while (matcher.find()) {
count++;
}
-
return count;
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpExtractAllFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpExtractAllFunction.java
index df4ef721d1a..375387a71da 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpExtractAllFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpExtractAllFunction.java
@@ -31,9 +31,8 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
-import java.util.regex.PatternSyntaxException;
-import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.REGEXP_PATTERN_CACHE;
+import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.getRegexpMatcher;
/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_EXTRACT_ALL}. */
@Internal
@@ -49,25 +48,22 @@ public class RegexpExtractAllFunction extends BuiltInScalarFunction {
public @Nullable ArrayData eval(
@Nullable StringData str, @Nullable StringData regex, @Nullable Number extractIndex) {
- if (str == null || regex == null || extractIndex == null) {
+ if (extractIndex == null || extractIndex.longValue() < 0) {
return null;
}
- Matcher matcher;
- try {
- matcher = REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
- } catch (PatternSyntaxException e) {
+ Matcher matcher = getRegexpMatcher(str, regex);
+ if (matcher == null) {
return null;
}
- long groupIndex = extractIndex.longValue();
- if (groupIndex < 0 || matcher.groupCount() < groupIndex) {
+ if (matcher.groupCount() < extractIndex.longValue()) {
return null;
}
List<StringData> list = new ArrayList<>();
while (matcher.find()) {
- list.add(BinaryStringData.fromString(matcher.group((int) groupIndex)));
+ list.add(BinaryStringData.fromString(matcher.group(extractIndex.intValue())));
}
return new GenericArrayData(list.toArray());
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java
index 2a233c042b1..cf4dbe50005 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java
@@ -26,9 +26,8 @@ import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
import javax.annotation.Nullable;
import java.util.regex.Matcher;
-import java.util.regex.PatternSyntaxException;
-import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.REGEXP_PATTERN_CACHE;
+import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.getRegexpMatcher;
/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_INSTR}. */
@Internal
@@ -39,17 +38,10 @@ public class RegexpInstrFunction extends BuiltInScalarFunction {
}
public @Nullable Integer eval(@Nullable StringData str, @Nullable StringData regex) {
- if (str == null || regex == null) {
+ Matcher matcher = getRegexpMatcher(str, regex);
+ if (matcher == null) {
return null;
}
-
- Matcher matcher;
- try {
- matcher = REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
- } catch (PatternSyntaxException e) {
- return null;
- }
-
return matcher.find() ? matcher.start() + 1 : 0;
}
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpSubstrFunction.java
similarity index 61%
copy from flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java
copy to flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpSubstrFunction.java
index 2a233c042b1..86b833de443 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpInstrFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpSubstrFunction.java
@@ -20,36 +20,28 @@ package org.apache.flink.table.runtime.functions.scalar;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.SpecializedFunction.SpecializedContext;
import javax.annotation.Nullable;
import java.util.regex.Matcher;
-import java.util.regex.PatternSyntaxException;
-import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.REGEXP_PATTERN_CACHE;
+import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.getRegexpMatcher;
-/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_INSTR}. */
+/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_SUBSTR}. */
@Internal
-public class RegexpInstrFunction extends BuiltInScalarFunction {
+public class RegexpSubstrFunction extends BuiltInScalarFunction {
- public RegexpInstrFunction(SpecializedContext context) {
- super(BuiltInFunctionDefinitions.REGEXP_INSTR, context);
+ public RegexpSubstrFunction(SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.REGEXP_SUBSTR, context);
}
- public @Nullable Integer eval(@Nullable StringData str, @Nullable StringData regex) {
- if (str == null || regex == null) {
- return null;
- }
-
- Matcher matcher;
- try {
- matcher = REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
- } catch (PatternSyntaxException e) {
- return null;
- }
-
- return matcher.find() ? matcher.start() + 1 : 0;
+ public @Nullable StringData eval(@Nullable StringData str, @Nullable StringData regex) {
+ Matcher matcher = getRegexpMatcher(str, regex);
+ return matcher != null && matcher.find()
+ ? BinaryStringData.fromString(matcher.group(0))
+ : null;
}
}
(flink) 02/05: [FLINK-35931][table] Add the built-in function REGEXP_EXTRACT_ALL
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 7fa4ebad057bca765b661a553f4ccc2df7861a05
Author: dylanhz <dy...@163.com>
AuthorDate: Fri Aug 16 11:37:31 2024 +0800
[FLINK-35931][table] Add the built-in function REGEXP_EXTRACT_ALL
---
docs/data/sql_functions.yml | 11 +-
docs/data/sql_functions_zh.yml | 10 ++
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 18 ++++
.../pyflink/table/tests/test_expression.py | 10 +-
.../flink/table/api/internal/BaseExpressions.java | 28 ++++++
.../functions/BuiltInFunctionDefinitions.java | 23 +++++
.../planner/functions/RegexpFunctionsITCase.java | 111 ++++++++++++++++++++-
.../table/runtime/functions/SqlFunctionUtils.java | 2 +-
.../functions/scalar/RegexpExtractAllFunction.java | 75 ++++++++++++++
10 files changed, 282 insertions(+), 7 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 9d6d8b2acef..70823fc582c 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -358,7 +358,16 @@ string:
not exceed the number of the defined groups.
E.g. REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 2)" returns "bar".
-
+ - sql: REGEXP_EXTRACT_ALL(str, regex[, extractIndex])
+ table: str.regexpExtractAll(regex[, extractIndex])
+ description: |
+ Extracts all the substrings in str that match the regex expression and correspond to the regex group extractIndex.
+
+ regex may contain multiple groups. extractIndex indicates which regex group to extract and starts from 1, also the default value if not specified. 0 means matching the entire regular expression.
+
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>, extractIndex <TINYINT | SMALLINT | INTEGER | BIGINT>`
+
+ Returns an `ARRAY<STRING>` representation of all the matched substrings. `NULL` if any of the arguments are `NULL` or invalid.
- sql: INITCAP(string)
table: STRING.initCap()
description: Returns a new form of STRING with the first character of each word converted to uppercase and the rest characters to lowercase. Here a word means a sequences of alphanumeric characters.
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index e26c3b44267..f2629ca1096 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -425,6 +425,16 @@ string:
将字符串 STRING1 按照 STRING2 正则表达式的规则拆分，返回指定 INTEGER1 处位置的字符串。正则表达式匹配组索引从 1 开始，
0 表示匹配整个正则表达式。此外，正则表达式匹配组索引不应超过定义的组数。
例如 `REGEXP_EXTRACT('foothebar', 'foo(.*?)(bar)', 2)` 返回 `"bar"`。
+ - sql: REGEXP_EXTRACT_ALL(str, regex[, extractIndex])
+ table: str.regexpExtractAll(regex[, extractIndex])
+ description: |
+ 提取字符串 str 中与正则表达式 regex 匹配且与 extractIndex 组对应的所有子串。
+
+ regex 可以包含多个组。extractIndex 用于指示要提取哪个正则组，索引从 1 开始，也作为未指定时的默认值。0 表示匹配整个正则表达式。
+
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>, extractIndex <TINYINT | SMALLINT | INTEGER | BIGINT>`
+
+ 返回一个 `ARRAY<STRING>`，表示所有匹配的子串。如果任何参数为 `NULL`或非法，则返回 `NULL`。
- sql: INITCAP(string)
table: STRING.initCap()
description: |
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
index 050feb15b30..60a5631e854 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -180,6 +180,7 @@ string functions
Expression.regexp
Expression.regexp_replace
Expression.regexp_extract
+ Expression.regexp_extract_all
Expression.from_base64
Expression.to_base64
Expression.ascii
diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
index d69bcd60277..1e924c508e1 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1225,6 +1225,24 @@ class Expression(Generic[T]):
else:
return _ternary_op("regexpExtract")(self, regex, extract_index)
+ def regexp_extract_all(self, regex, extract_index=None) -> 'Expression':
+ """
+ Extracts all the substrings in str that match the regex expression and correspond to the
+ regex group extract_index.
+ regex may contain multiple groups. extract_index indicates which regex group to extract and
+ starts from 1, also the default value if not specified. 0 means matching the entire
+ regular expression.
+ null if any of the arguments are null or invalid.
+
+ :param regex: A STRING expression with a matching pattern.
+ :param extract_index: An optional INTEGER expression with default 1.
+ :return: An ARRAY<STRING> of all the matched substrings.
+ """
+ if extract_index is None:
+ return _binary_op("regexpExtractAll")(self, regex)
+ else:
+ return _ternary_op("regexpExtractAll")(self, regex, extract_index)
+
@property
def from_base64(self) -> 'Expression[str]':
"""
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index 0449a9a14a8..3af9c27732d 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -146,9 +146,6 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
self.assertEqual('lpad(a, 4, b)', str(expr1.lpad(4, expr2)))
self.assertEqual('rpad(a, 4, b)', str(expr1.rpad(4, expr2)))
self.assertEqual('overlay(a, b, 6, 2)', str(expr1.overlay(expr2, 6, 2)))
- self.assertEqual("regexp(a, b)", str(expr1.regexp(expr2)))
- self.assertEqual("regexpReplace(a, b, 'abc')", str(expr1.regexp_replace(expr2, 'abc')))
- self.assertEqual('regexpExtract(a, b, 3)', str(expr1.regexp_extract(expr2, 3)))
self.assertEqual('fromBase64(a)', str(expr1.from_base64))
self.assertEqual('toBase64(a)', str(expr1.to_base64))
self.assertEqual('ascii(a)', str(expr1.ascii))
@@ -176,6 +173,13 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
self.assertEqual('ELT(3, a, b, c)', str(lit(3).elt(expr1, expr2, expr3)))
self.assertEqual("PRINTF('%d %s', a, b)", str(lit("%d %s").printf(expr1, expr2)))
+ # regexp functions
+ self.assertEqual("regexp(a, b)", str(expr1.regexp(expr2)))
+ self.assertEqual('regexpExtract(a, b, 3)', str(expr1.regexp_extract(expr2, 3)))
+ self.assertEqual('REGEXP_EXTRACT_ALL(a, b)', str(expr1.regexp_extract_all(expr2)))
+ self.assertEqual('REGEXP_EXTRACT_ALL(a, b, 3)', str(expr1.regexp_extract_all(expr2, 3)))
+ self.assertEqual("regexpReplace(a, b, 'abc')", str(expr1.regexp_replace(expr2, 'abc')))
+
# temporal functions
self.assertEqual('cast(a, DATE)', str(expr1.to_date))
self.assertEqual('cast(a, TIME(0))', str(expr1.to_time))
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 39c84ad3923..7ea2848e348 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -161,6 +161,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.PROCTI
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.RADIANS;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_EXTRACT;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_EXTRACT_ALL;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_REPLACE;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REPEAT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REPLACE;
@@ -1166,6 +1167,33 @@ public abstract class BaseExpressions<InType, OutType> {
unresolvedCall(REGEXP_EXTRACT, toExpr(), objectToExpression(regex)));
}
+ /**
+ * Extracts all the substrings in {@code str} that match the {@code regex} expression and
+ * correspond to the regex group {@code extractIndex}. <br>
+ * {@code regex} may contain multiple groups. {@code extractIndex} indicates which regex group
+ * to extract and starts from 1, also the default value if not specified. 0 means matching the
+ * entire regular expression.
+ *
+ * @param regex A STRING expression with a matching pattern.
+ * @param extractIndex An optional INTEGER expression with default 1.
+ * @return An ARRAY<STRING> of all the matched substrings. <br>
+ * null if any of the arguments are null or invalid.
+ */
+ public OutType regexpExtractAll(InType regex, InType extractIndex) {
+ return toApiSpecificExpression(
+ unresolvedCall(
+ REGEXP_EXTRACT_ALL,
+ toExpr(),
+ objectToExpression(regex),
+ objectToExpression(extractIndex)));
+ }
+
+ /** Extracts all the strings in str that match the regex expression. */
+ public OutType regexpExtractAll(InType regex) {
+ return toApiSpecificExpression(
+ unresolvedCall(REGEXP_EXTRACT_ALL, toExpr(), objectToExpression(regex)));
+ }
+
/**
* Returns a string by quotes a string as a JSON value and wrapping it with double quote
* characters.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index ad6f90ea5e7..a005aa938fb 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -1127,6 +1127,28 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(explicit(DataTypes.STRING().nullable()))
.build();
+ public static final BuiltInFunctionDefinition REGEXP_EXTRACT_ALL =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("REGEXP_EXTRACT_ALL")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ or(
+ sequence(
+ Arrays.asList("str", "regex"),
+ Arrays.asList(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.CHARACTER_STRING))),
+ sequence(
+ Arrays.asList("str", "regex", "extractIndex"),
+ Arrays.asList(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.INTEGER_NUMERIC)))))
+ .outputTypeStrategy(explicit(DataTypes.ARRAY(DataTypes.STRING())))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.RegexpExtractAllFunction")
+ .build();
+
public static final BuiltInFunctionDefinition JSON_QUOTE =
BuiltInFunctionDefinition.newBuilder()
.name("JSON_QUOTE")
@@ -1145,6 +1167,7 @@ public final class BuiltInFunctionDefinitions {
.runtimeClass(
"org.apache.flink.table.runtime.functions.scalar.JsonUnquoteFunction")
.build();
+
public static final BuiltInFunctionDefinition FROM_BASE64 =
BuiltInFunctionDefinition.newBuilder()
.name("fromBase64")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
index ab66e6e4030..3c1efb5c0c2 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
@@ -26,12 +26,12 @@ import java.util.stream.Stream;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;
-/** Test regex functions correct behaviour. */
+/** Test Regexp functions correct behaviour. */
class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
@Override
Stream<TestSetSpec> getTestSetSpecs() {
- return Stream.of(regexpExtractTestCases()).flatMap(s -> s);
+ return Stream.of(regexpExtractTestCases(), regexpExtractAllTestCases()).flatMap(s -> s);
}
private Stream<TestSetSpec> regexpExtractTestCases() {
@@ -50,4 +50,111 @@ class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
"ABC",
DataTypes.STRING().nullable()));
}
+
+ private Stream<TestSetSpec> regexpExtractAllTestCases() {
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.REGEXP_EXTRACT_ALL)
+ .onFieldsWithData(null, "abcdeabde", "100-200, 300-400")
+ .andDataTypes(DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING())
+ // null input
+ .testResult(
+ $("f0").regexpExtractAll($("f1")),
+ "REGEXP_EXTRACT_ALL(f0, f1)",
+ null,
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f1").regexpExtractAll($("f0")),
+ "REGEXP_EXTRACT_ALL(f1, f0)",
+ null,
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f1").regexpExtractAll($("f1"), null),
+ "REGEXP_EXTRACT_ALL(f1, f1, NULL)",
+ null,
+ DataTypes.ARRAY(DataTypes.STRING()))
+ // invalid regexp
+ .testResult(
+ $("f1").regexpExtractAll("("),
+ "REGEXP_EXTRACT_ALL(f1, '(')",
+ null,
+ DataTypes.ARRAY(DataTypes.STRING()))
+ // invalid extractIndex
+ .testResult(
+ $("f1").regexpExtractAll("(abcdeabde)", -1),
+ "REGEXP_EXTRACT_ALL(f1, '(abcdeabde)', -1)",
+ null,
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f1").regexpExtractAll("abcdeabde"),
+ "REGEXP_EXTRACT_ALL(f1, 'abcdeabde')",
+ null,
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f1").regexpExtractAll("(abcdeabde)", 2),
+ "REGEXP_EXTRACT_ALL(f1, '(abcdeabde)', 2)",
+ null,
+ DataTypes.ARRAY(DataTypes.STRING()))
+ // not found
+ .testResult(
+ $("f2").regexpExtractAll("[a-z]", 0),
+ "REGEXP_EXTRACT_ALL(f2, '[a-z]', 0)",
+ new String[] {},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ // optional rule
+ .testResult(
+ $("f1").regexpExtractAll("(abcdeabde)|([a-z]*)", 2),
+ "REGEXP_EXTRACT_ALL(f1, '(abcdeabde)|([a-z]*)', 2)",
+ new String[] {null, ""},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f1").regexpExtractAll("ab((c)|(.?))de", 2),
+ "REGEXP_EXTRACT_ALL(f1, 'ab((c)|(.?))de', 2)",
+ new String[] {"c", null},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ // normal cases
+ .testResult(
+ $("f1").regexpExtractAll("(ab)([a-z]+)(e)", 2),
+ "REGEXP_EXTRACT_ALL(f1, '(ab)([a-z]+)(e)', 2)",
+ new String[] {"cdeabd"},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f1").regexpExtractAll("", 0),
+ "REGEXP_EXTRACT_ALL(f1, '', 0)",
+ new String[] {"", "", "", "", "", "", "", "", "", ""},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f2").regexpExtractAll("(\\d+)-(\\d+)", 0),
+ "REGEXP_EXTRACT_ALL(f2, '(\\d+)-(\\d+)', 0)",
+ new String[] {"100-200", "300-400"},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f2").regexpExtractAll("(\\d+)-(\\d+)", 1),
+ "REGEXP_EXTRACT_ALL(f2, '(\\d+)-(\\d+)', 1)",
+ new String[] {"100", "300"},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f2").regexpExtractAll("(\\d+)-(\\d+)", 2),
+ "REGEXP_EXTRACT_ALL(f2, '(\\d+)-(\\d+)', 2)",
+ new String[] {"200", "400"},
+ DataTypes.ARRAY(DataTypes.STRING()))
+ .testResult(
+ $("f2").regexpExtractAll("(\\d+).*", 1),
+ "REGEXP_EXTRACT_ALL(f2, '(\\d+).*', 1)",
+ new String[] {"100"},
+ DataTypes.ARRAY(DataTypes.STRING())),
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.REGEXP_EXTRACT_ALL, "Validation Error")
+ .onFieldsWithData(1024)
+ .andDataTypes(DataTypes.INT())
+ .testTableApiValidationError(
+ $("f0").regexpExtractAll("1024"),
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_EXTRACT_ALL(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)\n"
+ + "REGEXP_EXTRACT_ALL(str <CHARACTER_STRING>, regex <CHARACTER_STRING>, extractIndex <INTEGER_NUMERIC>)")
+ .testSqlValidationError(
+ "REGEXP_EXTRACT_ALL(f0, '1024')",
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_EXTRACT_ALL(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)\n"
+ + "REGEXP_EXTRACT_ALL(str <CHARACTER_STRING>, regex <CHARACTER_STRING>, extractIndex <INTEGER_NUMERIC>)"));
+ }
}
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
index 2d6a3124e63..d51dd9f1cb9 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
@@ -64,7 +64,7 @@ public class SqlFunctionUtils {
private static final Logger LOG = LoggerFactory.getLogger(SqlFunctionUtils.class);
- private static final ThreadLocalCache<String, Pattern> REGEXP_PATTERN_CACHE =
+ public static final ThreadLocalCache<String, Pattern> REGEXP_PATTERN_CACHE =
new ThreadLocalCache<String, Pattern>() {
@Override
public Pattern getNewInstance(String regex) {
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpExtractAllFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpExtractAllFunction.java
new file mode 100644
index 00000000000..df4ef721d1a
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpExtractAllFunction.java
@@ -0,0 +1,75 @@
+/*
+ * 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.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.GenericArrayData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryStringData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.PatternSyntaxException;
+
+import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.REGEXP_PATTERN_CACHE;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_EXTRACT_ALL}. */
+@Internal
+public class RegexpExtractAllFunction extends BuiltInScalarFunction {
+
+ public RegexpExtractAllFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.REGEXP_EXTRACT_ALL, context);
+ }
+
+ public @Nullable ArrayData eval(@Nullable StringData str, @Nullable StringData regex) {
+ return eval(str, regex, 1);
+ }
+
+ public @Nullable ArrayData eval(
+ @Nullable StringData str, @Nullable StringData regex, @Nullable Number extractIndex) {
+ if (str == null || regex == null || extractIndex == null) {
+ return null;
+ }
+
+ Matcher matcher;
+ try {
+ matcher = REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
+ } catch (PatternSyntaxException e) {
+ return null;
+ }
+
+ long groupIndex = extractIndex.longValue();
+ if (groupIndex < 0 || matcher.groupCount() < groupIndex) {
+ return null;
+ }
+
+ List<StringData> list = new ArrayList<>();
+ while (matcher.find()) {
+ list.add(BinaryStringData.fromString(matcher.group((int) groupIndex)));
+ }
+
+ return new GenericArrayData(list.toArray());
+ }
+}
(flink) 03/05: [FLINK-35932][table] Add the built-in function REGEXP_COUNT
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8189bba195326d899a9830d3a19b50cafa092666
Author: dylanhz <dy...@163.com>
AuthorDate: Thu Aug 1 17:20:52 2024 +0800
[FLINK-35932][table] Add the built-in function REGEXP_COUNT
---
docs/data/sql_functions.yml | 22 ++++--
docs/data/sql_functions_zh.yml | 22 ++++--
.../docs/reference/pyflink.table/expressions.rst | 1 +
flink-python/pyflink/table/expression.py | 11 +++
.../pyflink/table/tests/test_expression.py | 1 +
.../flink/table/api/internal/BaseExpressions.java | 14 ++++
.../functions/BuiltInFunctionDefinitions.java | 15 ++++
.../planner/functions/RegexpFunctionsITCase.java | 83 +++++++++++++++++++++-
.../functions/scalar/RegexpCountFunction.java | 60 ++++++++++++++++
9 files changed, 214 insertions(+), 15 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 70823fc582c..29b0c27ebcd 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -337,16 +337,14 @@ string:
- sql: REPLACE(string1, string2, string3)
table: STRING1.replace(STRING2, STRING3)
description: Returns a new string which replaces all the occurrences of STRING2 with STRING3 (non-overlapping) from STRING1. E.g., 'hello world'.replace('world', 'flink') returns 'hello flink'; 'ababab'.replace('abab', 'z') returns 'zab'.
- - sql: TRANSLATE(expr, fromStr, toStr)
- table: expr.translate(fromStr, toStr)
+ - sql: REGEXP_COUNT(str, regex)
+ table: str.regexpCount(regex)
description: |
- Translate an expr where all characters in fromStr have been replaced with those in toStr.
+ Returns the number of times str matches the regex pattern. regex must be a Java regular expression.
- If toStr has a shorter length than fromStr, unmatched characters are removed.
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
- expr <CHAR | VARCHAR>, fromStr <CHAR | VARCHAR>, toStr <CHAR | VARCHAR>
-
- Returns a STRING of translated expr.
+ Returns an `INTEGER` representation of the number of matches. `NULL` if any of the arguments are `NULL` or regex is invalid.
- sql: REGEXP_EXTRACT(string1, string2[, integer])
table: STRING1.regexpExtract(STRING2[, INTEGER1])
description: |
@@ -368,6 +366,16 @@ string:
`str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>, extractIndex <TINYINT | SMALLINT | INTEGER | BIGINT>`
Returns an `ARRAY<STRING>` representation of all the matched substrings. `NULL` if any of the arguments are `NULL` or invalid.
+ - sql: TRANSLATE(expr, fromStr, toStr)
+ table: expr.translate(fromStr, toStr)
+ description: |
+ Translate an expr where all characters in fromStr have been replaced with those in toStr.
+
+ If toStr has a shorter length than fromStr, unmatched characters are removed.
+
+ `expr <CHAR | VARCHAR>, fromStr <CHAR | VARCHAR>, toStr <CHAR | VARCHAR>`
+
+ Returns a `STRING` of translated expr.
- sql: INITCAP(string)
table: STRING.initCap()
description: Returns a new form of STRING with the first character of each word converted to uppercase and the rest characters to lowercase. Here a word means a sequences of alphanumeric characters.
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index f2629ca1096..76607687f17 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -409,16 +409,14 @@ string:
返回一个新字符串，它用 STRING3（非重叠）替换 STRING1 中所有出现的 STRING2。
例如 `'hello world'.replace('world', 'flink')` 返回 `'hello flink'`；
`'ababab'.replace('abab', 'z')` 返回 `'zab'`。
- - sql: TRANSLATE(expr, fromStr, toStr)
- table: expr.translate(fromStr, toStr)
+ - sql: REGEXP_COUNT(str, regex)
+ table: str.regexpCount(regex)
description: |
- 将 expr 中所有出现在 fromStr 之中的字符替换为 toStr 中的相应字符。
+ 返回字符串 str 匹配正则表达式模式 regex 的次数。regex 必须是一个 Java 正则表达式。
- 如果 toStr 的长度短于 fromStr，则未匹配的字符将被移除。
+ `str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>`
- expr <CHAR | VARCHAR>, fromStr <CHAR | VARCHAR>, toStr <CHAR | VARCHAR>
-
- 返回 STRING 格式的转换结果。
+ 返回一个 `INTEGER` 表示匹配成功的次数。如果任何参数为 `NULL` 或 regex 非法，则返回 `NULL`。
- sql: REGEXP_EXTRACT(string1, string2[, integer])
table: STRING1.regexpExtract(STRING2[, INTEGER1])
description: |
@@ -435,6 +433,16 @@ string:
`str <CHAR | VARCHAR>, regex <CHAR | VARCHAR>, extractIndex <TINYINT | SMALLINT | INTEGER | BIGINT>`
返回一个 `ARRAY<STRING>`，表示所有匹配的子串。如果任何参数为 `NULL`或非法，则返回 `NULL`。
+ - sql: TRANSLATE(expr, fromStr, toStr)
+ table: expr.translate(fromStr, toStr)
+ description: |
+ 将 expr 中所有出现在 fromStr 之中的字符替换为 toStr 中的相应字符。
+
+ 如果 toStr 的长度短于 fromStr，则未匹配的字符将被移除。
+
+ `expr <CHAR | VARCHAR>, fromStr <CHAR | VARCHAR>, toStr <CHAR | VARCHAR>`
+
+ 返回 `STRING` 格式的转换结果。
- sql: INITCAP(string)
table: STRING.initCap()
description: |
diff --git a/flink-python/docs/reference/pyflink.table/expressions.rst b/flink-python/docs/reference/pyflink.table/expressions.rst
index 60a5631e854..8698e884e1a 100644
--- a/flink-python/docs/reference/pyflink.table/expressions.rst
+++ b/flink-python/docs/reference/pyflink.table/expressions.rst
@@ -178,6 +178,7 @@ string functions
Expression.rpad
Expression.overlay
Expression.regexp
+ Expression.regexp_count
Expression.regexp_replace
Expression.regexp_extract
Expression.regexp_extract_all
diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
index 1e924c508e1..5975a79502b 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1204,6 +1204,17 @@ class Expression(Generic[T]):
"""
return _binary_op("regexp")(self, regex)
+ def regexp_count(self, regex) -> 'Expression':
+ """
+ Returns the number of times str matches the regex pattern.
+ regex must be a Java regular expression.
+ null if any of the arguments are null or regex is invalid.
+
+ :param regex: A STRING expression with a matching pattern.
+ :return: An INTEGER representation of the number of matches.
+ """
+ return _binary_op("regexpCount")(self, regex)
+
def regexp_replace(self,
regex: Union[str, 'Expression[str]'],
replacement: Union[str, 'Expression[str]']) -> 'Expression[str]':
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index 3af9c27732d..fb94a02ec7d 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -175,6 +175,7 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
# regexp functions
self.assertEqual("regexp(a, b)", str(expr1.regexp(expr2)))
+ self.assertEqual("REGEXP_COUNT(a, b)", str(expr1.regexp_count(expr2)))
self.assertEqual('regexpExtract(a, b, 3)', str(expr1.regexp_extract(expr2, 3)))
self.assertEqual('REGEXP_EXTRACT_ALL(a, b)', str(expr1.regexp_extract_all(expr2)))
self.assertEqual('REGEXP_EXTRACT_ALL(a, b, 3)', str(expr1.regexp_extract_all(expr2, 3)))
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
index 7ea2848e348..5092937e7df 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/BaseExpressions.java
@@ -160,6 +160,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.PRINTF
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.PROCTIME;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.RADIANS;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_COUNT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_EXTRACT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_EXTRACT_ALL;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP_REPLACE;
@@ -1136,6 +1137,19 @@ public abstract class BaseExpressions<InType, OutType> {
return toApiSpecificExpression(unresolvedCall(REGEXP, toExpr(), objectToExpression(regex)));
}
+ /**
+ * Returns the number of times {@code str} matches the {@code regex} pattern. {@code regex} must
+ * be a Java regular expression.
+ *
+ * @param regex A STRING expression with a matching pattern.
+ * @return An INTEGER representation of the number of matches. <br>
+ * null if any of the arguments are null or {@code regex} is invalid.
+ */
+ public OutType regexpCount(InType regex) {
+ return toApiSpecificExpression(
+ unresolvedCall(REGEXP_COUNT, toExpr(), objectToExpression(regex)));
+ }
+
/**
* Returns a string with all substrings that match the regular expression consecutively being
* replaced.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index a005aa938fb..e726051a4c4 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -1110,6 +1110,21 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.STRING())))
.build();
+ public static final BuiltInFunctionDefinition REGEXP_COUNT =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("REGEXP_COUNT")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ Arrays.asList("str", "regex"),
+ Arrays.asList(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.CHARACTER_STRING))))
+ .outputTypeStrategy(explicit(DataTypes.INT()))
+ .runtimeClass(
+ "org.apache.flink.table.runtime.functions.scalar.RegexpCountFunction")
+ .build();
+
public static final BuiltInFunctionDefinition REGEXP_EXTRACT =
BuiltInFunctionDefinition.newBuilder()
.name("regexpExtract")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
index 3c1efb5c0c2..3809f98637f 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
@@ -25,13 +25,94 @@ import java.util.stream.Stream;
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;
+import static org.apache.flink.table.api.Expressions.lit;
/** Test Regexp functions correct behaviour. */
class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
@Override
Stream<TestSetSpec> getTestSetSpecs() {
- return Stream.of(regexpExtractTestCases(), regexpExtractAllTestCases()).flatMap(s -> s);
+ return Stream.of(
+ regexpCountTestCases(),
+ regexpExtractTestCases(),
+ regexpExtractAllTestCases())
+ .flatMap(s -> s);
+ }
+
+ private Stream<TestSetSpec> regexpCountTestCases() {
+ return Stream.of(
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.REGEXP_COUNT)
+ .onFieldsWithData(null, "abcdeabde")
+ .andDataTypes(DataTypes.STRING(), DataTypes.STRING())
+ // null input
+ .testResult(
+ $("f0").regexpCount($("f1")),
+ "REGEXP_COUNT(f0, f1)",
+ null,
+ DataTypes.INT())
+ .testResult(
+ $("f1").regexpCount($("f0")),
+ "REGEXP_COUNT(f1, f0)",
+ null,
+ DataTypes.INT())
+ // invalid regexp
+ .testResult(
+ $("f1").regexpCount("("),
+ "REGEXP_COUNT(f1, '(')",
+ null,
+ DataTypes.INT())
+ // normal cases
+ .testResult(
+ lit("hello world! Hello everyone!").regexpCount("Hello"),
+ "REGEXP_COUNT('hello world! Hello everyone!', 'Hello')",
+ 1,
+ DataTypes.INT())
+ .testResult(
+ lit("abcabcabc").regexpCount("abcab"),
+ "REGEXP_COUNT('abcabcabc', 'abcab')",
+ 1,
+ DataTypes.INT())
+ .testResult(
+ lit("abcd").regexpCount("z"),
+ "REGEXP_COUNT('abcd', 'z')",
+ 0,
+ DataTypes.INT())
+ .testResult(
+ lit("^abc").regexpCount("\\^abc"),
+ "REGEXP_COUNT('^abc', '\\^abc')",
+ 1,
+ DataTypes.INT())
+ .testResult(
+ lit("a.b.c.d").regexpCount("\\."),
+ "REGEXP_COUNT('a.b.c.d', '\\.')",
+ 3,
+ DataTypes.INT())
+ .testResult(
+ lit("a*b*c*d").regexpCount("\\*"),
+ "REGEXP_COUNT('a*b*c*d', '\\*')",
+ 3,
+ DataTypes.INT())
+ .testResult(
+ lit("abc123xyz456").regexpCount("\\d"),
+ "REGEXP_COUNT('abc123xyz456', '\\d')",
+ 6,
+ DataTypes.INT())
+ .testResult(
+ lit("Helloworld! Hello everyone!").regexpCount("\\bHello\\b"),
+ "REGEXP_COUNT('Helloworld! Hello everyone!', '\\bHello\\b')",
+ 1,
+ DataTypes.INT()),
+ TestSetSpec.forFunction(BuiltInFunctionDefinitions.REGEXP_COUNT, "Validation Error")
+ .onFieldsWithData(1024)
+ .andDataTypes(DataTypes.INT())
+ .testTableApiValidationError(
+ $("f0").regexpCount("1024"),
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_COUNT(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)")
+ .testSqlValidationError(
+ "REGEXP_COUNT(f0, '1024')",
+ "Invalid input arguments. Expected signatures are:\n"
+ + "REGEXP_COUNT(str <CHARACTER_STRING>, regex <CHARACTER_STRING>)"));
}
private Stream<TestSetSpec> regexpExtractTestCases() {
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpCountFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpCountFunction.java
new file mode 100644
index 00000000000..48f9cdb731e
--- /dev/null
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/scalar/RegexpCountFunction.java
@@ -0,0 +1,60 @@
+/*
+ * 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.table.runtime.functions.scalar;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.SpecializedFunction;
+
+import javax.annotation.Nullable;
+
+import java.util.regex.Matcher;
+import java.util.regex.PatternSyntaxException;
+
+import static org.apache.flink.table.runtime.functions.SqlFunctionUtils.REGEXP_PATTERN_CACHE;
+
+/** Implementation of {@link BuiltInFunctionDefinitions#REGEXP_COUNT}. */
+@Internal
+public class RegexpCountFunction extends BuiltInScalarFunction {
+
+ public RegexpCountFunction(SpecializedFunction.SpecializedContext context) {
+ super(BuiltInFunctionDefinitions.REGEXP_COUNT, context);
+ }
+
+ public @Nullable Integer eval(@Nullable StringData str, @Nullable StringData regex) {
+ if (str == null || regex == null) {
+ return null;
+ }
+
+ Matcher matcher;
+ try {
+ matcher = REGEXP_PATTERN_CACHE.get(regex.toString()).matcher(str.toString());
+ } catch (PatternSyntaxException e) {
+ return null;
+ }
+
+ int count = 0;
+ while (matcher.find()) {
+ count++;
+ }
+
+ return count;
+ }
+}
(flink) 01/05: [FLINK-35931][table-planner] Use a separate test class for the regexp family of functions
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lincoln pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 697b4b26010c6e7014bc48f8b1bf1ad5ee1ca55e
Author: dylanhz <dy...@163.com>
AuthorDate: Fri Aug 16 10:38:09 2024 +0800
[FLINK-35931][table-planner] Use a separate test class for the regexp family of functions
---
.../planner/functions/RegexpFunctionsITCase.java | 53 ++++++++++++++++++++++
.../planner/functions/StringFunctionsITCase.java | 25 +---------
2 files changed, 54 insertions(+), 24 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
new file mode 100644
index 00000000000..ab66e6e4030
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/RegexpFunctionsITCase.java
@@ -0,0 +1,53 @@
+/*
+ * 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.table.planner.functions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+
+import java.util.stream.Stream;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+
+/** Test regex functions correct behaviour. */
+class RegexpFunctionsITCase extends BuiltInFunctionTestBase {
+
+ @Override
+ Stream<TestSetSpec> getTestSetSpecs() {
+ return Stream.of(regexpExtractTestCases()).flatMap(s -> s);
+ }
+
+ private Stream<TestSetSpec> regexpExtractTestCases() {
+ return Stream.of(
+ TestSetSpec.forFunction(
+ BuiltInFunctionDefinitions.REGEXP_EXTRACT, "Check return type")
+ .onFieldsWithData("22", "ABC")
+ .testResult(
+ call("regexpExtract", $("f0"), "[A-Z]+"),
+ "REGEXP_EXTRACT(f0,'[A-Z]+')",
+ null,
+ DataTypes.STRING().nullable())
+ .testResult(
+ call("regexpExtract", $("f1"), "[A-Z]+"),
+ "REGEXP_EXTRACT(f1, '[A-Z]+')",
+ "ABC",
+ DataTypes.STRING().nullable()));
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StringFunctionsITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StringFunctionsITCase.java
index d1baff469e4..08a390b0897 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StringFunctionsITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/StringFunctionsITCase.java
@@ -31,7 +31,6 @@ import java.time.temporal.ChronoUnit;
import java.util.stream.Stream;
import static org.apache.flink.table.api.Expressions.$;
-import static org.apache.flink.table.api.Expressions.call;
import static org.apache.flink.table.api.Expressions.lit;
/** Test String functions correct behaviour. */
@@ -39,12 +38,7 @@ class StringFunctionsITCase extends BuiltInFunctionTestBase {
@Override
Stream<TestSetSpec> getTestSetSpecs() {
- return Stream.of(
- bTrimTestCases(),
- eltTestCases(),
- printfTestCases(),
- regexpExtractTestCases(),
- translateTestCases())
+ return Stream.of(bTrimTestCases(), eltTestCases(), printfTestCases(), translateTestCases())
.flatMap(s -> s);
}
@@ -316,23 +310,6 @@ class StringFunctionsITCase extends BuiltInFunctionTestBase {
+ "PRINTF(format <CHARACTER_STRING>, obj <ANY>...)"));
}
- private Stream<TestSetSpec> regexpExtractTestCases() {
- return Stream.of(
- TestSetSpec.forFunction(
- BuiltInFunctionDefinitions.REGEXP_EXTRACT, "Check return type")
- .onFieldsWithData("22", "ABC")
- .testResult(
- call("regexpExtract", $("f0"), "[A-Z]+"),
- "REGEXP_EXTRACT(f0,'[A-Z]+')",
- null,
- DataTypes.STRING().nullable())
- .testResult(
- call("regexpExtract", $("f1"), "[A-Z]+"),
- "REGEXP_EXTRACT(f1, '[A-Z]+')",
- "ABC",
- DataTypes.STRING().nullable()));
- }
-
private Stream<TestSetSpec> translateTestCases() {
return Stream.of(
TestSetSpec.forFunction(BuiltInFunctionDefinitions.TRANSLATE)