[
https://issues.apache.org/jira/browse/FLINK-5990?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15937659#comment-15937659
]
ASF GitHub Bot commented on FLINK-5990:
---------------------------------------
Github user sunjincheng121 commented on the issue:
https://github.com/apache/flink/pull/3585
Hi @fhueske, in our point of view, register timer should be correlated to
the event time of the CURRENT data, but not currentWaterMark (which is
correlated to the event time of the PREVIOUS datas) . Using the PREVIOUS event
time (`currentWatermark + 1`) to register the timer for CURRENT data may cause
some problems IMO. I do not have whole lots of counterexamples for now, just
intuitively think this is not the correct way. One example I have is:
Data line -> ... 5 7 4 3 2
Watermark line -> ... 5 4 3 2 1
In this case, assuming when watermark is `3`，we get `4` and `7`, then we
use ` currentWatermark + 1` to register timer `timer(3+1)` for both `4` and
`7`. When watermark goes to `4`，timer `timer(3+1)` be triggered (then `4` and
`7` will be emitted). After this, data `5` comes (while the watermark is still
`4`), then we register timer `timer(4+1)` for 5. This is incorrect, as we
should always ensure data `5` is triggered before data `7`. I run a test case,
and printed the results of [event time][currentWatermark] as below. I hope it
can help to explain my point.
Datas:
```
(1L, 1, "Hello"),
(2L, 2, "Hello"),
(4L, 4, "Hello"),
(3L, 3, "Hello"),
(5L, 3, "Hello"),
(7L, 7, "Hello"),
(6L, 7, "Hello"),
(8L, 8, "Hello World"),
(7L, 8, "Hello"),
(5L, 5, "Hello"),
(20L, 20, "Hello World"),
(9L, 9, "Hello World"))
```
Event time of each data & Watermark: (newTS_currentWM[event
time][currentWatermark])
```
newTS_currentWM[1][-9223372036854775808]
newTS_currentWM[2][-1]
newTS_currentWM[4][0]
newTS_currentWM[3][2]
newTS_currentWM[5][2]
newTS_currentWM[7][3] // a data with event time 7, it will be registered
timer to (3+1)
newTS_currentWM[6][5] // a data with event time 6, it will be registered
timer to (5+1)
newTS_currentWM[8][5]
newTS_currentWM[7][6]
newTS_currentWM[5][6]
newTS_currentWM[20][6]
newTS_currentWM[9][18]
```
Another important reason of using allowedLateness to configure the
watermark in my design is that, in our point of view, watermark is a semantics
which is correlated to the stream lateness SLA. User exactly knows the tuple of
(SLA, allowedLateness). For instance:
******************************************
A system which always ensure the order:
SLA(100%) = Watermark(eventTime)
******************************************
For flink system process out of order data, and user have different
watermarks indicating different SLA levels:
(SLA, allowedLateness) = Watermark(eventTime - allowedLateness)
(SLA(80%), 0) = Watermark(eventTime)
(SLA(90%), 2) = Watermark(eventTime - 2)
(SLA(95%), 5) = Watermark(eventTime - 5)
We want to keep the interface to let user configure the
allowedLateness(AssignerWithPeriodicWatermarks|AssignerWithPunctuatedWatermarks),
thereby the correlated SLA.
What do you think ?
Thanks,
SunJincheng
> Add [partitioned] event time OVER ROWS BETWEEN x PRECEDING aggregation to SQL
> -----------------------------------------------------------------------------
>
> Key: FLINK-5990
> URL: https://issues.apache.org/jira/browse/FLINK-5990
> Project: Flink
> Issue Type: Sub-task
> Components: Table API & SQL
> Reporter: sunjincheng
> Assignee: sunjincheng
>
> The goal of this issue is to add support for OVER ROWS aggregations on event
> time streams to the SQL interface.
> Queries similar to the following should be supported:
> {code}
> SELECT
> a,
> SUM(b) OVER (PARTITION BY c ORDER BY rowTime() ROWS BETWEEN 2 PRECEDING AND
> CURRENT ROW) AS sumB,
> MIN(b) OVER (PARTITION BY c ORDER BY rowTime() ROWS BETWEEN 2 PRECEDING AND
> CURRENT ROW) AS minB
> FROM myStream
> {code}
> The following restrictions should initially apply:
> - All OVER clauses in the same SELECT clause must be exactly the same.
> - The PARTITION BY clause is required
> - The ORDER BY clause may only have rowTime() as parameter. rowTime() is a
> parameterless scalar function that just indicates event time mode.
> - UNBOUNDED PRECEDING is not supported (see FLINK-5803)
> - FOLLOWING is not supported.
> The restrictions will be resolved in follow up issues. If we find that some
> of the restrictions are trivial to address, we can add the functionality in
> this issue as well.
> This issue includes:
> - Design of the DataStream operator to compute OVER ROW aggregates
> - Translation from Calcite's RelNode representation (LogicalProject with
> RexOver expression).
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
