[
https://issues.apache.org/jira/browse/FLINK-5658?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15889387#comment-15889387
]
Yuhong Hong commented on FLINK-5658:
------------------------------------
Hi [~fhueske], as we talked before, We simply discard late arriving records.
I plan to use watermark, and discard record before current watermark.
like
{code}
A(Time:10) -> agg(A)
B(Time:11) -> agg(A,B)
C(Time:12) -> agg(A,B,C)
-------watermark(Time:12)----------
D(Time:11) -> _ //just discard
E(Time:13) -> agg(A,B,C,E)
{code}
But as the records that not be discarded, the result will be different
according to the order they come,
such as
{code}
A(Time:10) -> agg(A) B(Time:11) -> agg(B) C(Time:12)
-> agg(C)
B(Time:11) -> agg(A,B) A(Time:10) -> agg(A) B(Time:11)
-> agg(B)
C(Time:12) -> agg(A,B,C) C(Time:12) -> agg(B,C) A(Time:10)
-> agg(A)
{code}
so i think a retraction operation may be needed to make sure they have same
results.
{code}
B(Time:11) -> agg(B)
A(Time:10) -> agg(A), agg(A,B)
C(Time:12) -> agg(A,B,C)
{code}
What do you thnk? hope can give some suggestion.
> Add event time OVER RANGE BETWEEN UNBOUNDED PRECEDING aggregation to SQL
> ------------------------------------------------------------------------
>
> Key: FLINK-5658
> URL: https://issues.apache.org/jira/browse/FLINK-5658
> Project: Flink
> Issue Type: Sub-task
> Components: Table API & SQL
> Reporter: Fabian Hueske
> Assignee: Yuhong Hong
>
> The goal of this issue is to add support for OVER RANGE 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() RANGE BETWEEN UNBOUNDED
> PRECEDING AND CURRENT ROW) AS sumB,
> MIN(b) OVER (PARTITION BY c ORDER BY rowTime() RANGE BETWEEN UNBOUNDED
> 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 rowTime() as parameter. rowTime() is a
> parameterless scalar function that just indicates processing time mode.
> - bounded PRECEDING is not supported (see FLINK-5655)
> - 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)
