twalthr commented on code in PR #28763:
URL: https://github.com/apache/flink/pull/28763#discussion_r3602525627
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java:
##########
@@ -123,22 +124,27 @@ public void onMatch(RelOptRuleCall call) {
"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));
Review Comment:
Can we get rid of this check? Having more than one time attribute should be
fine, as long as watermarks are flowing? We should follow the principle of as
little errors as possible.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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
+ *
org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalLateralSnapshotJoinRule}
Review Comment:
nit:
```suggestion
* BatchPhysicalLateralSnapshotJoinRule}
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.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
+ *
org.apache.flink.table.planner.plan.rules.logical.LogicalJoinToLateralSnapshotJoinRule})
into a
+ * regular batch {@link BatchPhysicalHashJoin} for batch execution.
+ *
+ * <p>In batch all input is bounded and append-only (batch rejects
non-insert-only sources up
Review Comment:
```suggestion
* <p>In batch all input is bounded and append-only (batch rejects or in the
future materializes non-insert-only sources up
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.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
+ *
org.apache.flink.table.planner.plan.rules.logical.LogicalJoinToLateralSnapshotJoinRule})
into a
Review Comment:
nit: please use fully qualified imports, very annoying for readability
otherwise
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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
+ *
org.apache.flink.table.planner.plan.rules.physical.batch.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 testInnerJoinWithCteBuildSide() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testSnapshotArgumentsAreDropped() {
Review Comment:
remove?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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
+ *
org.apache.flink.table.planner.plan.rules.physical.batch.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 testInnerJoinWithCteBuildSide() {
Review Comment:
remove? we can assume CTEs work?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/join/LateralSnapshotJoinITCase.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.runtime.batch.sql.join;
+
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * End-to-end tests for {@link
+ *
org.apache.flink.table.planner.plan.rules.physical.batch.BatchPhysicalLateralSnapshotJoinRule}:
+ * in batch mode a {@code LATERAL SNAPSHOT} join is executed as a regular join
of the probe side
+ * against the (final) build side.
+ */
+class LateralSnapshotJoinITCase {
Review Comment:
Can we use the `SemanticTestsBase` for this?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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
+ *
org.apache.flink.table.planner.plan.rules.physical.batch.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 testInnerJoinWithCteBuildSide() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testSnapshotArgumentsAreDropped() {
+ // The idle-timeout and state-ttl arguments are streaming-only; in
batch they are dropped
+ // and the plan is a plain join, identical to 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)), "
+ + "load_completed_idle_timeout => INTERVAL '10'
SECOND, "
+ + "state_ttl => INTERVAL '1' DAY"
+ + ")) AS s "
+ + "ON probe.pk = s.bk");
+ }
+
+ @Test
+ void testDefaultCompileTime() {
Review Comment:
same as inner join, remove?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.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
+ *
org.apache.flink.table.planner.plan.rules.physical.batch.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 testInnerJoinWithCteBuildSide() {
+ util.verifyRelPlan(
+ "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");
+ }
+
+ @Test
+ void testSnapshotArgumentsAreDropped() {
+ // The idle-timeout and state-ttl arguments are streaming-only; in
batch they are dropped
+ // and the plan is a plain join, identical to 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)), "
+ + "load_completed_idle_timeout => INTERVAL '10'
SECOND, "
+ + "state_ttl => INTERVAL '1' DAY"
+ + ")) AS s "
+ + "ON probe.pk = s.bk");
+ }
+
+ @Test
+ void testDefaultCompileTime() {
+ util.verifyRelPlan(
+ "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(input =>
TABLE b)) 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");
+ }
+
+ @Test
+ void testRejectInvalidSnapshotArgument() {
Review Comment:
remove? let's avoid duplicate tests. this test should be about the specifics
of batch mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
