[FLINK-39785][table] Add e2e tests for LateralSnapshotJoin

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java

@@ -257,15 +257,24 @@ public static class FromElementSourceFunctionWithWatermark
 
         private final TerminatingLogic terminating;
 
+        /** Sleep for {@link #sleepTimeMillis} after emitting every {@code sleepAfterElements}. */
+        private final int sleepAfterElements;
+
+        private final long sleepTimeMillis;
+
         public FromElementSourceFunctionWithWatermark(
                 String tableName,
                 TypeSerializer<RowData> serializer,
                 Iterable<RowData> elements,
                 WatermarkStrategy<RowData> watermarkStrategy,
-                TerminatingLogic terminating)
+                TerminatingLogic terminating,
+                int sleepAfterElements,
+                long sleepTimeMillis)
                 throws IOException {
             this.tableName = tableName;
             this.terminating = terminating;
+            this.sleepAfterElements = sleepAfterElements;
+            this.sleepTimeMillis = sleepTimeMillis;
             ByteArrayOutputStream baos = new ByteArrayOutputStream();
             DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(baos);
 
@@ -325,6 +334,13 @@ public RelativeClock getInputActivityClock() {
                     generator.onEvent(next, Long.MIN_VALUE, output);
                     generator.onPeriodicEmit(output);
                 }
+
+                // If enabled, throttle emission of values
+                if (sleepAfterElements > 0
+                        && sleepTimeMillis > 0
+                        && numElementsEmitted % sleepAfterElements == 0) {
+                    Thread.sleep(sleepTimeMillis);
+                }
             }
 
             if (terminating == TerminatingLogic.INFINITE) {

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java

@@ -716,7 +716,9 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                                     partitions,
                                     readableMetadata,
                                     null,
-                                    enableAggregatePushDown);
+                                    enableAggregatePushDown,
+                                    sleepAfterElements,
+                                    sleepTimeMillis);
                     source.setEnableMetadataFilterPushDown(enableMetadataFilterPushDown);
                     return source;
                 } else {
@@ -1750,6 +1752,8 @@ private static class TestValuesScanTableSourceWithWatermarkPushDown
             extends TestValuesScanTableSource
             implements SupportsWatermarkPushDown, SupportsSourceWatermark {
         private final String tableName;
+        private final int sleepAfterElements;
+        private final long sleepTimeMillis;
 
         private WatermarkStrategy<RowData> watermarkStrategy = WatermarkStrategy.noWatermarks();
 
@@ -1771,7 +1775,9 @@ private TestValuesScanTableSourceWithWatermarkPushDown(
                 List<Map<String, String>> allPartitions,
                 Map<String, DataType> readableMetadata,
                 @Nullable int[] projectedMetadataFields,
-                boolean enableAggregatePushDown) {
+                boolean enableAggregatePushDown,
+                int sleepAfterElements,
+                long sleepTimeMillis) {
             super(
                     producedDataType,
                     changelogMode,
@@ -1792,6 +1798,8 @@ private TestValuesScanTableSourceWithWatermarkPushDown(
                     projectedMetadataFields,
                     enableAggregatePushDown);
             this.tableName = tableName;
+            this.sleepAfterElements = sleepAfterElements;
+            this.sleepTimeMillis = sleepTimeMillis;
         }
 
         @Override
@@ -1817,7 +1825,13 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderCon
             try {
                 return SourceFunctionProvider.of(
                         new TestValuesRuntimeFunctions.FromElementSourceFunctionWithWatermark(
-                                tableName, serializer, values, watermarkStrategy, terminating),
+                                tableName,
+                                serializer,
+                                values,
+                                watermarkStrategy,
+                                terminating,
+                                sleepAfterElements,
+                                sleepTimeMillis),
                         false);
             } catch (IOException e) {
                 throw new TableException("Fail to init source function", e);
@@ -1845,7 +1859,9 @@ public DynamicTableSource copy() {
                             allPartitions,
                             readableMetadata,
                             projectedMetadataFields,
-                            enableAggregatePushDown);
+                            enableAggregatePushDown,
+                            sleepAfterElements,
+                            sleepTimeMillis);
             newSource.watermarkStrategy = watermarkStrategy;
             newSource.setEnableMetadataFilterPushDown(enableMetadataFilterPushDown);
             return newSource;

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinRestoreTest.java

@@ -0,0 +1,40 @@
+/*
+ * 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.stream;
+
+import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase;
+import org.apache.flink.table.test.program.TableTestProgram;
+
+import java.util.Arrays;
+import java.util.List;
+
+/** Restore tests for {@link StreamExecLateralSnapshotJoin}. */
+public class LateralSnapshotJoinRestoreTest extends RestoreTestBase {
+
+    public LateralSnapshotJoinRestoreTest() {
+        super(StreamExecLateralSnapshotJoin.class);
+    }
+
+    @Override
+    public List<TableTestProgram> programs() {
+        return Arrays.asList(
+                LateralSnapshotJoinTestPrograms.LATERAL_SNAPSHOT_JOIN_INNER,
+                LateralSnapshotJoinTestPrograms.LATERAL_SNAPSHOT_JOIN_LEFT);
+    }
+}

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java

@@ -0,0 +1,145 @@
+/*
+ * 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.stream;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+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 testing {@link StreamExecLateralSnapshotJoin}.
+ *
+ * <p>Both programs load the whole build side before the savepoint: the {@code 'user_time'} gate at
+ * {@code 00:00:03} is reached by the last build-side row, so the operator has flipped to the JOIN
+ * phase and its build-side snapshot is materialized and frozen when the stop-with-savepoint fires
+ * (after the "before restore" rows have been joined). After restore, no further build changes
+ * arrive; the "after restore" probe rows join the restored snapshot, which verifies that the
+ * materialized build state and the LOAD/JOIN phase (union operator state) survive the savepoint.
+ */
+public class LateralSnapshotJoinTestPrograms {
+
+    static final String[] PROBE_SCHEMA = {
+        "pk STRING",
+        "pv INT",
+        "pts_str STRING",
+        "pts AS TO_TIMESTAMP(pts_str)",
+        "WATERMARK FOR pts AS pts"
+    };
+
+    static final String[] BUILD_SCHEMA = {
+        "bk STRING",
+        "bv INT",
+        "bts_str STRING",
+        "bts AS TO_TIMESTAMP(bts_str)",
+        "WATERMARK FOR bts AS bts"
+    };
+
+    static final String[] SINK_SCHEMA = {"pk STRING", "pv INT", "bk STRING", "bv INT"};
+
+    // Two rows for key 'a' exercise the per-key multi-set; the last row's watermark (00:00:03)
+    // reaches the gate and flips the operator to JOIN.
+    static final Row[] BUILD_BEFORE_DATA = {
+        Row.of("a", 10, "2020-01-01 00:00:01"),
+        Row.of("b", 20, "2020-01-01 00:00:02"),
+        Row.of("a", 11, "2020-01-01 00:00:03")
+    };
+
+    static final Row[] PROBE_BEFORE_DATA = {
+        Row.of("a", 100, "2020-01-01 00:00:06"), Row.of("b", 200, "2020-01-01 00:00:07")
+    };
+
+    // 'a' matches the restored snapshot; 'c' has no match.
+    static final Row[] PROBE_AFTER_DATA = {
+        Row.of("a", 101, "2020-01-01 00:00:10"), Row.of("c", 300, "2020-01-01 00:00:11")
+    };
+
+    private static final String SNAPSHOT_BUILD =
+            "LATERAL TABLE(SNAPSHOT("
+                    + "input => TABLE b, "
+                    + "load_completed_condition => 'user_time', "
+                    + "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:03' AS TIMESTAMP_LTZ(3))"
+                    + ")) AS s ON probe.pk = s.bk";
+
+    public static final TableTestProgram LATERAL_SNAPSHOT_JOIN_INNER =
+            TableTestProgram.of(
+                            "lateral-snapshot-join-inner",
+                            "validates a LATERAL SNAPSHOT inner join across a restore")
+                    .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("probe")
+                                    .addSchema(PROBE_SCHEMA)
+                                    .producedBeforeRestore(PROBE_BEFORE_DATA)
+                                    .producedAfterRestore(PROBE_AFTER_DATA)
+                                    .build())
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("b")
+                                    .addSchema(BUILD_SCHEMA)
+                                    .producedBeforeRestore(BUILD_BEFORE_DATA)
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink")
+                                    .addSchema(SINK_SCHEMA)
+                                    .consumedBeforeRestore(
+                                            "+I[a, 100, a, 10]",
+                                            "+I[a, 100, a, 11]",
+                                            "+I[b, 200, b, 20]")
+                                    .consumedAfterRestore("+I[a, 101, a, 10]", "+I[a, 101, a, 11]")
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink SELECT probe.pk, probe.pv, s.bk, s.bv "
+                                    + "FROM probe JOIN "
+                                    + SNAPSHOT_BUILD)
+                    .build();
+
+    public static final TableTestProgram LATERAL_SNAPSHOT_JOIN_LEFT =
+            TableTestProgram.of(
+                            "lateral-snapshot-join-left",
+                            "validates a LATERAL SNAPSHOT left join across a restore")
+                    .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("probe")
+                                    .addSchema(PROBE_SCHEMA)
+                                    .producedBeforeRestore(PROBE_BEFORE_DATA)
+                                    .producedAfterRestore(PROBE_AFTER_DATA)
+                                    .build())
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("b")
+                                    .addSchema(BUILD_SCHEMA)
+                                    .producedBeforeRestore(BUILD_BEFORE_DATA)
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink")
+                                    .addSchema(SINK_SCHEMA)
+                                    .consumedBeforeRestore(
+                                            "+I[a, 100, a, 10]",
+                                            "+I[a, 100, a, 11]",
+                                            "+I[b, 200, b, 20]")
+                                    .consumedAfterRestore(
+                                            "+I[a, 101, a, 10]",
+                                            "+I[a, 101, a, 11]",
+                                            "+I[c, 300, null, null]")
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink SELECT probe.pk, probe.pv, s.bk, s.bv "
+                                    + "FROM probe LEFT JOIN "
+                                    + SNAPSHOT_BUILD)
+                    .build();
+}

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestCompleteness.java

@@ -19,7 +19,6 @@
 package org.apache.flink.table.planner.plan.nodes.exec.testutils;
 
 import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
-import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecLateralSnapshotJoin;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecPythonAsyncCalc;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecPythonCalc;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecPythonCorrelate;
@@ -50,11 +49,6 @@ public class RestoreTestCompleteness {
     private static final Set<Class<? extends ExecNode<?>>> SKIP_EXEC_NODES =
             new HashSet<Class<? extends ExecNode<?>>>() {
                 {
-                    // TODO: FLINK-39781 - the LATERAL SNAPSHOT runtime operator is still a stub,
-                    //   so a restore test cannot generate a savepoint yet. Remove this entry and
-                    //   add LateralSnapshotJoinRestoreTest once the operator is implemented.
-                    add(StreamExecLateralSnapshotJoin.class);
-
                     /** Ignoring python based exec nodes temporarily. */
                     add(StreamExecPythonCalc.class);
                     add(StreamExecPythonCorrelate.class);