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 <[email protected]> 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]])
