[
https://issues.apache.org/jira/browse/FLINK-5990?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15934844#comment-15934844
]
ASF GitHub Bot commented on FLINK-5990:
---------------------------------------
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/3585
Hi @sunjincheng121, thanks for this PR.
To be honest, I don't completely understand the implementation of the
`RowsClauseBoundedOverProcessFunction`.
I thought about another design, that I would like to discuss (not
considering process time which should be addressed in a separate
ProcessFunction, IMO.):
- we have three state objects: 1) the accumulator row, 2) a MapState[Long,
List[Row]] for not processed data (`toProcess`), 3) a MapState[Long, List[Row]]
for processed data which needs to be retracted (`toRetract`).
- processElement() puts the element in the `toProcess` MapState with the
original timestamp and registers a timer for `currentWatermark() + 1`. Hence,
we only have a single timer which triggers when the next watermark is reached.
- onTimer() is called for the next watermark. We get an iterator over the
`toProcess` MapState. For RocksDB the iterator is sorted on the key. We
sort-insert the records from the iterator into a `LinkedList` (since the
iterator is sorted for RocksDB this will be simple append. For other state
backends it will be more expensive but we can tolerate that, IMO). We do the
same for `toRetract` MapState. So we have two sorted lists for data to
accumulate and to retract. We go over both sorted lists and accumulate and
retract for each step using the accumulator state. Then we emit the new row and
move the emitted row from the `toProcess` MapState to the `toRetract` MapState.
This design has the benefit of using RocksDB to sort. Moreover, we could
also put only those fields into the toRetract state that need to be retracted
instead of the full row.
What do you think about this approach?
Thanks, Fabian
> 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)
