You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fhueske (via GitHub)" <gi...@apache.org> on 2026/09/15 15:36:36 UTC
[PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
fhueske opened a new pull request, #29195:
URL: https://github.com/apache/flink/pull/29195
## What is the purpose of the change
We can simplify the signature and usage of the `SNAPSHOT` function by removing the `load_completed_condition` argument.
Instead we infer the condition from the presence of the `load_completed_time` argument.
## Brief change log
* removed `load_completed_condition` argument and all related checks
* adjusted inference logic to use presence of `load_completed_time` argument to determine condition
* adjusted all related tests (incl. golden plan xml files)
* adjusted the documentation
## Verifying this change
* run existing tests
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no (`SNAPSHOT` function is not released yet)
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes (change function signature)
- If yes, how is the feature documented? docs were adjusted
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Sonnet 5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29195:
URL: https://github.com/apache/flink/pull/29195#issuecomment-5683471283
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4750060334ec1e70b448af6c89747a781ca5624a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4750060334ec1e70b448af6c89747a781ca5624a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4750060334ec1e70b448af6c89747a781ca5624a UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on PR #29195:
URL: https://github.com/apache/flink/pull/29195#issuecomment-5694216168
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on PR #29195:
URL: https://github.com/apache/flink/pull/29195#issuecomment-5694226676
@twalthr I've added a commit to derive the same timestamp for all SNAPSHOT functions in a query.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29195:
URL: https://github.com/apache/flink/pull/29195#discussion_r4017851962
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java:
##########
@@ -169,37 +162,20 @@ public void onMatch(RelOptRuleCall call) {
LateralSnapshotTypeStrategy.STATE_TTL_ARG_INDEX,
LateralSnapshotTypeStrategy.STATE_TTL_ARG_NAME);
- // Resolve load_completed_time according to load_completed_condition. The default
- // 'compile_time' uses the wall-clock time at planning; 'user_time' uses the user-provided
- // load_completed_time (which the type strategy guarantees is present for 'user_time').
- final String condition =
- conditionLiteral == null ? null : conditionLiteral.getValueAs(String.class);
+ // The presence of load_completed_time determines the load-completion mode: if the user
+ // provided it, the load phase completes at the specified event time ('user_time');
+ // otherwise it completes when the build-side event time exceeds the wall-clock time the
+ // query is compiled ('compile_time').
+ // The effective load completed condition is carried for explain output.
final Long loadCompletedTime;
- if (condition == null
- || LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_COMPILE_TIME.equals(
- condition)) {
- loadCompletedTime = System.currentTimeMillis();
- } else if (LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_USER_TIME.equals(
- condition)) {
- loadCompletedTime =
- loadCompletedTimeLiteral == null
- ? null
- : loadCompletedTimeLiteral.getValueAs(Long.class);
- if (loadCompletedTime == null) {
- throw new ValidationException(
- "SNAPSHOT requires 'load_completed_time' when "
- + "'load_completed_condition' is 'user_time'.");
- }
+ final String loadCompletedCondition;
+ if (loadCompletedTimeLiteral != null) {
+ loadCompletedTime = loadCompletedTimeLiteral.getValueAs(Long.class);
+ loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_USER_TIME;
} else {
- throw new ValidationException(
- String.format("Unknown SNAPSHOT 'load_completed_condition': '%s'.", condition));
+ loadCompletedTime = System.currentTimeMillis();
Review Comment:
should all lateral snapshot functions use the same timestamp? this would make sense and aligns with NOW() semantics of batch systems. let's make this a follow up. we can follow batch and create a config option that get's set when planning starts.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java:
##########
@@ -46,6 +46,19 @@
@Internal
public final class LateralSnapshotJoinUtil {
+ /**
+ * Label for {@code FlinkLogicalLateralSnapshotJoin#getLoadCompletedCondition()} when the load
Review Comment:
who uses getLoadCompletedCondition after this change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske merged PR #29195:
URL: https://github.com/apache/flink/pull/29195
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on code in PR #29195:
URL: https://github.com/apache/flink/pull/29195#discussion_r4023337143
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java:
##########
@@ -46,6 +46,19 @@
@Internal
public final class LateralSnapshotJoinUtil {
+ /**
+ * Label for {@code FlinkLogicalLateralSnapshotJoin#getLoadCompletedCondition()} when the load
Review Comment:
It's only used for EXPLAIN to see whether the load_completed_time was manually set or automatically set at compile time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on code in PR #29195:
URL: https://github.com/apache/flink/pull/29195#discussion_r4023475540
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java:
##########
@@ -169,37 +162,20 @@ public void onMatch(RelOptRuleCall call) {
LateralSnapshotTypeStrategy.STATE_TTL_ARG_INDEX,
LateralSnapshotTypeStrategy.STATE_TTL_ARG_NAME);
- // Resolve load_completed_time according to load_completed_condition. The default
- // 'compile_time' uses the wall-clock time at planning; 'user_time' uses the user-provided
- // load_completed_time (which the type strategy guarantees is present for 'user_time').
- final String condition =
- conditionLiteral == null ? null : conditionLiteral.getValueAs(String.class);
+ // The presence of load_completed_time determines the load-completion mode: if the user
+ // provided it, the load phase completes at the specified event time ('user_time');
+ // otherwise it completes when the build-side event time exceeds the wall-clock time the
+ // query is compiled ('compile_time').
+ // The effective load completed condition is carried for explain output.
final Long loadCompletedTime;
- if (condition == null
- || LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_COMPILE_TIME.equals(
- condition)) {
- loadCompletedTime = System.currentTimeMillis();
- } else if (LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_USER_TIME.equals(
- condition)) {
- loadCompletedTime =
- loadCompletedTimeLiteral == null
- ? null
- : loadCompletedTimeLiteral.getValueAs(Long.class);
- if (loadCompletedTime == null) {
- throw new ValidationException(
- "SNAPSHOT requires 'load_completed_time' when "
- + "'load_completed_condition' is 'user_time'.");
- }
+ final String loadCompletedCondition;
+ if (loadCompletedTimeLiteral != null) {
+ loadCompletedTime = loadCompletedTimeLiteral.getValueAs(Long.class);
+ loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_USER_TIME;
} else {
- throw new ValidationException(
- String.format("Unknown SNAPSHOT 'load_completed_condition': '%s'.", condition));
+ loadCompletedTime = System.currentTimeMillis();
Review Comment:
That's a good point!
I think it shouldn't be too difficult to implement that now. I'll give it a try
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org