[FLINK-15602][table-planner-blink] Padding zeros for TIMESTAMP type t…

…o respect the precision when casting timestamp to varchar This closes apache#10877
tongcheng-elong · Jan 23, 2020 · 84e76b0 · 84e76b0
1 parent 8b485eb
commit 84e76b0
Show file tree

Hide file tree

Showing 14 changed files with 117 additions and 101 deletions.
diff --git a/...ka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableTestBase.java b/...ka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableTestBase.java
@@ -160,8 +160,8 @@ public void testKafkaSourceSink() throws Exception {
  }
 
  List<String> expected = Arrays.asList(
- "2019-12-12 00:00:05,2019-12-12 00:00:04.004,3,50.00",
- "2019-12-12 00:00:10,2019-12-12 00:00:06.006,2,5.33");
+ "2019-12-12 00:00:05.000,2019-12-12 00:00:04.004,3,50.00",
+ "2019-12-12 00:00:10.000,2019-12-12 00:00:06.006,2,5.33");
 
  assertEquals(expected, TestingSinkFunction.rows);
 

diff --git a/...ts-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientKafkaITCase.java b/...ts-common-kafka/src/test/java/org/apache/flink/tests/util/kafka/SQLClientKafkaITCase.java
@@ -42,22 +42,18 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
-import java.security.MessageDigest;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.apache.flink.util.StringUtils.byteToHexString;
-
 /**
  * End-to-end test for the kafka SQL connectors.
  */
@@ -222,14 +218,12 @@ private void checkCsvResultFile() throws Exception {
  String lines = new String(bytes, Charsets.UTF_8);
  if (lines.split("\n").length == 4) {
  success = true;
- // Check the MD5SUM of the result file.
- // Expected results:
- //
- // 2018-03-12 08:00:00.000,Alice,This was a warning.,2,Success constant folding.
- // 2018-03-12 09:00:00.000,Bob,This was another warning.,1,Success constant folding.
- // 2018-03-12 09:00:00.000,Steve,This was another info.,2,Success constant folding.
- // 2018-03-12 09:00:00.000,Alice,This was a info.,1,Success constant folding.
- Assert.assertEquals("MD5 checksum mismatch", "9b06d1f8c8b8dd4ce3341786897c8993", getMd5Sum(bytes));
+ String expected =
+ "2018-03-12 08:00:00.000,Alice,This was a warning.,2,Success constant folding.\n" +
+ "2018-03-12 09:00:00.000,Bob,This was another warning.,1,Success constant folding.\n" +
+ "2018-03-12 09:00:00.000,Steve,This was another info.,2,Success constant folding.\n" +
+ "2018-03-12 09:00:00.000,Alice,This was a info.,1,Success constant folding.\n";
+ Assert.assertEquals(expected, lines);
  break;
  }
  } else {
@@ -239,15 +233,4 @@ private void checkCsvResultFile() throws Exception {
  }
  Assert.assertTrue("Timeout(" + (maxRetries * duration) + " sec) to read the correct CSV results.", success);
  }
-
- private static String getMd5Sum(byte[] bytes) throws Exception {
- MessageDigest md = MessageDigest.getInstance("MD5");
- try (ByteArrayInputStream is = new ByteArrayInputStream(bytes)) {
- byte[] buf = new byte[1024];
- for (int len = 0; (len = is.read(buf)) > 0; ) {
- md.update(buf, 0, len);
- }
- }
- return byteToHexString(md.digest());
- }
 }
diff --git a/...er-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala b/...er-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/BuiltInMethods.scala
@@ -218,12 +218,12 @@ object BuiltInMethods {
  val TIMESTAMP_TO_STRING = Types.lookupMethod(
  classOf[SqlDateTimeUtils],
  "timestampToString",
- classOf[SqlTimestamp])
+ classOf[SqlTimestamp], classOf[Int])
 
  val TIMESTAMP_TO_STRING_TIME_ZONE = Types.lookupMethod(
  classOf[SqlDateTimeUtils],
  "timestampToString",
- classOf[SqlTimestamp], classOf[TimeZone])
+ classOf[SqlTimestamp], classOf[TimeZone], classOf[Int])
 
  val TIMESTAMP_TO_TIMESTAMP_WITH_LOCAL_ZONE = Types.lookupMethod(
  classOf[SqlDateTimeUtils],

diff --git a/...link/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala b/...link/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
@@ -2325,11 +2325,13 @@ object ScalarOperatorGens {
  case TIMESTAMP_WITHOUT_TIME_ZONE => // including rowtime indicator
  // The interpreted string conforms to the definition of timestamp literal
  // SQL 2011 Part 2 Section 6.13 General Rules 11) d)
- s"${qualifyMethod(BuiltInMethods.TIMESTAMP_TO_STRING)}($operandTerm)"
+ val precision = fromType.asInstanceOf[TimestampType].getPrecision
+ s"${qualifyMethod(BuiltInMethods.TIMESTAMP_TO_STRING)}($operandTerm, $precision)"
  case TIMESTAMP_WITH_LOCAL_TIME_ZONE =>
  val method = qualifyMethod(BuiltInMethods.TIMESTAMP_TO_STRING_TIME_ZONE)
  val zone = ctx.addReusableTimeZone()
- s"$method($operandTerm, $zone)"
+ val precision = fromType.asInstanceOf[LocalZonedTimestampType].getPrecision
+ s"$method($operandTerm, $zone, $precision)"
  }
 
 }
diff --git a/...anner-blink/src/test/scala/org/apache/flink/table/planner/expressions/ArrayTypeTest.scala b/...anner-blink/src/test/scala/org/apache/flink/table/planner/expressions/ArrayTypeTest.scala
@@ -90,9 +90,13 @@ class ArrayTypeTest extends ArrayTypeTestBase {
  "ARRAY[TIME '14:15:16', TIME '17:18:19']",
  "[14:15:16, 17:18:19]")
 
- testAllApis(
+ // There is no timestamp literal function in Java String Table API,
+ // toTimestamp is casting string to TIMESTAMP(3) which is not the same to timestamp literal.
+ testTableApi(
  Array(localDateTime("1985-04-11 14:15:16"), localDateTime("2018-07-26 17:18:19")),
- "array('1985-04-11 14:15:16'.toTimestamp, '2018-07-26 17:18:19'.toTimestamp)",
+ "[1985-04-11 14:15:16, 2018-07-26 17:18:19]")
+
+ testSqlApi(
  "ARRAY[TIMESTAMP '1985-04-11 14:15:16', TIMESTAMP '2018-07-26 17:18:19']",
  "[1985-04-11 14:15:16, 2018-07-26 17:18:19]")
 

diff --git a/...ner-blink/src/test/scala/org/apache/flink/table/planner/expressions/DecimalTypeTest.scala b/...ner-blink/src/test/scala/org/apache/flink/table/planner/expressions/DecimalTypeTest.scala
@@ -189,6 +189,12 @@ class DecimalTypeTest extends ExpressionTestBase {
  BigDecimal("123456789.123456789123456789").cast(DataTypes.DOUBLE),
  "(123456789.123456789123456789p).cast(DOUBLE)",
  "1.2345678912345679E8")
+
+ // testing padding behaviour
+ testSqlApi(
+ "CAST(CAST(f67 AS DECIMAL(10, 5)) AS VARCHAR)",
+ "1.00000"
+ )
  }
 
  @Test

diff --git a/...planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/MapTypeTest.scala b/...planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/MapTypeTest.scala
@@ -85,13 +85,15 @@ class MapTypeTest extends MapTypeTestBase {
  "MAP[DATE '1985-04-11', TIME '14:15:16', DATE '2018-07-26', TIME '17:18:19']",
  "{1985-04-11=14:15:16, 2018-07-26=17:18:19}")
 
- testAllApis(
+ // There is no timestamp literal function in Java String Table API,
+ // toTimestamp is casting string to TIMESTAMP(3) which is not the same to timestamp literal.
+ testTableApi(
  map(valueLiteral(gLocalTime("14:15:16")), valueLiteral(localDateTime("1985-04-11 14:15:16")),
  valueLiteral(gLocalTime("17:18:19")), valueLiteral(localDateTime("2018-07-26 17:18:19"))),
- "map('14:15:16'.toTime, '1985-04-11 14:15:16'.toTimestamp, " +
-  "'17:18:19'.toTime, '2018-07-26 17:18:19'.toTimestamp)",
+ "{14:15:16=1985-04-11 14:15:16, 17:18:19=2018-07-26 17:18:19}")
+ testSqlApi(
  "MAP[TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16', " +
-  "TIME '17:18:19', TIMESTAMP '2018-07-26 17:18:19']",
+ "TIME '17:18:19', TIMESTAMP '2018-07-26 17:18:19']",
  "{14:15:16=1985-04-11 14:15:16, 17:18:19=2018-07-26 17:18:19}")
 
  testAllApis(

diff --git a/...planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/RowTypeTest.scala b/...planner-blink/src/test/scala/org/apache/flink/table/planner/expressions/RowTypeTest.scala
@@ -38,7 +38,7 @@ class RowTypeTest extends RowTypeTestBase {
  "(1,foo,true)")
 
  // special literal
- testAllApis(
+ testTableApi(
  row(
  localDate("1985-04-11"),
  gLocalTime("14:15:16"),
@@ -47,12 +47,7 @@ class RowTypeTest extends RowTypeTestBase {
  array(1, 2, 3),
  map("foo", "bar"),
  row(1, true)),
- "row('1985-04-11'.toDate, '14:15:16'.toTime, '1985-04-11 14:15:16'.toTimestamp, " +
- "0.1p, Array(1, 2, 3), Map('foo', 'bar'), row(1, true))",
- "ROW(DATE '1985-04-11', TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16', " +
- "CAST(0.1 AS DECIMAL(2, 1)), ARRAY[1, 2, 3], MAP['foo', 'bar'], row(1, true))",
  "(1985-04-11,14:15:16,1985-04-11 14:15:16,0.1,[1, 2, 3],{foo=bar},(1,true))")
-
  testSqlApi(
  "ROW(DATE '1985-04-11', TIME '14:15:16', TIMESTAMP '1985-04-11 14:15:16', " +
  "CAST(0.1 AS DECIMAL(2, 1)), ARRAY[1, 2, 3], MAP['foo', 'bar'], row(1, true))",

diff --git a/...blink/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala b/...blink/src/test/scala/org/apache/flink/table/planner/expressions/ScalarFunctionsTest.scala
@@ -3018,37 +3018,37 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  'f18.floor(TimeIntervalUnit.YEAR),
  "f18.floor(YEAR)",
  "FLOOR(f18 TO YEAR)",
- "1996-01-01 00:00:00")
+ "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")
+ "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")
+ "1996-11-10 00:00:00.000")
 
  testAllApis(
  'f18.floor(TimeIntervalUnit.HOUR),
  "f18.floor(HOUR)",
  "FLOOR(f18 TO HOUR)",
- "1996-11-10 06:00:00")
+ "1996-11-10 06:00:00.000")
 
  testAllApis(
  'f18.floor(TimeIntervalUnit.MINUTE),
  "f18.floor(MINUTE)",
  "FLOOR(f18 TO MINUTE)",
- "1996-11-10 06:55:00")
+ "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")
+ "1996-11-10 06:55:44.000")
 
  testAllApis(
  'f17.floor(TimeIntervalUnit.HOUR),
@@ -3084,37 +3084,37 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  'f18.ceil(TimeIntervalUnit.YEAR),
  "f18.ceil(YEAR)",
  "CEIL(f18 TO YEAR)",
- "1997-01-01 00:00:00")
+ "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")
+ "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")
+ "1996-11-11 00:00:00.000")
 
  testAllApis(
  'f18.ceil(TimeIntervalUnit.HOUR),
  "f18.ceil(HOUR)",
  "CEIL(f18 TO HOUR)",
- "1996-11-10 07:00:00")
+ "1996-11-10 07:00:00.000")
 
  testAllApis(
  'f18.ceil(TimeIntervalUnit.MINUTE),
  "f18.ceil(MINUTE)",
  "CEIL(f18 TO MINUTE)",
- "1996-11-10 06:56:00")
+ "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")
+ "1996-11-10 06:55:45.000")
 
  testAllApis(
  'f17.ceil(TimeIntervalUnit.HOUR),
@@ -3538,25 +3538,38 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  "timestampadd(DAY, 1, date '2016-06-15')",
  "2016-06-16")
 
- testAllApis("2016-06-15".toTimestamp - 1.hour,
+ // There is no timestamp literal function in Java String Table API,
+ // toTimestamp is casting string to TIMESTAMP(3) which is not the same to timestamp literal.
+ testTableApi("2016-06-15".toTimestamp - 1.hour,
  "'2016-06-15'.toTimestamp - 1.hour",
+ "2016-06-14 23:00:00.000")
+ testSqlApi(
  "timestampadd(HOUR, -1, date '2016-06-15')",
- "2016-06-14 23:00:00")
+ "2016-06-14 23:00:00.000000")
 
- testAllApis("2016-06-15".toTimestamp + 1.minute,
+ // There is no timestamp literal function in Java String Table API,
+ // toTimestamp is casting string to TIMESTAMP(3) which is not the same to timestamp literal.
+ testTableApi("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")
+ "2016-06-15 00:01:00.000")
+ testSqlApi("timestampadd(MINUTE, 1, date '2016-06-15')",
+ "2016-06-15 00:01:00.000000")
 
- testAllApis("2016-06-15".toTimestamp - 1.second,
+ // There is no timestamp literal function in Java String Table API,
+ // toTimestamp is casting string to TIMESTAMP(3) which is not the same to timestamp literal.
+ testTableApi("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")
+ "2016-06-14 23:59:59.000")
+ testSqlApi("timestampadd(SQL_TSI_SECOND, -1, date '2016-06-15')",
+ "2016-06-14 23:59:59.000000")
 
- testAllApis("2016-06-15".toTimestamp + 1.second,
+ // There is no timestamp literal function in Java String Table API,
+ // toTimestamp is casting string to TIMESTAMP(3) which is not the same to timestamp literal.
+ testTableApi("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")
+ "2016-06-15 00:00:01.000")
+ testSqlApi("timestampadd(SECOND, 1, date '2016-06-15')",
+ "2016-06-15 00:00:01.000000")
 
  testAllApis(nullOf(Types.SQL_TIMESTAMP) + 1.second,
  "nullOf(SQL_TIMESTAMP) + 1.second",
@@ -3615,9 +3628,9 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
  @Test
  def testToTimestamp(): Unit = {
  testSqlApi("to_timestamp('abc')", "null")
- testSqlApi("to_timestamp('2017-09-15 00:00:00')", "2017-09-15 00:00:00")
- testSqlApi("to_timestamp('20170915000000', 'yyyyMMddHHmmss')", "2017-09-15 00:00:00")
- testSqlApi("to_timestamp('2017-09-15', 'yyyy-MM-dd')", "2017-09-15 00:00:00")
+ testSqlApi("to_timestamp('2017-09-15 00:00:00')", "2017-09-15 00:00:00.000")
+ testSqlApi("to_timestamp('20170915000000', 'yyyyMMddHHmmss')", "2017-09-15 00:00:00.000")
+ testSqlApi("to_timestamp('2017-09-15', 'yyyy-MM-dd')", "2017-09-15 00:00:00.000")
  // test with null input
  testSqlApi("to_timestamp(cast(null as varchar))", "null")
  }