Hi,
I'd like to start an discussion about how Table API / SQL queries indicate
whether an operation is done in event or processing time.
1) Why do we need to indicate the time mode?
We need to distinguish event time and processing time mode for operations
in queries in order to have the semantics of a query fully defined.
This cannot be globally done in the TableEnvironment because some queries
explicitly request an expression such as the ORDER BY clause of an OVER
window with PRECEDING / FOLLOWING clauses.
So we need a way to specify something like the following query:
SELECT
a,
SUM(b) OVER (PARTITION BY c ORDER BY proctime ROWS BETWEEN 2
PRECEDING AND CURRENT ROW) AS sumB,
FROM myStream
where "proctime" indicates processing time. Equivalently "rowtime" would
indicate event time.
2) Current state
The current master branch implements time support only for grouping windows
in the Table API.
Internally, the Table API converts a 'rowtime symbol (which looks like a
regular attribute) into a special expression which indicates event-time.
For example:
table
.window(Tumble over 5.milli on 'rowtime as 'w)
.groupBy('a, 'w)
.select(...)
defines a tumbling event-time window.
Processing-time is indicated by omitting a time attribute
(table.window(Tumble over 5.milli as 'w) ).
3) How can we do that in SQL?
In SQL we cannot add special expressions without touching the parser which
we don't want to do because we want to stick to the SQL standard.
Therefore, I see only two options: adding system attributes or
(parameterless) built-in functions. I list some pros and cons of the
approaches below:
1. System Attributes:
+ most natural way to access a property of a record.
+ works with joins, because time attributes can be related to tables
- We need to ensure the attributes are not writable and always present in
streaming tables (i.e., they should be system defined attributes).
- Need to adapt existing Table API expressions (will not change the API but
some parts of the internal translation)
- Event time value must be set when the stream is converted, processing
time is evaluated on the fly
2. Built-in Functions
+ Users could try to modify time attributes which is not possible with
functions
- do not work with joins, because we need to address different relations
- not a natural way to access a property of a record
I think the only viable choice are system attributes, because built-in
functions cannot be used for joins.
However, system attributes are the more complex solution because they need
a better integration with Calcite's SQL validator (preventing user
attributes which are named rowtime for instance).
Since there are currently a several contributions on the way (such as SQL
OVER windows FLINK-5653 to FLINK-5658) that need time indicators, we need a
solution soon to be able to make progress.
There are two PRs, #3252 and #3271, which implement the built-in marker
functions proctime() and rowtime() and which could serve as a temporary
solution (since we do not work on joins yet).
I would like to suggest to use these functions as a starting point (once
the PRs are merged) and later change to the system attribute solution which
needs a bit more time to be implemented.
I talked with Timo today about this issue and he said he would like to
investigate how we can implement this as system functions properly
integrated with Calcite and the SQL Validator.
What do others think?
Best, Fabian
