[FLINK-36851][table] Introduce TemporalProcessTimeJoinOperator in TemporalJoin with Async State API #25777
Conversation
…poralJoin with Async State API
| maxRetentionTime, | ||
| isLeftOuterJoin); | ||
| if (config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_STATE_ENABLED)) { | ||
| return new AsyncStateTemporalProcessTimeJoinOperator( |
There was a problem hiding this comment.
The Jira mentions SQL and table and the component in the PR title mentions [TABLE], but the fix and test appear to be datastream only. Is there an intent that this effects table API and SQL?
Please could you update the documentation -as this is new changes to the API.
There was a problem hiding this comment.
Thanks for review. This PR is for completing a part of Flip-473. Its intention is similar to #25320 .
The original API has not been changed, it will only switch the TemporalProcessTimeJoinOperator under the FLINK-TABLE module to be executed by Asynchronous StateTemporalProcessTimeJoinOperator when table. exec.async-state.enable is set to true.
Is there anything else that needs to be modified? If so, please let me know. Thank you
There was a problem hiding this comment.
Thanks for review. This PR is for completing a part of Flip-473. Its intention is similar to #25320 .
The original API has not been changed, it will only switch the TemporalProcessTimeJoinOperator under the FLINK-TABLE module to be executed by Asynchronous StateTemporalProcessTimeJoinOperator when table. exec.async-state.enable is set to true.
Is there anything else that needs to be modified? If so, please let me know. Thank you
ok thanks for the clarification
| minRetentionTime, | ||
| maxRetentionTime, | ||
| isLeftOuterJoin); | ||
| if (config.get(ExecutionConfigOptions.TABLE_EXEC_ASYNC_STATE_ENABLED)) { |
There was a problem hiding this comment.
From the other PR, As a heads up, changes like this should update consumedOptions for the JSON serialization.
(Sorry for splitting the conversation across PRs.)
There was a problem hiding this comment.
The settings related to this configuration have been updated in #25320.
I don't quite understand the meaning of what you mean by 'update consumedOptions for the JSON serialization'. I would greatly appreciate it if you could give me some more tips.
There was a problem hiding this comment.
The annotations here: https://github.com/apache/flink/pull/25777/files#diff-76717e3a4ed9896387a0961a4ff297ccd8a7dd4ef19b49605cf0755fb1d0d1dbR73-R78 show provide information about how to serialize this node to JSON.
Checkout other examples and you'll see a field named consumedOptions which tells which options are used the node.
If I understand, you are adding a new option which is used by this class. As such, you need to advertise it in the annotation.
There was a problem hiding this comment.
Okay, thanks. I have added the corresponding explanation for consumedOptions.
Au-Miner
force-pushed
the
FLINK-36851
branch
from
5685f1f to
e6dd14c
Compare
| * <p>1) if your timers collide with clean up timers and you delete them, then state clean-up will | ||
| * not be performed, and | ||
| * | ||
| * <p>2) (this one is the reason why this class does not allow to override the onProcessingTime()) |
There was a problem hiding this comment.
nit: the brackets phrase would sit better at the end of 1)
There was a problem hiding this comment.
The 1) sentence and 2) sentence are continuous, meaning that the content on line 58 should be immediately followed by 'and', so I think there is no need to add it
|
Reviewed by Chi on 12/12/24. appears that this PR is healthily progressing |
What is the purpose of the change
Introduce TemporalProcessTimeJoinOperator in TemporalJoin with async state api.
Brief change log
Verifying this change
Existent tests and new added tests can verify this change.
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation