[
https://issues.apache.org/jira/browse/FLINK-5654?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15949408#comment-15949408
]
ASF GitHub Bot commented on FLINK-5654:
---------------------------------------
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/3641
Oh, I just remembered that the OVER RANGE semantics are slightly different.
All rows that arrive in the same millisecond need to get the same aggregation
value. So we need not only to retract all records which are too old but also
accumulate all records which are received in the same millisecond.
Therefore, we would need to redesign the `ProcessFunction` a bit.
- In `processElement()` we put the new record in the MapState and register
a processing time timer for current time + 1. This will create a call back on
`onTimer()` when current time + 1 is reached.
- When `onTimer()` is called, we process the rows of timestamp - 1, retract
all old values accumulate all new values and emit all rows of timestamp - 1.
The implementation of `onTimer()` can reuse most of what is currently done in
`processElement()`.
Does that make sense @rtudoran?
Do you want to make the change? Otherwise, I can also do it before merging.
Sorry for recognizing this just now :-/
Best, Fabian
> Add processing time OVER RANGE BETWEEN x PRECEDING aggregation to SQL
> ---------------------------------------------------------------------
>
> Key: FLINK-5654
> URL: https://issues.apache.org/jira/browse/FLINK-5654
> Project: Flink
> Issue Type: Sub-task
> Components: Table API & SQL
> Reporter: Fabian Hueske
> Assignee: radu
>
> The goal of this issue is to add support for OVER RANGE aggregations on
> processing 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 procTime() RANGE BETWEEN INTERVAL '1'
> HOUR PRECEDING AND CURRENT ROW) AS sumB,
> MIN(b) OVER (PARTITION BY c ORDER BY procTime() RANGE BETWEEN INTERVAL '1'
> HOUR 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 optional (no partitioning results in single
> threaded execution).
> - The ORDER BY clause may only have procTime() as parameter. procTime() is a
> parameterless scalar function that just indicates processing time mode.
> - UNBOUNDED PRECEDING is not supported (see FLINK-5657)
> - 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)
