This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 9c674275ca2 [FLINK-40541][table] Add on_time argument to SNAPSHOT for
LATERAL SNAPSHOT join
9c674275ca2 is described below
commit 9c674275ca28acc05b9fc5755634ef8c39335c4f
Author: Fabian Hueske <[email protected]>
AuthorDate: Mon Sep 7 09:13:17 2026 +0200
[FLINK-40541][table] Add on_time argument to SNAPSHOT for LATERAL SNAPSHOT
join
This closes #29098.
Co-Generated: Claude Opus 4.8 (1M context)
---
.../content.zh/docs/sql/reference/queries/joins.md | 10 +-
docs/content/docs/sql/reference/queries/joins.md | 22 +--
.../resolver/rules/ResolveCallByArgumentsRule.java | 5 +-
.../functions/BuiltInFunctionDefinitions.java | 1 +
.../table/types/inference/SystemTypeInference.java | 22 ++-
.../strategies/LateralSnapshotTypeStrategy.java | 70 ++++++++-
.../LateralSnapshotInputTypeStrategyTest.java | 120 ++++++++++++---
.../planner/calcite/FlinkCalciteSqlValidator.java | 18 ++-
.../planner/calcite/RelTimeIndicatorConverter.java | 1 +
.../logical/FlinkLogicalLateralSnapshotJoin.java | 11 ++
.../stream/StreamPhysicalLateralSnapshotJoin.java | 25 ++-
.../LogicalJoinToLateralSnapshotJoinRule.java | 148 ++++++++++--------
.../StreamPhysicalLateralSnapshotJoinRule.java | 1 +
.../LateralSnapshotJoinSemanticTestPrograms.java | 8 +-
.../stream/LateralSnapshotJoinTestPrograms.java | 2 +-
.../plan/stream/sql/ColumnExpansionTest.java | 90 +++++++++++
.../plan/stream/sql/SnapshotTableFunctionTest.java | 17 ++-
.../stream/sql/join/LateralSnapshotJoinTest.java | 169 ++++++++++++++++-----
.../stream/sql/join/LateralSnapshotJoinITCase.java | 6 +-
.../batch/sql/join/LateralSnapshotJoinTest.xml | 14 +-
.../stream/sql/join/LateralSnapshotJoinTest.xml | 40 ++---
21 files changed, 598 insertions(+), 202 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md
b/docs/content.zh/docs/sql/reference/queries/joins.md
index 2aad89d7588..b902a9a9929 100644
--- a/docs/content.zh/docs/sql/reference/queries/joins.md
+++ b/docs/content.zh/docs/sql/reference/queries/joins.md
@@ -323,7 +323,7 @@ For example, the following query enriches an append-only
stream of `orders` (the
SELECT o.order_id, o.currency, o.amount, r.rate
FROM orders AS o
-JOIN LATERAL SNAPSHOT(input => TABLE currency_rates) AS r
+JOIN LATERAL SNAPSHOT(input => TABLE currency_rates, on_time =>
DESCRIPTOR(update_time)) AS r
ON o.currency = r.currency;
order_id currency amount rate
@@ -368,6 +368,7 @@ SELECT [column_list]
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>, ]
@@ -379,13 +380,14 @@ The `SNAPSHOT` function accepts the following arguments:
| Argument | Type | Required | Description
|
| --- | --- | ---
|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `input` | TABLE | yes | The build-side table. It may use any changelog mode
(inserts, updates, and deletes). In streaming mode it must declare a
[watermark]({{< ref "docs/concepts/sql-table-concepts/time_attributes"
>}}#event-time).
|
+| `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_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).
|
-`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_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).
**Result and state characteristics**
@@ -397,7 +399,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
(`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.
+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 [...]
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 c5a5123d96a..5a0f6f4a9dc 100644
--- a/docs/content/docs/sql/reference/queries/joins.md
+++ b/docs/content/docs/sql/reference/queries/joins.md
@@ -328,7 +328,7 @@ For example, the following query enriches an append-only
stream of `orders` (the
SELECT o.order_id, o.currency, o.amount, r.rate
FROM orders AS o
-JOIN LATERAL SNAPSHOT(input => TABLE currency_rates) AS r
+JOIN LATERAL SNAPSHOT(input => TABLE currency_rates, on_time =>
DESCRIPTOR(update_time)) AS r
ON o.currency = r.currency;
order_id currency amount rate
@@ -377,6 +377,7 @@ SELECT [column_list]
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>, ]
@@ -386,15 +387,16 @@ ON probe_table.col = s.col
The `SNAPSHOT` function accepts the following arguments:
-| Argument | Type | Required | Description
|
-| --- | --- | ---
|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `input` | TABLE | yes | The build-side table. It may use any [changelog
mode]({{< ref "docs/sql/reference/queries/changelog" >}}) (inserts, updates,
and deletes). In streaming mode it must declare a [watermark]({{< ref
"docs/concepts/sql-table-concepts/time_attributes" >}}#event-time).
|
-| `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_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).
|
+| Argument | Type | Required | Description
|
+| --- | --- | ---
|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `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_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).
|
-`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_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).
**Result and state characteristics**
@@ -406,7 +408,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
(`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.
+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 [...]
Because every probe-side row is joined against the final, complete build side,
the batch result is **deterministic**.
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
index 7364536099a..9f28484f585 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
@@ -300,7 +300,10 @@ final class ResolveCallByArgumentsRule implements
ResolverRule {
}
SystemTypeInference.checkNoSystemArguments(
- inference.disableSystemArguments(), namedArgs.keySet(),
functionName);
+ inference.disableSystemArguments(),
+ namedArgs.keySet(),
+
declaredArgs.stream().map(StaticArgument::getName).collect(Collectors.toList()),
+ functionName);
fillInDefaultNamedArguments(declaredArgs, namedArgs);
fillInPtfSpecificNamedArguments(
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 a2a0cb5547f..fdfa00390d6 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
@@ -946,6 +946,7 @@ public final class BuiltInFunctionDefinitions {
StaticArgumentTrait.SUPPORT_UPDATES,
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(
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
index a1f9fbc5a9f..a4eadc77752 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/SystemTypeInference.java
@@ -139,17 +139,25 @@ public class SystemTypeInference {
public static void checkNoSystemArguments(
boolean sysArgsDisabled,
Collection<String> suppliedArgumentNames,
+ Collection<String> declaredArgumentNames,
String functionName) {
if (!sysArgsDisabled) {
return;
}
for (StaticArgument systemArg : PROCESS_TABLE_FUNCTION_SYSTEM_ARGS) {
- if (suppliedArgumentNames.contains(systemArg.getName())) {
+ final String systemArgName = systemArg.getName();
+ // A function that disabled the automatic system arguments may
still declare an argument
+ // with a reserved name. A supplied argument that matches such a
declared name is
+ // handled as an ordinary argument, not a system argument.
+ if (declaredArgumentNames.contains(systemArgName)) {
+ continue;
+ }
+ if (suppliedArgumentNames.contains(systemArgName)) {
throw new ValidationException(
String.format(
"Invalid function call. The '%s' argument is
not supported "
+ "because function '%s' does not use
system arguments.",
- systemArg.getName(), functionName));
+ systemArgName, functionName));
}
}
}
@@ -184,7 +192,9 @@ public class SystemTypeInference {
"Function requires a static signature that is not
overloaded and doesn't contain varargs.");
}
- checkReservedArgs(declaredArgs);
+ if (!disableSystemArgs) {
+ checkReservedArgs(declaredArgs);
+ }
checkMultipleTableArgs(declaredArgs);
checkPassThroughColumns(declaredArgs);
@@ -804,7 +814,11 @@ public class SystemTypeInference {
}
}
- private static boolean isUnsupportedOnTimeColumn(LogicalType type) {
+ /**
+ * Checks whether a column referenced by an {@code on_time} argument has
an unsupported data
+ * type. A supported column is a TIMESTAMP or TIMESTAMP_LTZ column up to
precision 3.
+ */
+ public static boolean isUnsupportedOnTimeColumn(LogicalType type) {
return !LogicalTypeChecks.canBeTimeAttributeType(type)
|| LogicalTypeChecks.getPrecision(type) > 3;
}
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 49bead4730d..24c22c85d27 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
@@ -31,8 +31,11 @@ import
org.apache.flink.table.types.inference.ConstantArgumentCount;
import org.apache.flink.table.types.inference.InputTypeStrategy;
import org.apache.flink.table.types.inference.Signature;
import org.apache.flink.table.types.inference.Signature.Argument;
+import org.apache.flink.table.types.inference.SystemTypeInference;
import org.apache.flink.table.types.inference.TypeStrategy;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.utils.DataTypeUtils;
+import org.apache.flink.types.ColumnList;
import java.util.List;
import java.util.Optional;
@@ -48,6 +51,7 @@ 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)
@@ -74,23 +78,28 @@ public final class LateralSnapshotTypeStrategy {
public static final String INPUT_ARG_NAME = "input";
+ /** The {@code on_time} DESCRIPTOR argument naming the build-side row-time
column. */
+ public static final int ON_TIME_ARG_INDEX = 1;
+
+ 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 = 1;
+ 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 = 2;
+ public static final int LOAD_COMPLETED_TIME_ARG_INDEX = 3;
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 = 3;
+ public static final int LOAD_COMPLETED_IDLE_TIMEOUT_ARG_INDEX = 4;
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 = 4;
+ public static final int STATE_TTL_ARG_INDEX = 5;
public static final String STATE_TTL_ARG_NAME = "state_ttl";
@@ -119,7 +128,7 @@ public final class LateralSnapshotTypeStrategy {
new InputTypeStrategy() {
@Override
public ArgumentCount getArgumentCount() {
- return ConstantArgumentCount.between(1, 5);
+ return ConstantArgumentCount.between(1, 6);
}
@Override
@@ -133,6 +142,7 @@ public final class LateralSnapshotTypeStrategy {
return List.of(
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(
@@ -188,6 +198,13 @@ public final class LateralSnapshotTypeStrategy {
throwOnFailure, "Argument 'input' of SNAPSHOT must be a
table.");
}
+ // Validate on_time if provided. Presence is enforced by the planner
rule for streaming.
+ final Optional<List<DataType>> timeColumnFailure =
+ validateOnTime(callContext, throwOnFailure);
+ if (timeColumnFailure != null) {
+ 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 =
@@ -233,6 +250,49 @@ public final class LateralSnapshotTypeStrategy {
return Optional.of(callContext.getArgumentDataTypes());
}
+ /**
+ * Validates {@code on_time} when present: it must name exactly one
existing TIMESTAMP or
+ * TIMESTAMP_LTZ column (precision up to 3). Returns {@code null} when the
argument is absent or
+ * valid; otherwise returns the failure result of {@link CallContext#fail}.
+ */
+ private static Optional<List<DataType>> validateOnTime(
+ final CallContext callContext, final boolean throwOnFailure) {
+ if (!isArgumentProvided(callContext, ON_TIME_ARG_INDEX)) {
+ return null;
+ }
+ final List<String> columns =
+ callContext
+ .getArgumentValue(ON_TIME_ARG_INDEX, ColumnList.class)
+ .map(ColumnList::getNames)
+ .orElse(List.of());
+ if (columns.size() != 1) {
+ return callContext.fail(
+ throwOnFailure,
+ "Argument 'on_time' of SNAPSHOT must reference exactly one
column.");
+ }
+ final String timeColumn = columns.get(0);
+ final DataType inputType =
callContext.getArgumentDataTypes().get(INPUT_ARG_INDEX);
+ final int idx = DataType.getFieldNames(inputType).indexOf(timeColumn);
+ if (idx < 0) {
+ return callContext.fail(
+ throwOnFailure,
+ "Argument 'on_time' of SNAPSHOT references column '%s'
which is not present "
+ + "in the input table.",
+ timeColumn);
+ }
+ final LogicalType columnType =
+
DataType.getFieldDataTypes(inputType).get(idx).getLogicalType();
+ if (SystemTypeInference.isUnsupportedOnTimeColumn(columnType)) {
+ return callContext.fail(
+ throwOnFailure,
+ "Argument 'on_time' of SNAPSHOT must reference a TIMESTAMP
or TIMESTAMP_LTZ "
+ + "column (up to precision 3), but column '%s' has
type '%s'.",
+ timeColumn,
+ columnType.asSummaryString());
+ }
+ return null;
+ }
+
private static boolean isArgumentProvided(final CallContext callContext,
final int index) {
return callContext.getArgumentDataTypes().size() > index
&& !callContext.isArgumentNull(index);
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 05012a3b277..23e132c0c93 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
@@ -22,6 +22,7 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
import org.apache.flink.table.types.inference.utils.TableSemanticsMock;
+import org.apache.flink.types.ColumnList;
import java.time.Duration;
import java.time.LocalDateTime;
@@ -41,17 +42,22 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
private static final DataType TABLE_TYPE =
DataTypes.ROW(
DataTypes.FIELD("k", DataTypes.STRING()),
- DataTypes.FIELD("v", DataTypes.INT()));
+ DataTypes.FIELD("v", DataTypes.INT()),
+ DataTypes.FIELD("ts", DataTypes.TIMESTAMP(3)));
private static final DataType STRING_TYPE = DataTypes.STRING();
+ private static final DataType DESCRIPTOR_TYPE = DataTypes.DESCRIPTOR();
private static final DataType TIMESTAMP_TYPE = DataTypes.TIMESTAMP(3);
private static final DataType INTERVAL_TYPE =
DataTypes.INTERVAL(DataTypes.SECOND());
+ private static final ColumnList ON_TIME = ColumnList.of("ts");
+
@Override
protected Stream<TestSpec> testData() {
return Stream.of(
//
----------------------------------------------------------------------------
- // Valid: just the build-side table.
+ // 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)",
@@ -60,16 +66,26 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
.expectArgumentTypes(TABLE_TYPE),
+ //
----------------------------------------------------------------------------
+ // Valid: input + on_time.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy("Valid: input + on_time",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .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, STRING_TYPE)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
STRING_TYPE)
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, "compile_time")
- .expectArgumentTypes(TABLE_TYPE, STRING_TYPE),
+ .calledWithLiteralAt(1, ON_TIME)
+ .calledWithLiteralAt(2, "compile_time")
+ .expectArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
STRING_TYPE),
//
----------------------------------------------------------------------------
// Valid: 'user_time' with a TIMESTAMP literal.
@@ -77,11 +93,14 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
TestSpec.forStrategy(
"Valid: condition='user_time' +
load_completed_time",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE,
TIMESTAMP_TYPE)
+ .calledWithArgumentTypes(
+ TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE,
TIMESTAMP_TYPE)
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, "user_time")
- .calledWithLiteralAt(2,
LocalDateTime.parse("2026-07-01T00:00:00.001"))
- .expectArgumentTypes(TABLE_TYPE, STRING_TYPE,
TIMESTAMP_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.
@@ -89,17 +108,20 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
TestSpec.forStrategy("Valid: full args",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
.calledWithArgumentTypes(
TABLE_TYPE,
+ DESCRIPTOR_TYPE,
STRING_TYPE,
TIMESTAMP_TYPE,
INTERVAL_TYPE,
INTERVAL_TYPE)
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, "user_time")
- .calledWithLiteralAt(2,
LocalDateTime.parse("2026-07-01T00:00:00.001"))
- .calledWithLiteralAt(3, Duration.ofSeconds(10))
- .calledWithLiteralAt(4, Duration.ofDays(1))
+ .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))
.expectArgumentTypes(
TABLE_TYPE,
+ DESCRIPTOR_TYPE,
STRING_TYPE,
TIMESTAMP_TYPE,
INTERVAL_TYPE,
@@ -123,15 +145,67 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
// Intentionally no table type registered at position
0.
.expectErrorMessage("Argument 'input' of SNAPSHOT must
be a table."),
+ //
----------------------------------------------------------------------------
+ // Invalid: on_time references an unknown column.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: on_time references unknown column",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, ColumnList.of("nonexistent"))
+ .expectErrorMessage(
+ "Argument 'on_time' of SNAPSHOT references
column 'nonexistent' "
+ + "which is not present in the input
table."),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: on_time references a non-timestamp column.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: on_time references non-timestamp
column",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, ColumnList.of("v"))
+ .expectErrorMessage(
+ "Argument 'on_time' of SNAPSHOT must reference
a TIMESTAMP or "
+ + "TIMESTAMP_LTZ column (up to
precision 3), but column 'v' "
+ + "has type 'INT'."),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: on_time references no column.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: on_time references no column",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE)
+ .calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
+ .calledWithLiteralAt(1, ColumnList.of())
+ .expectErrorMessage(
+ "Argument 'on_time' of SNAPSHOT must reference
exactly one column."),
+
+ //
----------------------------------------------------------------------------
+ // Invalid: on_time references more than one column.
+ //
----------------------------------------------------------------------------
+ TestSpec.forStrategy(
+ "Invalid: on_time references multiple columns",
+ LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE)
+ .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, STRING_TYPE)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
STRING_TYPE)
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, "user_time")
+ .calledWithLiteralAt(1, ON_TIME)
+ .calledWithLiteralAt(2, "user_time")
.expectErrorMessage(
"SNAPSHOT requires 'load_completed_time' when "
+ "'load_completed_condition' is
'user_time'."),
@@ -142,10 +216,12 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
TestSpec.forStrategy(
"Invalid: load_completed_time without explicit
'user_time'",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE,
TIMESTAMP_TYPE)
+ .calledWithArgumentTypes(
+ TABLE_TYPE, DESCRIPTOR_TYPE, STRING_TYPE,
TIMESTAMP_TYPE)
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, "compile_time")
- .calledWithLiteralAt(2,
LocalDateTime.parse("2026-07-01T00:00:00.001"))
+ .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'."),
@@ -156,9 +232,10 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
TestSpec.forStrategy(
"Invalid: unknown condition value",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
STRING_TYPE)
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
- .calledWithLiteralAt(1, "invalid_condition")
+ .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'."),
@@ -168,8 +245,9 @@ class LateralSnapshotInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
TestSpec.forStrategy(
"Invalid: non-literal
load_completed_condition",
LATERAL_SNAPSHOT_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, STRING_TYPE)
+ .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."));
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java
index 0eb925352f0..6f263ad6b1f 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java
@@ -34,7 +34,9 @@ import
org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction;
import org.apache.flink.table.planner.plan.FlinkCalciteCatalogReader;
import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
import org.apache.flink.table.planner.utils.ShortcutUtils;
+import org.apache.flink.table.types.inference.StaticArgument;
import org.apache.flink.table.types.inference.SystemTypeInference;
+import org.apache.flink.table.types.inference.TypeInference;
import org.apache.flink.table.types.logical.DecimalType;
import org.apache.calcite.plan.RelOptCluster;
@@ -470,8 +472,11 @@ public final class FlinkCalciteSqlValidator extends
FlinkSqlParsingValidator {
*/
private static void checkDisabledSystemArgs(SqlBasicCall call) {
final SqlOperator operator = call.getOperator();
- if (!(operator instanceof BridgingSqlFunction)
- || !((BridgingSqlFunction)
operator).getTypeInference().disableSystemArguments()) {
+ if (!(operator instanceof BridgingSqlFunction)) {
+ return;
+ }
+ final TypeInference typeInference = ((BridgingSqlFunction)
operator).getTypeInference();
+ if (!typeInference.disableSystemArguments()) {
return;
}
final Set<String> suppliedArgNames = new HashSet<>();
@@ -483,7 +488,14 @@ public final class FlinkCalciteSqlValidator extends
FlinkSqlParsingValidator {
}
}
}
- SystemTypeInference.checkNoSystemArguments(true, suppliedArgNames,
operator.getName());
+ // A function that disabled the automatic system arguments may still
declare an argument
+ // with a reserved name (e.g. SNAPSHOT declares `on_time`); such
declared names are allowed.
+ final Set<String> declaredArgNames =
+ typeInference.getStaticArguments().orElse(List.of()).stream()
+ .map(StaticArgument::getName)
+ .collect(Collectors.toSet());
+ SystemTypeInference.checkNoSystemArguments(
+ true, suppliedArgNames, declaredArgNames, operator.getName());
}
@Override
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
index 86954858c16..5ab9f8ad0cb 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/RelTimeIndicatorConverter.java
@@ -401,6 +401,7 @@ public final class RelTimeIndicatorConverter extends
RelHomogeneousShuttle {
newRight,
newCondition,
join.getJoinType(),
+ join.getRightTimeAttributeIndex(),
join.getLoadCompletedCondition(),
join.getLoadCompletedTime(),
join.getLoadCompletedIdleTimeoutMs(),
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalLateralSnapshotJoin.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalLateralSnapshotJoin.java
index 6391e66fb1b..7127cfc4ab5 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalLateralSnapshotJoin.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/logical/FlinkLogicalLateralSnapshotJoin.java
@@ -50,6 +50,8 @@ public class FlinkLogicalLateralSnapshotJoin extends Join
implements FlinkLogica
private final Long loadCompletedTime;
private final @Nullable Long loadCompletedIdleTimeoutMs;
private final @Nullable Long stateTtlMs;
+ // Build-side row-time column index (from the on_time argument); -1 in
batch.
+ private final int rightTimeAttributeIndex;
public FlinkLogicalLateralSnapshotJoin(
RelOptCluster cluster,
@@ -58,6 +60,7 @@ public class FlinkLogicalLateralSnapshotJoin extends Join
implements FlinkLogica
RelNode right,
RexNode condition,
JoinRelType joinType,
+ int rightTimeAttributeIndex,
String loadCompletedCondition,
Long loadCompletedTime,
@Nullable Long loadCompletedIdleTimeoutMs,
@@ -72,12 +75,17 @@ public class FlinkLogicalLateralSnapshotJoin extends Join
implements FlinkLogica
Collections.emptySet(),
joinType);
Preconditions.checkNotNull(loadCompletedTime, "loadCompletedTime must
not be null.");
+ this.rightTimeAttributeIndex = rightTimeAttributeIndex;
this.loadCompletedCondition = loadCompletedCondition;
this.loadCompletedTime = loadCompletedTime;
this.loadCompletedIdleTimeoutMs = loadCompletedIdleTimeoutMs;
this.stateTtlMs = stateTtlMs;
}
+ public int getRightTimeAttributeIndex() {
+ return rightTimeAttributeIndex;
+ }
+
public String getLoadCompletedCondition() {
return loadCompletedCondition;
}
@@ -109,6 +117,7 @@ public class FlinkLogicalLateralSnapshotJoin extends Join
implements FlinkLogica
right,
conditionExpr,
joinType,
+ rightTimeAttributeIndex,
loadCompletedCondition,
loadCompletedTime,
loadCompletedIdleTimeoutMs,
@@ -154,6 +163,7 @@ public class FlinkLogicalLateralSnapshotJoin extends Join
implements FlinkLogica
RelNode right,
RexNode condition,
JoinRelType joinType,
+ int rightTimeAttributeIndex,
String loadCompletedCondition,
Long loadCompletedTime,
@Nullable Long loadCompletedIdleTimeoutMs,
@@ -167,6 +177,7 @@ public class FlinkLogicalLateralSnapshotJoin extends Join
implements FlinkLogica
right,
condition,
joinType,
+ rightTimeAttributeIndex,
loadCompletedCondition,
loadCompletedTime,
loadCompletedIdleTimeoutMs,
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLateralSnapshotJoin.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLateralSnapshotJoin.java
index 2b1e8b9ebd6..9304544c7a8 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLateralSnapshotJoin.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalLateralSnapshotJoin.java
@@ -34,13 +34,11 @@ import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexNode;
import javax.annotation.Nullable;
import java.util.Collections;
-import java.util.List;
import static
org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig;
@@ -53,6 +51,7 @@ import static
org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConf
public class StreamPhysicalLateralSnapshotJoin extends CommonPhysicalJoin
implements StreamPhysicalRel {
+ private final int rightTimeAttributeIndex;
private final String loadCompletedCondition;
private final Long loadCompletedTime;
private final @Nullable Long loadCompletedIdleTimeoutMs;
@@ -65,11 +64,13 @@ public class StreamPhysicalLateralSnapshotJoin extends
CommonPhysicalJoin
RelNode rightRel,
RexNode condition,
JoinRelType joinType,
+ int rightTimeAttributeIndex,
String loadCompletedCondition,
Long loadCompletedTime,
@Nullable Long loadCompletedIdleTimeoutMs,
@Nullable Long stateTtlMs) {
super(cluster, traitSet, leftRel, rightRel, condition, joinType,
Collections.emptyList());
+ this.rightTimeAttributeIndex = rightTimeAttributeIndex;
Preconditions.checkNotNull(loadCompletedTime, "loadCompletedTime must
not be null.");
this.loadCompletedCondition = loadCompletedCondition;
this.loadCompletedTime = loadCompletedTime;
@@ -107,6 +108,7 @@ public class StreamPhysicalLateralSnapshotJoin extends
CommonPhysicalJoin
right,
conditionExpr,
joinType,
+ rightTimeAttributeIndex,
loadCompletedCondition,
loadCompletedTime,
loadCompletedIdleTimeoutMs,
@@ -129,17 +131,14 @@ public class StreamPhysicalLateralSnapshotJoin extends
CommonPhysicalJoin
@Override
public ExecNode<?> translateToExecNode() {
- // The build (right) side carries a watermark, so it must expose a
row-time attribute whose
- // field index drives the event-time-ordered application of buffered
build-side changes.
- final List<RelDataTypeField> rightFields =
getRight().getRowType().getFieldList();
- int rightTimeAttributeIndex = -1;
- for (int i = 0; i < rightFields.size(); i++) {
- if
(FlinkTypeFactory.isRowtimeIndicatorType(rightFields.get(i).getType())) {
- rightTimeAttributeIndex = i;
- break;
- }
- }
- if (rightTimeAttributeIndex < 0) {
+ // The build-side row-time column index (from the on_time argument)
drives the
+ // event-time-ordered application of buffered build-side changes.
Verify it points at an
+ // in-bounds row-time attribute.
+ final RelDataType rightRowType = getRight().getRowType();
+ if (rightTimeAttributeIndex < 0
+ || rightTimeAttributeIndex >= rightRowType.getFieldCount()
+ || !FlinkTypeFactory.isRowtimeIndicatorType(
+
rightRowType.getFieldList().get(rightTimeAttributeIndex).getType())) {
throw new TableException(
"The build (right) side of a LATERAL SNAPSHOT join must
have a row-time "
+ "attribute. This is a bug, please file an
issue.");
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 c4b1012df97..b50ff8e69b7 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
@@ -89,7 +89,7 @@ public class LogicalJoinToLateralSnapshotJoinRule
@Override
public void onMatch(RelOptRuleCall call) {
final FlinkLogicalJoin join = call.rel(0);
- final RelNode leftNode = join.getLeft();
+ final RelNode probeInputNode = join.getLeft();
final FlinkLogicalTableFunctionScan scan =
findSnapshotScan(join.getRight());
if (scan == null) {
// matches() guarantees a SNAPSHOT scan on the right, so this
cannot happen.
@@ -116,47 +116,23 @@ public class LogicalJoinToLateralSnapshotJoinRule
final RexCall snapshotCall = (RexCall) scan.getCall();
- // Resolve the raw build-side TABLE input the operator reads. A null
result means the
- // SNAPSHOT call is malformed, which cannot happen for a plan that
reached this rule.
- final RelNode rawTableInput = getSnapshotInputTable(scan);
- if (rawTableInput == null) {
- throw new TableException(
- "Could not resolve the TABLE input of the SNAPSHOT scan on
the build side of "
- + "a LATERAL SNAPSHOT join. This is a bug, please
file an issue.");
- }
- // The build-side row-time attribute drives the streaming operator's
LOAD phase. In batch
- // all input is bounded and the join degrades to a regular join (see
- // BatchPhysicalLateralSnapshotJoinRule), so no watermark is required.
- if (!ShortcutUtils.unwrapContext(join).isBatchMode()) {
- // The build-side input must declare exactly one watermark,
otherwise the operator
- // cannot determine when the LOAD phase is complete.
- final long rowtimeCount =
- rawTableInput.getRowType().getFieldList().stream()
- .filter(f ->
FlinkTypeFactory.isRowtimeIndicatorType(f.getType()))
- .count();
- if (rowtimeCount == 0) {
- throw new ValidationException(
- "LATERAL SNAPSHOT requires a watermark on the
build-side input.");
- }
- if (rowtimeCount > 1) {
- throw new ValidationException(
- String.format(
- "The build-side input of a LATERAL SNAPSHOT
join must not have more than one "
- + "row-time attribute, but found %d.",
- rowtimeCount));
- }
- }
-
- // Replace the SNAPSHOT TableFunctionScan with its input, preserving
any FlinkLogicalCalc
- // nodes that the optimizer placed above the scan.
- final RelNode rightNode = replaceSnapshotScan(join.getRight());
- if (rightNode == null) {
+ // Replace the SNAPSHOT TableFunctionScan with its TABLE input,
preserving any
+ // FlinkLogicalCalc nodes that the optimizer placed above the scan.
+ final RelNode buildInputNode = replaceSnapshotScan(join.getRight());
+ if (buildInputNode == null) {
throw new TableException(
"Could not rewrite the build side of a LATERAL SNAPSHOT
join by replacing the "
+ "SNAPSHOT scan with its TABLE input. This is a
bug, please file an "
+ "issue.");
}
+ // Resolve the build-side rowtime column named by the on_time
argument. Streaming
+ // requires it; in batch the join degrades to a regular join (see
+ // BatchPhysicalLateralSnapshotJoinRule) and no watermark is needed.
+ final boolean isBatch =
ShortcutUtils.unwrapContext(join).isBatchMode();
+ final int rightTimeAttributeIndex =
+ resolveOnTimeIndex(snapshotCall, buildInputNode, isBatch);
+
final List<RexNode> operands = snapshotCall.getOperands();
final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
final RexExecutor executor =
join.getCluster().getPlanner().getExecutor();
@@ -232,12 +208,12 @@ public class LogicalJoinToLateralSnapshotJoinRule
intervalMillis(stateTtlLiteral,
LateralSnapshotTypeStrategy.STATE_TTL_ARG_NAME);
// The original join condition's field types were resolved against the
SNAPSHOT scan's
- // materialized output, but rightNode (its raw TABLE input) still
exposes the build-side
- // row-time attribute as an indicator (see replaceSnapshotScan).
Retype the condition to
- // the actual left+right input types.
+ // materialized output, but buildInputNode (its raw TABLE input) still
exposes the
+ // build-side rowtime attribute as an indicator (see
replaceSnapshotScan). Retype the
+ // condition to the actual left+right input types.
final List<RelDataTypeField> leftRightFields = new ArrayList<>();
- leftRightFields.addAll(leftNode.getRowType().getFieldList());
- leftRightFields.addAll(rightNode.getRowType().getFieldList());
+ leftRightFields.addAll(probeInputNode.getRowType().getFieldList());
+ leftRightFields.addAll(buildInputNode.getRowType().getFieldList());
final RexNode rebasedCondition =
join.getCondition()
.accept(
@@ -253,24 +229,23 @@ public class LogicalJoinToLateralSnapshotJoinRule
// build-side input.
final RelNode node =
FlinkLogicalLateralSnapshotJoin.create(
- leftNode,
- rightNode,
+ probeInputNode,
+ buildInputNode,
rebasedCondition,
joinType,
+ rightTimeAttributeIndex,
loadCompletedCondition,
loadCompletedTime,
loadCompletedIdleTimeoutMs,
stateTtlMs);
final int origRightCount =
unwrap(join.getRight()).getRowType().getFieldCount();
- final int newRightCount = rightNode.getRowType().getFieldCount();
+ final int newRightCount = buildInputNode.getRowType().getFieldCount();
final boolean isRowtimeFieldAdded = newRightCount > origRightCount;
if (isRowtimeFieldAdded) {
- // If the build-side projection stripped the row-time attribute,
replaceSnapshotScan
- // re-appended it as a trailing column so it reaches the operator.
In that case the node
- // has extra trailing column(s) that a wrapper Calc projects away
to restore the
- // original join's output type. Otherwise, the node's output type
already matches the
- // original join.
+ // If a build-side projection stripped the rowtime attribute,
rebaseCalc re-appended it
+ // as a trailing column so it reaches the operator. Project that
column away with a
+ // wrapper Calc to restore the original join's output type.
final RelDataType originalOutputType = join.getRowType();
final List<RexNode> wrapperProjects = new ArrayList<>();
for (int i = 0; i < originalOutputType.getFieldCount(); i++) {
@@ -351,18 +326,26 @@ public class LogicalJoinToLateralSnapshotJoinRule
* Walks the right subtree replacing the {@link
FlinkLogicalTableFunctionScan} (the SNAPSHOT
* scan) with the scan's TABLE input, while preserving any {@link
FlinkLogicalCalc} nodes
* stacked above the scan. The SNAPSHOT type strategy materializes the
build-side time
- * attributes, so the scan's output type differs from its input's (the
build-side row-time
- * attribute is a plain timestamp on the scan output but a row-time
indicator on the raw input).
+ * attributes, so the scan's output type differs from its input's (the
build-side rowtime
+ * attribute is a plain timestamp on the scan output but a rowtime
indicator on the raw input).
* Each preserved Calc was built against the materialized scan output, so
its {@link RexProgram}
- * is rebased onto the raw (row-time-bearing) input type, which lets the
row-time attribute flow
+ * is rebased onto the raw (rowtime-bearing) input type, which lets the
rowtime attribute flow
* through to the operator.
*/
@Nullable
private static RelNode replaceSnapshotScan(RelNode node) {
final RelNode current = unwrap(node);
if (current instanceof FlinkLogicalTableFunctionScan) {
- // the top node is the TableFunctionScan, return its table input
argument
- return getSnapshotInputTable((FlinkLogicalTableFunctionScan)
current);
+ // Resolve the raw build-side TABLE input the operator reads. A
null result means the
+ // SNAPSHOT call is malformed, which cannot happen for a plan that
reached this rule.
+ final RelNode tableInput =
+ getSnapshotInputTable((FlinkLogicalTableFunctionScan)
current);
+ if (tableInput == null) {
+ throw new TableException(
+ "Could not resolve the TABLE input of the SNAPSHOT
scan on the build side of "
+ + "a LATERAL SNAPSHOT join. This is a bug,
please file an issue.");
+ }
+ return tableInput;
}
if (current instanceof FlinkLogicalCalc) {
// the top node is a calc that needs to be rebased
@@ -376,15 +359,55 @@ public class LogicalJoinToLateralSnapshotJoinRule
return null;
}
+ /**
+ * Resolves the build-side rowtime column named by the {@code on_time}
argument to its field
+ * index in {@code buildInputNode}. Streaming requires the argument and
the referenced column to
+ * be a rowtime attribute; batch does not use a rowtime attribute and
returns {@code -1} when
+ * the argument is absent.
+ */
+ private static int resolveOnTimeIndex(
+ RexCall snapshotCall, RelNode buildInputNode, boolean isBatch) {
+ final List<RexNode> operands = snapshotCall.getOperands();
+ final int argIndex = LateralSnapshotTypeStrategy.ON_TIME_ARG_INDEX;
+ final RexNode timeColumnArg = argIndex < operands.size() ?
operands.get(argIndex) : null;
+ if (timeColumnArg == null || timeColumnArg.isA(SqlKind.DEFAULT)) {
+ if (!isBatch) {
+ throw new ValidationException(
+ "LATERAL SNAPSHOT requires the 'on_time' argument to
identify the "
+ + "build-side rowtime attribute.");
+ }
+ return -1;
+ }
+ // The type strategy (LateralSnapshotTypeStrategy#validateOnTime)
already validated that
+ // on_time is a single-column DESCRIPTOR referencing an existing
TIMESTAMP/TIMESTAMP_LTZ
+ // column, so the operand structure is safe to read here.
+ final String timeColName =
+ RexLiteral.stringValue((RexLiteral) ((RexCall)
timeColumnArg).getOperands().get(0));
+ final int timeColIdx =
buildInputNode.getRowType().getFieldNames().indexOf(timeColName);
+ if (isBatch) {
+ // Batch degrades to a regular join and does not use the rowtime
attribute.
+ return timeColIdx;
+ }
+ // A rowtime column named by on_time is always retained in the build
input, so a missing
+ // index means the referenced column is not a rowtime attribute.
+ if (timeColIdx < 0
+ || !FlinkTypeFactory.isRowtimeIndicatorType(
+
buildInputNode.getRowType().getFieldList().get(timeColIdx).getType())) {
+ throw new ValidationException(
+ String.format(
+ "Argument 'on_time' of SNAPSHOT must reference a
rowtime attribute "
+ + "(a column with a watermark), but column
'%s' is not one.",
+ timeColName));
+ }
+ return timeColIdx;
+ }
+
/**
* Rebuilds {@code calc}'s {@link RexProgram} so it reads from {@code
newInput} (whose
- * build-side time attributes are still row-time indicators) instead of
the materialized
- * SNAPSHOT scan output it was originally built against. Input references
are retyped to the new
- * input's field types; the projection/condition expressions and output
field names are
- * otherwise preserved.
- *
- * <p>If the projection dropped the build-side row-time attribute, it is
re-appended as a
- * trailing column so it is available for the snapshot join operator.
+ * build-side time attributes are still rowtime indicators) instead of the
materialized SNAPSHOT
+ * scan output it was originally built against. Input references are
retyped to the new input's
+ * field types; the projection/condition expressions and output field
names are otherwise
+ * preserved.
*/
private static RelNode rebaseCalc(FlinkLogicalCalc calc, RelNode newInput)
{
final RexProgram program = calc.getProgram();
@@ -410,7 +433,8 @@ public class LogicalJoinToLateralSnapshotJoinRule
:
program.expandLocalRef(program.getCondition()).accept(retyper);
final List<String> fieldNames = new
ArrayList<>(program.getOutputRowType().getFieldNames());
- // Re-append the build-side row-time attribute if this projection
dropped it.
+ // Re-append the build-side rowtime attribute if this projection
dropped it, so it reaches
+ // the operator even when the outer query does not select it.
final boolean exposesRowtime =
newProjects.stream()
.anyMatch(p ->
FlinkTypeFactory.isRowtimeIndicatorType(p.getType()));
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
index cde7e68e81a..ec8010deed1 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
@@ -68,6 +68,7 @@ public class StreamPhysicalLateralSnapshotJoinRule extends
ConverterRule {
newRight,
join.getCondition(),
join.getJoinType(),
+ join.getRightTimeAttributeIndex(),
join.getLoadCompletedCondition(),
join.getLoadCompletedTime(),
join.getLoadCompletedIdleTimeoutMs(),
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 bb45bde761c..f8864578acf 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
@@ -92,7 +92,7 @@ public class LateralSnapshotJoinSemanticTestPrograms {
+ "SELECT probe.pk, probe.pv, s.bk, s.bv "
+ "FROM probe "
+ " JOIN LATERAL TABLE(SNAPSHOT("
- + " input => TABLE b, "
+ + " 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")
@@ -135,7 +135,7 @@ public class LateralSnapshotJoinSemanticTestPrograms {
.build())
.runSql(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b, "
+ + "input => TABLE b, on_time =>
DESCRIPTOR(bts), "
+ MID_FLIP
+ ") AS s ON probe.pk = s.bk")
.build();
@@ -262,7 +262,7 @@ public class LateralSnapshotJoinSemanticTestPrograms {
.runSql(
"INSERT INTO sink SELECT probe.pk, probe.pv, s.bk,
s.bv "
+ "FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b"
+ + "input => TABLE b, on_time =>
DESCRIPTOR(bts)"
+ ") AS s ON probe.pk = s.bk")
.build();
@@ -354,7 +354,7 @@ public class LateralSnapshotJoinSemanticTestPrograms {
+ " FROM probe "
+ joinType
+ " LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ flip
+ ") AS s ON "
+ condition;
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 fcae8c89c62..b1fae558aab 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
@@ -96,7 +96,7 @@ public class LateralSnapshotJoinTestPrograms {
private static final String SNAPSHOT_BUILD =
"LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 79d1c3eacc2..1d96ad811e8 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
@@ -452,6 +452,96 @@ class ColumnExpansionTest {
+ "argument 'r2' has data type 'INT'.");
}
+ @Test
+ void testLateralSnapshotJoinWithOnTimeOnHiddenMetadataColumn() {
+ tableEnv.getConfig()
+ .set(
+ TABLE_COLUMN_EXPANSION_STRATEGY,
+ List.of(EXCLUDE_DEFAULT_VIRTUAL_METADATA_COLUMNS));
+
+ tableEnv.executeSql(
+ "CREATE TABLE snapshot_probe (\n"
+ + " pk STRING,\n"
+ + " pts TIMESTAMP(3),\n"
+ + " WATERMARK FOR pts AS pts\n"
+ + ") WITH ('connector' = 'values', 'bounded' =
'false')");
+
+ // The build-side watermark is declared on a virtual metadata column
that
+ // EXCLUDE_DEFAULT_VIRTUAL_METADATA_COLUMNS hides from SELECT *.
+ tableEnv.executeSql(
+ "CREATE TABLE snapshot_build_hidden (\n"
+ + " bk STRING,\n"
+ + " bv INT,\n"
+ + " rt TIMESTAMP_LTZ(3) METADATA VIRTUAL,\n"
+ + " WATERMARK FOR rt AS rt\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false',\n"
+ + " 'readable-metadata' = 'rt:TIMESTAMP_LTZ(3)'\n"
+ + ")");
+
+ 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";
+
+ // rt is selected due to the on_time descriptor, even though it is a
hidden metadata column.
+ assertColumnNames(sql, "pk", "pts", "bk", "bv", "rt");
+ // The row-time attribute named by on_time is recognized, so the join
plans as a LATERAL
+ // SNAPSHOT join instead of being rejected for a missing build-side
watermark.
+ assertThat(tableEnv.explainSql(sql)).contains("LateralSnapshotJoin");
+ }
+
+ @Test
+ void testLateralSnapshotJoinWithOnTimeOnPushedDownHiddenMetadataColumn() {
+ tableEnv.getConfig()
+ .set(
+ TABLE_COLUMN_EXPANSION_STRATEGY,
+ List.of(EXCLUDE_DEFAULT_VIRTUAL_METADATA_COLUMNS));
+
+ tableEnv.executeSql(
+ "CREATE TABLE snapshot_probe (\n"
+ + " pk STRING,\n"
+ + " pts TIMESTAMP(3),\n"
+ + " WATERMARK FOR pts AS pts\n"
+ + ") WITH ('connector' = 'values', 'bounded' =
'false')");
+
+ // Same hidden-metadata watermark, but enable-watermark-push-down
folds it into the source;
+ // disable-lookup keeps the values source a pure scan source so the
watermark is pushed into
+ // the scan.
+ tableEnv.executeSql(
+ "CREATE TABLE snapshot_build_pushed (\n"
+ + " bk STRING,\n"
+ + " bv INT,\n"
+ + " rt TIMESTAMP_LTZ(3) METADATA VIRTUAL,\n"
+ + " WATERMARK FOR rt AS rt\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'bounded' = 'false',\n"
+ + " 'disable-lookup' = 'true',\n"
+ + " 'enable-watermark-push-down' = 'true',\n"
+ + " 'scan.watermark.emit.strategy' = 'on-event',\n"
+ + " 'readable-metadata' = 'rt:TIMESTAMP_LTZ(3)'\n"
+ + ")");
+
+ 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";
+
+ // rt is selected due to the on_time descriptor, even though it is a
hidden metadata column.
+ assertColumnNames(sql, "pk", "pts", "bk", "bv", "rt");
+ final String plan = tableEnv.explainSql(sql);
+ assertThat(plan).contains("LateralSnapshotJoin");
+ // watermarkEmitStrategy only appears on the TableSourceScan when the
watermark is folded
+ // into the source, confirming the pushed-down row-time attribute
reaches the operator.
+ assertThat(plan).contains("watermarkEmitStrategy=[on-event]");
+ }
+
@DataTypeHint("ROW<out STRING>")
public static class PassThroughPtf extends ProcessTableFunction<Row> {
@SuppressWarnings("unused")
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 722e96d0abb..3f4b636b4bd 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
@@ -77,7 +77,7 @@ public class SnapshotTableFunctionTest extends TableTestBase {
.explainSql(
"SELECT o.order_id, o.amount, r.rate "
+ "FROM Orders AS o, LATERAL SNAPSHOT("
- + "input => TABLE Rates, "
+ + "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 "
@@ -94,7 +94,7 @@ public class SnapshotTableFunctionTest extends TableTestBase {
"CREATE VIEW OrdersWithRate AS "
+ "SELECT o.order_id, o.amount, r.rate "
+ "FROM Orders AS o, LATERAL SNAPSHOT("
- + "input => TABLE Rates, "
+ + "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 "
@@ -111,7 +111,7 @@ public class SnapshotTableFunctionTest extends
TableTestBase {
.explainSql(
"SELECT o.order_id, o.amount, r.rate "
+ "FROM Orders AS o, LATERAL SNAPSHOT("
- + "input => TABLE RatesView, "
+ + "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 "
@@ -121,20 +121,21 @@ public class SnapshotTableFunctionTest extends
TableTestBase {
@Test
void testSystemArgumentsNotAllowed() {
- // SNAPSHOT disables the implicit system arguments (e.g. `on_time`).
Passing one in a
- // LATERAL context must be rejected because the argument is not part
of the function
- // signature.
+ // SNAPSHOT disables the implicit system arguments. It declares its
own `on_time` argument
+ // (so that name is accepted), but the other system arguments such as
`uid` remain
+ // unsupported and must be rejected.
assertThatThrownBy(
() ->
util.verifyRelPlan(
"SELECT o.order_id "
+ "FROM Orders AS o, LATERAL
SNAPSHOT("
+ "input => TABLE Rates, "
- + "on_time =>
DESCRIPTOR(rate_time)) AS r "
+ + "on_time =>
DESCRIPTOR(rate_time), "
+ + "uid => 'my_uid') AS r "
+ "WHERE o.currency =
r.currency"))
.satisfies(
anyCauseMatches(
- "The 'on_time' argument is not supported
because function "
+ "The 'uid' argument is not supported because
function "
+ "'SNAPSHOT' does not use system
arguments."));
}
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 323ed9beb91..4d389ecd1c5 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
@@ -91,7 +91,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoin() {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -102,7 +102,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testLeftJoin() {
util.verifyRelPlan(
"SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -113,7 +113,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithIdleTimeoutAndStateTtl() {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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, "
@@ -126,7 +126,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithNonEquiCondition() {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -137,7 +137,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithCompositeKeys() {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -148,7 +148,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithTimeAttributeInCondition() {
util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -160,7 +160,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.verifyRelPlan(
"WITH cte AS (SELECT bk, bv + 1 AS bv, bts FROM b) "
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE cte, "
+ + "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 "
@@ -171,7 +171,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithoutBuildTimeColumn() {
util.verifyRelPlan(
"SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -182,7 +182,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testLeftJoinWithoutBuildTimeColumn() {
util.verifyRelPlan(
"SELECT probe.pk, probe.pv, s.bv FROM probe LEFT JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -202,7 +202,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
+ ") WITH ('connector' = 'values', 'bounded' =
'false')");
util.verifyRelPlan(
"SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b_proctime, "
+ + "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 "
@@ -219,7 +219,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
// materialized) build time attribute as `bts`.
final String derived =
"(SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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)";
@@ -259,7 +259,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.tableEnv()
.explainSql(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b_upsert, "
+ + "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");
@@ -271,7 +271,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testNonEquiConditionCompilesEndToEnd() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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";
@@ -284,7 +284,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.tableEnv()
.explainSql(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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");
@@ -294,7 +294,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
@Test
void testInnerJoinWithDefaultCompileTimeCompilesEndToEnd() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b)
AS s "
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b,
on_time => DESCRIPTOR(bts)) AS s "
+ "ON probe.pk = s.bk";
// Compile via the table environment without verifying the plan XML
(since
// load_completed_time embeds wall-clock millis at planning).
@@ -309,7 +309,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithExplicitCompileTimeCompilesEndToEnd() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ "load_completed_condition => 'compile_time'"
+ ") AS s ON probe.pk = s.bk";
assertThat(util.tableEnv().explainSql(sql))
@@ -324,7 +324,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
//
------------------------------------------------------------------------------------------
@Test
- void testRejectBuildSideWithoutWatermark() {
+ void testRejectOnTimeWithoutWatermark() {
util.tableEnv()
.executeSql(
"CREATE TABLE b_no_wm ("
@@ -334,7 +334,29 @@ public class LateralSnapshotJoinTest extends TableTestBase
{
+ ") WITH ('connector' = 'values', 'bounded' =
'false')");
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b_no_wm, "
+ + "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";
+ assertThatThrownBy(() -> util.verifyRelPlan(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Argument 'on_time' of SNAPSHOT must reference a
rowtime attribute");
+ }
+
+ @Test
+ void testRejectOnTimeWithoutWatermarkWhenBuildTimeColumnPruned() {
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE b_no_wm ("
+ + " bk STRING,"
+ + " bv INT,"
+ + " bts TIMESTAMP(3)"
+ + ") WITH ('connector' = 'values', 'bounded' =
'false')");
+ 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 "
@@ -342,7 +364,79 @@ public class LateralSnapshotJoinTest extends TableTestBase
{
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
- "LATERAL SNAPSHOT requires a watermark on the
build-side input.");
+ "Argument 'on_time' of SNAPSHOT must reference a
rowtime attribute");
+ }
+
+ @Test
+ 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";
+ assertThatThrownBy(() -> util.verifyRelPlan(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("LATERAL SNAPSHOT requires the 'on_time'
argument");
+ }
+
+ @Test
+ 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";
+ assertThatThrownBy(() -> util.verifyRelPlan(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasStackTraceContaining(
+ "Argument 'on_time' of SNAPSHOT references column
'nonexistent'");
+ }
+
+ @Test
+ 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";
+ assertThatThrownBy(() -> util.verifyRelPlan(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasStackTraceContaining("must reference a TIMESTAMP or
TIMESTAMP_LTZ column");
+ }
+
+ @Test
+ void testRejectProctimeOnTimeColumn() {
+ // A proctime column is a TIMESTAMP_LTZ(3), so it passes the
type-strategy check, but it is
+ // a proctime (not a rowtime) attribute and cannot drive the load
phase, so the rule
+ // rejects it.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE b_proctime_wm ("
+ + " bk STRING,"
+ + " bv INT,"
+ + " bts TIMESTAMP(3),"
+ + " pt AS PROCTIME(),"
+ + " WATERMARK FOR bts AS bts"
+ + ") WITH ('connector' = 'values', 'bounded' =
'false')");
+ // Select s.pt so the proctime column is forced into the build-side
expansion and reaches
+ // the rule as a proctime (not rowtime) indicator.
+ 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";
+ assertThatThrownBy(() -> util.verifyRelPlan(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasStackTraceContaining(
+ "Argument 'on_time' of SNAPSHOT must reference a
rowtime attribute");
}
@Test
@@ -362,7 +456,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
+ ")");
final String sql =
"SELECT * FROM probe_updates JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -377,7 +471,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectMissingEqualityPredicate() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 "
@@ -392,7 +486,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectNonConstantCondition() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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))
@@ -405,7 +499,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectNonConstantLoadCompletedTime() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ "load_completed_condition => 'user_time', "
+ "load_completed_time => CURRENT_TIMESTAMP"
+ ") AS s ON probe.pk = s.bk";
@@ -419,7 +513,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectNonConstantIdleTimeout() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 => "
@@ -435,7 +529,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectNonConstantStateTtl() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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 => "
@@ -451,7 +545,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectYearMonthIntervalStateTtl() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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"
@@ -465,7 +559,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectNegativeIdleTimeout() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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"
@@ -480,7 +574,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testRejectNegativeStateTtl() {
final String sql =
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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"
@@ -499,7 +593,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinJsonPlan() {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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");
@@ -509,7 +603,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testLeftJoinJsonPlan() {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe LEFT JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b, "
+ + "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");
@@ -519,7 +613,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithIdleTimeoutAndStateTtlJsonPlan() {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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, "
@@ -531,7 +625,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithCompositeKeysJsonPlan() {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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");
@@ -541,7 +635,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testInnerJoinWithNonEquiConditionJsonPlan() {
util.verifyJsonPlan(
"INSERT INTO sink SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, "
+ + "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");
@@ -558,7 +652,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
.getConfig()
.set(ExecutionConfigOptions.IDLE_STATE_RETENTION,
Duration.ofHours(12));
- assertThat(resolveStateTtlMs("SNAPSHOT(input => TABLE b)"))
+ assertThat(resolveStateTtlMs("SNAPSHOT(input => TABLE b, on_time =>
DESCRIPTOR(bts))"))
.isEqualTo(Duration.ofHours(12).toMillis());
}
@@ -568,7 +662,9 @@ public class LateralSnapshotJoinTest extends TableTestBase {
.getConfig()
.set(ExecutionConfigOptions.IDLE_STATE_RETENTION,
Duration.ofHours(12));
- assertThat(resolveStateTtlMs("SNAPSHOT(input => TABLE b, state_ttl =>
INTERVAL '1' DAY)"))
+ assertThat(
+ resolveStateTtlMs(
+ "SNAPSHOT(input => TABLE b, on_time =>
DESCRIPTOR(bts), state_ttl => INTERVAL '1' DAY)"))
.isEqualTo(Duration.ofDays(1).toMillis());
}
@@ -576,7 +672,8 @@ public class LateralSnapshotJoinTest extends TableTestBase {
void testStateTtlDisabledWhenNeitherArgNorPipelineTtlSet() {
util.tableEnv().getConfig().set(ExecutionConfigOptions.IDLE_STATE_RETENTION,
Duration.ZERO);
- assertThat(resolveStateTtlMs("SNAPSHOT(input => TABLE b)")).isZero();
+ assertThat(resolveStateTtlMs("SNAPSHOT(input => TABLE b, on_time =>
DESCRIPTOR(bts))"))
+ .isZero();
}
/**
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 7715e6e1d41..36f32230f0f 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
@@ -110,7 +110,7 @@ public class LateralSnapshotJoinITCase extends
StreamingWithStateTestBase {
final List<Row> actual =
collect(
"SELECT probe.pk, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b, "
+ + "input => TABLE b, on_time =>
DESCRIPTOR(bts), "
+ MID_FLIP
+ ") AS s ON probe.pk = s.bk");
@@ -132,7 +132,7 @@ public class LateralSnapshotJoinITCase extends
StreamingWithStateTestBase {
final List<Row> actual =
collect(
"SELECT probe.pk, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b, "
+ + "input => TABLE b, on_time =>
DESCRIPTOR(bts), "
+ MID_FLIP
+ ") AS s ON probe.pk = s.bk");
@@ -168,7 +168,7 @@ public class LateralSnapshotJoinITCase extends
StreamingWithStateTestBase {
sortedByProbeId(
collect(
"SELECT probe.pv, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
- + "input => TABLE b, "
+ + "input => TABLE b, on_time =>
DESCRIPTOR(bts), "
+ MID_FLIP
+ ") AS s ON probe.pk = s.bk"));
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 d69cb1baf0e..8f6ec484a13 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())], rowType=[RecordType(VARCHAR(2147483647) bk,
INTEGER bv, TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) pt)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(),
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())], rowType=[RecordType(VARCHAR(2147483647) bk,
INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(),
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]])
]]>
@@ -75,7 +75,7 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk,
pv, pts, bk, bv, b
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),
_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(),
_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)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -99,7 +99,7 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk,
pv, pts, bk, bv, b
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),
_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(),
_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)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -123,7 +123,7 @@ HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), =(pv,
bv))], select=[pk, pv
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),
_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(),
_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)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -147,7 +147,7 @@ HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), >(pv,
bv))], select=[pk, pv
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),
_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(),
_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)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
]]>
@@ -173,7 +173,7 @@ Calc(select=[pk, pv, bv])
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),
_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(),
_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)])
+- 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 f75754f8a09..885d12b4b1a 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, 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_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]]>
</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),
_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'), _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)])
+- 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, 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_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]]>
</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),
_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'), _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)])
+- 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,
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_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]]>
</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),
_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'), _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)])
+- 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, 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_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]]>
</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),
_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'), _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)])
+- 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,
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_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]]>
</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),
_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'), _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)])
+- 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,
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_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]]>
</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),
_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'), _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)])
+- 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,
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_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]]>
</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),
_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'), _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)])
+- 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, 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_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]]>
</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),
_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'), _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)])
+- 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, 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_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]]>
</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),
_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'), _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)])
+- 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, 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_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]]>
</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),
_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'), _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)])
+- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+- LogicalTableScan(table=[[default_catalog, default_database, b]])
