You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by "lorenzo.affetti.ververica.com via user" <us...@flink.apache.org> on 2024/03/05 08:57:12 UTC
Re: Temporal join on rolling aggregate
I do agree with Ron Liu.
This would definitely need a FLIP as it would impact SQL and extend it with the equivalent of TimestampAssigners in the Java API.
Is there any existing JIRA here, or is anybody willing to drive a FLIP?
On Feb 26, 2024 at 02:36 +0100, Ron liu <ro...@gmail.com>, wrote:
> +1,
> But I think this should be a more general requirement, that is, support for
> declaring watermarks in query, which can be declared for any type of
> source, such as table, view. Similar to databricks provided [1], this needs
> a FLIP.
>
> [1]
> https://docs.databricks.com/en/sql/language-manual/sql-ref-syntax-qry-select-watermark.html
>
> Best,
> Ron
Re: Temporal join on rolling aggregate
Posted by Matthias Broecheler <ma...@datasqrl.com>.
Hey Sebastien et al,
have you tried rewriting the rolling aggregate as a window-over query? A
window-over aggregation creates an append-only stream which should preserve
the timestamp/watermark of the source. You can then add a deduplication
<https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/deduplication/>
to
create a versioned state that you can use in a temporal join. The partition
key of the deduplication becomes the primary key to join on.
I think what you are running into is that Flink creates a change-stream for
a group-by aggregation that has retractions. Temporal joins only work on
"versioned state" or lookup tables.
However, I think you have a valid point in a more foundational limitation
of FlinkSQL: It is currently not possible to create an append-only stream
from a change stream without going through the DataStream API. It would be
incredibly useful to support this natively in FlinkSQL. And Calcite has
support for the STREAM <https://calcite.apache.org/docs/stream.html>
keyword to do this. Materialize (i.e. differential dataflow) has a somewhat
related construct in their SQL called a subscription
<https://materialize.com/docs/sql/subscribe/>.
We added support for this in DataSQRL
<https://www.datasqrl.com/docs/reference/sqrl/stream/#convert-state-to-stream>
(which
is a streaming database compiler that generates Flink jobs) using a syntax
that looks like this:
STREAM ON UPDATE AS (*YOUR BASE QUERY*)
to create a append-only stream from a change-stream by essentially dropping
all retractions and deletes (you can also do STREAM ON DELETE to get only
deletes, etc). However, I think this might be a feature that should live in
FlinkSQL instead and we'd be happy to create a FLIP and donate our
implementation if there is interest.
Cheers,
Matthias
On Mon, Mar 18, 2024 at 3:01 AM Sebastien <se...@erreur404.ch> wrote:
> Hi everyone,
>
> Before digging into what it would it take to implement a general solution,
> I narrowed down the scope to write a fix which makes the query mentioned in
> the thread work. Here are some findings:
>
> - For the temporal join logic, it's not the watermark that matters but
> having a TimeIndicatorRelDataType column in the input relation. To address
> that, in the PR below, we introduced a call to the LAST_VALUE aggregate
> function to bring a timestamp column to the view. That makes the query
> works, but we think it is not enough. It would probably require a distinct
> aggregate function or a new syntax to be able to support more general use
> cases.
> - There is a relationship between the way the logical operators are
> reordered, the way the special Flink's Timestamp time is materialized and
> the watermark assigner.
> - I also looked into the flink-sql-parser and I found out that Flink has
> customized the parsing of the CREATE and DROP statements (
> https://github.com/apache/flink/blob/master/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd#L627-L638)
> (and Calcite supports as well support customizations for ALTER statements)
> but Calcite does not seem to support changes to the SELECT syntax (see
> https://issues.apache.org/jira/browse/CALCITE-4979). I mention it because
> I think it will inform what could be done syntax-wise.
>
> and a PR that highlights the changes with the complete investigation
> https://github.com/apache/flink/pull/24506
>
> This is more of a demonstration and I am looking to get feedback from
> someone who has more experience with the codebase.
>
> Thanks,
> Seb
>
> On Tue, Mar 5, 2024, at 10:07, Gyula Fóra wrote:
>
> Hi Everyone!
>
> I have discussed this with Sébastien Chevalley, he is going to prepare and
> drive the FLIP while I will assist him along the way.
>
> Thanks
> Gyula
>
> On Tue, Mar 5, 2024 at 9:57 AM <lo...@ververica.com> wrote:
>
> I do agree with Ron Liu.
> This would definitely need a FLIP as it would impact SQL and extend it
> with the equivalent of TimestampAssigners in the Java API.
>
> Is there any existing JIRA here, or is anybody willing to drive a FLIP?
> On Feb 26, 2024 at 02:36 +0100, Ron liu <ro...@gmail.com>, wrote:
>
> +1,
> But I think this should be a more general requirement, that is, support for
> declaring watermarks in query, which can be declared for any type of
> source, such as table, view. Similar to databricks provided [1], this needs
> a FLIP.
>
> [1]
>
> https://docs.databricks.com/en/sql/language-manual/sql-ref-syntax-qry-select-watermark.html
>
> Best,
> Ron
>
>
>
Re: Temporal join on rolling aggregate
Posted by Sebastien <se...@erreur404.ch>.
Hi everyone,
Before digging into what it would it take to implement a general solution, I narrowed down the scope to write a fix which makes the query mentioned in the thread work. Here are some findings:
- For the temporal join logic, it's not the watermark that matters but having a TimeIndicatorRelDataType column in the input relation. To address that, in the PR below, we introduced a call to the LAST_VALUE aggregate function to bring a timestamp column to the view. That makes the query works, but we think it is not enough. It would probably require a distinct aggregate function or a new syntax to be able to support more general use cases.
- There is a relationship between the way the logical operators are reordered, the way the special Flink's Timestamp time is materialized and the watermark assigner.
- I also looked into the flink-sql-parser and I found out that Flink has customized the parsing of the CREATE and DROP statements (https://github.com/apache/flink/blob/master/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd#L627-L638) (and Calcite supports as well support customizations for ALTER statements) but Calcite does not seem to support changes to the SELECT syntax (see https://issues.apache.org/jira/browse/CALCITE-4979). I mention it because I think it will inform what could be done syntax-wise.
and a PR that highlights the changes with the complete investigation https://github.com/apache/flink/pull/24506
This is more of a demonstration and I am looking to get feedback from someone who has more experience with the codebase.
Thanks,
Seb
On Tue, Mar 5, 2024, at 10:07, Gyula Fóra wrote:
> Hi Everyone!
>
> I have discussed this with Sébastien Chevalley, he is going to prepare and drive the FLIP while I will assist him along the way.
>
> Thanks
> Gyula
>
> On Tue, Mar 5, 2024 at 9:57 AM <lo...@ververica.com> wrote:
>> I do agree with Ron Liu.
>> This would definitely need a FLIP as it would impact SQL and extend it with the equivalent of TimestampAssigners in the Java API.
>>
>> Is there any existing JIRA here, or is anybody willing to drive a FLIP?
>> On Feb 26, 2024 at 02:36 +0100, Ron liu <ro...@gmail.com>, wrote:
>>> +1,
>>> But I think this should be a more general requirement, that is, support for
>>> declaring watermarks in query, which can be declared for any type of
>>> source, such as table, view. Similar to databricks provided [1], this needs
>>> a FLIP.
>>>
>>> [1]
>>> https://docs.databricks.com/en/sql/language-manual/sql-ref-syntax-qry-select-watermark.html
>>>
>>> Best,
>>> Ron
Re: Temporal join on rolling aggregate
Posted by Sebastien <se...@erreur404.ch>.
Hi everyone,
Before digging into what it would it take to implement a general solution, I narrowed down the scope to write a fix which makes the query mentioned in the thread work. Here are some findings:
- For the temporal join logic, it's not the watermark that matters but having a TimeIndicatorRelDataType column in the input relation. To address that, in the PR below, we introduced a call to the LAST_VALUE aggregate function to bring a timestamp column to the view. That makes the query works, but we think it is not enough. It would probably require a distinct aggregate function or a new syntax to be able to support more general use cases.
- There is a relationship between the way the logical operators are reordered, the way the special Flink's Timestamp time is materialized and the watermark assigner.
- I also looked into the flink-sql-parser and I found out that Flink has customized the parsing of the CREATE and DROP statements (https://github.com/apache/flink/blob/master/flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd#L627-L638) (and Calcite supports as well support customizations for ALTER statements) but Calcite does not seem to support changes to the SELECT syntax (see https://issues.apache.org/jira/browse/CALCITE-4979). I mention it because I think it will inform what could be done syntax-wise.
and a PR that highlights the changes with the complete investigation https://github.com/apache/flink/pull/24506
This is more of a demonstration and I am looking to get feedback from someone who has more experience with the codebase.
Thanks,
Seb
On Tue, Mar 5, 2024, at 10:07, Gyula Fóra wrote:
> Hi Everyone!
>
> I have discussed this with Sébastien Chevalley, he is going to prepare and drive the FLIP while I will assist him along the way.
>
> Thanks
> Gyula
>
> On Tue, Mar 5, 2024 at 9:57 AM <lo...@ververica.com> wrote:
>> I do agree with Ron Liu.
>> This would definitely need a FLIP as it would impact SQL and extend it with the equivalent of TimestampAssigners in the Java API.
>>
>> Is there any existing JIRA here, or is anybody willing to drive a FLIP?
>> On Feb 26, 2024 at 02:36 +0100, Ron liu <ro...@gmail.com>, wrote:
>>> +1,
>>> But I think this should be a more general requirement, that is, support for
>>> declaring watermarks in query, which can be declared for any type of
>>> source, such as table, view. Similar to databricks provided [1], this needs
>>> a FLIP.
>>>
>>> [1]
>>> https://docs.databricks.com/en/sql/language-manual/sql-ref-syntax-qry-select-watermark.html
>>>
>>> Best,
>>> Ron
Re: Temporal join on rolling aggregate
Posted by Gyula Fóra <gy...@gmail.com>.
Hi Everyone!
I have discussed this with Sébastien Chevalley, he is going to prepare and
drive the FLIP while I will assist him along the way.
Thanks
Gyula
On Tue, Mar 5, 2024 at 9:57 AM <lo...@ververica.com> wrote:
> I do agree with Ron Liu.
> This would definitely need a FLIP as it would impact SQL and extend it
> with the equivalent of TimestampAssigners in the Java API.
>
> Is there any existing JIRA here, or is anybody willing to drive a FLIP?
> On Feb 26, 2024 at 02:36 +0100, Ron liu <ro...@gmail.com>, wrote:
>
> +1,
> But I think this should be a more general requirement, that is, support for
> declaring watermarks in query, which can be declared for any type of
> source, such as table, view. Similar to databricks provided [1], this needs
> a FLIP.
>
> [1]
>
> https://docs.databricks.com/en/sql/language-manual/sql-ref-syntax-qry-select-watermark.html
>
> Best,
> Ron
>
>
Re: Temporal join on rolling aggregate
Posted by Gyula Fóra <gy...@gmail.com>.
Hi Everyone!
I have discussed this with Sébastien Chevalley, he is going to prepare and
drive the FLIP while I will assist him along the way.
Thanks
Gyula
On Tue, Mar 5, 2024 at 9:57 AM <lo...@ververica.com> wrote:
> I do agree with Ron Liu.
> This would definitely need a FLIP as it would impact SQL and extend it
> with the equivalent of TimestampAssigners in the Java API.
>
> Is there any existing JIRA here, or is anybody willing to drive a FLIP?
> On Feb 26, 2024 at 02:36 +0100, Ron liu <ro...@gmail.com>, wrote:
>
> +1,
> But I think this should be a more general requirement, that is, support for
> declaring watermarks in query, which can be declared for any type of
> source, such as table, view. Similar to databricks provided [1], this needs
> a FLIP.
>
> [1]
>
> https://docs.databricks.com/en/sql/language-manual/sql-ref-syntax-qry-select-watermark.html
>
> Best,
> Ron
>
>