Status

Current state: under discussion

Discussion thread: 

JIRA: 

Released: 1.13.0

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Currently some temporal function behaviors are wired to users. 

• When users use a PROCTIME() in SQL, the return value of PROCTIME() has a timezone offset with the wall-clock time in users' local time zone, users need to add their local time zone offset manually to get expected local timestamp(e.g: Users in Germany need to +1h to get expected local timestamp). 
• Users can not use CURRENT_DATE/CURRENT_TIME/CURRENT_TIMESTAMP  to get wall-clock timestamp in local time zone, and thus they need write UDF in their SQL just for implementing a simple filter like WHERE date_col =  CURRENT_DATE. 

• Another common case  is the time window  with day interval based on PROCTIME(), user plan to put all data from one day into the same window, but the window is assigned using timestamp in UTC+0 timezone rather than the session timezone which leads to the window starts with an offset(e.g: Users in China need to add -8h in their business sql start and then +8h when output the result, the conversion like a magic for users). 

These problems come from the fact that lots of time-related functions like PROCTIME(), NOW(), CURRENT_DATE, CURRENT_TIME and CURRENT_TIMESTAMP are returning time values based on UTC+0 time zone.

This FLIP aims to consistent the timestamp function behavior and eventually improve the usability.

Public Interfaces 

As we knew some functions' behavior is wrong currently, but after we correct these function, the legacy behavior should still work in old code.

This FLIP introduce an option for compatibility consideration, given two option names, I prefer the first one, the second one is like Impala style[1].

table.exec.fallback-legacy-time-function = false/true 

table.exec.use-utc-for-unixtime-conversion = false/true    

Proposed Changes

1. Correct the return value of time functions

I invested all Flink time-related functions current behavior and compared with other DB vendors like Pg,Presto, Hive, Spark, Snowflake,  I made an excel [2] to organize them well.

To correct our current behavior, we need to make the function return type clear, especially for timestamp type, i.e.

• TIMESTAMP/TIMESTAMP WITHOUT TIME ZONE
• TIMESTAMP WITH LOCAL TIME ZONE
• TIMESTAMP WITH TIME ZONE

In order to understand these types better, I wrote a document[3]. BTW, Flink also keeps same semantics for three timestamp types comparing with Hadoop ecosystem.

From my investigation, to correct this time functions' behavior, we have several options
(1) change the function return type
(2) change the function return value
(3) change them both.

All of those way are valid because SQL:2011 does not specify the function return type and every SQL engine vendor has its own implementation[2], for example CURRENT_TIMESTAMP.

This FLIP proposes option 2 which only change the return value, the proposed changes as following, the current wall-clock is 2020-12-29 07:52:52 in Beijing time(UTC+8):

After the proposal, the function NOW(), CURRENT_TIMESTAMP and LOCALTIMESTAMP become  synonyms, the function CURRENT_TIME and LOCALTIME become synonyms, you can also lookup all time function behaviors in reference [2]. 

2. Disable CAST between NUMERIC  and TIMESTAMP

Currently, the following CAST conversion behaviors are wrong which does not consider the session time zone. It should use session time zone when cast between NUMERIC type and TIMESTAMP type if there’re strong requirements to support this , the numeric type include TINYINT, SMALLINT, INT, BIGINT, FLOAT, DOUBLE, they keep same conversion behavior, for example the cast between BIGINT and TIMESTAMP. Although we didn’t expose this feature in the document, some users may use them.

These cast conversions have wrong behavior and problematic semantics,  because SQL:2011 does not contains these cast specification and we never expose to user.

This FLIP propose to disable the cast between numeric and timestamp,  and introduce a the function TO_TIMESTAMP(seconds) function which accepts a BIGINT seconds and return a TIMESTAMP in local time zone. 

3. Consider the timezone offset in conversion between Table and DataStream

The row time field is stored  in a separate field with Long type of StreamRecord in DataStream world, and the value with the timestamp is the milliseconds since java epoch(1970-01-01 00:00:00 UTC+0), which definitely represents an Instant semantics. 

When converting a Table contains row time from/to DataStream, the timestamp in StreamRecord will exchange with Table’s row time column.

We need to consider the time zone offset just like the cast conversion between the TIMESTAMP with wall-clock semantics and BIGINT with Instant semantics.  This way can make the both DataStream and SQL/API users feel intuitive.

4. Introduce an option that enable fallback to legacy behavior

 The default value of table.exec.fallback-legacy-time-function is 'false' which means

(1) the return values of function

• CURRENT_DATE
• CURRENT_TIME
• CURRENT_TIMESTAMP
• NOW()
• PROCTIME()

consider the local timezone.

(2) The cast between NUMERIC and TIMESTAMP is forbidden

(3) The session timezone offset will be considered when convert a Table from/to DataStream.

Users can set the option to 'true', at this moment:

(1) These functions would keep the legacy behavior.

(2) The cast between NUMERIC and TIMESTAMP is supported.

(3)The session timezone offset will not be considered when convert a Table from/to DataStream.

General Implementations

1.Change the codegen implementations for above five functions/cast conversions according to the value of introduced table option: table.exec.fallback-legacy-time-function

2.Consider the session time zone offset when materializing a PROCTIME() attribute, including PROCTIME_MATERIALIZE function and the processing timestamp used to register timer in the window operator, because after change the return value of PROCTIME(）function,

the window based on processing time should be triggered by the changed processing time value.

Compatibility, Deprecation, and Migration Plan

• Compatibility

This is an incompatible change, we introduce SQL/API option  table.exec.fallback-legacy-time-function for compacting current wrong behavior, and set it to ‘false’. If users want to keep the legacy behavior, they need to set it to ’true’ manually, this would be add to release note. 

• Migration Plan

After the proposal is finished, the above user cases will work smoothly.  Assume users' local time zone is UTC+8, the wall-clock is 2020-12-29 07:52:52.

• user case 1 :

Flink SQL> SELECT NOW(), PROCTIME(), CURRENT_TIMESTAMP, CURRENT_DATE, CURRENT_TIME;
-- output:
+-------------------------+-------------------------+-------------------------+--------------+--------------+
|                   NOW() |              PROCTIME() |       CURRENT_TIMESTAMP | CURRENT_DATE | CURRENT_TIME |
+-------------------------+-------------------------+-------------------------+--------------+--------------+
|     2020-12-29T07:52:52 |     2020-12-29T07:52:52 |     2020-12-29T07:52:52 |   2020-12-29 | 07:52:52.236 |
+-------------------------+-------------------------+-------------------------+--------------+--------------+

• user case 2: 

Flink SQL> SELECT TUMBLE_START(proctime, INTERVAL ‘1’ DAY),
         >        TUMBLE_END(proctime, INTERVAL ‘1’ DAY),
         >        count(userId) as cnt
         >    FROM userLog
         >    GROUP BY TUMBLE_WINDOW(proctime, INTERVAL ‘1’ DAY);

-- output:
+-------------------------+-------------------------+-------------------------+ 
|            TUMBLE_START |              TUMBLE_END |           count(userId) | 
+-------------------------+-------------------------+-------------------------+ 
|     2020-12-29T00:00:00 |     2020-12-30T00:00:00 |     		  		  100 |
+-------------------------+-------------------------+-------------------------+ 

• user case 3:

Flink SQL> SELECT  *
         >    FROM userLog
         >    WHERE date >= CURRENT_DATE;

-- in the query, records earlier than 2020-12-29 will not be output.
+-------------------------+-------------------------+-------------------------+ 
|                    date |                  log_ts |                    user | 
+-------------------------+-------------------------+-------------------------+ 
|             2020-12-29  |     2020-12-29T00:00:00 |     		  		Alice |
+-------------------------+-------------------------+-------------------------+ 
|             2020-12-29  |     2020-12-29T00:00:01 |     		  		  Bob |
+-------------------------+-------------------------+-------------------------+ 
|             2020-12-29  |     2020-12-29T00:00:02 |     		  		  Tom |
+-------------------------+-------------------------+-------------------------+ 

• int the query 

records earlier than 2020-12-29 will not be output.

Test Plan

Will add plan tests, unit tests, window operator harness tests as well as IT tests. 

Rejected Alternatives

1. Change the return type of function CURRENT_TIMESTAMP/NOW()/PROCETIME() to TIMESTAMP WITH TIME ZONE. This proposal needs to introduce a new type TIMESTAMP WITH TIME ZONE, and we think there are no enough benefits. If we do this, the return type of function CURRENT_TIME must be TIME WITH TIME ZONE for consistent consideration, we need to introduce another type.

2. Change the return type of function  CURRENT_TIMESTAMP/NOW()/PROCETIME() to TIMESTAMP WITH LOCAL TIME ZONE. This proposal will lead to a embarrassed situation for function CURRENT_TIME, because no DB vendor use type TIME WITH LOCAL TIME ZONE yet, and the time semantic of TIMESTAMP WITH LOCAL TIME ZONE is instant which is too complex to understand for normal users.

References:

1. https://docs.cloudera.com/documentation/enterprise/latest/topics/impala_timezone.html

      2. https://docs.google.com/document/d/1iY3eatV8LBjmF0gWh2JYrQR0FlTadsSeuCsksOVp_iA/edit?usp=sharing

      3. https://docs.google.com/spreadsheets/d/1T178krh9xG-WbVpN7mRVJ8bzFnaSJx3l-eg1EWZe_X4/edit?usp=sharing