You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2019/08/08 12:38:23 UTC
[flink] 01/09: [FLINK-13561][table-planner-blink] Remove some
builtin datetime functions which can be covered by existing functions
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 28260cdd70aee0a4169a068887feae71a81ec4c0
Author: Jark Wu <im...@gmail.com>
AuthorDate: Wed Aug 7 13:36:40 2019 +0800
[FLINK-13561][table-planner-blink] Remove some builtin datetime functions which can be covered by existing functions
Removes DATE_FORMAT_TZ, DATE_ADD,DATE_SUB, DATEDIFF, FROM_TIMESTAMP, TO_TIMESTAMP_TZ builtin functions which can be covered by existing functions.
---
.../functions/sql/FlinkSqlOperatorTable.java | 60 ----------------------
.../planner/codegen/calls/BuiltInMethods.scala | 16 ------
.../planner/codegen/calls/FunctionGenerator.scala | 23 ---------
.../planner/codegen/calls/StringCallGen.scala | 34 ------------
.../planner/expressions/TemporalTypesTest.scala | 37 ++-----------
.../planner/runtime/batch/sql/CalcITCase.scala | 29 -----------
.../runtime/stream/sql/CorrelateITCase.scala | 2 +-
7 files changed, 6 insertions(+), 195 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
index e201400..d81b2f1 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
@@ -659,38 +659,6 @@ public class FlinkSqlOperatorTable extends ReflectiveSqlOperatorTable {
OperandTypes.family(SqlTypeFamily.NUMERIC, SqlTypeFamily.STRING)),
SqlFunctionCategory.TIMEDATE);
- public static final SqlFunction DATEDIFF = new SqlFunction(
- "DATEDIFF",
- SqlKind.OTHER_FUNCTION,
- ReturnTypes.INTEGER_NULLABLE,
- null,
- OperandTypes.or(
- OperandTypes.family(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.STRING),
- OperandTypes.family(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.TIMESTAMP),
- OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.TIMESTAMP),
- OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.STRING)),
- SqlFunctionCategory.TIMEDATE);
-
- public static final SqlFunction DATE_SUB = new SqlFunction(
- "DATE_SUB",
- SqlKind.OTHER_FUNCTION,
- VARCHAR_2000_NULLABLE,
- null,
- OperandTypes.or(
- OperandTypes.family(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.INTEGER),
- OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.INTEGER)),
- SqlFunctionCategory.TIMEDATE);
-
- public static final SqlFunction DATE_ADD = new SqlFunction(
- "DATE_ADD",
- SqlKind.OTHER_FUNCTION,
- VARCHAR_2000_NULLABLE,
- null,
- OperandTypes.or(
- OperandTypes.family(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.INTEGER),
- OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.INTEGER)),
- SqlFunctionCategory.TIMEDATE);
-
public static final SqlFunction IF = new SqlFunction(
"IF",
SqlKind.OTHER_FUNCTION,
@@ -800,14 +768,6 @@ public class FlinkSqlOperatorTable extends ReflectiveSqlOperatorTable {
OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.STRING)),
SqlFunctionCategory.TIMEDATE);
- public static final SqlFunction FROM_TIMESTAMP = new SqlFunction(
- "FROM_TIMESTAMP",
- SqlKind.OTHER_FUNCTION,
- ReturnTypes.cascade(ReturnTypes.BIGINT, SqlTypeTransforms.TO_NULLABLE),
- null,
- OperandTypes.family(SqlTypeFamily.TIMESTAMP),
- SqlFunctionCategory.TIMEDATE);
-
public static final SqlFunction TO_DATE = new SqlFunction(
"TO_DATE",
SqlKind.OTHER_FUNCTION,
@@ -819,26 +779,6 @@ public class FlinkSqlOperatorTable extends ReflectiveSqlOperatorTable {
OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.STRING)),
SqlFunctionCategory.TIMEDATE);
- public static final SqlFunction TO_TIMESTAMP_TZ = new SqlFunction(
- "TO_TIMESTAMP_TZ",
- SqlKind.OTHER_FUNCTION,
- ReturnTypes.cascade(ReturnTypes.explicit(SqlTypeName.TIMESTAMP), SqlTypeTransforms.FORCE_NULLABLE),
- null,
- OperandTypes.or(
- OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.STRING),
- OperandTypes.family(SqlTypeFamily.STRING, SqlTypeFamily.STRING, SqlTypeFamily.STRING)),
- SqlFunctionCategory.TIMEDATE);
-
- public static final SqlFunction DATE_FORMAT_TZ = new SqlFunction(
- "DATE_FORMAT_TZ",
- SqlKind.OTHER_FUNCTION,
- ReturnTypes.cascade(ReturnTypes.explicit(SqlTypeName.VARCHAR), SqlTypeTransforms.TO_NULLABLE),
- InferTypes.RETURN_TYPE,
- OperandTypes.or(
- OperandTypes.family(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.STRING, SqlTypeFamily.STRING),
- OperandTypes.family(SqlTypeFamily.TIMESTAMP, SqlTypeFamily.STRING)),
- SqlFunctionCategory.TIMEDATE);
-
public static final SqlFunction CONVERT_TZ = new SqlFunction(
"CONVERT_TZ",
SqlKind.OTHER_FUNCTION,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
index 74b388a..dd91b52 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
@@ -460,22 +460,6 @@ object BuiltInMethods {
"timestampCeil",
classOf[TimeUnitRange], classOf[Long], classOf[TimeZone])
- val STRING_TO_TIMESTAMP_TZ = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
- "toTimestampTz",
- classOf[String], classOf[String])
-
- val STRING_TO_TIMESTAMP_FORMAT_TZ = Types.lookupMethod(
- classOf[SqlDateTimeUtils],
- "toTimestampTz",
- classOf[String], classOf[String], classOf[String])
-
- val DATE_FORMAT_LONG_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateFormatTz", classOf[Long], classOf[String])
-
- val DATE_FORMAT_LONG_STRING_ZONE = Types.lookupMethod(
- classOf[SqlDateTimeUtils], "dateFormatTz", classOf[Long], classOf[String], classOf[String])
-
val CONVERT_TZ = Types.lookupMethod(
classOf[SqlDateTimeUtils],
"convertTz",
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
index 29c2643..348907a 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/FunctionGenerator.scala
@@ -648,12 +648,6 @@ object FunctionGenerator {
Seq(TIMESTAMP_WITH_LOCAL_TIME_ZONE),
BuiltInMethods.UNIX_TIMESTAMP_TS)
- addSqlFunctionMethod(
- DATEDIFF,
- Seq(TIMESTAMP_WITHOUT_TIME_ZONE,
- TIMESTAMP_WITHOUT_TIME_ZONE),
- BuiltInMethods.DATEDIFF_T_T)
-
// This sequence must be in sync with [[NumericOrDefaultReturnTypeInference]]
val numericTypes = Seq(
INTEGER,
@@ -817,11 +811,6 @@ object FunctionGenerator {
Seq(DECIMAL),
BuiltInMethods.DECIMAL_TO_TIMESTAMP)
- addSqlFunctionMethod(
- FROM_TIMESTAMP,
- Seq(TIMESTAMP_WITHOUT_TIME_ZONE),
- BuiltInMethods.TIMESTAMP_TO_BIGINT)
-
INTEGRAL_TYPES foreach (
dt => addSqlFunctionMethod(
FROM_UNIXTIME,
@@ -842,18 +831,6 @@ object FunctionGenerator {
addSqlFunctionMethod(FROM_UNIXTIME, Seq(BIGINT, VARCHAR), BuiltInMethods.FROM_UNIXTIME_FORMAT)
addSqlFunctionMethod(FROM_UNIXTIME, Seq(BIGINT, CHAR), BuiltInMethods.FROM_UNIXTIME_FORMAT)
- addSqlFunctionMethod(DATE_SUB, Seq(VARCHAR, INTEGER), BuiltInMethods.DATE_SUB_S)
- addSqlFunctionMethod(DATE_SUB, Seq(CHAR, INTEGER), BuiltInMethods.DATE_SUB_S)
-
- addSqlFunctionMethod(
- DATE_SUB, Seq(TIMESTAMP_WITHOUT_TIME_ZONE, INTEGER), BuiltInMethods.DATE_SUB_T)
-
- addSqlFunctionMethod(DATE_ADD, Seq(VARCHAR, INTEGER), BuiltInMethods.DATE_ADD_S)
- addSqlFunctionMethod(DATE_ADD, Seq(CHAR, INTEGER), BuiltInMethods.DATE_ADD_S)
-
- addSqlFunctionMethod(
- DATE_ADD, Seq(TIMESTAMP_WITHOUT_TIME_ZONE, INTEGER), BuiltInMethods.DATE_ADD_T)
-
// ----------------------------------------------------------------------------------------------
/**
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala
index c542173..676eef6 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/StringCallGen.scala
@@ -206,18 +206,6 @@ object StringCallGen {
isCharacterString(operands(1).resultType) =>
methodGen(BuiltInMethods.UNIX_TIMESTAMP_FORMAT)
- case DATEDIFF if isTimestamp(operands.head.resultType) &&
- isCharacterString(operands(1).resultType) =>
- methodGen(BuiltInMethods.DATEDIFF_T_S)
-
- case DATEDIFF if isCharacterString(operands.head.resultType) &&
- isCharacterString(operands(1).resultType) =>
- methodGen(BuiltInMethods.DATEDIFF_S_S)
-
- case DATEDIFF if isCharacterString(operands.head.resultType) &&
- isTimestamp(operands(1).resultType) =>
- methodGen(BuiltInMethods.DATEDIFF_S_T)
-
case DATE_FORMAT if operands.size == 2 &&
isTimestamp(operands.head.resultType) &&
isCharacterString(operands(1).resultType) =>
@@ -239,28 +227,6 @@ object StringCallGen {
isCharacterString(operands(2).resultType) =>
methodGen(BuiltInMethods.DATE_FORMAT_STRING_STRING_STRING)
- case TO_TIMESTAMP_TZ if operands.size == 2 &&
- isCharacterString(operands.head.resultType) &&
- isCharacterString(operands(1).resultType) =>
- methodGen(BuiltInMethods.STRING_TO_TIMESTAMP_TZ)
-
- case TO_TIMESTAMP_TZ if operands.size == 3 &&
- isCharacterString(operands.head.resultType) &&
- isCharacterString(operands(1).resultType) &&
- isCharacterString(operands(2).resultType) =>
- methodGen(BuiltInMethods.STRING_TO_TIMESTAMP_FORMAT_TZ)
-
- case DATE_FORMAT_TZ if operands.size == 2 &&
- isTimestamp(operands.head.resultType) &&
- isCharacterString(operands(1).resultType) =>
- methodGen(BuiltInMethods.DATE_FORMAT_LONG_ZONE)
-
- case DATE_FORMAT_TZ if operands.size == 3 &&
- isTimestamp(operands.head.resultType) &&
- isCharacterString(operands(1).resultType) &&
- isCharacterString(operands(2).resultType) =>
- methodGen(BuiltInMethods.DATE_FORMAT_LONG_STRING_ZONE)
-
case CONVERT_TZ if operands.size == 3 &&
isCharacterString(operands.head.resultType) &&
isCharacterString(operands(1).resultType) &&
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala
index b2d72b1..90bdc92 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/TemporalTypesTest.scala
@@ -839,28 +839,14 @@ class TemporalTypesTest extends ExpressionTestBase {
@Test
def testNullableCases(): Unit = {
- testSqlApi(
- "DATE_FORMAT_TZ(TO_TIMESTAMP(cast(NUll as bigInt)), 'yyyy/MM/dd HH:mm:ss', 'Asia/Shanghai')",
- nullable)
-
- testSqlApi("CONVERT_TZ(cast(NUll as varchar), 'yyyy-MM-dd HH:mm:ss', 'UTC', 'Asia/Shanghai')",
+ testSqlApi("CONVERT_TZ(cast(NULL as varchar), 'UTC', 'Asia/Shanghai')",
nullable)
- testSqlApi("FROM_TIMESTAMP(f13)", nullable)
-
- testSqlApi("DATE_FORMAT(cast(NUll as varchar), 'yyyy/MM/dd HH:mm:ss')", nullable)
-
- testSqlApi("UNIX_TIMESTAMP(TO_TIMESTAMP(cast(NUll as bigInt)))", nullable)
-
- testSqlApi("FROM_UNIXTIME(cast(NUll as bigInt))", nullable)
+ testSqlApi("DATE_FORMAT(cast(NULL as varchar), 'yyyy/MM/dd HH:mm:ss')", nullable)
- testSqlApi("TO_DATE(cast(NUll as varchar))", nullable)
+ testSqlApi("FROM_UNIXTIME(cast(NULL as bigInt))", nullable)
- testSqlApi("TO_TIMESTAMP_TZ(cast(NUll as varchar), 'Asia/Shanghai')", nullable)
-
- testSqlApi(
- "DATE_FORMAT_TZ(cast(NUll as timestamp), 'yyyy/MM/dd HH:mm:ss', 'Asia/Shanghai')",
- nullable)
+ testSqlApi("TO_DATE(cast(NULL as varchar))", nullable)
}
@Test
@@ -869,7 +855,6 @@ class TemporalTypesTest extends ExpressionTestBase {
testSqlApi(s"DATE_FORMAT('$invalidStr', 'yyyy/MM/dd HH:mm:ss')", nullable)
testSqlApi(s"TO_TIMESTAMP('$invalidStr', 'yyyy-mm-dd')", nullable)
testSqlApi(s"TO_DATE('$invalidStr')", nullable)
- testSqlApi(s"TO_TIMESTAMP_TZ('$invalidStr', 'Asia/Shanghai')", nullable)
testSqlApi(
s"CONVERT_TZ('$invalidStr', 'yyyy-MM-dd HH:mm:ss', 'UTC', 'Asia/Shanghai')",
nullable)
@@ -882,7 +867,6 @@ class TemporalTypesTest extends ExpressionTestBase {
s"DATE_FORMAT('$invalidStr', 'yyyy/MM/dd HH:mm:ss')",
s"TO_TIMESTAMP('$invalidStr', 'yyyy-mm-dd')",
s"TO_DATE('$invalidStr')",
- s"TO_TIMESTAMP_TZ('$invalidStr', 'Asia/Shanghai')",
s"CONVERT_TZ('$invalidStr', 'yyyy-MM-dd HH:mm:ss', 'UTC', 'Asia/Shanghai')")
cases.foreach {
@@ -892,20 +876,9 @@ class TemporalTypesTest extends ExpressionTestBase {
}
@Test
- def testTimeZoneFunction(): Unit = {
- testSqlApi("TO_TIMESTAMP_TZ('2018-03-14 11:00:00', 'Asia/Shanghai')", "2018-03-14 03:00:00.000")
- testSqlApi("TO_TIMESTAMP_TZ('2018-03-14 11:00:00', 'yyyy-MM-dd HH:mm:ss', 'Asia/Shanghai')",
- "2018-03-14 03:00:00.000")
-
+ def testConvertTZ(): Unit = {
testSqlApi("CONVERT_TZ('2018-03-14 11:00:00', 'yyyy-MM-dd HH:mm:ss', 'UTC', 'Asia/Shanghai')",
"2018-03-14 19:00:00")
-
- testSqlApi("TO_TIMESTAMP_TZ(f14, 'UTC')", "null")
-
- // Note that, if timezone is invalid, here we follow the default behavior of JDK's getTimeZone()
- // It will use UTC timezone by default.
- // TODO: it is would be better to report the error at compiling stage. timezone/format codegen
- testSqlApi("TO_TIMESTAMP_TZ('2018-03-14 11:00:00', 'invalid_tz')", "2018-03-14 11:00:00.000")
}
// ----------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
index e01d567..8e3e82c 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/CalcITCase.scala
@@ -1168,35 +1168,6 @@ class CalcITCase extends BatchTestBase {
}
@Test
- def testDateDiff(): Unit = {
- checkResult("SELECT" +
- " DATEDIFF('2017-12-14 01:00:34', '2016-12-14 12:00:00')," +
- " DATEDIFF(TIMESTAMP '2017-12-14 01:00:23', '2016-08-14 12:00:00')," +
- " DATEDIFF('2017-12-14 09:00:23', TIMESTAMP '2013-08-19 11:00:00')," +
- " DATEDIFF(TIMESTAMP '2017-12-14 09:00:23', TIMESTAMP '2018-08-19 11:00:00')" +
- " FROM testTable WHERE a = TRUE",
- Seq(row(365, 487, 1578, -248)))
- }
-
- @Test
- def testDateSub(): Unit = {
- checkResult("SELECT" +
- " DATE_SUB(TIMESTAMP '2017-12-14 09:00:23', 3)," +
- " DATE_SUB('2017-12-14 09:00:23', -3)" +
- " FROM testTable WHERE a = TRUE",
- Seq(row("2017-12-11", "2017-12-17")))
- }
-
- @Test
- def testDateAdd(): Unit = {
- checkResult("SELECT" +
- " DATE_ADD('2017-12-14', 4)," +
- " DATE_ADD(TIMESTAMP '2017-12-14 09:10:20',-4)" +
- " FROM testTable WHERE a = TRUE",
- Seq(row("2017-12-18", "2017-12-10")))
- }
-
- @Test
def testToDate(): Unit = {
checkResult("SELECT" +
" TO_DATE(CAST(null AS VARCHAR))," +
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala
index 4268460..fa6be95 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CorrelateITCase.scala
@@ -214,7 +214,7 @@ class CorrelateITCase extends StreamingTestBase {
val sql =
"""
|SELECT * FROM TMP1
- |where DATE_ADD(v, 3) > DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd')
+ |where TIMESTAMP_ADD(day, 3, v) > DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd')
""".stripMargin
val sink = new TestingAppendSink