You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2026/09/16 12:26:44 UTC
(flink) 01/02: [FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument
This is an automated email from the ASF dual-hosted git repository.
fhueske pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c4e485a22d5e1e9c4b66ca273d10f4e8e1425734
Author: Fabian Hueske <fh...@confluent.io>
AuthorDate: Tue Sep 15 14:43:56 2026 +0200
[FLINK-40666][table] Simplify SNAPSHOT function by removing load_completed_condition argument
Whether a LATERAL SNAPSHOT join's load phase completes at compile time
or at a user-specified time is now inferred from whether
load_completed_time is provided, instead of a separate
load_completed_condition argument.
Generated-By: Claude Sonnet 5
---
.../content.zh/docs/sql/reference/queries/joins.md | 10 +-
docs/content/docs/sql/reference/queries/joins.md | 10 +-
.../flink/sql/parser/FlinkSqlParserImplTest.java | 4 +-
.../functions/BuiltInFunctionDefinitions.java | 2 -
.../strategies/LateralSnapshotTypeStrategy.java | 91 ++----------------
.../LateralSnapshotInputTypeStrategyTest.java | 102 ++-------------------
.../LogicalJoinToLateralSnapshotJoinRule.java | 50 +++-------
.../plan/utils/LateralSnapshotJoinUtil.java | 13 +++
.../batch/sql/join/LateralSnapshotJoinTest.java | 6 --
.../batch/LateralSnapshotJoinTestPrograms.java | 1 -
.../LateralSnapshotJoinSemanticTestPrograms.java | 14 +--
.../stream/LateralSnapshotJoinTestPrograms.java | 5 +-
.../plan/stream/sql/ColumnExpansionTest.java | 2 -
.../plan/stream/sql/SnapshotTableFunctionTest.java | 3 -
.../stream/sql/join/LateralSnapshotJoinTest.java | 60 ------------
.../stream/sql/join/LateralSnapshotJoinITCase.java | 8 +-
.../batch/sql/join/LateralSnapshotJoinTest.xml | 24 ++---
.../stream/sql/join/LateralSnapshotJoinTest.xml | 40 ++++----
18 files changed, 94 insertions(+), 351 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md b/docs/content.zh/docs/sql/reference/queries/joins.md
index b902a9a9929..eaacc8eee5e 100644
--- a/docs/content.zh/docs/sql/reference/queries/joins.md
+++ b/docs/content.zh/docs/sql/reference/queries/joins.md
@@ -349,7 +349,7 @@ A `LATERAL SNAPSHOT` join operates in two phases to avoid joining probe-side row
During the *load phase*, the operator accumulates the build-side changes into state until the load-completion condition is met, without emitting any results yet. Probe-side rows that arrive during the load phase are buffered. The load phase completes when one of the following occurs:
-- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user (`load_completed_condition => 'user_time'`) or automatically set to the wall-clock time when the query is compiled (`load_completed_condition => 'compile_time'`), or
+- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user, or, if `load_completed_time` is not provided, automatically set to the wall-clock time when the query is compiled, or
- as a fallback, the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up).
When the load phase completes, the operator transitions to the *join phase*: all buffered probe-side rows are joined against the current build-side state and emitted.
@@ -369,7 +369,6 @@ FROM probe_table
[LEFT] JOIN LATERAL SNAPSHOT(
input => TABLE build_table,
[ on_time => DESCRIPTOR(<rowtime_column>), ]
- [ load_completed_condition => <'compile_time' | 'user_time'>, ]
[ load_completed_time => <timestamp_ltz>, ]
[ load_completed_idle_timeout => <interval>, ]
[ state_ttl => <interval> ]) AS s
@@ -382,12 +381,11 @@ The `SNAPSHOT` function accepts the following arguments:
| --- | --- | --- |-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `input` | TABLE | yes | The build-side table. It may use any changelog mode (inserts, updates, and deletes). |
| `on_time` | DESCRIPTOR | no | Declares a build-side rowtime column that defines the order in which the build-side changes are applied. The referenced column must exist in `input` and be a `TIMESTAMP` or `TIMESTAMP_LTZ` column (up to precision 3) that is declared as a [watermarked rowtime attribute]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). The argument is **required for streaming queries**. |
-| `load_completed_condition` | STRING | no | Determines when the initial load phase completes. One of `'compile_time'` (default) or `'user_time'`. With `'compile_time'`, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. With `'user_time'`, it completes once the build-side watermark reaches the explicit `load_completed_time`. |
-| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. |
+| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. If omitted, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. |
| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback to complete the load phase. The transition to the join phase happens when the build-side watermark does not advance for more than the configured interval. |
| `state_ttl` | INTERVAL | no | Retention time for build-side state. Join keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). |
-`on_time`, `load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below).
+`on_time`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below).
**Result and state characteristics**
@@ -399,7 +397,7 @@ The build-side state grows with the number of distinct build-side keys, and duri
**Batch mode**
-In batch mode, a `LATERAL SNAPSHOT` join is executed as a regular (`INNER` or `LEFT`) join between the probe side and the complete build side. Batch execution reads the entire build side before joining, so there is no load phase and no incremental state build-up. The streaming-specific arguments (`on_time`, `load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl`) are accepted but have no effect, and the build side does not need to declare a water [...]
+In batch mode, a `LATERAL SNAPSHOT` join is executed as a regular (`INNER` or `LEFT`) join between the probe side and the complete build side. Batch execution reads the entire build side before joining, so there is no load phase and no incremental state build-up. The streaming-specific arguments (`on_time`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl`) are accepted but have no effect, and the build side does not need to declare a watermark or provide a `on_time`.
Because every probe-side row is joined against the final, complete build side, the batch result is **deterministic**.
diff --git a/docs/content/docs/sql/reference/queries/joins.md b/docs/content/docs/sql/reference/queries/joins.md
index 5a0f6f4a9dc..c815b0b585f 100644
--- a/docs/content/docs/sql/reference/queries/joins.md
+++ b/docs/content/docs/sql/reference/queries/joins.md
@@ -356,7 +356,7 @@ it first loads the build side up to a well-defined point in time (the *load phas
During the *load phase*, the operator accumulates the build-side changes into state until the load-completion condition is met, without emitting any results yet.
Probe-side rows that arrive during the load phase are buffered. The load phase completes when one of the following occurs:
-- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user (`load_completed_condition => 'user_time'`) or automatically set to the wall-clock time when the query is compiled (`load_completed_condition => 'compile_time'`), or
+- the build-side watermark reaches a configured `load_completed_time`. This time is either explicitly set by the user, or, if `load_completed_time` is not provided, automatically set to the wall-clock time when the query is compiled, or
- as a fallback, the `load_completed_idle_timeout` elapses in processing time without the build-side watermark advancing (which handles build sides that become idle during start-up).
When the load phase completes, the operator transitions to the *join phase*: all buffered probe-side rows are joined against the current build-side state and emitted.
@@ -378,7 +378,6 @@ FROM probe_table
[LEFT] JOIN LATERAL SNAPSHOT(
input => TABLE build_table,
[ on_time => DESCRIPTOR(<rowtime_column>), ]
- [ load_completed_condition => <'compile_time' | 'user_time'>, ]
[ load_completed_time => <timestamp_ltz>, ]
[ load_completed_idle_timeout => <interval>, ]
[ state_ttl => <interval> ]) AS s
@@ -391,12 +390,11 @@ The `SNAPSHOT` function accepts the following arguments:
| --- | --- | --- |-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `input` | TABLE | yes | The build-side table. It may use any [changelog mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates, and deletes). |
| `on_time` | DESCRIPTOR | no | Declares a build-side rowtime column that defines the order in which the build-side changes are applied. The referenced column must exist in `input` and be a `TIMESTAMP` or `TIMESTAMP_LTZ` column (up to precision 3) that is declared as a [watermarked rowtime attribute]({{< ref "docs/concepts/sql-table-concepts/time_attributes" >}}#event-time). The argument is **required for streaming queries**. |
-| `load_completed_condition` | STRING | no | Determines when the initial load phase completes. One of `'compile_time'` (default) or `'user_time'`. With `'compile_time'`, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. With `'user_time'`, it completes once the build-side watermark reaches the explicit `load_completed_time`. |
-| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. Required when `load_completed_condition` is `'user_time'` and must not be set otherwise. |
+| `load_completed_time` | TIMESTAMP_LTZ(3) | no | The build-side event time that completes the load phase. If omitted, the load phase completes once the build-side watermark reaches the wall-clock time at which the query was compiled. |
| `load_completed_idle_timeout` | INTERVAL | no | A processing-time fallback to complete the load phase. The transition to the join phase happens when the build-side watermark does not advance for more than the configured interval. |
| `state_ttl` | INTERVAL | no | Retention time for build-side state. Join keys that are not accessed within this duration become eligible for eviction. Only applied during the join phase. Defaults to the pipeline's [state TTL]({{< ref "docs/dev/table/config" >}}#table-exec-state-ttl). |
-`on_time`, `load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below).
+`on_time`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl` only affect streaming execution and are ignored in batch mode (see **Batch mode** below).
**Result and state characteristics**
@@ -408,7 +406,7 @@ The build-side state grows with the number of distinct build-side keys, and duri
**Batch mode**
-In batch mode, a `LATERAL SNAPSHOT` join is executed as a regular (`INNER` or `LEFT`) join between the probe side and the complete build side. Batch execution reads the entire build side before joining, so there is no load phase and no incremental state build-up. The streaming-specific arguments (`on_time`, `load_completed_condition`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl`) are accepted but have no effect, and the build side does not need to declare a water [...]
+In batch mode, a `LATERAL SNAPSHOT` join is executed as a regular (`INNER` or `LEFT`) join between the probe side and the complete build side. Batch execution reads the entire build side before joining, so there is no load phase and no incremental state build-up. The streaming-specific arguments (`on_time`, `load_completed_time`, `load_completed_idle_timeout`, and `state_ttl`) are accepted but have no effect, and the build side does not need to declare a watermark or provide a `on_time`.
Because every probe-side row is joined against the final, complete build side, the batch result is **deterministic**.
diff --git a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
index 01f8c61541d..5338102c312 100644
--- a/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
+++ b/flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java
@@ -4167,13 +4167,13 @@ class FlinkSqlParserImplTest extends SqlParserTest {
// Named arguments and TABLE-typed arg passed to the function.
sql("select * from t, lateral snapshot("
+ "input => table s, "
- + "load_completed_condition => 'on_time')")
+ + "load_completed_idle_timeout => interval '10' second)")
.ok(
"SELECT *\n"
+ "FROM `T`,\n"
+ "LATERAL TABLE(`SNAPSHOT`("
+ "`INPUT` => (TABLE `S`), "
- + "`LOAD_COMPLETED_CONDITION` => 'on_time'))");
+ + "`LOAD_COMPLETED_IDLE_TIMEOUT` => INTERVAL '10' SECOND))");
// LATERAL fn(...) as the first FROM entry (no preceding table).
sql("select * from lateral ramp(3)").ok("SELECT *\n" + "FROM LATERAL TABLE(`RAMP`(3))");
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 18f166d0ba6..53ec8d16078 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -963,8 +963,6 @@ public final class BuiltInFunctionDefinitions {
StaticArgumentTrait.REQUIRE_UPDATE_BEFORE,
StaticArgumentTrait.REQUIRE_FULL_DELETE)),
StaticArgument.scalar("on_time", DataTypes.DESCRIPTOR(), true),
- StaticArgument.scalar(
- "load_completed_condition", DataTypes.STRING(), true),
StaticArgument.scalar(
"load_completed_time", DataTypes.TIMESTAMP_LTZ(3), true),
StaticArgument.scalar(
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java
index 24c22c85d27..892cedc3887 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotTypeStrategy.java
@@ -39,7 +39,6 @@ import org.apache.flink.types.ColumnList;
import java.util.List;
import java.util.Optional;
-import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -52,21 +51,12 @@ import java.util.stream.IntStream;
* <ul>
* <li>{@code input} (TABLE, required)
* <li>{@code on_time} (DESCRIPTOR, optional; required for streaming, enforced by the rule)
- * <li>{@code load_completed_condition} (STRING literal, optional, default {@code 'compile_time'},
- * allowed values: {@code 'compile_time'}, {@code 'user_time'})
- * <li>{@code load_completed_time} (TIMESTAMP_LTZ(3), optional)
+ * <li>{@code load_completed_time} (TIMESTAMP_LTZ(3), optional; when absent, the planner rule
+ * defaults the load completed time to the wall-clock time when the query is compiled)
* <li>{@code load_completed_idle_timeout} (INTERVAL SECOND, optional)
* <li>{@code state_ttl} (INTERVAL SECOND, optional)
* </ul>
*
- * <p>and ensures cross-argument consistency:
- *
- * <ul>
- * <li>{@code load_completed_condition='user_time'} requires {@code load_completed_time}.
- * <li>{@code load_completed_condition='compile_time'} (or unset) forbids {@code
- * load_completed_time}.
- * </ul>
- *
* <p>The output type forwards the input table's row type, but materializes any rowtime attribute
* indicator into a regular timestamp.
*/
@@ -83,43 +73,21 @@ public final class LateralSnapshotTypeStrategy {
public static final String ON_TIME_ARG_NAME = "on_time";
- /** The {@code load_completed_condition} STRING argument. */
- public static final int LOAD_COMPLETED_CONDITION_ARG_INDEX = 2;
-
- public static final String LOAD_COMPLETED_CONDITION_ARG_NAME = "load_completed_condition";
-
/** The {@code load_completed_time} TIMESTAMP_LTZ argument. */
- public static final int LOAD_COMPLETED_TIME_ARG_INDEX = 3;
+ public static final int LOAD_COMPLETED_TIME_ARG_INDEX = 2;
public static final String LOAD_COMPLETED_TIME_ARG_NAME = "load_completed_time";
/** The {@code load_completed_idle_timeout} INTERVAL argument. */
- public static final int LOAD_COMPLETED_IDLE_TIMEOUT_ARG_INDEX = 4;
+ public static final int LOAD_COMPLETED_IDLE_TIMEOUT_ARG_INDEX = 3;
public static final String LOAD_COMPLETED_IDLE_TIMEOUT_ARG_NAME = "load_completed_idle_timeout";
/** The {@code state_ttl} INTERVAL argument. */
- public static final int STATE_TTL_ARG_INDEX = 5;
+ public static final int STATE_TTL_ARG_INDEX = 4;
public static final String STATE_TTL_ARG_NAME = "state_ttl";
- /** Default value for {@code load_completed_condition}. */
- public static final String LOAD_COMPLETED_CONDITION_COMPILE_TIME = "compile_time";
-
- /**
- * Allowed value for {@code load_completed_condition} that requires {@code load_completed_time}.
- */
- public static final String LOAD_COMPLETED_CONDITION_USER_TIME = "user_time";
-
- private static final Set<String> VALID_LOAD_COMPLETED_CONDITIONS =
- Set.of(LOAD_COMPLETED_CONDITION_COMPILE_TIME, LOAD_COMPLETED_CONDITION_USER_TIME);
-
- /** Stable, human-readable rendering of {@link #VALID_LOAD_COMPLETED_CONDITIONS}. */
- private static final String VALID_LOAD_COMPLETED_CONDITIONS_DESC =
- String.format(
- "'%s', '%s'",
- LOAD_COMPLETED_CONDITION_COMPILE_TIME, LOAD_COMPLETED_CONDITION_USER_TIME);
-
// --------------------------------------------------------------------------------------------
// Input validation
// --------------------------------------------------------------------------------------------
@@ -128,7 +96,7 @@ public final class LateralSnapshotTypeStrategy {
new InputTypeStrategy() {
@Override
public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.between(1, 6);
+ return ConstantArgumentCount.between(1, 5);
}
@Override
@@ -143,7 +111,6 @@ public final class LateralSnapshotTypeStrategy {
Signature.of(
Argument.of(INPUT_ARG_NAME, "TABLE"),
Argument.of(ON_TIME_ARG_NAME, "DESCRIPTOR"),
- Argument.of(LOAD_COMPLETED_CONDITION_ARG_NAME, "STRING"),
Argument.of(LOAD_COMPLETED_TIME_ARG_NAME, "TIMESTAMP_LTZ(3)"),
Argument.of(
LOAD_COMPLETED_IDLE_TIMEOUT_ARG_NAME,
@@ -205,48 +172,6 @@ public final class LateralSnapshotTypeStrategy {
return timeColumnFailure;
}
- // Reject non-literal load_completed_condition explicitly: the planner needs the value
- // at compile time to decide between 'compile_time' and 'user_time'.
- final boolean hasLoadCompletedCondition =
- isArgumentProvided(callContext, LOAD_COMPLETED_CONDITION_ARG_INDEX);
- if (isProvidedNonLiteral(callContext, LOAD_COMPLETED_CONDITION_ARG_INDEX)) {
- return callContext.fail(
- throwOnFailure,
- "Argument 'load_completed_condition' of SNAPSHOT must be a STRING literal.");
- }
- // Get condition and default to 'compile_time' if not provided
- final String condition =
- hasLoadCompletedCondition
- ? callContext
- .getArgumentValue(LOAD_COMPLETED_CONDITION_ARG_INDEX, String.class)
- .orElse(LOAD_COMPLETED_CONDITION_COMPILE_TIME)
- : LOAD_COMPLETED_CONDITION_COMPILE_TIME;
- // Reject invalid condition value
- if (!VALID_LOAD_COMPLETED_CONDITIONS.contains(condition)) {
- return callContext.fail(
- throwOnFailure,
- "Argument 'load_completed_condition' of SNAPSHOT must be one of %s but was '%s'.",
- VALID_LOAD_COMPLETED_CONDITIONS_DESC,
- condition);
- }
-
- final boolean hasLoadCompletedTime =
- isArgumentProvided(callContext, LOAD_COMPLETED_TIME_ARG_INDEX);
-
- // Cross-argument consistency: condition <-> load_completed_time
- if (LOAD_COMPLETED_CONDITION_USER_TIME.equals(condition) && !hasLoadCompletedTime) {
- return callContext.fail(
- throwOnFailure,
- "SNAPSHOT requires 'load_completed_time' when "
- + "'load_completed_condition' is 'user_time'.");
- }
- if (!LOAD_COMPLETED_CONDITION_USER_TIME.equals(condition) && hasLoadCompletedTime) {
- return callContext.fail(
- throwOnFailure,
- "SNAPSHOT does not accept 'load_completed_time' when "
- + "'load_completed_condition' is not 'user_time'.");
- }
-
return Optional.of(callContext.getArgumentDataTypes());
}
@@ -298,9 +223,5 @@ public final class LateralSnapshotTypeStrategy {
&& !callContext.isArgumentNull(index);
}
- private static boolean isProvidedNonLiteral(final CallContext callContext, final int index) {
- return isArgumentProvided(callContext, index) && !callContext.isArgumentLiteral(index);
- }
-
private LateralSnapshotTypeStrategy() {}
}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotInputTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotInputTypeStrategyTest.java
index 23e132c0c93..33223bd6465 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotInputTypeStrategyTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/LateralSnapshotInputTypeStrategyTest.java
@@ -33,9 +33,7 @@ import static org.apache.flink.table.types.inference.strategies.SpecificInputTyp
/**
* Tests for {@link SpecificInputTypeStrategies#LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY}.
*
- * <p>Validates the named-argument signature of the {@code SNAPSHOT} table function, including the
- * cross-argument consistency between {@code load_completed_condition} and {@code
- * load_completed_time}.
+ * <p>Validates the named-argument signature of the {@code SNAPSHOT} table function.
*/
class LateralSnapshotInputTypeStrategyTest extends InputTypeStrategiesTestBase {
@@ -59,9 +57,7 @@ class LateralSnapshotInputTypeStrategyTest extends InputTypeStrategiesTestBase {
// Valid: just the build-side table (on_time is optional at this layer; the
// planner rule enforces it for streaming).
// ----------------------------------------------------------------------------
- TestSpec.forStrategy(
- "Valid: input only (default condition)",
- LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ TestSpec.forStrategy("Valid: input only", LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
.calledWithArgumentTypes(TABLE_TYPE)
.calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
.expectArgumentTypes(TABLE_TYPE),
@@ -75,33 +71,6 @@ class LateralSnapshotInputTypeStrategyTest extends InputTypeStrategiesTestBase {
.calledWithLiteralAt(1, ON_TIME)
.expectArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE),
- // ----------------------------------------------------------------------------
- // Valid: explicit 'compile_time' condition without load_completed_time.
- // ----------------------------------------------------------------------------
- TestSpec.forStrategy(
- "Valid: condition='compile_time'",
- LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE)
- .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, ON_TIME)
- .calledWithLiteralAt(2, "compile_time")
- .expectArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE),
-
- // ----------------------------------------------------------------------------
- // Valid: 'user_time' with a TIMESTAMP literal.
- // ----------------------------------------------------------------------------
- TestSpec.forStrategy(
- "Valid: condition='user_time' + load_completed_time",
- LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(
- TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE, TIMESTAMP_TYPE)
- .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, ON_TIME)
- .calledWithLiteralAt(2, "user_time")
- .calledWithLiteralAt(3, LocalDateTime.parse("2026-07-01T00:00:00.001"))
- .expectArgumentTypes(
- TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE, TIMESTAMP_TYPE),
-
// ----------------------------------------------------------------------------
// Valid: full named-arg form with idle timeout and TTL.
// ----------------------------------------------------------------------------
@@ -109,20 +78,17 @@ class LateralSnapshotInputTypeStrategyTest extends InputTypeStrategiesTestBase {
.calledWithArgumentTypes(
TABLE_TYPE,
DESCRIPTOR_TYPE,
- STRING_TYPE,
TIMESTAMP_TYPE,
INTERVAL_TYPE,
INTERVAL_TYPE)
.calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
.calledWithLiteralAt(1, ON_TIME)
- .calledWithLiteralAt(2, "user_time")
- .calledWithLiteralAt(3, LocalDateTime.parse("2026-07-01T00:00:00.001"))
- .calledWithLiteralAt(4, Duration.ofSeconds(10))
- .calledWithLiteralAt(5, Duration.ofDays(1))
+ .calledWithLiteralAt(2, LocalDateTime.parse("2026-07-01T00:00:00.001"))
+ .calledWithLiteralAt(3, Duration.ofSeconds(10))
+ .calledWithLiteralAt(4, Duration.ofDays(1))
.expectArgumentTypes(
TABLE_TYPE,
DESCRIPTOR_TYPE,
- STRING_TYPE,
TIMESTAMP_TYPE,
INTERVAL_TYPE,
INTERVAL_TYPE),
@@ -194,62 +160,6 @@ class LateralSnapshotInputTypeStrategyTest extends InputTypeStrategiesTestBase {
.calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
.calledWithLiteralAt(1, ColumnList.of("ts", "k"))
.expectErrorMessage(
- "Argument 'on_time' of SNAPSHOT must reference exactly one column."),
-
- // ----------------------------------------------------------------------------
- // Invalid: 'user_time' condition requires load_completed_time.
- // ----------------------------------------------------------------------------
- TestSpec.forStrategy(
- "Invalid: condition='user_time' without load_completed_time",
- LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE)
- .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, ON_TIME)
- .calledWithLiteralAt(2, "user_time")
- .expectErrorMessage(
- "SNAPSHOT requires 'load_completed_time' when "
- + "'load_completed_condition' is 'user_time'."),
-
- // ----------------------------------------------------------------------------
- // Invalid: load_completed_time requires 'user_time' condition.
- // ----------------------------------------------------------------------------
- TestSpec.forStrategy(
- "Invalid: load_completed_time without explicit 'user_time'",
- LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(
- TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE, TIMESTAMP_TYPE)
- .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, ON_TIME)
- .calledWithLiteralAt(2, "compile_time")
- .calledWithLiteralAt(3, LocalDateTime.parse("2026-07-01T00:00:00.001"))
- .expectErrorMessage(
- "SNAPSHOT does not accept 'load_completed_time' when "
- + "'load_completed_condition' is not 'user_time'."),
-
- // ----------------------------------------------------------------------------
- // Invalid: unknown condition value.
- // ----------------------------------------------------------------------------
- TestSpec.forStrategy(
- "Invalid: unknown condition value",
- LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE)
- .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, ON_TIME)
- .calledWithLiteralAt(2, "invalid_condition")
- .expectErrorMessage(
- "Argument 'load_completed_condition' of SNAPSHOT must be one of 'compile_time', 'user_time' but was 'invalid_condition'."),
-
- // ----------------------------------------------------------------------------
- // Invalid: load_completed_condition provided as a non-literal expression.
- // ----------------------------------------------------------------------------
- TestSpec.forStrategy(
- "Invalid: non-literal load_completed_condition",
- LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE)
- .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, ON_TIME)
- // Intentionally no literal provided for load_completed_condition
- .expectErrorMessage(
- "Argument 'load_completed_condition' of SNAPSHOT must be a STRING literal."));
+ "Argument 'on_time' of SNAPSHOT must reference exactly one column."));
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java
index b50ff8e69b7..0b48c9920ea 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java
@@ -58,8 +58,8 @@ import java.util.List;
* Rewrites a {@link FlinkLogicalJoin} whose right side is a {@link FlinkLogicalTableFunctionScan}
* backed by the built-in {@code SNAPSHOT} function into a dedicated {@link
* FlinkLogicalLateralSnapshotJoin}. The right-side input becomes the actual TABLE argument of the
- * SNAPSHOT call. The SNAPSHOT-specific arguments (load_completed_condition, load_completed_time,
- * load_completed_idle_timeout, state_ttl) are carried as fields on the new node.
+ * SNAPSHOT call. The SNAPSHOT-specific arguments (load_completed_time, load_completed_idle_timeout,
+ * state_ttl) are carried as fields on the new node.
*
* <p>By the time this rule fires, Calcite's decorrelator has already converted the original {@code
* LogicalCorrelate} into a {@code LogicalJoin} (because SNAPSHOT does not actually reference any
@@ -140,13 +140,6 @@ public class LogicalJoinToLateralSnapshotJoinRule
// All scalar SNAPSHOT arguments must be constant expressions, so we constant-fold each one
// and reject anything that does not reduce to a literal. The 'input' TABLE argument
// (index 0) is exempt.
- final RexLiteral conditionLiteral =
- foldToLiteral(
- rexBuilder,
- executor,
- operands,
- LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_ARG_INDEX,
- LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_ARG_NAME);
final RexLiteral loadCompletedTimeLiteral =
foldToLiteral(
rexBuilder,
@@ -169,37 +162,20 @@ public class LogicalJoinToLateralSnapshotJoinRule
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();
+ loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_COMPILE_TIME;
}
-
- // The effective condition (defaulting to 'compile_time') is carried for explain output.
- final String loadCompletedCondition =
- condition == null
- ? LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_COMPILE_TIME
- : condition;
final Long loadCompletedIdleTimeoutMs =
intervalMillis(
idleTimeoutLiteral,
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java
index aa2f8a22cc9..61fa3471c2b 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java
@@ -46,6 +46,19 @@ import java.util.List;
@Internal
public final class LateralSnapshotJoinUtil {
+ /**
+ * Label for {@code FlinkLogicalLateralSnapshotJoin#getLoadCompletedCondition()} when the load
+ * phase completes at the wall-clock time the query was compiled (the default, used when the
+ * user does not provide {@code load_completed_time}).
+ */
+ public static final String LOAD_COMPLETED_CONDITION_COMPILE_TIME = "compile_time";
+
+ /**
+ * Label for {@code FlinkLogicalLateralSnapshotJoin#getLoadCompletedCondition()} when the load
+ * phase completes at a user-provided {@code load_completed_time}.
+ */
+ public static final String LOAD_COMPLETED_CONDITION_USER_TIME = "user_time";
+
/**
* {@code true} when {@code definition} is the {@link BuiltInFunctionDefinitions#SNAPSHOT}
* built-in.
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
index 214e84c880b..11aefe68908 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
@@ -73,7 +73,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -84,7 +83,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -95,7 +93,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk AND probe.pv = s.bv");
@@ -106,7 +103,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk AND probe.pv > s.bv");
@@ -117,7 +113,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -165,7 +160,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pv > s.bv";
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
index 9f5d53c8720..f33e94e5c5c 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
@@ -121,7 +121,6 @@ public class LateralSnapshotJoinTestPrograms {
"INSERT INTO sink SELECT pk, pv, bk, bv FROM probe JOIN LATERAL "
+ "SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'compile_time', "
+ "load_completed_idle_timeout => INTERVAL '10' SECOND, "
+ "state_ttl => INTERVAL '1' DAY"
+ ") AS s ON probe.pk = s.bk")
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java
index f8864578acf..275323610ec 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java
@@ -45,15 +45,16 @@ import java.util.stream.IntStream;
*/
public class LateralSnapshotJoinSemanticTestPrograms {
- /** The {@code 'user_time'} condition reached mid-stream by the build-side flip-trigger row. */
+ /**
+ * The build-side watermark reaches the configured {@code load_completed_time} gate mid-stream
+ * via the flip-trigger row.
+ */
private static final String MID_FLIP =
- "load_completed_condition => 'user_time', "
- + "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3))";
+ "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3))";
/** A far-future flip condition: the flip happens only at end of all input. */
private static final String END_FLIP =
- "load_completed_condition => 'user_time', "
- + "load_completed_time => CAST(TIMESTAMP '2100-01-01 00:00:00' AS TIMESTAMP_LTZ(3))";
+ "load_completed_time => CAST(TIMESTAMP '2100-01-01 00:00:00' AS TIMESTAMP_LTZ(3))";
/** Event time of the flip-trigger row; equal to the {@link #MID_FLIP} timestamp. */
private static final String FLIP_TRIGGER_TS = "00:00:10";
@@ -93,7 +94,6 @@ public class LateralSnapshotJoinSemanticTestPrograms {
+ "FROM probe "
+ " JOIN LATERAL TABLE(SNAPSHOT("
+ " input => TABLE b, on_time => DESCRIPTOR(bts), "
- + " load_completed_condition => 'user_time', "
+ " load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3)))) AS s "
+ " ON probe.pk = s.bk")
.build();
@@ -258,7 +258,7 @@ public class LateralSnapshotJoinSemanticTestPrograms {
"+I[a, 100, a, 11]",
"+I[b, 200, b, 20]")
.build())
- // No options: 'load_completed_condition' defaults to 'compile_time'.
+ // No load_completed_time: the load phase defaults to 'compile_time'.
.runSql(
"INSERT INTO sink SELECT probe.pk, probe.pv, s.bk, s.bv "
+ "FROM probe JOIN LATERAL SNAPSHOT("
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java
index b1fae558aab..5e92fbfb201 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java
@@ -26,8 +26,8 @@ import org.apache.flink.types.Row;
/**
* {@link TableTestProgram} definitions for testing {@link StreamExecLateralSnapshotJoin}.
*
- * <p>The programs cover a savepoint taken in each of the operator's two phases; the {@code
- * 'user_time'} gate is at {@code 00:00:03} in both.
+ * <p>The programs cover a savepoint taken in each of the operator's two phases; the
+ * load_completed_time gate is at {@code 00:00:03} in both.
*
* <ul>
* <li>{@link #LATERAL_SNAPSHOT_JOIN_PHASE_LOAD}: the transition to JOIN is not triggered before
@@ -97,7 +97,6 @@ public class LateralSnapshotJoinTestPrograms {
private static final String SNAPSHOT_BUILD =
"LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:03' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk";
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ColumnExpansionTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ColumnExpansionTest.java
index 1d96ad811e8..5a9a388b77d 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ColumnExpansionTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ColumnExpansionTest.java
@@ -483,7 +483,6 @@ class ColumnExpansionTest {
final String sql =
"SELECT * FROM snapshot_probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE snapshot_build_hidden, on_time => DESCRIPTOR(rt), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON snapshot_probe.pk = s.bk";
@@ -529,7 +528,6 @@ class ColumnExpansionTest {
final String sql =
"SELECT * FROM snapshot_probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE snapshot_build_pushed, on_time => DESCRIPTOR(rt), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON snapshot_probe.pk = s.bk";
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java
index 3f4b636b4bd..a5e7ef0fe3c 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java
@@ -78,7 +78,6 @@ public class SnapshotTableFunctionTest extends TableTestBase {
"SELECT o.order_id, o.amount, r.rate "
+ "FROM Orders AS o, LATERAL SNAPSHOT("
+ "input => TABLE Rates, on_time => DESCRIPTOR(rate_time), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS r "
+ "WHERE o.currency = r.currency");
@@ -95,7 +94,6 @@ public class SnapshotTableFunctionTest extends TableTestBase {
+ "SELECT o.order_id, o.amount, r.rate "
+ "FROM Orders AS o, LATERAL SNAPSHOT("
+ "input => TABLE Rates, on_time => DESCRIPTOR(rate_time), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS r "
+ "WHERE o.currency = r.currency");
@@ -112,7 +110,6 @@ public class SnapshotTableFunctionTest extends TableTestBase {
"SELECT o.order_id, o.amount, r.rate "
+ "FROM Orders AS o, LATERAL SNAPSHOT("
+ "input => TABLE RatesView, on_time => DESCRIPTOR(rate_time), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS r "
+ "WHERE o.currency = r.currency");
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java
index 4d389ecd1c5..16a9aa322c8 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java
@@ -92,7 +92,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ")) AS s "
+ "ON probe.pk = s.bk");
@@ -103,7 +102,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -114,7 +112,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), "
+ "load_completed_idle_timeout => INTERVAL '10' SECOND, "
+ "state_ttl => INTERVAL '1' DAY"
@@ -127,7 +124,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk AND probe.pv > s.bv");
@@ -138,7 +134,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk AND probe.pv = s.bv");
@@ -149,7 +144,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk AND probe.pts >= s.bts");
@@ -161,7 +155,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
"WITH cte AS (SELECT bk, bv + 1 AS bv, bts FROM b) "
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE cte, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -172,7 +165,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -183,7 +175,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT probe.pk, probe.pv, s.bv FROM probe LEFT JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -203,7 +194,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b_proctime, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk");
@@ -220,7 +210,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String derived =
"(SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk)";
// Windowing over the build-side time column is rejected: it is materialized, not a time
@@ -260,7 +249,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
.explainSql(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b_upsert, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk");
assertThat(plan).contains("ChangelogNormalize");
@@ -272,7 +260,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk AND probe.pv > s.bv AND probe.pts >= s.bts";
assertThat(util.tableEnv().explainSql(sql)).contains("LateralSnapshotJoin");
@@ -285,7 +272,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
.explainSql(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk");
assertThat(plan).contains("LateralSnapshotJoin");
@@ -305,20 +291,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
.contains("where=[=(pk, bk)]");
}
- @Test
- void testInnerJoinWithExplicitCompileTimeCompilesEndToEnd() {
- final String sql =
- "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'compile_time'"
- + ") AS s ON probe.pk = s.bk";
- assertThat(util.tableEnv().explainSql(sql))
- .contains("LateralSnapshotJoin")
- .contains("loadCompletedCondition=[compile_time]")
- .contains("joinType=[InnerJoin]")
- .contains("where=[=(pk, bk)]");
- }
-
// ------------------------------------------------------------------------------------------
// Validation: rejection paths
// ------------------------------------------------------------------------------------------
@@ -335,7 +307,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b_no_wm, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk";
@@ -357,7 +328,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b_no_wm, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk";
@@ -372,7 +342,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk";
@@ -386,7 +355,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(nonexistent), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk";
@@ -401,7 +369,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bv), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk";
@@ -429,7 +396,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT probe.pk, s.bk, s.pt FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b_proctime_wm, on_time => DESCRIPTOR(pt), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pk = s.bk";
@@ -457,7 +423,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe_updates JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe_updates.pk = s.bk";
@@ -472,7 +437,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
+ "ON probe.pv > s.bv";
@@ -482,25 +446,11 @@ public class LateralSnapshotJoinTest extends TableTestBase {
"LATERAL SNAPSHOT join requires at least one equality predicate.");
}
- @Test
- void testRejectNonConstantCondition() {
- final String sql =
- "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => CAST(CURRENT_TIMESTAMP AS STRING)"
- + ") AS s ON probe.pk = s.bk";
- assertThatThrownBy(() -> util.verifyRelPlan(sql))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining("Invalid function call")
- .hasMessageContaining("SNAPSHOT");
- }
-
@Test
void testRejectNonConstantLoadCompletedTime() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CURRENT_TIMESTAMP"
+ ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
@@ -514,7 +464,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), "
+ "load_completed_idle_timeout => "
+ "CASE WHEN RAND() > 0.5 THEN INTERVAL '10' SECOND ELSE INTERVAL '20' SECOND END"
@@ -530,7 +479,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), "
+ "state_ttl => "
+ "CASE WHEN RAND() > 0.5 THEN INTERVAL '1' DAY ELSE INTERVAL '2' DAY END"
@@ -546,7 +494,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), "
+ "state_ttl => INTERVAL '1' YEAR"
+ ") AS s ON probe.pk = s.bk";
@@ -560,7 +507,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), "
+ "load_completed_idle_timeout => INTERVAL -'10' SECOND"
+ ") AS s ON probe.pk = s.bk";
@@ -575,7 +521,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), "
+ "state_ttl => INTERVAL -'10' MINUTE"
+ ") AS s ON probe.pk = s.bk";
@@ -594,7 +539,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk");
}
@@ -604,7 +548,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk");
}
@@ -614,7 +557,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), "
+ "load_completed_idle_timeout => INTERVAL '10' SECOND, "
+ "state_ttl => INTERVAL '1' DAY"
@@ -626,7 +568,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk AND probe.pv = s.bv");
}
@@ -636,7 +577,6 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s ON probe.pk = s.bk AND probe.pv > s.bv");
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java
index 36f32230f0f..785ce09f802 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java
@@ -58,10 +58,12 @@ import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThatList;
@ExtendWith(ParameterizedTestExtension.class)
public class LateralSnapshotJoinITCase extends StreamingWithStateTestBase {
- /** The {@code 'user_time'} condition reached mid-stream by the build-side flip-trigger row. */
+ /**
+ * The build-side watermark reaches the configured {@code load_completed_time} gate mid-stream
+ * via the flip-trigger row.
+ */
private static final String MID_FLIP =
- "load_completed_condition => 'user_time', "
- + "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3))";
+ "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3))";
/** Event time of the flip-trigger row; equal to the {@link #MID_FLIP} timestamp. */
private static final String FLIP_TRIGGER_TS = "00:00:10";
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
index 8f6ec484a13..8abf54d79f8 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
@@ -25,7 +25,7 @@ limitations under the License.
LogicalProject(pk=[$0], bk=[$3], bv=[$4], pt=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) pt)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) pt)])
+- LogicalProject(bk=[$0], bv=[$1], pt=[$2])
+- LogicalProject(bk=[$0], bv=[$1], pt=[PROCTIME()])
+- LogicalTableScan(table=[[default_catalog, default_database, b_proctime]])
@@ -51,7 +51,7 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, bk, bv, pt], build
LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b_no_wm]])
]]>
@@ -68,14 +68,14 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts, bk, bv, b
</TestCase>
<TestCase name="testInnerJoin">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -92,14 +92,14 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts, bk, bv, b
</TestCase>
<TestCase name="testInnerJoinWithCompositeKeys">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv = s.bv]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv = s.bv]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[AND(=($0, $3), =($1, $4))], joinType=[inner])
:- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -116,14 +116,14 @@ HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), =(pv, bv))], select=[pk, pv
</TestCase>
<TestCase name="testInnerJoinWithNonEquiCondition">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv > s.bv]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv > s.bv]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[AND(=($0, $3), >($1, $4))], joinType=[inner])
:- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -140,14 +140,14 @@ HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), >(pv, bv))], select=[pk, pv
</TestCase>
<TestCase name="testInnerJoinWithoutBuildTimeColumn">
<Resource name="sql">
- <![CDATA[SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(pk=[$0], pv=[$1], bv=[$4])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -166,14 +166,14 @@ Calc(select=[pk, pv, bv])
</TestCase>
<TestCase name="testLeftJoin">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[left])
:- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml
index 885d12b4b1a..cb7016f34c7 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml
@@ -18,7 +18,7 @@ limitations under the License.
<Root>
<TestCase name="testBuildSideProctimeIsMaterialized">
<Resource name="sql">
- <![CDATA[SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b_proctime, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b_proctime, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -26,7 +26,7 @@ LogicalProject(pk=[$0], bk=[$3], bv=[$4], pt=[$6])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts, TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) pt)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts, TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) pt)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2], pt=[$3])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2], pt=[PROCTIME()])
@@ -51,7 +51,7 @@ Calc(select=[pk, bk, bv, pt])
</TestCase>
<TestCase name="testInnerJoin">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -59,7 +59,7 @@ LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
@@ -79,7 +79,7 @@ LateralSnapshotJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts
</TestCase>
<TestCase name="testInnerJoinWithCompositeKeys">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv = s.bv]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv = s.bv]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -87,7 +87,7 @@ LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[AND(=($0, $3), =($1, $4))], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
@@ -107,7 +107,7 @@ LateralSnapshotJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), =(pv, bv))], sel
</TestCase>
<TestCase name="testInnerJoinWithCteBuildSide">
<Resource name="sql">
- <![CDATA[WITH cte AS (SELECT bk, bv + 1 AS bv, bts FROM b) SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE cte, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[WITH cte AS (SELECT bk, bv + 1 AS bv, bts FROM b) SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE cte, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -115,7 +115,7 @@ LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalProject(bk=[$0], bv=[+($1, 1)], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
@@ -137,7 +137,7 @@ LateralSnapshotJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts
</TestCase>
<TestCase name="testInnerJoinWithIdleTimeoutAndStateTtl">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), load_completed_idle_timeout => INTERVAL '10' SECOND, state_ttl => INTERVAL '1' DAY) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3)), load_completed_idle_timeout => INTERVAL '10' SECOND, state_ttl => INTERVAL '1' DAY) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -145,7 +145,7 @@ LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, 10000:INTERVAL SECOND, 86400000:INTERVAL DAY)], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, 10000:INTERVAL SECOND, 86400000:INTERVAL DAY)], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
@@ -165,7 +165,7 @@ LateralSnapshotJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts
</TestCase>
<TestCase name="testInnerJoinWithNonEquiCondition">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv > s.bv]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pv > s.bv]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -173,7 +173,7 @@ LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[AND(=($0, $3), >($1, $4))], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
@@ -193,7 +193,7 @@ LateralSnapshotJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), >(pv, bv))], sel
</TestCase>
<TestCase name="testInnerJoinWithTimeAttributeInCondition">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pts >= s.bts]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk AND probe.pts >= s.bts]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -201,7 +201,7 @@ LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[AND(=($0, $3), >=($2, $5))], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
@@ -221,7 +221,7 @@ LateralSnapshotJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), >=(pts, bts))],
</TestCase>
<TestCase name="testInnerJoinWithoutBuildTimeColumn">
<Resource name="sql">
- <![CDATA[SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -229,7 +229,7 @@ LogicalProject(pk=[$0], pv=[$1], bv=[$4])
+- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
@@ -251,7 +251,7 @@ Calc(select=[pk, pv, bv])
</TestCase>
<TestCase name="testLeftJoin">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -259,7 +259,7 @@ LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
+- LogicalJoin(condition=[=($0, $3)], joinType=[left])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
@@ -279,7 +279,7 @@ LateralSnapshotJoin(joinType=[LeftOuterJoin], where=[=(pk, bk)], select=[pk, pv,
</TestCase>
<TestCase name="testLeftJoinWithoutBuildTimeColumn">
<Resource name="sql">
- <![CDATA[SELECT probe.pk, probe.pv, s.bv FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_condition => 'user_time', load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT probe.pk, probe.pv, s.bv FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -287,7 +287,7 @@ LogicalProject(pk=[$0], pv=[$1], bv=[$4])
+- LogicalJoin(condition=[=($0, $3)], joinType=[left])
:- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
: +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
- +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), _UTF-16LE'user_time', CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])