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/10/08 07:43:46 UTC
[flink] branch master updated: [FLINK-29354][table] Support TO_DATE and TO_TIMESTAMP built-in function in the Table API
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 946949eda11 [FLINK-29354][table] Support TO_DATE and TO_TIMESTAMP built-in function in the Table API
946949eda11 is described below
commit 946949eda11184c02f56272a2fb24200b6badb29
Author: zhangjingcun <cu...@163.com>
AuthorDate: Tue Oct 4 17:58:32 2022 +0800
[FLINK-29354][table] Support TO_DATE and TO_TIMESTAMP built-in function in the Table API
This closes #20948.
---
docs/data/sql_functions.yml | 2 +
docs/data/sql_functions_zh.yml | 2 +
flink-python/pyflink/table/expressions.py | 29 ++++++++++++++
.../pyflink/table/tests/test_expression.py | 9 ++++-
.../org/apache/flink/table/api/Expressions.java | 44 ++++++++++++++++++++++
.../table/api/ImplicitExpressionConversions.scala | 30 +++++++++++++++
.../functions/BuiltInFunctionDefinitions.java | 26 +++++++++++++
.../expressions/converter/DirectConvertRule.java | 4 ++
.../planner/expressions/TemporalTypesTest.scala | 11 +++++-
9 files changed, 155 insertions(+), 2 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index e91b3ef5cd2..a2b467f54e1 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -495,11 +495,13 @@ temporal:
- sql: UNIX_TIMESTAMP(string1[, string2])
description: 'Converts date time string string1 in format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config.'
- sql: TO_DATE(string1[, string2])
+ table: toDate(STRING1[, STRING2])
description: Converts a date string string1 with format string2 (by default 'yyyy-MM-dd') to a date.
- sql: TO_TIMESTAMP_LTZ(numeric, precision)
table: toTimestampLtz(NUMERIC, PRECISION)
description: "Converts a epoch seconds or epoch milliseconds to a TIMESTAMP_LTZ, the valid precision is 0 or 3, the 0 represents TO_TIMESTAMP_LTZ(epochSeconds, 0), the 3 represents TO_TIMESTAMP_LTZ(epochMilliseconds, 3)."
- sql: TO_TIMESTAMP(string1[, string2])
+ table: toTimestamp(STRING1[, STRING2])
description: "Converts date time string string1 with format string2 (by default: 'yyyy-MM-dd HH:mm:ss') under the 'UTC+0' time zone to a timestamp."
- sql: CURRENT_WATERMARK(rowtime)
description: |
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index c9b825e8914..a0667c4604b 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -623,6 +623,7 @@ temporal:
使用表配置中指定的时区将格式为 string2 的日期时间字符串 string1（如果未指定默认情况下：yyyy-MM-dd HH:mm:ss）
转换为 Unix 时间戳（以秒为单位）。
- sql: TO_DATE(string1[, string2])
+ table: toDate(STRING1[, STRING2])
description: 将格式为 string2（默认为 'yyyy-MM-dd'）的字符串 string1 转换为日期。
- sql: TO_TIMESTAMP_LTZ(numeric, precision)
table: toTimestampLtz(numeric, PRECISION)
@@ -630,6 +631,7 @@ temporal:
将纪元秒或纪元毫秒转换为 TIMESTAMP_LTZ，有效精度为 0 或 3，0 代表 `TO_TIMESTAMP_LTZ(epochSeconds, 0)`，
3 代表` TO_TIMESTAMP_LTZ(epochMilliseconds, 3)`。
- sql: TO_TIMESTAMP(string1[, string2])
+ table: toTimestamp(STRING1[, STRING2])
description: 将 'UTC+0' 时区下格式为 string2（默认为：'yyyy-MM-dd HH:mm:ss'）的字符串 string1 转换为时间戳。
- sql: CURRENT_WATERMARK(rowtime)
description: |
diff --git a/flink-python/pyflink/table/expressions.py b/flink-python/pyflink/table/expressions.py
index 33114183746..70cac00d9c3 100644
--- a/flink-python/pyflink/table/expressions.py
+++ b/flink-python/pyflink/table/expressions.py
@@ -272,6 +272,35 @@ def local_timestamp() -> Expression:
return _leaf_op("localTimestamp")
+def to_date(date_str: str, format: str = None) -> Expression:
+ """
+ Converts the date string with the given format (by default 'yyyy-MM-dd') to a date.
+
+ :param date_str: The date string
+ :param format: The format of the string
+ :return: The date value with DATE type.
+ """
+ if format is None:
+ return _unary_op("toDate", date_str)
+ else:
+ return _binary_op("toDate", date_str, format)
+
+
+def to_timestamp(timestamp_str: str, format: str = None) -> Expression:
+ """
+ Converts the date time string with the given format (by default: 'yyyy-MM-dd HH:mm:ss')
+ under the 'UTC+0' time zone to a timestamp.
+
+ :param timestamp_str: The date time string
+ :param format: The format of the string
+ :return: The date value with TIMESTAMP type.
+ """
+ if format is None:
+ return _unary_op("toTimestamp", timestamp_str)
+ else:
+ return _binary_op("toTimestamp", timestamp_str, format)
+
+
def to_timestamp_ltz(numeric_epoch_time, precision) -> Expression:
"""
Converts a numeric type epoch time to TIMESTAMP_LTZ.
diff --git a/flink-python/pyflink/table/tests/test_expression.py b/flink-python/pyflink/table/tests/test_expression.py
index a2423c20111..ad10f528ac7 100644
--- a/flink-python/pyflink/table/tests/test_expression.py
+++ b/flink-python/pyflink/table/tests/test_expression.py
@@ -26,7 +26,7 @@ from pyflink.table.expressions import (col, lit, range_, and_, or_, current_date
timestamp_diff, array, row, map_, row_interval, pi, e,
rand, rand_integer, atan2, negative, concat, concat_ws, uuid,
null_of, log, if_then_else, with_columns, call,
- to_timestamp_ltz, from_unixtime)
+ to_timestamp_ltz, from_unixtime, to_date, to_timestamp)
from pyflink.testing.test_case_utils import PyFlinkTestCase
@@ -257,7 +257,14 @@ class PyFlinkBatchExpressionTests(PyFlinkTestCase):
self.assertEqual('currentTimestamp()', str(current_timestamp()))
self.assertEqual('localTime()', str(local_time()))
self.assertEqual('localTimestamp()', str(local_timestamp()))
+ self.assertEqual("toDate('2018-03-18')", str(to_date('2018-03-18')))
+ self.assertEqual("toDate('2018-03-18', 'yyyy-MM-dd')",
+ str(to_date('2018-03-18', 'yyyy-MM-dd')))
self.assertEqual('toTimestampLtz(123, 0)', str(to_timestamp_ltz(123, 0)))
+ self.assertEqual("toTimestamp('1970-01-01 08:01:40')",
+ str(to_timestamp('1970-01-01 08:01:40')))
+ self.assertEqual("toTimestamp('1970-01-01 08:01:40', 'yyyy-MM-dd HH:mm:ss')",
+ str(to_timestamp('1970-01-01 08:01:40', 'yyyy-MM-dd HH:mm:ss')))
self.assertEqual("temporalOverlaps(cast('2:55:00', TIME(0)), 3600000, "
"cast('3:30:00', TIME(0)), 7200000)",
str(temporal_overlaps(
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
index 29d30858fda..fb523648ef8 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Expressions.java
@@ -303,6 +303,50 @@ public final class Expressions {
return apiCall(BuiltInFunctionDefinitions.LOCAL_TIMESTAMP);
}
+ /**
+ * Converts the given date string with format 'yyyy-MM-dd' to {@link DataTypes#DATE()}.
+ *
+ * @param dateStr The date string.
+ * @return The date value of {@link DataTypes#DATE()} type.
+ */
+ public static ApiExpression toDate(Object dateStr) {
+ return apiCall(BuiltInFunctionDefinitions.TO_DATE, dateStr);
+ }
+
+ /**
+ * Converts the date string with the specified format to {@link DataTypes#DATE()}.
+ *
+ * @param dateStr The date string.
+ * @param format The format of the string.
+ * @return The date value of {@link DataTypes#DATE()} type.
+ */
+ public static ApiExpression toDate(Object dateStr, Object format) {
+ return apiCall(BuiltInFunctionDefinitions.TO_DATE, dateStr, format);
+ }
+
+ /**
+ * Converts the given date time string with format 'yyyy-MM-dd HH:mm:ss' under the 'UTC+0' time
+ * zone to {@link DataTypes#TIMESTAMP()}.
+ *
+ * @param timestampStr The date time string.
+ * @return The timestamp value with {@link DataTypes#TIMESTAMP()} type.
+ */
+ public static ApiExpression toTimestamp(Object timestampStr) {
+ return apiCall(BuiltInFunctionDefinitions.TO_TIMESTAMP, timestampStr);
+ }
+
+ /**
+ * Converts the given time string with the specified format under the 'UTC+0' time zone to
+ * {@link DataTypes#TIMESTAMP()}.
+ *
+ * @param timestampStr The date time string.
+ * @param format The format of the string.
+ * @return The timestamp value with {@link DataTypes#TIMESTAMP()} type.
+ */
+ public static ApiExpression toTimestamp(Object timestampStr, Object format) {
+ return apiCall(BuiltInFunctionDefinitions.TO_TIMESTAMP, timestampStr, format);
+ }
+
/**
* Converts a numeric type epoch time to {@link DataTypes#TIMESTAMP_LTZ(int)}.
*
diff --git a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
index 90261590143..8638dc6cdda 100644
--- a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
+++ b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
@@ -481,6 +481,36 @@ trait ImplicitExpressionConversions {
Expressions.localTimestamp()
}
+ /**
+ * Converts the date string with format 'yyyy-MM-dd' to a date value of [[DataTypes.DATE]] type.
+ */
+ def toDate(dateStr: Expression): Expression = {
+ Expressions.toDate(dateStr)
+ }
+
+ /**
+ * Converts the date string with the specified format to a date value of [[DataTypes.DATE]] type.
+ */
+ def toDate(dateStr: Expression, format: Expression): Expression = {
+ Expressions.toDate(dateStr, format)
+ }
+
+ /**
+ * Converts the date time string with format 'yyyy-MM-dd HH:mm:ss' under the 'UTC+0' time zone
+ * to a timestamp value of [[DataTypes.TIMESTAMP]].
+ */
+ def toTimestamp(timestampStr: Expression): Expression = {
+ Expressions.toTimestamp(timestampStr)
+ }
+
+ /**
+ * Converts the date time string with the specified format under the 'UTC+0' time zone
+ * to a timestamp value of [[DataTypes.TIMESTAMP]].
+ */
+ def toTimestamp(timestampStr: Expression, format: Expression): Expression = {
+ Expressions.toTimestamp(timestampStr, format)
+ }
+
/**
* Converts a numeric type epoch time to [[DataTypes#TIMESTAMP_LTZ]].
*
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 23b293a9c5f..875f4c0aaca 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
@@ -1548,6 +1548,19 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(nullableIfArgs(explicit(STRING())))
.build();
+ public static final BuiltInFunctionDefinition TO_DATE =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("toDate")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ or(
+ sequence(logical(LogicalTypeFamily.CHARACTER_STRING)),
+ sequence(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.CHARACTER_STRING))))
+ .outputTypeStrategy(nullableIfArgs(explicit(DATE())))
+ .build();
+
public static final BuiltInFunctionDefinition TO_TIMESTAMP_LTZ =
BuiltInFunctionDefinition.newBuilder()
.name("toTimestampLtz")
@@ -1559,6 +1572,19 @@ public final class BuiltInFunctionDefinitions {
.outputTypeStrategy(SpecificTypeStrategies.TO_TIMESTAMP_LTZ)
.build();
+ public static final BuiltInFunctionDefinition TO_TIMESTAMP =
+ BuiltInFunctionDefinition.newBuilder()
+ .name("toTimestamp")
+ .kind(SCALAR)
+ .inputTypeStrategy(
+ or(
+ sequence(logical(LogicalTypeFamily.CHARACTER_STRING)),
+ sequence(
+ logical(LogicalTypeFamily.CHARACTER_STRING),
+ logical(LogicalTypeFamily.CHARACTER_STRING))))
+ .outputTypeStrategy(nullableIfArgs(explicit(TIMESTAMP(3))))
+ .build();
+
// --------------------------------------------------------------------------------------------
// Collection functions
// --------------------------------------------------------------------------------------------
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 47aad8acb16..b293e642248 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
@@ -207,9 +207,13 @@ public class DirectConvertRule implements CallExpressionConvertRule {
BuiltInFunctionDefinitions.DATE_FORMAT, FlinkSqlOperatorTable.DATE_FORMAT);
DEFINITION_OPERATOR_MAP.put(
BuiltInFunctionDefinitions.FROM_UNIXTIME, FlinkSqlOperatorTable.FROM_UNIXTIME);
+ DEFINITION_OPERATOR_MAP.put(
+ BuiltInFunctionDefinitions.TO_DATE, FlinkSqlOperatorTable.TO_DATE);
DEFINITION_OPERATOR_MAP.put(
BuiltInFunctionDefinitions.TO_TIMESTAMP_LTZ,
FlinkSqlOperatorTable.TO_TIMESTAMP_LTZ);
+ DEFINITION_OPERATOR_MAP.put(
+ BuiltInFunctionDefinitions.TO_TIMESTAMP, FlinkSqlOperatorTable.TO_TIMESTAMP);
// catalog functions
DEFINITION_OPERATOR_MAP.put(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala
index e9824afe835..ceeaf953e87 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala
@@ -611,7 +611,16 @@ class TemporalTypesTest extends ExpressionTestBase {
testSqlApi("CAST('12:44:31' AS TIME)", "12:44:31")
testSqlApi("CAST('2018-03-18' AS DATE)", "2018-03-18")
testSqlApi("TIME '12:44:31'", "12:44:31")
- testSqlApi("TO_DATE('2018-03-18')", "2018-03-18")
+
+ testAllApis(toDate("2018-03-18", "yyyy-MM-dd"), "TO_DATE('2018-03-18')", "2018-03-18")
+ testAllApis(
+ toTimestamp("1970-01-01 08:01:40"),
+ "TO_TIMESTAMP('1970-01-01 08:01:40')",
+ "1970-01-01 08:01:40.000")
+ testAllApis(
+ toTimestamp("1970-01-01 08:01:40", "yyyy-MM-dd HH:mm:ss"),
+ "TO_TIMESTAMP('1970-01-01 08:01:40', 'yyyy-MM-dd HH:mm:ss')",
+ "1970-01-01 08:01:40.000")
// EXTRACT
// testSqlApi("TO_DATE(1521331200)", "2018-03-18")