From 4750060334ec1e70b448af6c89747a781ca5624a Mon Sep 17 00:00:00 2001 From: Fabian Hueske Date: Tue, 15 Sep 2026 14:43:56 +0200 Subject: [PATCH 1/2] [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 --- .../docs/sql/reference/queries/joins.md | 10 +- .../docs/sql/reference/queries/joins.md | 10 +- .../sql/parser/FlinkSqlParserImplTest.java | 4 +- .../functions/BuiltInFunctionDefinitions.java | 2 - .../LateralSnapshotTypeStrategy.java | 91 ++-------------- .../LateralSnapshotInputTypeStrategyTest.java | 102 ++---------------- .../LogicalJoinToLateralSnapshotJoinRule.java | 50 +++------ .../plan/utils/LateralSnapshotJoinUtil.java | 13 +++ .../sql/join/LateralSnapshotJoinTest.java | 6 -- .../LateralSnapshotJoinTestPrograms.java | 1 - ...teralSnapshotJoinSemanticTestPrograms.java | 14 +-- .../LateralSnapshotJoinTestPrograms.java | 5 +- .../plan/stream/sql/ColumnExpansionTest.java | 2 - .../stream/sql/SnapshotTableFunctionTest.java | 3 - .../sql/join/LateralSnapshotJoinTest.java | 60 ----------- .../sql/join/LateralSnapshotJoinITCase.java | 8 +- .../sql/join/LateralSnapshotJoinTest.xml | 24 ++--- .../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 b902a9a992994..eaacc8eee5e36 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(), ] - [ load_completed_condition => <'compile_time' | 'user_time'>, ] [ load_completed_time => , ] [ load_completed_idle_timeout => , ] [ state_ttl => ]) 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 watermark or provide a `on_time`. +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 5a0f6f4a9dcb9..c815b0b585f9a 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(), ] - [ load_completed_condition => <'compile_time' | 'user_time'>, ] [ load_completed_time => , ] [ load_completed_idle_timeout => , ] [ state_ttl => ]) 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 watermark or provide a `on_time`. +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 01f8c61541d42..5338102c312ce 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 @@ void testLateralImplicitTableFunction() { // 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 1b16c3f52bd4d..2d29024d9cb74 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 @@ -947,8 +947,6 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL) 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 24c22c85d276d..892cedc388718 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 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 @@ *
    *
  • {@code input} (TABLE, required) *
  • {@code on_time} (DESCRIPTOR, optional; required for streaming, enforced by the rule) - *
  • {@code load_completed_condition} (STRING literal, optional, default {@code 'compile_time'}, - * allowed values: {@code 'compile_time'}, {@code 'user_time'}) - *
  • {@code load_completed_time} (TIMESTAMP_LTZ(3), optional) + *
  • {@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) *
  • {@code load_completed_idle_timeout} (INTERVAL SECOND, optional) *
  • {@code state_ttl} (INTERVAL SECOND, optional) *
* - *

and ensures cross-argument consistency: - * - *

    - *
  • {@code load_completed_condition='user_time'} requires {@code load_completed_time}. - *
  • {@code load_completed_condition='compile_time'} (or unset) forbids {@code - * load_completed_time}. - *
- * *

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 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 List getExpectedSignatures(final FunctionDefinition definition 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 @@ private static Optional> validateInputs( 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 @@ private static boolean isArgumentProvided(final CallContext callContext, final i && !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 23e132c0c93db..33223bd64655f 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 @@ /** * Tests for {@link SpecificInputTypeStrategies#LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY}. * - *

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}. + *

Validates the named-argument signature of the {@code SNAPSHOT} table function. */ class LateralSnapshotInputTypeStrategyTest extends InputTypeStrategiesTestBase { @@ -59,9 +57,7 @@ protected Stream testData() { // 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 @@ protected Stream testData() { .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 @@ protected Stream testData() { .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 @@ protected Stream testData() { .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 b50ff8e69b7ee..0b48c9920ea80 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 @@ * 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. * *

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 void onMatch(RelOptRuleCall call) { // 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 void onMatch(RelOptRuleCall call) { LateralSnapshotTypeStrategy.STATE_TTL_ARG_INDEX, LateralSnapshotTypeStrategy.STATE_TTL_ARG_NAME); - // Resolve load_completed_time according to load_completed_condition. The default - // 'compile_time' uses the wall-clock time at planning; 'user_time' uses the user-provided - // load_completed_time (which the type strategy guarantees is present for 'user_time'). - final String condition = - conditionLiteral == null ? null : conditionLiteral.getValueAs(String.class); + // The presence of load_completed_time determines the load-completion mode: if the user + // provided it, the load phase completes at the specified event time ('user_time'); + // otherwise it completes when the build-side event time exceeds the wall-clock time the + // query is compiled ('compile_time'). + // The effective load completed condition is carried for explain output. final Long loadCompletedTime; - if (condition == null - || LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_COMPILE_TIME.equals( - condition)) { - loadCompletedTime = System.currentTimeMillis(); - } else if (LateralSnapshotTypeStrategy.LOAD_COMPLETED_CONDITION_USER_TIME.equals( - condition)) { - loadCompletedTime = - loadCompletedTimeLiteral == null - ? null - : loadCompletedTimeLiteral.getValueAs(Long.class); - if (loadCompletedTime == null) { - throw new ValidationException( - "SNAPSHOT requires 'load_completed_time' when " - + "'load_completed_condition' is 'user_time'."); - } + final String loadCompletedCondition; + if (loadCompletedTimeLiteral != null) { + loadCompletedTime = loadCompletedTimeLiteral.getValueAs(Long.class); + loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_USER_TIME; } else { - throw new ValidationException( - String.format("Unknown SNAPSHOT 'load_completed_condition': '%s'.", condition)); + loadCompletedTime = System.currentTimeMillis(); + 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 aa2f8a22cc9f1..61fa3471c2bbd 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 @@ @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 214e84c880bf5..11aefe689088a 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 @@ void testInnerJoin() { 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 @@ void testLeftJoin() { 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 @@ void testInnerJoinWithCompositeKeys() { 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 @@ void testInnerJoinWithNonEquiCondition() { 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 @@ void testInnerJoinWithoutBuildTimeColumn() { 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 @@ void testRejectMissingEqualityPredicate() { 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 9f5d53c8720e3..f33e94e5c5c02 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 @@ private static SourceTestStep build() { "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 f8864578acf07..275323610ec5d 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 @@ */ 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 b1fae558aabba..5e92fbfb201fd 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 @@ /** * {@link TableTestProgram} definitions for testing {@link StreamExecLateralSnapshotJoin}. * - *

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. + *

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. * *

    *
  • {@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 1d96ad811e83c..5a9a388b77da7 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 @@ void testLateralSnapshotJoinWithOnTimeOnHiddenMetadataColumn() { 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 @@ void testLateralSnapshotJoinWithOnTimeOnPushedDownHiddenMetadataColumn() { 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 3f4b636b4bd27..a5e7ef0fe3c2d 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 @@ void testLateralContext() { "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 @@ void testLateralContextInView() { + "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 @@ void testSnapshotWithViewArgument() { "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 4d389ecd1c59f..16a9aa322c831 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 @@ void testInnerJoin() { 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 @@ void testLeftJoin() { 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 @@ void testInnerJoinWithIdleTimeoutAndStateTtl() { 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 @@ void testInnerJoinWithNonEquiCondition() { 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 @@ void testInnerJoinWithCompositeKeys() { 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 @@ void testInnerJoinWithTimeAttributeInCondition() { 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 @@ void testInnerJoinWithCteBuildSide() { "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 @@ void testInnerJoinWithoutBuildTimeColumn() { 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 @@ void testLeftJoinWithoutBuildTimeColumn() { 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 @@ void testBuildSideProctimeIsMaterialized() { 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 @@ void testBuildRowtimeIsNotForwarded() { 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 @@ void testInnerJoinWithUpsertBuildSourceMaterializesRetractions() { .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 @@ void testNonEquiConditionCompilesEndToEnd() { 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 @@ void testFoldableConstantArgs() { .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 @@ void testInnerJoinWithDefaultCompileTimeCompilesEndToEnd() { .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 @@ void testRejectOnTimeWithoutWatermark() { 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 @@ void testRejectOnTimeWithoutWatermarkWhenBuildTimeColumnPruned() { 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 @@ void testRejectMissingOnTime() { 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 @@ void testRejectUnknownOnTimeColumn() { 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 @@ void testRejectNonTimestampOnTimeColumn() { 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 @@ void testRejectProctimeOnTimeColumn() { 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 @@ void testRejectProbeSideNotAppendOnly() { 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 @@ void testRejectMissingEqualityPredicate() { 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 @@ void testRejectMissingEqualityPredicate() { "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 @@ void testRejectNonConstantIdleTimeout() { 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 @@ void testRejectNonConstantStateTtl() { 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 @@ void testRejectYearMonthIntervalStateTtl() { 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 @@ void testRejectNegativeIdleTimeout() { 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 @@ void testRejectNegativeStateTtl() { 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 @@ void testInnerJoinJsonPlan() { 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 @@ void testLeftJoinJsonPlan() { 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 @@ void testInnerJoinWithIdleTimeoutAndStateTtlJsonPlan() { 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 @@ void testInnerJoinWithCompositeKeysJsonPlan() { 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 @@ void testInnerJoinWithNonEquiConditionJsonPlan() { 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 36f32230f0f46..785ce09f802e2 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 @@ @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 8f6ec484a139c..8abf54d79f857 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 - 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]]> + TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]> @@ -92,14 +92,14 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts, bk, bv, b - 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]]> + 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]]> @@ -116,14 +116,14 @@ HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), =(pv, bv))], select=[pk, pv - 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]]> + 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]]> ($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 - 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]]> + TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]> @@ -166,14 +166,14 @@ Calc(select=[pk, pv, bv]) - 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]]> + TABLE b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk]]> 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 885d12b4b1a9a..cb7016f34c7f4 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. - 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]]> + 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]]> - 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]]> + 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]]> - 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]]> + 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]]> - 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]]> + 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]]> - 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]]> + 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]]> - 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]]> + 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]]> ($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 - 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]]> + 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]]> =($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))], - 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]]> + 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]]> - 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]]> + 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]]> - 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]]> + 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]]> Date: Wed, 16 Sep 2026 09:50:37 +0200 Subject: [PATCH 2/2] [FLINK-40666][table] Derive default SNAPSHOT load_completed_time from query-start epoch Reuses the query-start epoch time that is set once before the compilation, instead of calling System.currentTimeMillis() per rule match, so multiple compile_time LATERAL SNAPSHOT joins in the same query agree on load_completed_time. Generated-By: Claude Sonnet 5 --- .../LogicalJoinToLateralSnapshotJoinRule.java | 4 ++- .../plan/utils/LateralSnapshotJoinUtil.java | 17 +++++++++ .../sql/join/LateralSnapshotJoinTest.java | 33 +++++++++++++++++ .../utils/LateralSnapshotJoinUtilTest.java | 35 +++++++++++++++++++ 4 files changed, 88 insertions(+), 1 deletion(-) 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 0b48c9920ea80..6edb403f93427 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 @@ -173,7 +173,9 @@ public void onMatch(RelOptRuleCall call) { loadCompletedTime = loadCompletedTimeLiteral.getValueAs(Long.class); loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_USER_TIME; } else { - loadCompletedTime = System.currentTimeMillis(); + loadCompletedTime = + LateralSnapshotJoinUtil.resolveDefaultLoadCompletedTime( + ShortcutUtils.unwrapTableConfig(call)); loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_COMPILE_TIME; } final Long loadCompletedIdleTimeoutMs = 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 61fa3471c2bbd..764dca00c3748 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 @@ -19,12 +19,14 @@ package org.apache.flink.table.planner.plan.utils; import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.functions.BuiltInFunctionDefinition; import org.apache.flink.table.functions.BuiltInFunctionDefinitions; import org.apache.flink.table.functions.FunctionDefinition; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction; import org.apache.flink.table.planner.plan.schema.TimeIndicatorRelDataType; +import org.apache.flink.table.planner.utils.InternalConfigOptions; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.TimestampType; @@ -59,6 +61,21 @@ public final class LateralSnapshotJoinUtil { */ public static final String LOAD_COMPLETED_CONDITION_USER_TIME = "user_time"; + /** + * Resolves the {@code load_completed_time} used by a {@code compile_time} LATERAL SNAPSHOT + * join, i.e. one where the user did not provide the argument. + * + *

    Uses {@link InternalConfigOptions#TABLE_QUERY_START_EPOCH_TIME}: {@link + * org.apache.flink.table.planner.delegation.PlannerBase#beforeTranslation} records it once + * before optimization begins, so every SNAPSHOT call reduced during that optimization run + * resolves to the same value. If the config is absent, we fall back to wall clock time. + */ + public static long resolveDefaultLoadCompletedTime(TableConfig tableConfig) { + return tableConfig + .getOptional(InternalConfigOptions.TABLE_QUERY_START_EPOCH_TIME) + .orElseGet(System::currentTimeMillis); + } + /** * {@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/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 16a9aa322c831..81ac5c8ed8031 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 @@ -35,7 +35,11 @@ import java.time.Duration; import java.time.ZoneId; +import java.util.HashSet; import java.util.List; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -291,6 +295,35 @@ void testInnerJoinWithDefaultCompileTimeCompilesEndToEnd() { .contains("where=[=(pk, bk)]"); } + @Test + void testMultipleDefaultCompileTimeSnapshotJoinsShareTheSameLoadCompletedTime() { + util.tableEnv() + .executeSql( + "CREATE TABLE b2 (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts" + + ") WITH ('connector' = 'values', 'bounded' = 'false')"); + // Two LATERAL SNAPSHOT joins in the same query, neither providing load_completed_time: + // both must resolve to the same wall-clock compile-time timestamp. + final String sql = + "SELECT probe.pk, s1.bv, s2.bv FROM probe " + + "JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts)) AS s1 " + + "ON probe.pk = s1.bk " + + "JOIN LATERAL SNAPSHOT(input => TABLE b2, on_time => DESCRIPTOR(bts)) AS s2 " + + "ON probe.pk = s2.bk"; + final String plan = util.tableEnv().explainSql(sql); + + // assert that both LateralSnapshotJoins use the same loadCompletedTime + final Matcher matcher = Pattern.compile("loadCompletedTime=\\[(\\d+)]").matcher(plan); + final Set loadCompletedTimes = new HashSet<>(); + while (matcher.find()) { + loadCompletedTimes.add(matcher.group(1)); + } + assertThat(loadCompletedTimes).as("plan:%n%s", plan).hasSize(1); + } + // ------------------------------------------------------------------------------------------ // Validation: rejection paths // ------------------------------------------------------------------------------------------ diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java index fb1dd9649180b..f7aa9d3e31eb1 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java @@ -18,9 +18,11 @@ package org.apache.flink.table.planner.plan.utils; +import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.calcite.FlinkTypeSystem; import org.apache.flink.table.planner.plan.schema.TimeIndicatorRelDataType; +import org.apache.flink.table.planner.utils.InternalConfigOptions; import org.apache.calcite.rel.core.JoinRelType; import org.apache.calcite.rel.type.RelDataType; @@ -28,6 +30,7 @@ import org.apache.calcite.sql.type.BasicSqlType; import org.apache.calcite.sql.type.SqlTypeName; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; @@ -159,6 +162,38 @@ void testDeriveRowTypeIgnoresStaleOriginalTypeNullability(boolean nullableTimeCo .isEqualTo(nullableTimeCol); } + /** + * When the planner has recorded a query-start epoch time (as {@link + * org.apache.flink.table.planner.delegation.PlannerBase#beforeTranslation} does before + * optimization begins), every SNAPSHOT call in the query must resolve its default {@code + * load_completed_time} to that same value. + */ + @Test + void testResolveDefaultLoadCompletedTimeUsesQueryStartEpochTimeWhenSet() { + final TableConfig tableConfig = TableConfig.getDefault(); + tableConfig.set(InternalConfigOptions.TABLE_QUERY_START_EPOCH_TIME, 1_700_000_000_000L); + + assertThat(LateralSnapshotJoinUtil.resolveDefaultLoadCompletedTime(tableConfig)) + .isEqualTo(1_700_000_000_000L); + } + + /** + * Outside of query translation (e.g. {@code TableTestUtil#verifyRelPlan} in tests) no + * query-start epoch time is recorded; the resolution must still fall back to the wall clock + * rather than fail. + */ + @Test + void testResolveDefaultLoadCompletedTimeFallsBackToWallClockWhenUnset() { + final TableConfig tableConfig = TableConfig.getDefault(); + + final long before = System.currentTimeMillis(); + final long loadCompletedTime = + LateralSnapshotJoinUtil.resolveDefaultLoadCompletedTime(tableConfig); + final long after = System.currentTimeMillis(); + + assertThat(loadCompletedTime).isBetween(before, after); + } + private RelDataType varchar(boolean nullable) { return typeFactory.createTypeWithNullability( typeFactory.createSqlType(SqlTypeName.VARCHAR), nullable);