This is an automated email from the ASF dual-hosted git repository.
fhueske pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 52ef7d53dbd [FLINK-40885][table] Support LATERAL SNAPSHOT join in the
Table API (#29368)
52ef7d53dbd is described below
commit 52ef7d53dbd3a2f8b2c64607dc253e9c97e08aef
Author: Fabian Hueske <[email protected]>
AuthorDate: Thu Oct 8 15:38:52 2026 +0200
[FLINK-40885][table] Support LATERAL SNAPSHOT join in the Table API (#29368)
Enable the processing-time LATERAL SNAPSHOT temporal join to be expressed
through
Table API via Table.joinLateral / Table.leftOuterJoinLateral with a
call("SNAPSHOT", ...) build side,
producing the same optimized plan and operator
(StreamExecLateralSnapshotJoin) as the equivalent SQL.
Also adds Table API join documentation.
Generated-by: Claude Code, Opus 4.8
---
docs/content.zh/docs/dev/table/tableApi.md | 41 +++++
docs/content/docs/dev/table/tableApi.md | 41 +++++
.../utils/CorrelatedFunctionTableFactory.java | 18 +-
.../operations/utils/JoinOperationFactory.java | 15 +-
.../planner/plan/QueryOperationConverter.java | 151 ++++++++++------
.../api/QueryOperationSqlSerializationTest.java | 4 +-
.../table/api/QueryOperationTestPrograms.java | 85 +++++++++
.../LateralSnapshotJoinSemanticTestPrograms.java | 52 ++++++
.../stream/LateralSnapshotJoinSemanticTests.java | 3 +-
.../sql/join/LateralSnapshotJoinTableApiTest.java | 201 +++++++++++++++++++++
10 files changed, 553 insertions(+), 58 deletions(-)
diff --git a/docs/content.zh/docs/dev/table/tableApi.md
b/docs/content.zh/docs/dev/table/tableApi.md
index 2ed8f2e395b..a88533fa0af 100644
--- a/docs/content.zh/docs/dev/table/tableApi.md
+++ b/docs/content.zh/docs/dev/table/tableApi.md
@@ -1152,6 +1152,47 @@ val result = orders
{{< /tab >}}
{{< /tabs >}}
+#### Lateral Snapshot Join
+
+{{< label "Batch" >}} {{< label "Streaming" >}}
+
+Enriches an append-only table with the current state of an updating table by
calling the built-in `SNAPSHOT` [process table function]({{< ref
"docs/sql/functions/built-in-functions" >}}) as the build (right) side of a
`LATERAL` join. Each probe-side row is joined with the build-side state that is
current when the row is processed. Both inner and left outer joins are
supported, and the predicate must contain at least one equality condition.
+
+See [LATERAL SNAPSHOT join]({{< ref "docs/sql/reference/queries/joins"
>}}#lateral-snapshot-join) for the full list of `SNAPSHOT` arguments and the
join semantics.
+
+{{< tabs "snapshotjoin" >}}
+{{< tab "Java" >}}
+```java
+Table orders = tableEnv.from("Orders");
+Table currencyRates = tableEnv.from("CurrencyRates");
+
+// inner join: enrich every order with the current conversion rate
+Table result = orders
+ .joinLateral(
+ call(
+ "SNAPSHOT",
+ currencyRates.asArgument("input"),
+ descriptor("update_time").asArgument("on_time")),
+ $("currency").isEqual($("r_currency")));
+
+// left outer join: unmatched orders are preserved and padded with nulls
+Table leftResult = orders
+ .leftOuterJoinLateral(
+ call(
+ "SNAPSHOT",
+ currencyRates.asArgument("input"),
+ descriptor("update_time").asArgument("on_time")),
+ $("currency").isEqual($("r_currency")));
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+Currently not supported in Scala Table API.
+{{< /tab >}}
+{{< tab "Python" >}}
+Currently not supported in Python Table API.
+{{< /tab >}}
+{{< /tabs >}}
+
{{< top >}}
### Set 操作
diff --git a/docs/content/docs/dev/table/tableApi.md
b/docs/content/docs/dev/table/tableApi.md
index 118befaaadd..f50b150c711 100644
--- a/docs/content/docs/dev/table/tableApi.md
+++ b/docs/content/docs/dev/table/tableApi.md
@@ -1151,6 +1151,47 @@ Currently not supported in Python Table API.
{{< /tab >}}
{{< /tabs >}}
+#### Lateral Snapshot Join
+
+{{< label "Batch" >}} {{< label "Streaming" >}}
+
+Enriches an append-only table with the current state of an updating table by
calling the built-in `SNAPSHOT` [process table function]({{< ref
"docs/sql/functions/built-in-functions" >}}) as the build (right) side of a
`LATERAL` join. Each probe-side row is joined with the build-side state that is
current when the row is processed. Both inner and left outer joins are
supported, and the predicate must contain at least one equality condition.
+
+See [LATERAL SNAPSHOT join]({{< ref "docs/sql/reference/queries/joins"
>}}#lateral-snapshot-join) for the full list of `SNAPSHOT` arguments and the
join semantics.
+
+{{< tabs "snapshotjoin" >}}
+{{< tab "Java" >}}
+```java
+Table orders = tableEnv.from("Orders");
+Table currencyRates = tableEnv.from("CurrencyRates");
+
+// inner join: enrich every order with the current conversion rate
+Table result = orders
+ .joinLateral(
+ call(
+ "SNAPSHOT",
+ currencyRates.asArgument("input"),
+ descriptor("update_time").asArgument("on_time")),
+ $("currency").isEqual($("r_currency")));
+
+// left outer join: unmatched orders are preserved and padded with nulls
+Table leftResult = orders
+ .leftOuterJoinLateral(
+ call(
+ "SNAPSHOT",
+ currencyRates.asArgument("input"),
+ descriptor("update_time").asArgument("on_time")),
+ $("currency").isEqual($("r_currency")));
+```
+{{< /tab >}}
+{{< tab "Scala" >}}
+Currently not supported in Scala Table API.
+{{< /tab >}}
+{{< tab "Python" >}}
+Currently not supported in Python Table API.
+{{< /tab >}}
+{{< /tabs >}}
+
{{< top >}}
### Set Operations
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/CorrelatedFunctionTableFactory.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/CorrelatedFunctionTableFactory.java
index 0a477a3d0d8..7d957d657f6 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/CorrelatedFunctionTableFactory.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/CorrelatedFunctionTableFactory.java
@@ -27,6 +27,7 @@ import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.ExpressionUtils;
import org.apache.flink.table.expressions.ResolvedExpression;
import
org.apache.flink.table.expressions.utils.ResolvedExpressionDefaultVisitor;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.FunctionKind;
import org.apache.flink.table.operations.CorrelatedFunctionQueryOperation;
@@ -39,8 +40,10 @@ import java.util.Collections;
import java.util.List;
import static java.util.stream.Collectors.toList;
+import static org.apache.flink.table.expressions.ApiExpressionUtils.isFunction;
import static
org.apache.flink.table.expressions.ApiExpressionUtils.isFunctionOfKind;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.AS;
+import static
org.apache.flink.table.functions.BuiltInFunctionDefinitions.SNAPSHOT;
/** Utility class for creating a valid {@link
CorrelatedFunctionQueryOperation} operation. */
@Internal
@@ -58,6 +61,14 @@ final class CorrelatedFunctionTableFactory {
return callExpr.accept(calculatedTableCreator);
}
+ /**
+ * Whether the given function is {@code SNAPSHOT}, currently the only
{@link
+ * FunctionKind#PROCESS_TABLE} function that is valid as the build side of
a lateral join.
+ */
+ static boolean isSnapshot(FunctionDefinition definition) {
+ return definition.equals(BuiltInFunctionDefinitions.SNAPSHOT);
+ }
+
private static class FunctionTableCallVisitor
extends
ResolvedExpressionDefaultVisitor<CorrelatedFunctionQueryOperation> {
private final List<String> leftTableFieldNames;
@@ -91,11 +102,14 @@ final class CorrelatedFunctionTableFactory {
+ alias)))
.collect(toList());
- if (!isFunctionOfKind(children.get(0), FunctionKind.TABLE)) {
+ // SNAPSHOT is PROCESS_TABLE rather than TABLE but is valid in a
LATERAL join.
+ Expression firstChild = children.get(0);
+ if (!isFunctionOfKind(firstChild, FunctionKind.TABLE)
+ && !isFunction(firstChild, SNAPSHOT)) {
throw fail();
}
- CallExpression tableCall = (CallExpression) children.get(0);
+ CallExpression tableCall = (CallExpression) firstChild;
return createFunctionCall(tableCall, aliases,
tableCall.getResolvedChildren());
}
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/JoinOperationFactory.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/JoinOperationFactory.java
index 5b5cd81d0c4..5ab89c7bda2 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/JoinOperationFactory.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/JoinOperationFactory.java
@@ -80,8 +80,21 @@ final class JoinOperationFactory {
JoinType joinType,
ResolvedExpression condition,
boolean correlated) {
- boolean alwaysTrue = ExpressionUtils.extractValue(condition,
Boolean.class).orElse(false);
+ // Unlike other lateral joins with a table function (which only allow
an empty/literal-true
+ // condition), a lateral SNAPSHOT join supports an ON predicate.
+ final boolean isLateralSnapshot =
+ correlated
+ && right instanceof CorrelatedFunctionQueryOperation
+ && CorrelatedFunctionTableFactory.isSnapshot(
+ ((CorrelatedFunctionQueryOperation) right)
+ .getResolvedFunction()
+ .getDefinition());
+ if (isLateralSnapshot) {
+ return;
+ }
+ final boolean alwaysTrue =
+ ExpressionUtils.extractValue(condition,
Boolean.class).orElse(false);
if (alwaysTrue) {
return;
}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
index 468430dbc7b..10531fd28e8 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
@@ -93,6 +93,7 @@ import
org.apache.flink.table.planner.plan.schema.DataStreamTable$;
import org.apache.flink.table.planner.plan.schema.LegacyTableSourceTable;
import org.apache.flink.table.planner.plan.schema.TypedFlinkTableFunction;
import org.apache.flink.table.planner.plan.stats.FlinkStatistic;
+import org.apache.flink.table.planner.plan.utils.LateralSnapshotJoinUtil;
import org.apache.flink.table.planner.sources.TableSourceUtil;
import org.apache.flink.table.planner.utils.ShortcutUtils;
import org.apache.flink.table.runtime.groupwindow.NamedWindowProperty;
@@ -284,7 +285,17 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
@Override
public RelNode visit(JoinQueryOperation join) {
final Set<CorrelationId> corSet;
- if (join.isCorrelated()) {
+ // Check if this is a LATERAL SNAPSHOT join. The SNAPSHOT function
never references the
+ // outer row. Emit a plain LogicalJoin instead of a Correlate, so
that
+ // LogicalJoinToLateralSnapshotJoinRule matches.
+ final QueryOperation right = join.getChildren().get(1);
+ final boolean isLateralSnapshotJoin =
+ right instanceof CorrelatedFunctionQueryOperation
+ && LateralSnapshotJoinUtil.isSnapshotFunction(
+ ((CorrelatedFunctionQueryOperation) right)
+ .getResolvedFunction()
+ .getDefinition());
+ if (join.isCorrelated() && !isLateralSnapshotJoin) {
corSet =
Collections.singleton(relBuilder.peek().getCluster().createCorrel());
} else {
corSet = Collections.emptySet();
@@ -352,44 +363,11 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
.map(
resolvedArg -> {
if (resolvedArg instanceof
TableReferenceExpression) {
- final TableReferenceExpression
tableRef =
- (TableReferenceExpression)
resolvedArg;
- final LogicalType tableArgType =
-
tableRef.getOutputDataType().getLogicalType();
- final RelDataType rowType =
-
typeFactory.buildRelNodeRowType(
- (RowType)
tableArgType);
- final int[] partitionKeys;
- final int[] orderKeys;
- final SortOrder[] sortOrders;
- if (tableRef.getQueryOperation()
- instanceof
PartitionQueryOperation) {
- final PartitionQueryOperation
partitionOperation =
-
(PartitionQueryOperation)
-
tableRef.getQueryOperation();
- partitionKeys =
-
partitionOperation.getPartitionKeys();
- orderKeys =
partitionOperation.getOrderKeys();
- final SortDirection[]
directions =
-
partitionOperation.getOrderDirections();
- sortOrders = new
SortOrder[directions.length];
- for (int i = 0; i <
directions.length; i++) {
- sortOrders[i] =
-
SortOrder.fromSortDirection(
-
directions[i]);
- }
- } else {
- partitionKeys = new int[0];
- orderKeys = new int[0];
- sortOrders = new SortOrder[0];
- }
final RexTableArgCall tableArgCall
=
- new RexTableArgCall(
- rowType,
+ buildTableArgCall(
+
(TableReferenceExpression) resolvedArg,
inputStack.size(),
- partitionKeys,
- orderKeys,
- sortOrders);
+ typeFactory);
inputStack.add(relBuilder.build());
return tableArgCall;
}
@@ -400,22 +378,9 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
// relBuilder.build() works in LIFO fashion, this restores the
original input order
Collections.reverse(inputStack);
- final BridgingSqlFunction sqlFunction =
- BridgingSqlFunction.of(relBuilder.getCluster(),
contextFunction);
-
- final RexNode call =
- relBuilder
- .getRexBuilder()
- .makeCall(outputRelDataType, sqlFunction,
rexNodeArgs);
- final RelNode functionScan =
- LogicalTableFunctionScan.create(
- relBuilder.getCluster(),
- inputStack,
- call,
- null,
- outputRelDataType,
- Collections.emptySet());
- relBuilder.push(functionScan);
+ relBuilder.push(
+ createTableFunctionScan(
+ contextFunction, rexNodeArgs, inputStack,
outputRelDataType));
return relBuilder.build();
}
@@ -430,6 +395,11 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
public RelNode visit(CorrelatedFunctionQueryOperation
correlatedFunction) {
final ContextResolvedFunction contextFunction =
correlatedFunction.getResolvedFunction();
+ // SNAPSHOT is currently the only PTF valid in a lateral join and
hence the only
+ // function accepting table args.
+ if
(LateralSnapshotJoinUtil.isSnapshotFunction(contextFunction.getDefinition())) {
+ return
convertCorrelatedFunctionWithTableArgs(correlatedFunction);
+ }
final List<RexNode> parameters =
convertToRexNodes(correlatedFunction.getArguments());
final FunctionDefinition functionDefinition =
contextFunction.getDefinition();
@@ -455,6 +425,81 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
return relBuilder.build();
}
+ private RelNode convertCorrelatedFunctionWithTableArgs(
+ CorrelatedFunctionQueryOperation correlatedFunction) {
+ final FlinkTypeFactory typeFactory =
ShortcutUtils.unwrapTypeFactory(relBuilder);
+ final List<RelNode> inputs = new ArrayList<>();
+ final List<RexNode> args = new ArrayList<>();
+ for (ResolvedExpression arg : correlatedFunction.getArguments()) {
+ if (arg instanceof TableReferenceExpression) {
+ final TableReferenceExpression tableRef =
(TableReferenceExpression) arg;
+ args.add(buildTableArgCall(tableRef, inputs.size(),
typeFactory));
+
inputs.add(tableRef.getQueryOperation().accept(QueryOperationConverter.this));
+ } else {
+ args.add(convertExprToRexNode(arg));
+ }
+ }
+ // Derive the output row type like the FROM-clause PTF path
+ // (FunctionQueryOperation#getOutputDataType).
+ final RelDataType outputType =
+ typeFactory.buildRelNodeRowType(
+ (RowType)
+
DataTypeUtils.fromResolvedSchemaPreservingTimeAttributes(
+
correlatedFunction.getResolvedSchema())
+ .getLogicalType());
+ return createTableFunctionScan(
+ correlatedFunction.getResolvedFunction(), args, inputs,
outputType);
+ }
+
+ /**
+ * Converts a {@link TableReferenceExpression} table argument into a
{@link
+ * RexTableArgCall}, carrying PARTITION BY / ORDER BY keys when the
referenced operation is
+ * a {@link PartitionQueryOperation}.
+ */
+ private RexTableArgCall buildTableArgCall(
+ TableReferenceExpression tableRef, int inputIndex,
FlinkTypeFactory typeFactory) {
+ final RelDataType rowType =
+ typeFactory.buildRelNodeRowType(
+ (RowType)
tableRef.getOutputDataType().getLogicalType());
+ final int[] partitionKeys;
+ final int[] orderKeys;
+ final SortOrder[] sortOrders;
+ if (tableRef.getQueryOperation() instanceof
PartitionQueryOperation) {
+ final PartitionQueryOperation partitionOperation =
+ (PartitionQueryOperation) tableRef.getQueryOperation();
+ partitionKeys = partitionOperation.getPartitionKeys();
+ orderKeys = partitionOperation.getOrderKeys();
+ final SortDirection[] directions =
partitionOperation.getOrderDirections();
+ sortOrders = new SortOrder[directions.length];
+ for (int i = 0; i < directions.length; i++) {
+ sortOrders[i] = SortOrder.fromSortDirection(directions[i]);
+ }
+ } else {
+ partitionKeys = new int[0];
+ orderKeys = new int[0];
+ sortOrders = new SortOrder[0];
+ }
+ return new RexTableArgCall(rowType, inputIndex, partitionKeys,
orderKeys, sortOrders);
+ }
+
+ /** Builds a {@link LogicalTableFunctionScan} for a function call with
table arguments. */
+ private RelNode createTableFunctionScan(
+ ContextResolvedFunction function,
+ List<RexNode> args,
+ List<RelNode> inputs,
+ RelDataType outputType) {
+ final BridgingSqlFunction sqlFunction =
+ BridgingSqlFunction.of(relBuilder.getCluster(), function);
+ final RexNode call =
relBuilder.getRexBuilder().makeCall(outputType, sqlFunction, args);
+ return LogicalTableFunctionScan.create(
+ relBuilder.getCluster(),
+ inputs,
+ call,
+ null,
+ outputType,
+ Collections.emptySet());
+ }
+
private RelNode convertLegacyTableFunction(
CorrelatedFunctionQueryOperation correlatedFunction,
TableFunctionDefinition functionDefinition,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
index 7b5795fe9a7..7924486711a 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
@@ -91,7 +91,9 @@ public class QueryOperationSqlSerializationTest implements
TableTestProgramRunne
QueryOperationTestPrograms.INLINE_FUNCTION_SERIALIZATION,
QueryOperationTestPrograms.ML_PREDICT_MODEL_API,
QueryOperationTestPrograms.ASYNC_ML_PREDICT_MODEL_API,
-
QueryOperationTestPrograms.ASYNC_ML_PREDICT_TABLE_API_MAP_EXPRESSION_CONFIG);
+
QueryOperationTestPrograms.ASYNC_ML_PREDICT_TABLE_API_MAP_EXPRESSION_CONFIG,
+ QueryOperationTestPrograms.SNAPSHOT_INNER_JOIN,
+ QueryOperationTestPrograms.SNAPSHOT_LEFT_JOIN);
}
@ParameterizedTest
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
index 6c11367955f..fb41ab9c8ac 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
@@ -40,6 +40,7 @@ import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
import java.math.BigDecimal;
+import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.LocalTime;
@@ -1878,4 +1879,88 @@ public class QueryOperationTestPrograms {
lit(1).asArgument("i")),
"sink")
.build();
+
+ private static final String SNAPSHOT_PROBE_DDL =
+ "CREATE TABLE probe (pk STRING, pv INT, pts TIMESTAMP(3),
WATERMARK FOR pts AS pts)"
+ + " WITH ('connector' = 'values', 'bounded' = 'false')";
+
+ private static final String SNAPSHOT_BUILD_DDL =
+ "CREATE TABLE b (bk STRING, bv INT, bts TIMESTAMP(3), WATERMARK
FOR bts AS bts)"
+ + " WITH ('connector' = 'values', 'bounded' = 'false',"
+ + " 'changelog-mode' = 'I,UB,UA,D')";
+
+ private static ApiExpression snapshotLoadCompletedTime() {
+ return
lit(LocalDateTime.parse("2026-07-01T00:00:00")).cast(DataTypes.TIMESTAMP_LTZ(3));
+ }
+
+ static final TableTestProgram SNAPSHOT_INNER_JOIN =
+ TableTestProgram.of(
+ "lateral-snapshot-inner-join-serialization",
+ "verifies SQL serialization of a LATERAL SNAPSHOT
inner join")
+ .setupSql(SNAPSHOT_PROBE_DDL)
+ .setupSql(SNAPSHOT_BUILD_DDL)
+ .runTableApi(
+ env ->
+ env.from("probe")
+ .joinLateral(
+ call(
+ "SNAPSHOT",
+
env.from("b").asArgument("input"),
+ descriptor("bts")
+
.asArgument("on_time")),
+ $("pk").isEqual($("bk")))
+ .select($("pk"), $("pv"), $("bk"),
$("bv")),
+ "sink")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`pk`, `$$T_PROJECT`.`pv`,
`$$T_PROJECT`.`bk`, `$$T_PROJECT`.`bv` FROM (\n"
+ + " SELECT `$$T1_JOIN`.`pk`,
`$$T1_JOIN`.`pv`, `$$T1_JOIN`.`pts`, `$$T_LAT`.`bk`, `$$T_LAT`.`bv`,
`$$T_LAT`.`bts` FROM (\n"
+ + " SELECT `$$T_SOURCE`.`pk`,
`$$T_SOURCE`.`pv`, `$$T_SOURCE`.`pts` FROM
`default_catalog`.`default_database`.`probe` $$T_SOURCE\n"
+ + " ) $$T1_JOIN INNER JOIN \n"
+ + " LATERAL TABLE(SNAPSHOT((\n"
+ + " SELECT `$$T_SOURCE`.`bk`,
`$$T_SOURCE`.`bv`, `$$T_SOURCE`.`bts` FROM
`default_catalog`.`default_database`.`b` $$T_SOURCE\n"
+ + " ), DESCRIPTOR(`bts`), DEFAULT,
DEFAULT, DEFAULT)) $$T_LAT(`bk`, `bv`, `bts`) ON `$$T1_JOIN`.`pk` =
`$$T_LAT`.`bk`\n"
+ + ") $$T_PROJECT")
+ .build();
+
+ static final TableTestProgram SNAPSHOT_LEFT_JOIN =
+ TableTestProgram.of(
+ "lateral-snapshot-left-join-serialization",
+ "verifies SQL serialization of a LATERAL SNAPSHOT
left outer join with a "
+ + "filtered build side and all optional
arguments passed as "
+ + "scrambled named arguments")
+ .setupSql(SNAPSHOT_PROBE_DDL)
+ .setupSql(SNAPSHOT_BUILD_DDL)
+ .runTableApi(
+ env ->
+ env.from("probe")
+ .leftOuterJoinLateral(
+ call(
+ "SNAPSHOT",
+
snapshotLoadCompletedTime()
+
.asArgument(
+
"load_completed_time"),
+
lit(Duration.ofDays(1))
+
.asArgument("state_ttl"),
+
descriptor("bts").asArgument("on_time"),
+
lit(Duration.ofSeconds(10))
+
.asArgument(
+
"load_completed_idle_timeout"),
+ env.from("b")
+
.filter($("bv").isGreater(10))
+
.asArgument("input")),
+ $("pk").isEqual($("bk")))
+ .select($("pk"), $("pv"), $("bk"),
$("bv")),
+ "sink")
+ .runSql(
+ "SELECT `$$T_PROJECT`.`pk`, `$$T_PROJECT`.`pv`,
`$$T_PROJECT`.`bk`, `$$T_PROJECT`.`bv` FROM (\n"
+ + " SELECT `$$T1_JOIN`.`pk`,
`$$T1_JOIN`.`pv`, `$$T1_JOIN`.`pts`, `$$T_LAT`.`bk`, `$$T_LAT`.`bv`,
`$$T_LAT`.`bts` FROM (\n"
+ + " SELECT `$$T_SOURCE`.`pk`,
`$$T_SOURCE`.`pv`, `$$T_SOURCE`.`pts` FROM
`default_catalog`.`default_database`.`probe` $$T_SOURCE\n"
+ + " ) $$T1_JOIN LEFT OUTER JOIN \n"
+ + " LATERAL TABLE(SNAPSHOT((\n"
+ + " SELECT `$$T_FILTER`.`bk`,
`$$T_FILTER`.`bv`, `$$T_FILTER`.`bts` FROM (\n"
+ + " SELECT
`$$T_SOURCE`.`bk`, `$$T_SOURCE`.`bv`, `$$T_SOURCE`.`bts` FROM
`default_catalog`.`default_database`.`b` $$T_SOURCE\n"
+ + " ) $$T_FILTER WHERE
`$$T_FILTER`.`bv` > 10\n"
+ + " ), DESCRIPTOR(`bts`),
CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP(3) WITH LOCAL TIME ZONE),
INTERVAL '0 00:00:10' DAY(1) TO SECOND(0), INTERVAL '1 00:00:00' DAY(1) TO
SECOND(0))) $$T_LAT(`bk`, `bv`, `bts`) ON `$$T1_JOIN`.`pk` = `$$T_LAT`.`bk`\n"
+ + ") $$T_PROJECT")
+ .build();
}
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 275323610ec..0d951924c87 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
@@ -18,6 +18,8 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
+import org.apache.flink.table.api.ApiExpression;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.test.program.SinkTestStep;
import org.apache.flink.table.test.program.SourceTestStep;
import org.apache.flink.table.test.program.TableTestProgram;
@@ -33,6 +35,11 @@ import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+import static org.apache.flink.table.api.Expressions.descriptor;
+import static org.apache.flink.table.api.Expressions.lit;
+
/**
* Deterministic result {@link TableTestProgram} definitions for the {@code
LATERAL SNAPSHOT}
* processing-time temporal join ({@link StreamExecLateralSnapshotJoin}).
@@ -336,6 +343,42 @@ public class LateralSnapshotJoinSemanticTestPrograms {
.runSql(innerJoin("probe.pv, s.bv", MID_FLIP, "probe.pk =
s.bk"))
.build();
+ //
------------------------------------------------------------------------------------------
+ // Table API
+ //
+ // The Table API is a thin wrapper that converges on the same
plan/operator as SQL (asserted by
+ // LateralSnapshotJoinTableApiTest), so a single execution smoke is enough
here.
+ //
------------------------------------------------------------------------------------------
+
+ public static final TableTestProgram INNER_JOIN_TABLE_API =
+ TableTestProgram.of(
+ "lateral-snapshot-inner-join-table-api",
+ "LATERAL SNAPSHOT inner join expressed through the
Table API")
+ .setupTableSource(throttledProbe(defaultProbe(), 40L))
+
.setupTableSource(appendBuild(withFlipTrigger(defaultBuild())))
+ .setupTableSink(
+ keyValueSink()
+ .consumedValues(
+ "+I[a, 100, a, 10]",
+ "+I[a, 100, a, 11]",
+ "+I[b, 200, b, 20]")
+ .build())
+ .runTableApi(
+ env ->
+ env.from("probe")
+ .joinLateral(
+ call(
+ "SNAPSHOT",
+
env.from("b").asArgument("input"),
+
descriptor("bts").asArgument("on_time"),
+
loadCompletedTime(FLIP_TRIGGER_TS)
+
.asArgument(
+
"load_completed_time")),
+ $("pk").isEqual($("bk")))
+ .select($("pk"), $("pv"), $("bk"),
$("bv")),
+ "sink")
+ .build();
+
//
------------------------------------------------------------------------------------------
// Helpers
//
------------------------------------------------------------------------------------------
@@ -442,4 +485,13 @@ public class LateralSnapshotJoinSemanticTestPrograms {
private static LocalDateTime ts(String time) {
return LocalDateTime.parse("2020-01-01T" + time);
}
+
+ /**
+ * Builds the {@code load_completed_time} argument as a Table API
expression equivalent to the
+ * SQL {@code CAST(TIMESTAMP '2020-01-01 <time>' AS TIMESTAMP_LTZ(3))}
used by the SQL programs
+ * for the same flip timestamp.
+ */
+ private static ApiExpression loadCompletedTime(String time) {
+ return lit(ts(time)).cast(DataTypes.TIMESTAMP_LTZ(3));
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
index 6536f34a591..5ab4917eaea 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
@@ -38,6 +38,7 @@ public class LateralSnapshotJoinSemanticTests extends
SemanticTestBase {
LateralSnapshotJoinSemanticTestPrograms.FLIP_AT_END,
LateralSnapshotJoinSemanticTestPrograms.DEFAULT_COMPILE_TIME,
LateralSnapshotJoinSemanticTestPrograms.LIVE_JOIN,
- LateralSnapshotJoinSemanticTestPrograms.BUFFERED_THEN_DRAINED);
+ LateralSnapshotJoinSemanticTestPrograms.BUFFERED_THEN_DRAINED,
+ LateralSnapshotJoinSemanticTestPrograms.INNER_JOIN_TABLE_API);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTableApiTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTableApiTest.java
new file mode 100644
index 00000000000..544e21bf7e8
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTableApiTest.java
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.stream.sql.join;
+
+import org.apache.flink.table.api.ApiExpression;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.TableTestBase;
+import org.apache.flink.table.planner.utils.TableTestUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+import static org.apache.flink.table.api.Expressions.descriptor;
+import static org.apache.flink.table.api.Expressions.lit;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies that the Table API surface of the {@code LATERAL SNAPSHOT} join
produces the exact same
+ * optimized physical plan as the equivalent SQL.
+ *
+ * <p>This is the primary coverage for the Table API surface: it converges on
the same {@code
+ * LateralSnapshotJoin} {@code RelNode} / {@code
StreamExecLateralSnapshotJoin} operator as SQL,
+ * whose execution semantics are already fully covered by the SQL semantic
tests ({@code
+ * LateralSnapshotJoinSemanticTests}) and {@code LateralSnapshotJoinITCase}.
{@code
+ * LateralSnapshotJoinSemanticTestPrograms#INNER_JOIN_TABLE_API} keeps a
single execution smoke on
+ * top of this.
+ */
+public class LateralSnapshotJoinTableApiTest extends TableTestBase {
+
+ private static final String LOAD_COMPLETED_TIME_SQL =
+ "CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))";
+
+ private static final ApiExpression LOAD_COMPLETED_TIME_TABLE =
+
lit(LocalDateTime.parse("2026-07-01T00:00:00")).cast(DataTypes.TIMESTAMP_LTZ(3));
+
+ private TableTestUtil util;
+
+ @BeforeEach
+ void setup() {
+ final TableConfig config = TableConfig.getDefault();
+ config.setLocalTimeZone(ZoneId.of("UTC"));
+ util = streamTestUtil(config);
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE probe ("
+ + " pk STRING,"
+ + " pv INT,"
+ + " pts TIMESTAMP(3),"
+ + " WATERMARK FOR pts AS pts"
+ + ") WITH ('connector' = 'values', 'bounded' =
'false')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE b ("
+ + " bk STRING,"
+ + " bv INT,"
+ + " bts TIMESTAMP(3),"
+ + " WATERMARK FOR bts AS bts"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'bounded' = 'false',"
+ + " 'changelog-mode' = 'I,UB,UA,D'"
+ + ")");
+ }
+
+ //
------------------------------------------------------------------------------------------
+ // Plan equivalence: the Table API surface should produce the same
optimized plan as SQL.
+ //
------------------------------------------------------------------------------------------
+
+ @Test
+ void testInnerJoinPlanParity() {
+ final String sql =
+ "SELECT probe.pk, probe.pv, s.bk, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => "
+ + LOAD_COMPLETED_TIME_SQL
+ + ") AS s "
+ + "ON probe.pk = s.bk";
+
+ final Table apiResult =
+ util.tableEnv()
+ .from("probe")
+ .joinLateral(
+ call(
+ "SNAPSHOT",
+
util.tableEnv().from("b").asArgument("input"),
+
descriptor("bts").asArgument("on_time"),
+ LOAD_COMPLETED_TIME_TABLE.asArgument(
+ "load_completed_time")),
+ $("pk").isEqual($("bk")))
+ .select($("pk"), $("pv"), $("bk"), $("bv"));
+
+ assertSameOptimizedPhysicalPlan(sql, apiResult);
+ }
+
+ @Test
+ void testLeftOuterJoinPlanParity() {
+ final String sql =
+ "SELECT probe.pk, probe.pv, s.bk, s.bv FROM probe LEFT JOIN
LATERAL SNAPSHOT("
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => "
+ + LOAD_COMPLETED_TIME_SQL
+ + ") AS s "
+ + "ON probe.pk = s.bk";
+
+ final Table apiResult =
+ util.tableEnv()
+ .from("probe")
+ .leftOuterJoinLateral(
+ call(
+ "SNAPSHOT",
+
util.tableEnv().from("b").asArgument("input"),
+
descriptor("bts").asArgument("on_time"),
+ LOAD_COMPLETED_TIME_TABLE.asArgument(
+ "load_completed_time")),
+ $("pk").isEqual($("bk")))
+ .select($("pk"), $("pv"), $("bk"), $("bv"));
+
+ assertSameOptimizedPhysicalPlan(sql, apiResult);
+ }
+
+ @Test
+ void testAliasedColumnsPlanParity() {
+ // No narrowing projection on top: SELECT * keeps the aliased build
columns, so this
+ // asserts both plan parity and that the Table API and SQL agree on
the (aliased) schema.
+ final String sql =
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => "
+ + LOAD_COMPLETED_TIME_SQL
+ + ") AS s(k, v, t) "
+ + "ON probe.pk = s.k";
+
+ final Table apiResult =
+ util.tableEnv()
+ .from("probe")
+ .joinLateral(
+ call(
+ "SNAPSHOT",
+
util.tableEnv().from("b").asArgument("input"),
+
descriptor("bts").asArgument("on_time"),
+
LOAD_COMPLETED_TIME_TABLE.asArgument(
+ "load_completed_time"))
+ .as("k", "v", "t"),
+ $("pk").isEqual($("k")));
+
+ assertSameOptimizedPhysicalPlan(sql, apiResult);
+ assertThat(apiResult.getResolvedSchema())
+ .isEqualTo(util.tableEnv().sqlQuery(sql).getResolvedSchema());
+ }
+
+ //
------------------------------------------------------------------------------------------
+ // Helpers
+ //
------------------------------------------------------------------------------------------
+
+ private void assertSameOptimizedPhysicalPlan(String sql, Table apiResult) {
+ final String sqlPlan =
extractOptimizedPhysicalPlan(util.tableEnv().explainSql(sql));
+ final String apiPlan =
extractOptimizedPhysicalPlan(apiResult.explain());
+ assertThat(apiPlan).isEqualTo(sqlPlan);
+ }
+
+ /** Extracts the full "Optimized Physical Plan" from an {@code EXPLAIN}
output. */
+ private static String extractOptimizedPhysicalPlan(String plan) {
+ final String startMarker = "== Optimized Physical Plan ==";
+ final int start = plan.indexOf(startMarker);
+ if (start < 0) {
+ throw new AssertionError(
+ "No 'Optimized Physical Plan' section found in plan:\n" +
plan);
+ }
+ final int contentStart = start + startMarker.length();
+ final int end = plan.indexOf("== Optimized Execution Plan ==",
contentStart);
+ final String section =
+ end >= 0 ? plan.substring(contentStart, end) :
plan.substring(contentStart);
+ return
section.lines().map(String::stripTrailing).collect(Collectors.joining("\n")).strip();
+ }
+}
