[FLINK-11935][table-planner] Fix cast timestamp/date to string to avo…

…id Gregorian cutover

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala

@@ -624,10 +624,24 @@ object ScalarOperators {
  case (fromTp, toTp) if isArray(fromTp) && fromTp.getTypeClass == toTp.getTypeClass =>
  operand
 
- // Date/Time/Timestamp -> String
- case (dtt: SqlTimeTypeInfo[_], STRING_TYPE_INFO) =>
+ // Date -> String
+ case (SqlTimeTypeInfo.DATE, STRING_TYPE_INFO) =>
+ val method = qualifyMethod(BuiltInMethod.UNIX_DATE_TO_STRING.method)
  generateUnaryOperatorIfNotNull(nullCheck, targetType, operand) {
- (operandTerm) => s"${internalToTimePointCode(dtt, operandTerm)}.toString()"
+ (operandTerm) => s"$method($operandTerm)"
+ }
+
+ // Time -> String
+ case (SqlTimeTypeInfo.TIME, STRING_TYPE_INFO) =>
+ generateUnaryOperatorIfNotNull(nullCheck, targetType, operand) {
+ (operandTerm) => s"${internalToTimePointCode(operand.resultType, operandTerm)}.toString()"
+ }
+
+ // Timestamp -> String
+ case (SqlTimeTypeInfo.TIMESTAMP, STRING_TYPE_INFO) =>
+ val method = qualifyMethod(BuiltInMethod.UNIX_TIMESTAMP_TO_STRING.method)
+ generateUnaryOperatorIfNotNull(nullCheck, targetType, operand) {
+ (operandTerm) => s"$method($operandTerm, 3)"
  }
 
  // Interval Months -> String

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala

@@ -2339,31 +2339,31 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  'f18.floor(TimeIntervalUnit.YEAR),
  "f18.floor(YEAR)",
  "FLOOR(f18 TO YEAR)",
- "1996-01-01 00:00:00.0")
+ "1996-01-01 00:00:00.000")
 
  testAllApis(
  'f18.floor(TimeIntervalUnit.MONTH),
  "f18.floor(MONTH)",
  "FLOOR(f18 TO MONTH)",
- "1996-11-01 00:00:00.0")
+ "1996-11-01 00:00:00.000")
 
  testAllApis(
  'f18.floor(TimeIntervalUnit.DAY),
  "f18.floor(DAY)",
  "FLOOR(f18 TO DAY)",
- "1996-11-10 00:00:00.0")
+ "1996-11-10 00:00:00.000")
 
  testAllApis(
  'f18.floor(TimeIntervalUnit.MINUTE),
  "f18.floor(MINUTE)",
  "FLOOR(f18 TO MINUTE)",
- "1996-11-10 06:55:00.0")
+ "1996-11-10 06:55:00.000")
 
  testAllApis(
  'f18.floor(TimeIntervalUnit.SECOND),
  "f18.floor(SECOND)",
  "FLOOR(f18 TO SECOND)",
- "1996-11-10 06:55:44.0")
+ "1996-11-10 06:55:44.000")
 
  testAllApis(
  'f17.floor(TimeIntervalUnit.HOUR),
@@ -2399,31 +2399,31 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  'f18.ceil(TimeIntervalUnit.YEAR),
  "f18.ceil(YEAR)",
  "CEIL(f18 TO YEAR)",
- "1997-01-01 00:00:00.0")
+ "1997-01-01 00:00:00.000")
 
  testAllApis(
  'f18.ceil(TimeIntervalUnit.MONTH),
  "f18.ceil(MONTH)",
  "CEIL(f18 TO MONTH)",
- "1996-12-01 00:00:00.0")
+ "1996-12-01 00:00:00.000")
 
  testAllApis(
  'f18.ceil(TimeIntervalUnit.DAY),
  "f18.ceil(DAY)",
  "CEIL(f18 TO DAY)",
- "1996-11-11 00:00:00.0")
+ "1996-11-11 00:00:00.000")
 
  testAllApis(
  'f18.ceil(TimeIntervalUnit.MINUTE),
  "f18.ceil(MINUTE)",
  "CEIL(f18 TO MINUTE)",
- "1996-11-10 06:56:00.0")
+ "1996-11-10 06:56:00.000")
 
  testAllApis(
  'f18.ceil(TimeIntervalUnit.SECOND),
  "f18.ceil(SECOND)",
  "CEIL(f18 TO SECOND)",
- "1996-11-10 06:55:45.0")
+ "1996-11-10 06:55:45.000")
 
  testAllApis(
  'f17.ceil(TimeIntervalUnit.HOUR),
@@ -2810,22 +2810,22 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  testAllApis("2016-06-15".toTimestamp - 1.hour,
  "'2016-06-15'.toTimestamp - 1.hour",
  "timestampadd(HOUR, -1, date '2016-06-15')",
- "2016-06-14 23:00:00.0")
+ "2016-06-14 23:00:00.000")
 
  testAllApis("2016-06-15".toTimestamp + 1.minute,
  "'2016-06-15'.toTimestamp + 1.minute",
  "timestampadd(MINUTE, 1, date '2016-06-15')",
- "2016-06-15 00:01:00.0")
+ "2016-06-15 00:01:00.000")
 
  testAllApis("2016-06-15".toTimestamp - 1.second,
  "'2016-06-15'.toTimestamp - 1.second",
  "timestampadd(SQL_TSI_SECOND, -1, date '2016-06-15')",
- "2016-06-14 23:59:59.0")
+ "2016-06-14 23:59:59.000")
 
  testAllApis("2016-06-15".toTimestamp + 1.second,
  "'2016-06-15'.toTimestamp + 1.second",
  "timestampadd(SECOND, 1, date '2016-06-15')",
- "2016-06-15 00:00:01.0")
+ "2016-06-15 00:00:01.000")
 
  testAllApis(nullOf(Types.SQL_TIMESTAMP) + 1.second,
  "nullOf(SQL_TIMESTAMP) + 1.second",

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/expressions/TemporalTypesTest.scala

@@ -75,13 +75,13 @@ class TemporalTypesTest extends ExpressionTestBase {
  testTableApi(
  Timestamp.valueOf("2040-09-11 00:00:00.000"),
  "'2040-09-11 00:00:00.000'.toTimestamp",
- "2040-09-11 00:00:00.0")
+ "2040-09-11 00:00:00.000")
 
  testAllApis(
  "1500-04-30 12:00:00".cast(Types.SQL_TIMESTAMP),
  "'1500-04-30 12:00:00'.cast(SQL_TIMESTAMP)",
  "CAST('1500-04-30 12:00:00' AS TIMESTAMP)",
- "1500-04-30 12:00:00.0")
+ "1500-04-30 12:00:00.000")
  }
 
  @Test
@@ -171,13 +171,13 @@ class TemporalTypesTest extends ExpressionTestBase {
  'f0.cast(Types.SQL_TIMESTAMP),
  "f0.cast(SQL_TIMESTAMP)",
  "CAST(f0 AS TIMESTAMP)",
- "1990-10-14 00:00:00.0")
+ "1990-10-14 00:00:00.000")
 
  testAllApis(
  'f1.cast(Types.SQL_TIMESTAMP),
  "f1.cast(SQL_TIMESTAMP)",
  "CAST(f1 AS TIMESTAMP)",
- "1970-01-01 10:20:45.0")
+ "1970-01-01 10:20:45.000")
 
  testAllApis(
  'f2.cast(Types.SQL_DATE),
@@ -220,7 +220,7 @@ class TemporalTypesTest extends ExpressionTestBase {
  testTableApi(
  'f8.cast(Types.SQL_TIMESTAMP),
  "f8.cast(SQL_TIMESTAMP)",
- "2016-06-27 07:23:33.0")
+ "2016-06-27 07:23:33.000")
 
  testTableApi(
  'f8.cast(Types.SQL_TIMESTAMP).cast(Types.LONG),

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/expressions/UserDefinedScalarFunctionTest.scala

@@ -226,7 +226,7 @@ class UserDefinedScalarFunctionTest extends ExpressionTestBase {
  Func10('f6),
  "Func10(f6)",
  "Func10(f6)",
- "1990-10-14 12:10:10.0")
+ "1990-10-14 12:10:10.000")
  }
 
  @Test