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 f5aaf91f2db [FLINK-40158][table-planner] Support LATERAL SNAPSHOT join
in batch mode (#28763)
f5aaf91f2db is described below
commit f5aaf91f2db1f3eadffc325ad278680dc7aeac56
Author: Fabian Hueske <[email protected]>
AuthorDate: Mon Jul 20 16:08:32 2026 +0200
[FLINK-40158][table-planner] Support LATERAL SNAPSHOT join in batch mode
(#28763)
In batch, all input is bounded and append-only, so the processing-time
LATERAL SNAPSHOT join degenerates to a regular join of the probe side
against the (final) build side; the SNAPSHOT-specific arguments are
dropped. BatchPhysicalLateralSnapshotJoinRule converts the logical
snapshot join into a shuffle hash join that builds the (smaller) SNAPSHOT
side, mirroring StreamPhysicalLateralSnapshotJoinRule.
Generated-By: Claude Opus 4.8 (1M context)
---
.../LogicalJoinToLateralSnapshotJoinRule.java | 38 ++--
.../BatchPhysicalLateralSnapshotJoinRule.java | 108 ++++++++++++
.../planner/plan/rules/FlinkBatchRuleSets.scala | 12 +-
.../batch/sql/join/LateralSnapshotJoinTest.java | 189 ++++++++++++++++++++
.../LateralSnapshotJoinBatchSemanticTests.java | 42 +++++
.../batch/LateralSnapshotJoinTestPrograms.java | 129 ++++++++++++++
.../batch/sql/join/LateralSnapshotJoinTest.xml | 191 +++++++++++++++++++++
7 files changed, 691 insertions(+), 18 deletions(-)
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 f76154803a4..c4b1012df97 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
@@ -28,6 +28,7 @@ import
org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalLateralSnap
import
org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalTableFunctionScan;
import org.apache.flink.table.planner.plan.utils.FlinkRexUtil;
import org.apache.flink.table.planner.plan.utils.LateralSnapshotJoinUtil;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
import
org.apache.flink.table.types.inference.strategies.LateralSnapshotTypeStrategy;
import org.apache.calcite.plan.RelOptRuleCall;
@@ -123,22 +124,27 @@ public class LogicalJoinToLateralSnapshotJoinRule
"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 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));
+ // 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
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java
new file mode 100644
index 00000000000..9338ffb8feb
--- /dev/null
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java
@@ -0,0 +1,108 @@
+/*
+ * 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.rules.physical.batch;
+
+import org.apache.flink.table.planner.plan.nodes.FlinkConventions;
+import
org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalLateralSnapshotJoin;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalHashJoin;
+import
org.apache.flink.table.planner.plan.rules.logical.LogicalJoinToLateralSnapshotJoinRule;
+import
org.apache.flink.table.planner.plan.rules.physical.stream.StreamPhysicalLateralSnapshotJoinRule;
+import org.apache.flink.table.planner.plan.trait.FlinkRelDistribution;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.util.ImmutableIntList;
+
+/**
+ * Converts a {@link FlinkLogicalLateralSnapshotJoin} (created by {@link
+ * LogicalJoinToLateralSnapshotJoinRule}) into a regular batch {@link
BatchPhysicalHashJoin} for
+ * batch execution.
+ *
+ * <p>In batch all input is bounded and append-only (batch rejects or in the
future materializes
+ * non-insert-only sources up front), so the processing-time {@code LATERAL
SNAPSHOT} join
+ * degenerates to a regular join of the probe side against the (final) build
side; the
+ * SNAPSHOT-specific arguments are irrelevant and dropped. This rule mirrors
the streaming {@link
+ * StreamPhysicalLateralSnapshotJoinRule}, which converts the same logical
node to a dedicated
+ * stream operator.
+ *
+ * <p>The SNAPSHOT input is the build (right) side of the LATERAL join and is
the dimension-like
+ * side, expected to be smaller than the probe (left) side. The join is
therefore emitted as a
+ * shuffle hash join that builds on the right input.
+ */
+public class BatchPhysicalLateralSnapshotJoinRule extends ConverterRule {
+
+ public static final BatchPhysicalLateralSnapshotJoinRule INSTANCE =
+ new BatchPhysicalLateralSnapshotJoinRule(
+ Config.INSTANCE.withConversion(
+ FlinkLogicalLateralSnapshotJoin.class,
+ FlinkConventions.LOGICAL(),
+ FlinkConventions.BATCH_PHYSICAL(),
+ "BatchPhysicalLateralSnapshotJoinRule"));
+
+ private BatchPhysicalLateralSnapshotJoinRule(Config config) {
+ super(config);
+ }
+
+ @Override
+ public RelNode convert(RelNode rel) {
+ final FlinkLogicalLateralSnapshotJoin join =
(FlinkLogicalLateralSnapshotJoin) rel;
+ final RelTraitSet providedTraitSet =
+ rel.getTraitSet().replace(FlinkConventions.BATCH_PHYSICAL());
+
+ // Both inputs are hash-partitioned on their join keys (shuffle hash
join).
+ final JoinInfo joinInfo = join.analyzeCondition();
+ final RelNode newLeft = convertInput(join.getLeft(),
joinInfo.leftKeys);
+ final RelNode newRight = convertInput(join.getRight(),
joinInfo.rightKeys);
+
+ return new BatchPhysicalHashJoin(
+ join.getCluster(),
+ providedTraitSet,
+ newLeft,
+ newRight,
+ join.getCondition(),
+ join.getJoinType(),
+ // leftIsBuild = false: build the right (SNAPSHOT) side, the
smaller dimension side.
+ false,
+ // isBroadcast = false: shuffle hash join.
+ false,
+ // tryDistinctBuildRow = false: only relevant for semi/anti
joins.
+ false,
+ // withJobStrategyHint = false: not driven by a user join hint.
+ false);
+ }
+
+ /**
+ * Converts a join input to the batch-physical convention and requires it
to be hash-partitioned
+ * on the given join {@code keys} (or a singleton distribution if there
are none).
+ */
+ private static RelNode convertInput(RelNode input, ImmutableIntList keys) {
+ final FlinkRelDistribution distribution =
+ keys.isEmpty()
+ ? FlinkRelDistribution.SINGLETON()
+ : FlinkRelDistribution.hash(keys.toIntArray(), true);
+ final RelTraitSet requiredTraitSet =
+ input.getTraitSet()
+ .replace(FlinkConventions.BATCH_PHYSICAL())
+ .replace(distribution);
+ return RelOptRule.convert(input, requiredTraitSet);
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
index 3fb057bc08d..26553f24ddc 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala
@@ -18,7 +18,6 @@
package org.apache.flink.table.planner.plan.rules
import org.apache.flink.table.planner.plan.nodes.logical._
-import
org.apache.flink.table.planner.plan.rules.FlinkStreamRuleSets.SIMPLIFY_COALESCE_RULES
import org.apache.flink.table.planner.plan.rules.logical._
import
org.apache.flink.table.planner.plan.rules.physical.FlinkExpandConversionRule
import org.apache.flink.table.planner.plan.rules.physical.batch._
@@ -419,7 +418,13 @@ object FlinkBatchRuleSets {
// Avoid having async calls in multiple projections in a single calc.
AsyncCalcSplitRule.ONE_PER_CALC_SPLIT,
// remove output of rank number when it is not used by successor calc
- RedundantRankNumberColumnRemoveRule.INSTANCE
+ RedundantRankNumberColumnRemoveRule.INSTANCE,
+ // Rewrites a join over a LATERAL SNAPSHOT table function call into a
dedicated
+ // FlinkLogicalLateralSnapshotJoin
+ LogicalJoinToLateralSnapshotJoinRule.INSTANCE,
+ // Rejects SNAPSHOT scans that survived the rewrite above, i.e. SNAPSHOT
calls used outside a
+ // LATERAL context. Must run after LogicalJoinToLateralSnapshotJoinRule.
+ ForbidSnapshotOutsideLateralRule.INSTANCE
)
/** RuleSet to do physical optimize for batch */
@@ -460,6 +465,9 @@ object FlinkBatchRuleSets {
BatchPhysicalPythonWindowAggregateRule.INSTANCE,
// window tvf
BatchPhysicalWindowTableFunctionRule.INSTANCE,
+ // Converts a LATERAL SNAPSHOT join into a (shuffle hash) batch join with
the SNAPSHOT input as
+ // build side
+ BatchPhysicalLateralSnapshotJoinRule.INSTANCE,
// join
BatchPhysicalHashJoinRule.INSTANCE,
BatchPhysicalSortMergeJoinRule.INSTANCE,
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
new file mode 100644
index 00000000000..50416c1560f
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
@@ -0,0 +1,189 @@
+/*
+ * 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.batch.sql.join;
+
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.ValidationException;
+import
org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalLateralSnapshotJoinRule;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Plan tests for {@code LATERAL SNAPSHOT} joins in batch mode.
+ *
+ * <p>In batch all input is bounded and append-only, so the processing-time
{@code LATERAL SNAPSHOT}
+ * join degenerates to a regular join of the probe side against the (final)
build side. {@link
+ * BatchPhysicalLateralSnapshotJoinRule} performs this translation and the
SNAPSHOT-specific
+ * arguments are dropped.
+ */
+public class LateralSnapshotJoinTest extends TableTestBase {
+
+ private BatchTableTestUtil util;
+
+ @BeforeEach
+ void setup() {
+ util = batchTestUtil(TableConfig.getDefault());
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE probe ("
+ + " pk STRING,"
+ + " pv INT,"
+ + " pts TIMESTAMP(3),"
+ + " WATERMARK FOR pts AS pts"
+ + ") WITH ('connector' = 'values', 'bounded' =
'true')");
+
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE b ("
+ + " bk STRING,"
+ + " bv INT,"
+ + " bts TIMESTAMP(3),"
+ + " WATERMARK FOR bts AS bts"
+ + ") WITH ('connector' = 'values', 'bounded' =
'true')");
+ }
+
+ //
------------------------------------------------------------------------------------------
+ // Translation to a regular join
+ //
------------------------------------------------------------------------------------------
+
+ @Test
+ void testInnerJoin() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testLeftJoin() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testInnerJoinWithCompositeKeys() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testInnerJoinWithNonEquiCondition() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testInnerJoinWithoutBuildTimeColumn() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testBuildSideWithProctime() {
+ // A PROCTIME() build column is a time-attribute indicator in batch;
the rule materializes
+ // it (into a PROCTIME_MATERIALIZE call) when degrading to a regular
join.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE b_proctime ("
+ + " bk STRING,"
+ + " bv INT,"
+ + " 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 "
+ + "ON probe.pk = s.bk");
+ }
+
+ @Test
+ void testBuildSideWithoutWatermark() {
+ // A watermark on the build side is required in streaming (drives the
LOAD phase) but not in
+ // batch, where the build side is already final.
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE b_no_wm ("
+ + " bk STRING,"
+ + " bv INT,"
+ + " bts TIMESTAMP(3)"
+ + ") WITH ('connector' = 'values', 'bounded' =
'true')");
+ util.verifyRelPlan(
+ "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(input =>
TABLE b_no_wm)) AS s "
+ + "ON probe.pk = s.bk");
+ }
+
+ //
------------------------------------------------------------------------------------------
+ // Validation: rejection paths
+ //
------------------------------------------------------------------------------------------
+
+ @Test
+ void testRejectMissingEqualityPredicate() {
+ final String sql =
+ "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.pv > s.bv";
+ assertThatThrownBy(() -> util.verifyExecPlan(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "LATERAL SNAPSHOT join requires at least one equality
predicate.");
+ }
+
+ @Test
+ void testRejectSnapshotOutsideLateral() {
+ // SNAPSHOT used outside a LATERAL context is not rewritten into a
join;
+ // ForbidSnapshotOutsideLateralRule rejects the surviving SNAPSHOT
scan with a clear
+ // message.
+ assertThatThrownBy(() -> util.verifyExecPlan("SELECT * FROM
SNAPSHOT(input => TABLE b)"))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "The SNAPSHOT function can only be used as the build
side "
+ + "(right-hand side) of a LATERAL join");
+ }
+}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinBatchSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinBatchSemanticTests.java
new file mode 100644
index 00000000000..87725a60a26
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinBatchSemanticTests.java
@@ -0,0 +1,42 @@
+/*
+ * 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.nodes.exec.batch;
+
+import
org.apache.flink.table.planner.plan.nodes.exec.testutils.BatchSemanticTestBase;
+import
org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalLateralSnapshotJoinRule;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.List;
+
+/**
+ * Semantic tests for {@code LATERAL SNAPSHOT} joins in batch mode. {@link
+ * BatchPhysicalLateralSnapshotJoinRule} degrades the join to a regular join
of the probe side
+ * against the (final) build side.
+ */
+public class LateralSnapshotJoinBatchSemanticTests extends
BatchSemanticTestBase {
+
+ @Override
+ public List<TableTestProgram> programs() {
+ return List.of(
+ LateralSnapshotJoinTestPrograms.INNER_JOIN,
+ LateralSnapshotJoinTestPrograms.LEFT_JOIN,
+
LateralSnapshotJoinTestPrograms.INNER_JOIN_WITH_NON_EQUI_CONDITION,
+ LateralSnapshotJoinTestPrograms.SNAPSHOT_ARGUMENTS_IGNORED);
+ }
+}
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
new file mode 100644
index 00000000000..8b8be211952
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
@@ -0,0 +1,129 @@
+/*
+ * 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.nodes.exec.batch;
+
+import org.apache.flink.table.test.program.SinkTestStep;
+import org.apache.flink.table.test.program.SourceTestStep;
+import org.apache.flink.table.test.program.TableTestProgram;
+import org.apache.flink.types.Row;
+
+/**
+ * {@link TableTestProgram} definitions for semantically testing {@code
LATERAL SNAPSHOT} joins in
+ * batch mode, where the join degenerates to a regular join of the probe side
against the (final)
+ * build side.
+ */
+public class LateralSnapshotJoinTestPrograms {
+
+ private static SourceTestStep probe() {
+ return SourceTestStep.newBuilder("probe")
+ .addSchema("pk STRING", "pv INT")
+ .producedValues(Row.of("a", 1), Row.of("b", 2), Row.of("c", 3))
+ .build();
+ }
+
+ private static SourceTestStep build() {
+ // The key "a" appears twice: the batch join runs against the final
(complete) build side,
+ // so both "a" rows participate.
+ return SourceTestStep.newBuilder("b")
+ .addSchema("bk STRING", "bv INT")
+ .producedValues(Row.of("a", 10), Row.of("a", 11), Row.of("b",
20))
+ .build();
+ }
+
+ public static final TableTestProgram INNER_JOIN =
+ TableTestProgram.of(
+ "lateral-snapshot-join-inner",
+ "batch LATERAL SNAPSHOT inner join degrades to a
regular join")
+ .setupTableSource(probe())
+ .setupTableSource(build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pk STRING", "pv INT", "bk
STRING", "bv INT")
+ .consumedValues(
+ Row.of("a", 1, "a", 10),
+ Row.of("a", 1, "a", 11),
+ Row.of("b", 2, "b", 20))
+ .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")
+ .build();
+
+ public static final TableTestProgram LEFT_JOIN =
+ TableTestProgram.of(
+ "lateral-snapshot-join-left",
+ "batch LATERAL SNAPSHOT left join null-pads
unmatched probe rows")
+ .setupTableSource(probe())
+ .setupTableSource(build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pk STRING", "pv INT", "bk
STRING", "bv INT")
+ .consumedValues(
+ Row.of("a", 1, "a", 10),
+ Row.of("a", 1, "a", 11),
+ Row.of("b", 2, "b", 20),
+ Row.of("c", 3, null, null))
+ .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")
+ .build();
+
+ public static final TableTestProgram INNER_JOIN_WITH_NON_EQUI_CONDITION =
+ TableTestProgram.of(
+ "lateral-snapshot-join-non-equi",
+ "batch LATERAL SNAPSHOT join with an additional
non-equi predicate")
+ .setupTableSource(probe())
+ .setupTableSource(build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pk STRING", "pv INT", "bk
STRING", "bv INT")
+ .consumedValues(
+ Row.of("a", 1, "a", 11),
Row.of("b", 2, "b", 20))
+ .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 AND s.bv > 10")
+ .build();
+
+ public static final TableTestProgram SNAPSHOT_ARGUMENTS_IGNORED =
+ TableTestProgram.of(
+ "lateral-snapshot-join-arguments-ignored",
+ "streaming-only SNAPSHOT arguments do not affect
the batch result")
+ .setupTableSource(probe())
+ .setupTableSource(build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pk STRING", "pv INT", "bk
STRING", "bv INT")
+ .consumedValues(
+ Row.of("a", 1, "a", 10),
+ Row.of("a", 1, "a", 11),
+ Row.of("b", 2, "b", 20))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT pk, pv, bk, bv FROM probe
JOIN LATERAL "
+ + "TABLE(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")
+ .build();
+}
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
new file mode 100644
index 00000000000..5375c4f7eb3
--- /dev/null
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
@@ -0,0 +1,191 @@
+<?xml version="1.0" ?>
+<!--
+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.
+-->
+<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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+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)])
+ +- LogicalProject(bk=[$0], bv=[$1], pt=[$2])
+ +- LogicalProject(bk=[$0], bv=[$1], pt=[PROCTIME()])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
b_proctime]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, bk, bv, pt],
build=[right])
+:- Exchange(distribution=[hash[pk]])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe,
project=[pk], metadata=[]]], fields=[pk])
++- Calc(select=[bk, bv, PROCTIME_MATERIALIZE(PROCTIME()) AS pt])
+ +- Exchange(distribution=[hash[bk]])
+ +- TableSourceScan(table=[[default_catalog, default_database,
b_proctime]], fields=[bk, bv])
+]]>
+ </Resource>
+ </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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(),
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]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts, bk, bv,
bts], build=[right])
+:- Exchange(distribution=[hash[pk]])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[hash[bk]])
+ +- TableSourceScan(table=[[default_catalog, default_database, b_no_wm]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
_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]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, pts, bk, bv,
bts], build=[right])
+:- Exchange(distribution=[hash[pk]])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[hash[bk]])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[AND(=($0, $3), =($1, $4))], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
_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]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), =(pv, bv))], select=[pk,
pv, pts, bk, bv, bts], build=[right])
+:- Exchange(distribution=[hash[pk, pv]])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[hash[bk, bv]])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[AND(=($0, $3), >($1, $4))], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
_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]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[InnerJoin], where=[AND(=(pk, bk), >(pv, bv))], select=[pk,
pv, pts, bk, bv, bts], build=[right])
+:- Exchange(distribution=[hash[pk]])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[hash[bk]])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], bv=[$4])
++- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
_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]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[pk, pv, bv])
++- HashJoin(joinType=[InnerJoin], where=[=(pk, bk)], select=[pk, pv, bk, bv],
build=[right])
+ :- Exchange(distribution=[hash[pk]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, probe,
project=[pk, pv], metadata=[]]], fields=[pk, pv])
+ +- Exchange(distribution=[hash[bk]])
+ +- Calc(select=[bk, bv])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </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]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[=($0, $3)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
_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]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+HashJoin(joinType=[LeftOuterJoin], where=[=(pk, bk)], select=[pk, pv, pts, bk,
bv, bts], build=[right])
+:- Exchange(distribution=[hash[pk]])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[hash[bk]])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </TestCase>
+</Root>
