snuyanzin commented on code in PR #28664: URL: https://github.com/apache/flink/pull/28664#discussion_r3549549289
########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java: ########## @@ -0,0 +1,540 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.stream.sql.join; + +import org.apache.flink.table.api.TableConfig; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.planner.utils.TableTestBase; +import org.apache.flink.table.planner.utils.TableTestUtil; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.ZoneId; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Plan tests for the {@code LATERAL SNAPSHOT} processing-time temporal table join. */ +public class LateralSnapshotJoinTest extends TableTestBase { + + private TableTestUtil util; + + @BeforeEach + void setup() { + final TableConfig config = TableConfig.getDefault(); + config.setLocalTimeZone(ZoneId.of("UTC")); + util = streamTestUtil(config); + + util.tableEnv() + .executeSql( + "CREATE TABLE probe (" + + " pk STRING," + + " pv INT," + + " pts TIMESTAMP(3)," + + " WATERMARK FOR pts AS pts" + + ") WITH ('connector' = 'values', 'bounded' = 'false')"); + + util.tableEnv() + .executeSql( + "CREATE TABLE b (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts" + + ") WITH (" + + " 'connector' = 'values'," + + " 'bounded' = 'false'," + + " 'changelog-mode' = 'I,UB,UA,D'" + + ")"); + + // Sink for the JSON-plan tests; nullable columns accept the null-padded LEFT-join build + // side. + util.tableEnv() + .executeSql( + "CREATE TABLE sink (" + + " pk STRING," + + " pv INT," + + " pts TIMESTAMP(3)," + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)" + + ") WITH ('connector' = 'blackhole')"); + } + + @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 testInnerJoinWithIdleTimeoutAndStateTtl() { + 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 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 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 testInnerJoinWithTimeAttributeInCondition() { + 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.pts >= s.bts"); + } + + @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 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 testLeftJoinWithoutBuildTimeColumn() { + util.verifyRelPlan( + "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"); + } + + @Test + void testBuildSideProctimeIsMaterialized() { + util.tableEnv() + .executeSql( + "CREATE TABLE b_proctime (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " pt AS PROCTIME()," + + " 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(" + + "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"); + } + + // ------------------------------------------------------------------------------------------ + // Behavior and compilation smoke tests + // ------------------------------------------------------------------------------------------ + + @Test + void testBuildRowtimeIsNotForwarded() { + // 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(" + + "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)"; + // Windowing over the build-side time column is rejected: it is materialized, not a time + // attribute. + assertThatThrownBy( + () -> + util.verifyRelPlan( + "SELECT COUNT(*) FROM " + + derived + + " GROUP BY TUMBLE(bts, INTERVAL '1' MINUTE)")) + .hasMessageContaining("time attribute"); + // The probe-side time attribute is preserved and remains usable as event time. + util.tableEnv() + .explainSql( + "SELECT COUNT(*) FROM " + + derived + + " GROUP BY TUMBLE(pts, INTERVAL '1' MINUTE)"); + } + + @Test + void testInnerJoinWithUpsertBuildSourceMaterializesRetractions() { + util.tableEnv() + .executeSql( + "CREATE TABLE b_upsert (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts," + + " PRIMARY KEY (bk) NOT ENFORCED" + + ") WITH (" + + " 'connector' = 'values'," + + " 'bounded' = 'false'," + + " 'changelog-mode' = 'I,UA,D'" + + ")"); + final String plan = + util.tableEnv() + .explainSql( + "SELECT * FROM probe JOIN LATERAL TABLE(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"); + assertThat(plan).contains("ChangelogNormalize"); + assertThat(plan).doesNotContain("DropUpdateBefore"); + } + + @Test + void testNonEquiConditionCompilesEndToEnd() { + 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.pk = s.bk AND probe.pv > s.bv AND probe.pts >= s.bts"; + assertThat(util.tableEnv().explainSql(sql)).contains("LateralSnapshotJoin"); + } + + @Test + void testFoldableConstantArgs() { + final String plan = + util.tableEnv() + .explainSql( + "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"); + assertThat(plan).contains("LateralSnapshotJoin"); + } + + @Test + void testInnerJoinWithDefaultCompileTime_compilesEndToEnd() { Review Comment: ```suggestion void testInnerJoinWithDefaultCompileTimeCompilesEndToEnd() { ``` following the rest of the style here ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java: ########## @@ -0,0 +1,540 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.stream.sql.join; + +import org.apache.flink.table.api.TableConfig; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.planner.utils.TableTestBase; +import org.apache.flink.table.planner.utils.TableTestUtil; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.ZoneId; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Plan tests for the {@code LATERAL SNAPSHOT} processing-time temporal table join. */ +public class LateralSnapshotJoinTest extends TableTestBase { + + private TableTestUtil util; + + @BeforeEach + void setup() { + final TableConfig config = TableConfig.getDefault(); + config.setLocalTimeZone(ZoneId.of("UTC")); + util = streamTestUtil(config); + + util.tableEnv() + .executeSql( + "CREATE TABLE probe (" + + " pk STRING," + + " pv INT," + + " pts TIMESTAMP(3)," + + " WATERMARK FOR pts AS pts" + + ") WITH ('connector' = 'values', 'bounded' = 'false')"); + + util.tableEnv() + .executeSql( + "CREATE TABLE b (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts" + + ") WITH (" + + " 'connector' = 'values'," + + " 'bounded' = 'false'," + + " 'changelog-mode' = 'I,UB,UA,D'" + + ")"); + + // Sink for the JSON-plan tests; nullable columns accept the null-padded LEFT-join build + // side. + util.tableEnv() + .executeSql( + "CREATE TABLE sink (" + + " pk STRING," + + " pv INT," + + " pts TIMESTAMP(3)," + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)" + + ") WITH ('connector' = 'blackhole')"); + } + + @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 testInnerJoinWithIdleTimeoutAndStateTtl() { + 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 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 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 testInnerJoinWithTimeAttributeInCondition() { + 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.pts >= s.bts"); + } + + @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 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 testLeftJoinWithoutBuildTimeColumn() { + util.verifyRelPlan( + "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"); + } + + @Test + void testBuildSideProctimeIsMaterialized() { + util.tableEnv() + .executeSql( + "CREATE TABLE b_proctime (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " pt AS PROCTIME()," + + " 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(" + + "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"); + } + + // ------------------------------------------------------------------------------------------ + // Behavior and compilation smoke tests + // ------------------------------------------------------------------------------------------ + + @Test + void testBuildRowtimeIsNotForwarded() { + // 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(" + + "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)"; + // Windowing over the build-side time column is rejected: it is materialized, not a time + // attribute. + assertThatThrownBy( + () -> + util.verifyRelPlan( + "SELECT COUNT(*) FROM " + + derived + + " GROUP BY TUMBLE(bts, INTERVAL '1' MINUTE)")) + .hasMessageContaining("time attribute"); + // The probe-side time attribute is preserved and remains usable as event time. + util.tableEnv() + .explainSql( + "SELECT COUNT(*) FROM " + + derived + + " GROUP BY TUMBLE(pts, INTERVAL '1' MINUTE)"); + } + + @Test + void testInnerJoinWithUpsertBuildSourceMaterializesRetractions() { + util.tableEnv() + .executeSql( + "CREATE TABLE b_upsert (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts," + + " PRIMARY KEY (bk) NOT ENFORCED" + + ") WITH (" + + " 'connector' = 'values'," + + " 'bounded' = 'false'," + + " 'changelog-mode' = 'I,UA,D'" + + ")"); + final String plan = + util.tableEnv() + .explainSql( + "SELECT * FROM probe JOIN LATERAL TABLE(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"); + assertThat(plan).contains("ChangelogNormalize"); + assertThat(plan).doesNotContain("DropUpdateBefore"); + } + + @Test + void testNonEquiConditionCompilesEndToEnd() { + 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.pk = s.bk AND probe.pv > s.bv AND probe.pts >= s.bts"; + assertThat(util.tableEnv().explainSql(sql)).contains("LateralSnapshotJoin"); + } + + @Test + void testFoldableConstantArgs() { + final String plan = + util.tableEnv() + .explainSql( + "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"); + assertThat(plan).contains("LateralSnapshotJoin"); + } + + @Test + void testInnerJoinWithDefaultCompileTime_compilesEndToEnd() { + final String sql = + "SELECT * FROM probe JOIN LATERAL TABLE(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). + assertThat(util.tableEnv().explainSql(sql)) + .contains("LateralSnapshotJoin") + .contains("loadCompletedCondition=[compile_time]") + .contains("joinType=[InnerJoin]") + .contains("where=[=(pk, bk)]"); + } + + @Test + void testInnerJoinWithExplicitCompileTime_compilesEndToEnd() { Review Comment: ```suggestion void testInnerJoinWithExplicitCompileTimeCompilesEndToEnd() { ``` following the rest of the style here -- 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]
