fhueske commented on code in PR #28763: URL: https://github.com/apache/flink/pull/28763#discussion_r3604200187
########## 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: IMO the tests asserts that the same validation logic is applied for batch as for streaming (essentially that `LogicalJoinToLateralSnapshotJoinRule` is listed in `FlinkBatchRuleSets`) -- 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]
