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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
