[Flink-14599][table-planner-blink] Support precision of TimestampType in blink planner #10105

docete · 2019-11-06T12:16:49Z

What is the purpose of the change

Since FLINK-14080 introduced an internal representation(SqlTimestamp) of TimestampType with precision. This subtask will replace current long with SqlTimestamp, and let blink planner support precision of TimestampType.

Note:

Tables defined by DDL is not supported in this PR since de/serializing DataTypes lost precision information. Would open another ticket to cover.
Built-in time functions/operators which use long to compute is not supported in this PR too. Would open another ticket to cover.

Brief change log

The first commit(ee687db) introduces setTimestamp/getTimestamp interface to TypeGetterSetters/VectorizedColumnBatch and writeTimestamp interface to BinaryWriter
be1beee to 0b57ee3 replace current long with SqlTimestamp, and Fix all existing failure cases
915064c introduces LegacyTimestampTypeInfo and LegacyLocalDateTimeTypeInfo to represent Types.SQL_TIMESTAMP and Types.LOCAL_DATE_TIME with precision
3d1de92 supports precision of Timestamp type in table source
8751385 supports precision of Timestamp type for timestamp literals
90f3316 Make the default precision of Timestamp type as 6
b659001 Make the interpreted string of Timestamp type conforms to the definition of timestamp literal which defined in SQL standard

Verifying this change

This change is already covered by existing tests and new tests.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

flinkbot · 2019-11-06T12:19:39Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 212e234 (Wed Apr 15 11:35:06 UTC 2020)

Warnings:

1 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2019-11-06T12:59:08Z

CI report:

3d1de92 : SUCCESS Build
4246b62 : FAILURE Build
6127ca1 : SUCCESS Build
9e755e6 : FAILURE Build
9a4e8a2 : CANCELED Build
212e234 : FAILURE Build

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build

KurtYoung

Havn't finished yet...

KurtYoung · 2019-11-07T08:09:19Z

...table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala


  val STRING_UTIL: String = className[BinaryStringUtil]

+  val SQL_TIMESTAMP_TERM: String = className[SqlTimestamp]


change to SQL_TIMESTAMP, term normally used as variable name

KurtYoung · 2019-11-07T08:28:56Z

...anner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala


    // declaration
-    reusableMemberStatements.add(s"private long $fieldTerm;")
+    reusableMemberStatements.add(s"private $SQL_TIMESTAMP_TERM $fieldTerm;")


Why use SqlTimestamp when we want a ReusableLocalDateTime?

All java.sql.Timestamp and java.time.LocalDateTime in our engine should be represented as SqlTimestamp now.

KurtYoung · 2019-11-07T08:51:53Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala

+          case SqlTypeName.TIMESTAMP =>
+            val reducedValue = reduced.getField(reducedIdx)
+            val value = if (reducedValue != null) {
+              Long.box(reducedValue.asInstanceOf[SqlTimestamp].getMillisecond)


Why we reduce timestamp to Long but not SqlTimestamp?

Just want not break the original sql timestamp literal logic.
I think an individual commit would catch this, and let the sql timestamp literal support precision.

KurtYoung · 2019-11-07T08:55:28Z

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

      case TIMESTAMP_WITHOUT_TIME_ZONE =>
+        // TODO: support Timestamp(3) now
+        val fieldTerm = newName("timestamp")
        val millis = literalValue.asInstanceOf[Long]


Could you explain why when we generating a literal for TIMESTAMP type, the value will be passed in as Long?

KurtYoung · 2019-11-07T08:57:22Z

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

+         |  $contextTerm.timerService().currentProcessingTime());
         |""".stripMargin.trim
    // the proctime has been materialized, so it's TIMESTAMP now, not PROCTIME_INDICATOR
    GeneratedExpression(resultTerm, NEVER_NULL, resultCode, new TimestampType(3))


new TimestampType(3) -> resultType

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

KurtYoung

Thanks for this nice work! It makes Timestamp more close to sql standard. I left some comments.

@JingsongLi Please also take a look at this PR, especially code generation and blink-runtime part.

KurtYoung · 2019-11-09T06:07:40Z

.../src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java

+					datetime = valueLiteral.getValueAs(Timestamp.class)
+						.orElseThrow(() -> new TableException("Invalid literal.")).toLocalDateTime();
+				} else {
+					throw new TableException("Invalid literal.");


record the illegal class here?

KurtYoung · 2019-11-09T06:15:55Z

...e-planner-blink/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala

+        if (precision > 9 || precision < 0) {
          throw new TableException(
-            s"TIMESTAMP precision is not supported: ${relDataType.getPrecision}")
+            s"TIMESTAMP precision is not supported: ${precision}")


don't need {}

KurtYoung · 2019-11-09T06:16:35Z

...e-planner-blink/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeFactory.scala

      case TIMESTAMP =>
-        if (relDataType.getPrecision > 3) {
+        val precision = relDataType.getPrecision
+        if (precision > 9 || precision < 0) {


change 9 to TimestampType.MAX_PRECISION?

KurtYoung · 2019-11-09T06:17:34Z

...le-planner-blink/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeSystem.scala

      Int.MaxValue

+    // The maximal precision of TIMESTAMP is 3, change it to 9 to support nanoseconds precision
+    case SqlTypeName.TIMESTAMP => 9


change 9 to TimestampType.MAX_PRECISION?

BTW, please also check line #46, I remember our default precision for TIMESTAMP is 6?

Yes. But the default precision of Timestamp in Calcite is 3.
Should we change this behavior?

What is the impact of changing it to 9?

change it 6 and fix failures in 90f3316

KurtYoung · 2019-11-09T06:25:48Z

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

+        }
+
+        // TODO: we copied the logical of TimestampString::getMillisSinceEpoch since the copied
+        //  DateTimeUtils.ymdToJulian is wrong.


From the comment of DateTimeUtils, I think we already aware of CALCITE-1884, do you mean we didn't fix it in copied DateTimeUtils?

Yes. CALCITE-1884 is not fixed in our copied DateTimeUtils. And it's the root cause of some delicate cases, such as:
SELECT TIMESTAMP '1500-04-30 00:00:00.123456789' FROM docs; SELECT CAST('1500-04-30 00:00:00.123456789' AS DATETIME(9)) FROM docs;
should returns
1500-05-10T00:00:00.123456789

So how about fixing it in our copied DateTimeUtils?

Two reasons to not fixing our copied DateTimeUtils in this PR

two copies of DateTimeUtils should remain the same in legacy planner and blink planner

and the impact should be evaluated for both planner

FLINK-11935 should do this, and after that this copied code could be removed.

I would like to create another issue for this, since I don't think the author of FLINK-11935 will be aware of these codes and remove them.

KurtYoung · 2019-11-09T07:38:28Z

...lanner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala

        LocalTimeConverter.INSTANCE.toExternal(v)

      case TIMESTAMP_WITHOUT_TIME_ZONE =>
        val v = literal.getValueAs(classOf[java.lang.Long])


Why the literal for TIMESTAMP_WITHOUT_TIME_ZONE is long? I remember somewhere else you presume the literal is TimestampString

Good catch. It should be TimestampString and preserve the nanosecond precision.

KurtYoung · 2019-11-09T07:42:01Z

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

+
+    testSqlApi(
+      "ARRAY[TIMESTAMP '1985-04-11 14:15:16.1234567', TIMESTAMP '2018-07-26 17:18:19.1234567']",
+      "[1985-04-11T14:15:16.123456700, 2018-07-26T17:18:19.123456700]")


why the result isn't 1985-04-11T14:15:16.1234567?

We use the LocalDateTime.toString style to represent string-format TimestampType when precision > 3. It follows one of the following ISO-8601 formats:

uuuu-MM-dd'T'HH:mm:ss.SSSSSS
uuuu-MM-dd'T'HH:mm:ss.SSSSSSSSS

Can you check whether this fits the sql standard?

The interpreted string should conforms to the definition of timestamp literal(SQL 2011 Part 2 Section 6.13 General Rules (11) (d)). And i will fix it in next commit.

KurtYoung · 2019-11-09T07:42:22Z

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

+
+    testSqlApi(
+      "ARRAY[TIMESTAMP '1985-04-11 14:15:16.1234', TIMESTAMP '2018-07-26 17:18:19.1234']",
+      "[1985-04-11T14:15:16.123400, 2018-07-26T17:18:19.123400]")


why the result isn't 1985-04-11T14:15:16.1234?

KurtYoung · 2019-11-09T07:44:06Z

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

 import java.time.{Instant, ZoneId}
 import java.util.{Locale, TimeZone}

 class TemporalTypesTest extends ExpressionTestBase {


Could you extract all Timestamp related test to a dedicated one? TemporalTypesTest seems weird.

So divide it to TimestampTypeTest/DateTypeTest/TimeTypeTest ?

That would be great.

Double checked the TemporalTypesTest and find some datetime function can handle multiple datetime types(e.g. extract/floor etc). Split the class causes tests for one single function live in different places. So I would rename this as DateTimeTypesTest to avoid confusing.

KurtYoung · 2019-11-09T07:49:13Z

...e/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArray.java

+		assertIndexIsValid(pos);
+
+		if (SqlTimestamp.isCompact(precision)) {
+			return SqlTimestamp.fromEpochMillis(segments[0].getLong(getElementOffset(pos, 8)));


I think always reading from segments[0] is wrong

Good catch, it should be SegmentsUtil.getLong(segments, getElementOffset(pos, 8)

JingsongLi · 2019-11-11T11:30:22Z

Thanks @docete , I will review this these two days.

JingsongLi · 2019-11-11T12:14:46Z

Can we put nanos of SqlTimestamp to the fixed-length part in BinaryRow? In this way:
offsetAndNanos in fixed-length part and milliseconds in variable-length part.

docete · 2019-11-13T03:20:32Z

Can we put nanos of SqlTimestamp to the fixed-length part in BinaryRow? In this way:
offsetAndNanos in fixed-length part and milliseconds in variable-length part.

OK. It can save 4 bytes per high precision timestamp object.

KurtYoung · 2019-11-13T04:19:48Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala

 import java.io.File
 import java.util.TimeZone

+import org.apache.calcite.util.TimestampString


put this together with other calcite imports

KurtYoung · 2019-11-13T04:21:09Z

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

 import java.math.{BigDecimal => JBigDecimal}
+import java.lang.{Integer => JInteger}
+
+import org.apache.flink.table.runtime.functions.SqlDateTimeUtils


put this together with other flink imports

KurtYoung · 2019-11-13T06:48:48Z

...lanner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractor.scala

-        val v = literal.getValueAs(classOf[java.lang.Long])
-        LocalDateTimeConverter.INSTANCE.toExternal(v)
+        val timestampString = literal.getValueAs(classOf[TimestampString])
+        val millis = SqlDateTimeUtils.getMillisSinceEpoch(timestampString.toString)


Reuse the result of timestampString.toString

KurtYoung · 2019-11-13T06:54:53Z

...nner-blink/src/test/scala/org/apache/flink/table/planner/expressions/DateTimeTypesTest.scala

-      "CAST(f0 AS TIMESTAMP)",
-      "1990-10-14 00:00:00.000")
+      "CAST(f0 AS TIMESTAMP(3))",
+      "1990-10-14 00:00:00")


TIMESTAMP(3) should be print as 1990-10-14 00:00:00.000?

SQL 2011 defined the interpreted string value as the shortest character string that conforms to the definition of literal. And I checked other DBMSs:

MySQL 5.6: 'SELECT CAST(TIMESTAMP '1970-01-01 11:22:33' AS DATETIME(3))' => '1970-01-01T11:22:33Z'

PG 9.6: `SELECT CAST('1970-01-01 11:22:33' AS TIMESTAMP(3))' => '1970-01-01T11:22:33Z'

ORACLE 11g R2: select CAST(TIMESTAMP '1970-01-01 11:22:33' AS TIMESTAMP(3)) from log_table => '1970-01-01 11:22:33.0'

MSSQL 2017: SELECT CAST({ts '1970-01-01 11:22:33.000'} AS DATETIME), CAST({ts '1970-01-01 11:22:33.000'} AS DATETIME2) => '1970-01-01T11:22:33Z', '1970-01-01 11:22:33.0000000'

Only MSSQL's DATETIME2 will preserve tailing '0's and the precision can not be specified. IMO we shouldn't preserve the tailing '0's.

IMO 1990-10-14 00:00:00.000 is more intuitive, right?

Ignore the tailing '0's is more standard-compliant since it's the shortest to represent the literal?

I don't know. Does SQL standard describe something with this topic?

No. It just describe the cast specification from datetime type to char/varchar type:
If SD is a datetime data type or an interval data type then let Y be the shortest character string that conforms to the definition of <literal> in Subclause 5.3, “<literal>”, and such that the interpreted value of Y is SV and the interpreted precision of Y is the precision of SD. If SV is a negative interval, then <sign> shall be specified within <unquoted interval string> in the literal Y.

KurtYoung · 2019-11-13T07:02:06Z

flink-table/flink-table-runtime-blink/pom.xml

+		<!-- Use TimestampString to cover CAST TIMESTAMP to STRING -->
+		<dependency>
+			<groupId>org.apache.calcite</groupId>
+			<artifactId>calcite-core</artifactId>


I would prefer to not introduce calcite dependency for table-runtime module

Is there any harmful things to introduce calcite dependency?
The calcite's TimestampString provides standard-compliant processing of Timestamp literal which we can depend on.

I recalled depending on calcite in different places introduced some packaging problems during 1.9.

+1 to not introduce calcite to runtime.

+1 not introduce calcite. We are aiming to achieve Calcite-free in runtime to support different users' Calcite in a long term (shading Calcite is not easy).

KurtYoung · 2019-11-13T07:03:01Z

...r-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala

          case TIMESTAMP_WITHOUT_TIME_ZONE =>
-            generateOperatorIfNotNull(ctx, new TimestampType(), left, right) {
-              (l, r) => s"($l * ${MILLIS_PER_DAY}L) $op $r"
+            generateOperatorIfNotNull(ctx, new TimestampType(3), left, right) {


what do you think about this?

KurtYoung · 2019-11-13T07:03:30Z

...r-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala

    case (TIMESTAMP_WITHOUT_TIME_ZONE, BIGINT) =>
      generateUnaryOperatorIfNotNull(ctx, targetType, operand) {
-        operandTerm => s"((long) ($operandTerm / 1000))"
+        operandTerm => s"((long) ($operandTerm.getMillisecond() / 1000))"


what do you think about this?

KurtYoung · 2019-11-13T07:05:15Z

Please rebase to master, we should let travis verify all the tests since you modified a lot of them.

docete · 2019-11-13T08:30:04Z

@KurtYoung The Date plus Day-Time Interval case and Cast Timestamp to BigInt case, I prefer leave it to the next PR. The semantic of arithmetic and cast operator for date time type need to be double check. Mix the change of only one or two cases in the PR is incomplete.

By the way, the semantic of TIMESTAMPADD for DATE +/- INTERVAL_DAY_TIME(hours, minutes and seconds) seems OK to return a TIMESTAMP. What do you think?

…rface to TypeGetterSetters/VectorizedColumnBatch and writeTimestamp interface to BinaryWriter

…sentation of Timestamp type

…r Timestamp type

…ng and vice versa

…on for Timestamp type

…imestamp type

…-in function for Timestamp type

…sting Timestamp to Timestamp with local time zone

…elds

…n for Timestamp type

…ction

…p type in parameters or result

…kupTableFunction()

…ops in window

…and LegacyLocalDateTimeTypeInfo to hold precision on conversion

…e in table source

…e for timestamp literals

…stamp type as 6 which defined in SQL standard

…estamp type conforms to the definition of timestamp literal which defined in SQL standard

…er high precision timestamp

docete · 2019-11-13T08:55:55Z

Rebased to master.

JingsongLi

Thanks @docete .
Havn't finished yet...

JingsongLi · 2019-11-13T09:03:30Z

.../src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java

-				return relBuilder.getRexBuilder().makeTimestampLiteral(TimestampString.fromCalendarFields(
-						valueAsCalendar(extractValue(valueLiteral, java.sql.Timestamp.class))), 3);
+				TimestampType timestampType = (TimestampType) type;
+				Class<?> clazz = valueLiteral.getOutputDataType().getConversionClass();


First, these logical should in extractValue.

Second, we should just use valueLiteral.getValueAs.

We can just use valueLiteral.getValueAs(LocalDateTime.class).

JingsongLi · 2019-11-13T09:05:27Z

.../src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java

+				} else {
+					throw new TableException(String.format("Invalid literal of %s.", clazz.getCanonicalName()));
+				}
+				return relBuilder.getRexBuilder().makeTimestampLiteral(


We could have util in SqlDateTimeUtils or other class: toTimestampString(LocalDateTime)

JingsongLi · 2019-11-13T09:10:06Z

...-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/ExpressionReducer.scala

+            val value = if (reducedValue != null) {
+              val ts = reducedValue.asInstanceOf[SqlTimestamp]
+              val milliseconds = ts.getMillisecond
+              val nanoseconds = ts.toLocalDateTime.getNano


Why not just use LocalDateTime to TimestampString?

A little tricky here, take CAST('1500-04-30 00:00:00' AS TIMESTAMP(3)) as an example:
we use SqlDateTimeUtils.toTimestamp[1] to cast the string to SqlTimestamp and use ExpressionReducer[2] to make a Timestamp literal.

The [1] step considers Gregorian cutovers and again the [2] step uses TimestampString.fromMillisSinceEpoch which calls our copied DateTimeUtils.julianToString considers Gregorian cutovers too. Two wrongs make a correct result.

I will change the toTimestamp and this ExpressionReducer to ignore Gregorian cutovers in next PR and that will be the final one.

I think you can fix it, and it is very worthwhile to fix. Otherwise it will bring a lot of strange code.

JingsongLi · 2019-11-13T09:11:26Z

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

-        generateNonNullLiteral(literalType, millis + "L", millis)
+        val fieldTerm = newName("timestamp")
+        val timestampString = literalValue.asInstanceOf[TimestampString].toString
+        val millis = getMillisSinceEpoch(timestampString)


You should just use TimestampString to localDateTime.

How can TimestampString get a LocalDateTime?

In getMillisSinceEpoch, you have got the year, month, and etc.. Why not construct a LocalDateTime?

getMillisSinceEpoch is a util function of TimestampString, we copied it because our copied DateTimeUtils causes Gregorian cutovers issues. We should use TimestampString.getMillisSinceEpoch directly when we remove our copied DateTimeUtils. Do you mean we should not depend on TimestampString and parse the timestamp string by ourself?

JingsongLi · 2019-11-13T09:18:42Z

...link/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarFunctionCallGen.scala

    val resultTerm = ctx.addReusableLocalVariable(resultTypeTerm, "result")
-    val evalResult = s"$functionReference.eval(${parameters.map(_.resultTerm).mkString(", ")})"
+    val evalResult =
+      if (returnType.getTypeRoot == LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE) {


Types.TIMESTAMP can be represented as long is a very old document, now, TIMESTAMP_WITHOUT_TIME_ZONE can not be represented as long, you can take a look to TimestampType.INPUT_OUTPUT_CONVERSION.

I have no idea whether we should disable a UDX with long parameter to accept a Timestamp. What do you think @KurtYoung ?

JingsongLi · 2019-11-13T09:19:23Z

...link/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarFunctionCallGen.scala


    parameterClasses.zipWithIndex.zip(operands).map { case ((paramClass, i), operandExpr) =>
+      var newOperandExpr = operandExpr
+      if (operandExpr.resultType.getTypeRoot == LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE


JingsongLi · 2019-11-13T09:22:19Z

...r-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala

          case TIMESTAMP_WITHOUT_TIME_ZONE =>
-            generateOperatorIfNotNull(ctx, new TimestampType(), left, right) {
-              (l, r) => s"($l * ${MILLIS_PER_DAY}L) $op $r"
+            generateOperatorIfNotNull(ctx, new TimestampType(3), left, right) {


Agree with @KurtYoung . It not a clean way.
BTW, You can split this PR in other ways, such as format related changes and so on.

JingsongLi · 2019-11-13T09:24:10Z

...src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala

    val paraInternalType = fromDataTypeToLogicalType(parameterType)
    if (isAny(internal) && isAny(paraInternalType)) {
      getDefaultExternalClassForType(internal) == getDefaultExternalClassForType(paraInternalType)
+    } else if ((isTimestamp(internal) && isLong(paraInternalType))


ditto, logical type already defined the conversions.

JingsongLi · 2019-11-13T09:44:40Z

...e/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/BinaryArray.java

+			assert cursor > 0 : "invalid cursor " + cursor;
+
+			// zero-out the bytes
+			SegmentsUtil.setLong(segments, offset + cursor, 0L);


move this setLong to if (value == null). This is not decimal, timestamp is fix-length.
Others format too.

JingsongLi · 2019-11-13T09:46:50Z

flink-table/flink-table-runtime-blink/pom.xml

+		<!-- Use TimestampString to cover CAST TIMESTAMP to STRING -->
+		<dependency>
+			<groupId>org.apache.calcite</groupId>
+			<artifactId>calcite-core</artifactId>


+1 to not introduce calcite to runtime.

JingsongLi · 2019-11-13T10:00:24Z

Hi @docete , I think we should try our best to avoid merging the code in and fixing it in next PR.
A better way is to split the PR's dependency work as much as possible, so that the PR can be smaller.

wuchong

Thanks @docete . I left some commetns but didn't go through the PR in detail.

wuchong · 2019-11-13T09:55:17Z

...e-blink/src/main/java/org/apache/flink/table/runtime/types/LogicalTypeDataTypeConverter.java

 				} else if (typeInfo instanceof BigDecimalTypeInfo) {
 					BigDecimalTypeInfo decimalType = (BigDecimalTypeInfo) typeInfo;
 					return new DecimalType(decimalType.precision(), decimalType.scale());
+				} else if (typeInfo instanceof LegacyLocalDateTimeTypeInfo) {


The newly introduced LegacyLocalDateTimeTypeInfo is wired. In theory, we don't need such things unless somewhere converting DataType to TypeInformation and back again. I think we should find out the root cause why we need this and create JIRA to fix the root cause. And add comments on these classes with the JIRA id.

Otherwise, we don't know how to remove these temporary code in the future.

wuchong · 2019-11-13T10:00:32Z

.../flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/SqlTimestamp.java

 	}

+	/**
+	 * Obtains an instance of {@code SqlTimestamp} from a millisecond.


minor: I would suggest to use {@link SqlTimestamp}.

wuchong · 2019-11-14T04:16:50Z

flink-table/flink-table-runtime-blink/pom.xml

+		<!-- Use TimestampString to cover CAST TIMESTAMP to STRING -->
+		<dependency>
+			<groupId>org.apache.calcite</groupId>
+			<artifactId>calcite-core</artifactId>


+1 not introduce calcite. We are aiming to achieve Calcite-free in runtime to support different users' Calcite in a long term (shading Calcite is not easy).

wuchong · 2019-11-14T04:22:10Z

...le-planner-blink/src/main/scala/org/apache/flink/table/planner/calcite/FlinkTypeSystem.scala

    case SqlTypeName.VARCHAR | SqlTypeName.CHAR | SqlTypeName.VARBINARY | SqlTypeName.BINARY =>
      Int.MaxValue

+    // The maximal precision of TIMESTAMP is 3, change it to 9 to support nanoseconds precision


What is this "maximal precision of TIMESTAMP is 3" mean? Is that a typo?

wuchong · 2019-11-14T06:10:46Z

...e-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala

-      case TIMESTAMP_WITHOUT_TIME_ZONE => DataTypes.BIGINT
+      case TIMESTAMP_WITHOUT_TIME_ZONE =>
+        val dt = argTypes(0).asInstanceOf[TimestampType]
+        DataTypes.TIMESTAMP(dt.getPrecision).bridgedTo(classOf[SqlTimestamp])


This is a little performance sensitive, because it relates de/serialization.
If the precision is less than 3, we can use BIGINT to have a better performance.

Could you improve this a bit? Or create a issue and TODO for it.

wuchong · 2019-11-14T06:14:49Z

...nner-blink/src/test/scala/org/apache/flink/table/planner/expressions/DateTimeTypesTest.scala

+
+    testSqlApi(
+      "TIMESTAMP '1500-04-30 12:00:00.123456789'",
+    "1500-04-30 12:00:00.123456789")


wuchong · 2019-11-14T06:24:15Z

.../src/main/java/org/apache/flink/table/planner/expressions/converter/ExpressionConverter.java

-				return relBuilder.getRexBuilder().makeTimestampLiteral(TimestampString.fromCalendarFields(
-						valueAsCalendar(extractValue(valueLiteral, java.sql.Timestamp.class))), 3);
+				TimestampType timestampType = (TimestampType) type;
+				Class<?> clazz = valueLiteral.getOutputDataType().getConversionClass();


We can just use valueLiteral.getValueAs(LocalDateTime.class).

wuchong · 2019-11-14T06:29:41Z

...able-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

  }

-  def generateRowtimeAccess(
+  def generateTimestampAccess(


Why change the method name? I think the original method describe the logic more accurate.
At the first glance at the new method name, I though it is used to access timestamp field from a row or record.

docete · 2019-11-20T10:57:55Z

close this since it's splitted to two PRs

rmetzger added review=description? component=TableSQL/Planner labels Nov 6, 2019

KurtYoung reviewed Nov 7, 2019

View reviewed changes

KurtYoung reviewed Nov 9, 2019

View reviewed changes

KurtYoung reviewed Nov 13, 2019

View reviewed changes

docete added 18 commits November 13, 2019 16:44

[FLINK-14599][table-planner-blink] Add setTimestamp/getTimestamp inte…

f0192dc

…rface to TypeGetterSetters/VectorizedColumnBatch and writeTimestamp interface to BinaryWriter

[FLINK-14599][table-planner-blink] Use SqlTimestamp as internal repre…

90df5e4

…sentation of Timestamp type

[FLINK-14599][table-planner-blink] Use SqlTimestamp in CodeGenUtils

22fa480

[FLINK-14599][table-planner-blink] Fix extract() built-in function fo…

88082bb

…r Timestamp type

[FLINK-14599][table-planner-blink] Fix casting from Timestamp to Stri…

e6d75bc

…ng and vice versa

[FLINK-14599][table-planner-blink] Fix floor()/ceil() built-in functi…

0ec0b46

…on for Timestamp type

[FLINK-14599][table-planner-blink] Fix time interval arithmetic for T…

d44a523

…imestamp type

[FLINK-14599][table-planner-blink] Fix cast/reinterpret/compare built…

a7ff4db

…-in function for Timestamp type

[FLINK-14599][table-planner-blink] Fix Timestamp literal and fixup ca…

7cc6cf4

…sting Timestamp to Timestamp with local time zone

[FLINK-14599][table-planner-blink] Fix count distinct on timestamp fi…

886eac4

…elds

[FLINK-14599][table-planner-blink] Fix timestampdiff built-in functio…

a207224

…n for Timestamp type

[FLINK-14599][table-planner-blink] Fix toTimestamp built-in function

77d1789

[FLINK-14599][table-planner-blink] Fix current_timestamp built-in fun…

4a83ff6

…ction

[FLINK-14599][table-planner-blink] Fix udx which use long as Timestam…

1e4b5f4

…p type in parameters or result

[FLINK-14599][table-planner-blink] Fix LookupJoinTest::testInvalidLoo…

f79164e

…kupTableFunction()

[FLINK-14599][table-planner-blink] Fix comparator of Timestamp type

e76461e

[FLINK-14599][table-planner-blink] Fix ROWTIME/PROCTIME and window pr…

89e2ab2

…ops in window

[FLINK-14599][table-planner-blink] Fix PROCTIME in match recognize

d3b1bc9

docete added 11 commits November 13, 2019 16:48

[FLINK-14599][table-planner-blink] Introduce LegacyTimestampTypeInfo …

287c4a9

…and LegacyLocalDateTimeTypeInfo to hold precision on conversion

[FLINK-14599][table-planner-blink] Support precision of Timestamp typ…

52cad96

…e in table source

[FLINK-14599][table-planner-blink] Support precision of Timestamp typ…

11222cd

…e for timestamp literals

fixup: address comments

8860597

fixup: fix checkstyle

dfeca05

[FLINK-14599][table-planner-blink] Make the default precision of Time…

d17efb5

…stamp type as 6 which defined in SQL standard

[FLINK-14599][table-planner-blink] Make the interpreted string of Tim…

c3412c4

…estamp type conforms to the definition of timestamp literal which defined in SQL standard

fixup: ignore Gregorian shift

6b82048

fixup: address comments

aab2160

fixup: store nanoOfMillisecond in fixed-length part to save 4 bytes p…

ee2854a

…er high precision timestamp

fixup: address comments

212e234

docete force-pushed the FLINK-14599 branch from 9a4e8a2 to 212e234 Compare November 13, 2019 08:55

JingsongLi requested changes Nov 13, 2019

View reviewed changes

wuchong reviewed Nov 14, 2019

View reviewed changes

docete closed this Nov 20, 2019

docete deleted the FLINK-14599 branch November 20, 2019 10:58


		val STRING_UTIL: String = className[BinaryStringUtil]

		val SQL_TIMESTAMP_TERM: String = className[SqlTimestamp]

[Flink-14599][table-planner-blink] Support precision of TimestampType in blink planner #10105

[Flink-14599][table-planner-blink] Support precision of TimestampType in blink planner #10105

Uh oh!

Conversation

docete commented Nov 6, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

flinkbot commented Nov 6, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Automated Checks

Review Progress

Uh oh!

flinkbot commented Nov 6, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

KurtYoung left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

docete Nov 7, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

KurtYoung left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

docete commented Nov 6, 2019 •

edited

Loading

flinkbot commented Nov 6, 2019 •

edited

Loading

flinkbot commented Nov 6, 2019 •

edited

Loading

docete Nov 7, 2019 •

edited

Loading