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 a24de2e8589 [FLINK-40498][tests] Adjust LateralSnapshotJoin tests for
optional TABLE keyword (#29035)
a24de2e8589 is described below
commit a24de2e8589b74b2379fd270a78a531304651f7d
Author: Fabian Hueske <[email protected]>
AuthorDate: Fri Aug 28 16:32:21 2026 +0200
[FLINK-40498][tests] Adjust LateralSnapshotJoin tests for optional TABLE
keyword (#29035)
Drop the now-optional TABLE keyword from the LATERAL table-function calls in
the LATERAL SNAPSHOT test suite, using the "LATERAL SNAPSHOT(...)" form. One
plan test (testInnerJoin) and one semantic test program (INNER_JOIN) keep
the
explicit "LATERAL TABLE(SNAPSHOT(...))" form to retain coverage of that
syntax.
Co-Generated: Claude Opus 4.8 (1M context)
---
.../batch/sql/join/LateralSnapshotJoinTest.java | 30 ++---
.../batch/LateralSnapshotJoinTestPrograms.java | 10 +-
.../LateralSnapshotJoinSemanticTestPrograms.java | 22 ++--
.../stream/LateralSnapshotJoinTestPrograms.java | 4 +-
.../plan/stream/sql/SnapshotTableFunctionTest.java | 16 +--
.../stream/sql/join/LateralSnapshotJoinTest.java | 122 ++++++++++-----------
.../stream/sql/join/LateralSnapshotJoinITCase.java | 12 +-
.../batch/sql/join/LateralSnapshotJoinTest.xml | 14 +--
.../stream/sql/join/LateralSnapshotJoinTest.xml | 18 +--
9 files changed, 127 insertions(+), 121 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
index 50416c1560f..214e84c880b 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
@@ -71,55 +71,55 @@ public class LateralSnapshotJoinTest extends TableTestBase {
@Test
void testInnerJoin() {
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@Test
void testLeftJoin() {
util.verifyRelPlan(
- "SELECT * FROM probe LEFT JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@Test
void testInnerJoinWithCompositeKeys() {
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk AND probe.pv = s.bv");
}
@Test
void testInnerJoinWithNonEquiCondition() {
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk AND probe.pv > s.bv");
}
@Test
void testInnerJoinWithoutBuildTimeColumn() {
util.verifyRelPlan(
- "SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@@ -135,8 +135,8 @@ public class LateralSnapshotJoinTest extends TableTestBase {
+ " pt AS PROCTIME()"
+ ") WITH ('connector' = 'values', 'bounded' =
'true')");
util.verifyRelPlan(
- "SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
- + "input => TABLE b_proctime)) AS s "
+ "SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL
SNAPSHOT("
+ + "input => TABLE b_proctime) AS s "
+ "ON probe.pk = s.bk");
}
@@ -152,7 +152,7 @@ public class LateralSnapshotJoinTest extends TableTestBase {
+ " bts TIMESTAMP(3)"
+ ") WITH ('connector' = 'values', 'bounded' =
'true')");
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(input =>
TABLE b_no_wm)) AS s "
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE
b_no_wm) AS s "
+ "ON probe.pk = s.bk");
}
@@ -163,11 +163,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectMissingEqualityPredicate() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pv > s.bv";
assertThatThrownBy(() -> util.verifyExecPlan(sql))
.isInstanceOf(ValidationException.class)
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
index 8b8be211952..9f5d53c8720 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
@@ -62,7 +62,7 @@ public class LateralSnapshotJoinTestPrograms {
.build())
.runSql(
"INSERT INTO sink SELECT pk, pv, bk, bv FROM probe
JOIN LATERAL "
- + "TABLE(SNAPSHOT(input => TABLE b)) AS s
ON probe.pk = s.bk")
+ + "SNAPSHOT(input => TABLE b) AS s ON
probe.pk = s.bk")
.build();
public static final TableTestProgram LEFT_JOIN =
@@ -82,7 +82,7 @@ public class LateralSnapshotJoinTestPrograms {
.build())
.runSql(
"INSERT INTO sink SELECT pk, pv, bk, bv FROM probe
LEFT JOIN LATERAL "
- + "TABLE(SNAPSHOT(input => TABLE b)) AS s
ON probe.pk = s.bk")
+ + "SNAPSHOT(input => TABLE b) AS s ON
probe.pk = s.bk")
.build();
public static final TableTestProgram INNER_JOIN_WITH_NON_EQUI_CONDITION =
@@ -99,7 +99,7 @@ public class LateralSnapshotJoinTestPrograms {
.build())
.runSql(
"INSERT INTO sink SELECT pk, pv, bk, bv FROM probe
JOIN LATERAL "
- + "TABLE(SNAPSHOT(input => TABLE b)) AS s "
+ + "SNAPSHOT(input => TABLE b) AS s "
+ "ON probe.pk = s.bk AND s.bv > 10")
.build();
@@ -119,11 +119,11 @@ public class LateralSnapshotJoinTestPrograms {
.build())
.runSql(
"INSERT INTO sink SELECT pk, pv, bk, bv FROM probe
JOIN LATERAL "
- + "TABLE(SNAPSHOT("
+ + "SNAPSHOT("
+ "input => TABLE b, "
+ "load_completed_condition =>
'compile_time', "
+ "load_completed_idle_timeout => INTERVAL
'10' SECOND, "
+ "state_ttl => INTERVAL '1' DAY"
- + ")) AS s ON probe.pk = s.bk")
+ + ") AS s ON probe.pk = s.bk")
.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 ca2f3696938..bb45bde761c 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
@@ -88,8 +88,14 @@ public class LateralSnapshotJoinSemanticTestPrograms {
"+I[b, 200, b, 20]")
.build())
.runSql(
- innerJoin(
- "probe.pk, probe.pv, s.bk, s.bv",
MID_FLIP, "probe.pk = s.bk"))
+ "INSERT INTO sink "
+ + "SELECT probe.pk, probe.pv, s.bk, s.bv "
+ + "FROM probe "
+ + " JOIN LATERAL TABLE(SNAPSHOT("
+ + " input => TABLE b, "
+ + " load_completed_condition =>
'user_time', "
+ + " load_completed_time =>
CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3)))) AS s "
+ + " ON probe.pk = s.bk")
.build();
public static final TableTestProgram LEFT_JOIN =
@@ -128,10 +134,10 @@ public class LateralSnapshotJoinSemanticTestPrograms {
"+I[a, 100, 2020-01-01T00:01, a,
10, 2020-01-01T00:00:01]")
.build())
.runSql(
- "INSERT INTO sink SELECT * FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "INSERT INTO sink SELECT * FROM probe JOIN LATERAL
SNAPSHOT("
+ "input => TABLE b, "
+ MID_FLIP
- + ")) AS s ON probe.pk = s.bk")
+ + ") AS s ON probe.pk = s.bk")
.build();
public static final TableTestProgram COMPOSITE_KEYS =
@@ -255,9 +261,9 @@ public class LateralSnapshotJoinSemanticTestPrograms {
// No options: 'load_completed_condition' defaults to
'compile_time'.
.runSql(
"INSERT INTO sink SELECT probe.pk, probe.pv, s.bk,
s.bv "
- + "FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ + "FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b"
- + ")) AS s ON probe.pk = s.bk")
+ + ") AS s ON probe.pk = s.bk")
.build();
public static final TableTestProgram LIVE_JOIN =
@@ -347,10 +353,10 @@ public class LateralSnapshotJoinSemanticTestPrograms {
+ projection
+ " FROM probe "
+ joinType
- + " LATERAL TABLE(SNAPSHOT("
+ + " LATERAL SNAPSHOT("
+ "input => TABLE b, "
+ flip
- + ")) AS s ON "
+ + ") 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 b1c58be353b..fcae8c89c62 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
@@ -95,11 +95,11 @@ public class LateralSnapshotJoinTestPrograms {
static final Row[] LOAD_PROBE_AFTER_DATA = {Row.of("a", 101, "2020-01-01
00:00:10")};
private static final String SNAPSHOT_BUILD =
- "LATERAL TABLE(SNAPSHOT("
+ "LATERAL SNAPSHOT("
+ "input => TABLE b, "
+ "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";
+ + ") AS s ON probe.pk = s.bk";
// Restore taken while the operator is in LOAD phase: the savepoint
captures the partial build
// multi-set and the buffered probe row, the LOAD phase is recorded in
union operator state, and
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 eef09609aeb..722e96d0abb 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
@@ -76,11 +76,11 @@ public class SnapshotTableFunctionTest extends
TableTestBase {
util.tableEnv()
.explainSql(
"SELECT o.order_id, o.amount, r.rate "
- + "FROM Orders AS o, LATERAL
TABLE(SNAPSHOT("
+ + "FROM Orders AS o, LATERAL SNAPSHOT("
+ "input => TABLE Rates, "
+ "load_completed_condition =>
'user_time', "
+ "load_completed_time =>
CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
- + ")) AS r "
+ + ") AS r "
+ "WHERE o.currency = r.currency");
assertThat(plan).contains("LateralSnapshotJoin");
}
@@ -93,11 +93,11 @@ public class SnapshotTableFunctionTest extends
TableTestBase {
.executeSql(
"CREATE VIEW OrdersWithRate AS "
+ "SELECT o.order_id, o.amount, r.rate "
- + "FROM Orders AS o, LATERAL TABLE(SNAPSHOT("
+ + "FROM Orders AS o, LATERAL SNAPSHOT("
+ "input => TABLE Rates, "
+ "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP
'2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
- + ")) AS r "
+ + ") AS r "
+ "WHERE o.currency = r.currency");
final String plan = util.tableEnv().explainSql("SELECT * FROM
OrdersWithRate");
assertThat(plan).contains("LateralSnapshotJoin");
@@ -110,11 +110,11 @@ public class SnapshotTableFunctionTest extends
TableTestBase {
util.tableEnv()
.explainSql(
"SELECT o.order_id, o.amount, r.rate "
- + "FROM Orders AS o, LATERAL
TABLE(SNAPSHOT("
+ + "FROM Orders AS o, LATERAL SNAPSHOT("
+ "input => TABLE RatesView, "
+ "load_completed_condition =>
'user_time', "
+ "load_completed_time =>
CAST(TIMESTAMP '2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))"
- + ")) AS r "
+ + ") AS r "
+ "WHERE o.currency = r.currency");
assertThat(plan).contains("LateralSnapshotJoin");
}
@@ -128,9 +128,9 @@ public class SnapshotTableFunctionTest extends
TableTestBase {
() ->
util.verifyRelPlan(
"SELECT o.order_id "
- + "FROM Orders AS o, LATERAL
TABLE(SNAPSHOT("
+ + "FROM Orders AS o, LATERAL
SNAPSHOT("
+ "input => TABLE Rates, "
- + "on_time =>
DESCRIPTOR(rate_time))) AS r "
+ + "on_time =>
DESCRIPTOR(rate_time)) AS r "
+ "WHERE o.currency =
r.currency"))
.satisfies(
anyCauseMatches(
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 6b9725052fa..323ed9beb91 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
@@ -101,57 +101,57 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testLeftJoin() {
util.verifyRelPlan(
- "SELECT * FROM probe LEFT JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@Test
void testInnerJoinWithIdleTimeoutAndStateTtl() {
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@Test
void testInnerJoinWithNonEquiCondition() {
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk AND probe.pv > s.bv");
}
@Test
void testInnerJoinWithCompositeKeys() {
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk AND probe.pv = s.bv");
}
@Test
void testInnerJoinWithTimeAttributeInCondition() {
util.verifyRelPlan(
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk AND probe.pts >= s.bts");
}
@@ -159,33 +159,33 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
void testInnerJoinWithCteBuildSide() {
util.verifyRelPlan(
"WITH cte AS (SELECT bk, bv + 1 AS bv, bts FROM b) "
- + "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ + "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@Test
void testInnerJoinWithoutBuildTimeColumn() {
util.verifyRelPlan(
- "SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@Test
void testLeftJoinWithoutBuildTimeColumn() {
util.verifyRelPlan(
- "SELECT probe.pk, probe.pv, s.bv FROM probe LEFT JOIN LATERAL
TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@@ -201,11 +201,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
+ " WATERMARK FOR bts AS bts"
+ ") WITH ('connector' = 'values', 'bounded' =
'false')");
util.verifyRelPlan(
- "SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pk = s.bk");
}
@@ -218,11 +218,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
// Derived table whose SELECT * exposes the probe time attribute as
`pts` and the (now
// materialized) build time attribute as `bts`.
final String derived =
- "(SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "(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)";
+ + ") AS s ON probe.pk = s.bk)";
// Windowing over the build-side time column is rejected: it is
materialized, not a time
// attribute.
assertThatThrownBy(
@@ -258,11 +258,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
final String plan =
util.tableEnv()
.explainSql(
- "SELECT * FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b_upsert, "
+ "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");
+ + ") AS s ON probe.pk = s.bk");
assertThat(plan).contains("ChangelogNormalize");
assertThat(plan).doesNotContain("DropUpdateBefore");
}
@@ -270,11 +270,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testNonEquiConditionCompilesEndToEnd() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 AND
probe.pts >= s.bts";
+ + ") AS s ON probe.pk = s.bk AND probe.pv > s.bv AND
probe.pts >= s.bts";
assertThat(util.tableEnv().explainSql(sql)).contains("LateralSnapshotJoin");
}
@@ -283,18 +283,18 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
final String plan =
util.tableEnv()
.explainSql(
- "SELECT * FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "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");
+ + ") AS s ON probe.pk = s.bk");
assertThat(plan).contains("LateralSnapshotJoin");
}
@Test
void testInnerJoinWithDefaultCompileTimeCompilesEndToEnd() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(input =>
TABLE b)) AS s "
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b)
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).
@@ -308,10 +308,10 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testInnerJoinWithExplicitCompileTimeCompilesEndToEnd() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
+ "load_completed_condition => 'compile_time'"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThat(util.tableEnv().explainSql(sql))
.contains("LateralSnapshotJoin")
.contains("loadCompletedCondition=[compile_time]")
@@ -333,11 +333,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
+ " bts TIMESTAMP(3)"
+ ") WITH ('connector' = 'values', 'bounded' =
'false')");
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b_no_wm, "
+ "load_completed_condition => 'user_time', "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
- + ")) AS s "
+ + ") AS s "
+ "ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
@@ -361,11 +361,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
+ " 'changelog-mode' = 'I,UB,UA,D'"
+ ")");
final String sql =
- "SELECT * FROM probe_updates JOIN LATERAL TABLE(SNAPSHOT("
+ "SELECT * FROM probe_updates 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 "
+ + ") AS s "
+ "ON probe_updates.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.hasMessageContaining(
@@ -376,11 +376,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectMissingEqualityPredicate() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 "
+ + ") AS s "
+ "ON probe.pv > s.bv";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
@@ -391,10 +391,10 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectNonConstantCondition() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
+ "load_completed_condition => CAST(CURRENT_TIMESTAMP
AS STRING)"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining("Invalid function call")
@@ -404,11 +404,11 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectNonConstantLoadCompletedTime() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
+ "load_completed_condition => 'user_time', "
+ "load_completed_time => CURRENT_TIMESTAMP"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
@@ -418,13 +418,13 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectNonConstantIdleTimeout() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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 => "
+ "CASE WHEN RAND() > 0.5 THEN INTERVAL '10' SECOND
ELSE INTERVAL '20' SECOND END"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
@@ -434,13 +434,13 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectNonConstantStateTtl() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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)), "
+ "state_ttl => "
+ "CASE WHEN RAND() > 0.5 THEN INTERVAL '1' DAY ELSE
INTERVAL '2' DAY END"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
@@ -450,12 +450,12 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectYearMonthIntervalStateTtl() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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)), "
+ "state_ttl => INTERVAL '1' YEAR"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining("No match found for function signature
SNAPSHOT");
@@ -464,12 +464,12 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectNegativeIdleTimeout() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining(
@@ -479,12 +479,12 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testRejectNegativeStateTtl() {
final String sql =
- "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT("
+ "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)), "
+ "state_ttl => INTERVAL -'10' MINUTE"
- + ")) AS s ON probe.pk = s.bk";
+ + ") AS s ON probe.pk = s.bk";
assertThatThrownBy(() -> util.verifyRelPlan(sql))
.isInstanceOf(ValidationException.class)
.hasMessageContaining("Argument 'state_ttl' of SNAPSHOT must
not be negative");
@@ -498,53 +498,53 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
@Test
void testInnerJoinJsonPlan() {
util.verifyJsonPlan(
- "INSERT INTO sink SELECT * FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "INSERT INTO sink 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");
+ + ") AS s ON probe.pk = s.bk");
}
@Test
void testLeftJoinJsonPlan() {
util.verifyJsonPlan(
- "INSERT INTO sink SELECT * FROM probe LEFT JOIN LATERAL
TABLE(SNAPSHOT("
+ "INSERT INTO sink 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");
+ + ") AS s ON probe.pk = s.bk");
}
@Test
void testInnerJoinWithIdleTimeoutAndStateTtlJsonPlan() {
util.verifyJsonPlan(
- "INSERT INTO sink SELECT * FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "INSERT INTO sink 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");
+ + ") AS s ON probe.pk = s.bk");
}
@Test
void testInnerJoinWithCompositeKeysJsonPlan() {
util.verifyJsonPlan(
- "INSERT INTO sink SELECT * FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "INSERT INTO sink 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");
+ + ") AS s ON probe.pk = s.bk AND probe.pv = s.bv");
}
@Test
void testInnerJoinWithNonEquiConditionJsonPlan() {
util.verifyJsonPlan(
- "INSERT INTO sink SELECT * FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "INSERT INTO sink 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");
+ + ") AS s ON probe.pk = s.bk AND probe.pv > s.bv");
}
//
------------------------------------------------------------------------------------------
@@ -587,9 +587,9 @@ public class LateralSnapshotJoinTest extends TableTestBase {
util.tableEnv()
.compilePlanSql(
"INSERT INTO sink SELECT * FROM probe "
- + "JOIN LATERAL TABLE("
+ + "JOIN LATERAL "
+ snapshotCall
- + ") AS s ON probe.pk = s.bk");
+ + " AS s ON probe.pk = s.bk");
final List<Transformation<?>> transformations =
CompiledPlanUtils.toTransformations(util.tableEnv(), plan);
return findJoinOperator(transformations).getMinStateTtlMs();
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 58632e44326..7715e6e1d41 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
@@ -109,10 +109,10 @@ public class LateralSnapshotJoinITCase extends
StreamingWithStateTestBase {
final List<Row> actual =
collect(
- "SELECT probe.pk, s.bv FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "SELECT probe.pk, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
+ "input => TABLE b, "
+ MID_FLIP
- + ")) AS s ON probe.pk = s.bk");
+ + ") AS s ON probe.pk = s.bk");
assertThatList(actual).containsExactlyInAnyOrder(Row.of("a", 11));
}
@@ -131,10 +131,10 @@ public class LateralSnapshotJoinITCase extends
StreamingWithStateTestBase {
final List<Row> actual =
collect(
- "SELECT probe.pk, s.bv FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "SELECT probe.pk, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
+ "input => TABLE b, "
+ MID_FLIP
- + ")) AS s ON probe.pk = s.bk");
+ + ") AS s ON probe.pk = s.bk");
assertThatList(actual).containsExactlyInAnyOrder(Row.of("a", 11));
}
@@ -167,10 +167,10 @@ public class LateralSnapshotJoinITCase extends
StreamingWithStateTestBase {
final List<Row> byProbeId =
sortedByProbeId(
collect(
- "SELECT probe.pv, s.bv FROM probe JOIN LATERAL
TABLE(SNAPSHOT("
+ "SELECT probe.pv, s.bv FROM probe JOIN LATERAL
SNAPSHOT("
+ "input => TABLE b, "
+ MID_FLIP
- + ")) AS s ON probe.pk = s.bk"));
+ + ") AS s ON probe.pk = s.bk"));
assertThat(byProbeId).hasSize(probeCount);
assertMonotonicVersions(byProbeId);
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 5375c4f7eb3..d69cb1baf0e 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
@@ -18,7 +18,7 @@ limitations under the License.
<Root>
<TestCase name="testBuildSideWithProctime">
<Resource name="sql">
- <![CDATA[SELECT probe.pk, s.bk, s.bv, s.pt FROM probe JOIN LATERAL
TABLE(SNAPSHOT(input => TABLE b_proctime)) 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) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -44,7 +44,7 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk,
bk, bv, pt], build
</TestCase>
<TestCase name="testBuildSideWithoutWatermark">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(input => TABLE
b_no_wm)) AS s ON probe.pk = s.bk]]>
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE
b_no_wm) AS s ON probe.pk = s.bk]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -68,7 +68,7 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk,
pv, pts, bk, bv, b
</TestCase>
<TestCase name="testInnerJoin">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL 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 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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -92,7 +92,7 @@ HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk,
pv, pts, bk, bv, b
</TestCase>
<TestCase name="testInnerJoinWithCompositeKeys">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL 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 AND probe.pv = s.bv]]>
+ <![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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -116,7 +116,7 @@ HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), =(pv,
bv))], select=[pk, pv
</TestCase>
<TestCase name="testInnerJoinWithNonEquiCondition">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe JOIN LATERAL 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 AND probe.pv > s.bv]]>
+ <![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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -140,7 +140,7 @@ HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), >(pv,
bv))], select=[pk, pv
</TestCase>
<TestCase name="testInnerJoinWithoutBuildTimeColumn">
<Resource name="sql">
- <![CDATA[SELECT probe.pk, probe.pv, s.bv FROM probe JOIN LATERAL
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 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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -166,7 +166,7 @@ Calc(select=[pk, pv, bv])
</TestCase>
<TestCase name="testLeftJoin">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe LEFT 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 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]]>
</Resource>
<Resource name="ast">
<![CDATA[
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 4305878729b..f75754f8a09 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
TABLE(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, 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[
@@ -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 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 AND probe.pv = s.bv]]>
+ <![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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -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 TABLE(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, 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[
@@ -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 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)),
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,
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[
@@ -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 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 AND probe.pv > s.bv]]>
+ <![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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -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 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 AND probe.pts >= s.bts]]>
+ <![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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -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
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 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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -251,7 +251,7 @@ Calc(select=[pk, pv, bv])
</TestCase>
<TestCase name="testLeftJoin">
<Resource name="sql">
- <![CDATA[SELECT * FROM probe LEFT 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 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]]>
</Resource>
<Resource name="ast">
<![CDATA[
@@ -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
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 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]]>
</Resource>
<Resource name="ast">
<