You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2022/07/04 02:14:51 UTC
[flink] branch master updated: [FLINK-28298][table][python] Support left and right in Table API (#20104)
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new da720c3b97d [FLINK-28298][table][python] Support left and right in Table API (#20104)
da720c3b97d is described below
commit da720c3b97d5c3e781abb1e910a8337f6a5ee0e4
Author: Luning (Lucas) Wang <wa...@gmail.com>
AuthorDate: Mon Jul 4 10:14:44 2022 +0800
[FLINK-28298][table][python] Support left and right in Table API (#20104)
---
docs/data/sql_functions.yml | 2 ++
docs/data/sql_functions_zh.yml | 2 ++
flink-python/pyflink/table/expression.py | 12 ++++++++++++
.../pyflink/table/tests/test_expression.py | 2 ++
.../flink/table/api/internal/BaseExpressions.java | 12 ++++++++++++
.../functions/BuiltInFunctionDefinitions.java | 22 ++++++++++++++++++++++
.../expressions/converter/DirectConvertRule.java | 2 ++
.../planner/expressions/ScalarFunctionsTest.scala | 7 +++++++
8 files changed, 61 insertions(+)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 2ad4347b20c..7a8795566b0 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -341,8 +341,10 @@ string:
- sql: INSTR(string1, string2)
description: Returns the position of the first occurrence of string2 in string1. Returns NULL if any of arguments is NULL.
- sql: LEFT(string, integer)
+ table: STRING.LEFT(INT)
description: Returns the leftmost integer characters from the string. Returns EMPTY String if integer is negative. Returns NULL if any argument is NULL.
- sql: RIGHT(string, integer)
+ table: STRING.RIGHT(INT)
description: Returns the rightmost integer characters from the string. Returns EMPTY String if integer is negative. Returns NULL if any argument is NULL.
- sql: LOCATE(string1, string2[, integer])
description: Returns the position of the first occurrence of string1 in string2 after position integer. Returns 0 if not found. Returns NULL if any of arguments is NULL.
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index ad9940b3486..f9056bf62c3 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -434,10 +434,12 @@ string:
- sql: INSTR(string1, string2)
description: 返回 string2 在 string1 中第一次出现的位置。如果有任一参数为 `NULL`，则返回 `NULL`。
- sql: LEFT(string, integer)
+ table: STRING.LEFT(INT)
description: |
返回字符串中最左边的长度为 integer 值的字符串。如果 integer 为负，则返回 `EMPTY` 字符串。如果有任一参数
为 `NULL` 则返回 `NULL`。
- sql: RIGHT(string, integer)
+ table: STRING.RIGHT(INT)
description: |
返回字符串中最右边的长度为 integer 值的字符串。如果 integer 为负，则返回 `EMPTY` 字符串。如果有任一参数
为 `NULL` 则返回 `NULL`。
diff --git a/flink-python/pyflink/table/expression.py b/flink-python/pyflink/table/expression.py
index 1f7e64f9576..b22dd0f02b0 100644
--- a/flink-python/pyflink/table/expression.py
+++ b/flink-python/pyflink/table/expression.py
@@ -1212,6 +1212,18 @@ class Expression(Generic[T]):
"""
return _binary_op("encode")(self, charset)
+ def left(self, length: Union[int, 'Expression[int]']) -> 'Expression[str]':
+ """
+ Returns the leftmost integer characters from the input string.
+ """
+ return _binary_op("left")(self, length)
+
+ def right(self, length: Union[int, 'Expression[int]']) -> 'Expression[str]':
+ """
+ Returns the rightmost integer characters from the input string.
+ """
+ return _binary_op("right")(self, length)
+
@property
def ltrim(self) -> 'Expression[str]':
"""
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index dd494cb1cdf..e17f07ba963 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -147,6 +147,8 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
self.assertEqual('chr(a)', str(expr1.chr))
self.assertEqual("decode(a, 'utf-8')", str(expr1.decode('utf-8')))
self.assertEqual("encode(a, 'utf-8')", str(expr1.encode('utf-8')))
+ self.assertEqual('left(a, 2)', str(expr1.left(2)))
+ self.assertEqual('right(a, 2)', str(expr1.right(2)))
self.assertEqual('ltrim(a)', str(expr1.ltrim))
self.assertEqual('rtrim(a)', str(expr1.rtrim))
self.assertEqual('repeat(a, 3)', str(expr1.repeat(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 908a2ab3957..e68fd84c3f3 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
@@ -103,6 +103,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_E
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_QUERY;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.JSON_VALUE;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LAST_VALUE;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LEFT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LESS_THAN;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LIKE;
@@ -135,6 +136,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.REGEXP
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;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.RIGHT;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ROUND;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.ROWTIME;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.RPAD;
@@ -1048,6 +1050,16 @@ public abstract class BaseExpressions<InType, OutType> {
unresolvedCall(ENCODE, toExpr(), objectToExpression(charset)));
}
+ /** Returns the leftmost integer characters from the input string. */
+ public OutType left(InType len) {
+ return toApiSpecificExpression(unresolvedCall(LEFT, toExpr(), objectToExpression(len)));
+ }
+
+ /** Returns the rightmost integer characters from the input string. */
+ public OutType right(InType len) {
+ return toApiSpecificExpression(unresolvedCall(RIGHT, toExpr(), objectToExpression(len)));
+ }
+
/** Returns a string that removes the left whitespaces from the given string. */
public OutType ltrim() {
return toApiSpecificExpression(unresolvedCall(LTRIM, toExpr()));
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 bb00d2e45cb..645d8e5372b 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
@@ -760,6 +760,28 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(nullableIfArgs(explicit(DataTypes.BYTES())))
.build();
+ public static final BuiltInFunctionDefinition LEFT =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("left")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.INTEGER_NUMERIC)))
+ .outputTypeStrategy(nullableIfArgs(varyingString(argument(0))))
+ .build();
+
+ public static final BuiltInFunctionDefinition RIGHT =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("right")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ sequence(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.INTEGER_NUMERIC)))
+ .outputTypeStrategy(nullableIfArgs(varyingString(argument(0))))
+ .build();
+
public static final BuiltInFunctionDefinition UUID =
BuiltInFunctionDefinition.newBuilder()
.name("uuid")
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/DirectConvertRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/DirectConvertRule.java
index 010a0d062c2..5410191f5cd 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/DirectConvertRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/converter/DirectConvertRule.java
@@ -116,6 +116,8 @@ public class DirectConvertRule implements CallExpressionConvertRule {
BuiltInFunctionDefinitions.DECODE, FlinkSqlOperatorTable.DECODE);
DEFINITION_OPERATOR_MAP.put(
BuiltInFunctionDefinitions.ENCODE, FlinkSqlOperatorTable.ENCODE);
+ DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LEFT, FlinkSqlOperatorTable.LEFT);
+ DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.RIGHT, FlinkSqlOperatorTable.RIGHT);
DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.UUID, FlinkSqlOperatorTable.UUID);
DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.LTRIM, FlinkSqlOperatorTable.LTRIM);
DEFINITION_OPERATOR_MAP.put(BuiltInFunctionDefinitions.RTRIM, FlinkSqlOperatorTable.RTRIM);
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala
index 9136e9162f7..5e1995265fa 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala
@@ -195,6 +195,13 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
)
}
+ @Test
+ def testLeftAndRight(): Unit = {
+ val str = "Hello"
+ testAllApis(str.left(3), s"LEFT('$str', 3)", "Hel")
+ testAllApis(str.right(3), s"RIGHT('$str', 3)", "llo")
+ }
+
@Test
def testInstr(): Unit = {
testSqlApi("instr('Corporate Floor', 'or', 3, 2)", "14")