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:24 UTC
(flink) 04/05: [FLINK-35962][table] Add the built-in function REGEXP_INSTR
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;
+ }
+}