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
commit a77d210c99675043794aff6f5fe335e23557fa9f Author: Fabian Hueske <[email protected]> AuthorDate: Thu Jul 9 12:19:23 2026 +0200 [FLINK-39785][table] Add LATERAL SNAPSHOT e2e and restore tests Adds end-to-end coverage for the LATERAL SNAPSHOT processing-time temporal join: * LateralSnapshotJoinSemanticTests: Semantic tests LATERAL SNAPSHOT join * LateralSnapshotJoinITCase: non-deterministic result tests and tests over HEAP and ROCKSDB backends * LateralSnapshotJoinRestoreTest / LateralSnapshotJoinTestPrograms: savepoint restore tests * Configure UTC as local timezone for CommonSemanticTestBase and RestoreTestBase Generated-By: Claude Opus 4.8 (1M context) --- .../stream/LateralSnapshotJoinRestoreTest.java | 67 +++ .../LateralSnapshotJoinSemanticTestPrograms.java | 439 +++++++++++++++++++ .../stream/LateralSnapshotJoinSemanticTests.java | 43 ++ .../stream/LateralSnapshotJoinTestPrograms.java | 176 ++++++++ .../exec/testutils/CommonSemanticTestBase.java | 2 + .../plan/nodes/exec/testutils/RestoreTestBase.java | 2 + .../exec/testutils/RestoreTestCompleteness.java | 6 - .../stream/sql/join/LateralSnapshotJoinITCase.java | 290 +++++++++++++ .../plan/lateral-snapshot-join-inner.json | 469 +++++++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 19149 bytes .../plan/lateral-snapshot-join-load-phase.json | 468 ++++++++++++++++++++ .../savepoint/_metadata | Bin 0 -> 18950 bytes 12 files changed, 1956 insertions(+), 6 deletions(-) diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinRestoreTest.java new file mode 100644 index 00000000000..806ecf72ef3 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinRestoreTest.java @@ -0,0 +1,67 @@ +/* + * 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.factories.TestValuesTableFactory; +import org.apache.flink.table.planner.plan.nodes.exec.testutils.RestoreTestBase; +import org.apache.flink.table.test.program.SourceTestStep; +import org.apache.flink.table.test.program.TableTestProgram; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +/** Restore tests for {@link StreamExecLateralSnapshotJoin}. */ +public class LateralSnapshotJoinRestoreTest extends RestoreTestBase { + + // Bounds the wait for the LOAD-phase savepoint trigger so a stuck source fails fast. + private static final long SAVEPOINT_READY_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(5); + + public LateralSnapshotJoinRestoreTest() { + super(StreamExecLateralSnapshotJoin.class); + } + + @Override + public List<TableTestProgram> programs() { + return Arrays.asList( + LateralSnapshotJoinTestPrograms.LATERAL_SNAPSHOT_JOIN_PHASE_LOAD, + LateralSnapshotJoinTestPrograms.LATERAL_SNAPSHOT_JOIN_PHASE_JOIN); + } + + @Override + protected void awaitSavepointReady(TableTestProgram program, List<CompletableFuture<?>> futures) + throws Exception { + if (program != LateralSnapshotJoinTestPrograms.LATERAL_SNAPSHOT_JOIN_PHASE_LOAD) { + super.awaitSavepointReady(program, futures); + return; + } + // The operator emits nothing during its LOAD phase, so we can't use the default sink-based + // trigger. Instead, we gate on the number of records emitted by the sources. Once every + // source has emitted its before-restore rows, stop-with-savepoint drains them into keyed + // state before snapshotting. + for (SourceTestStep source : program.getSetupSourceTestSteps()) { + final int count = source.dataBeforeRestore.size(); + if (count > 0) { + TestValuesTableFactory.awaitSourceEmitted(source.name, count) + .get(SAVEPOINT_READY_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + } + } + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java new file mode 100644 index 00000000000..ca2f3696938 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java @@ -0,0 +1,439 @@ +/* + * 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.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; +import org.apache.flink.types.RowKind; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +/** + * Deterministic result {@link TableTestProgram} definitions for the {@code LATERAL SNAPSHOT} + * processing-time temporal join ({@link StreamExecLateralSnapshotJoin}). + * + * <p>Processing-time semantics make the join non-deterministic in general. To get stable results, + * each build source appends a non-matching "flip-trigger" row at the flip timestamp ({@link + * #FLIP_TRIGGER_TS}) while all real build rows are earlier. With per-record ({@code on-event}) + * watermarks the operator flips to the JOIN phase. Some programs throttle source emission ({@code + * source.sleep-*}) to place probes deterministically around the flip. + */ +public class LateralSnapshotJoinSemanticTestPrograms { + + /** The {@code 'user_time'} condition reached mid-stream by the build-side flip-trigger row. */ + private static final String MID_FLIP = + "load_completed_condition => 'user_time', " + + "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3))"; + + /** A far-future flip condition: the flip happens only at end of all input. */ + private static final String END_FLIP = + "load_completed_condition => 'user_time', " + + "load_completed_time => CAST(TIMESTAMP '2100-01-01 00:00:00' AS TIMESTAMP_LTZ(3))"; + + /** Event time of the flip-trigger row; equal to the {@link #MID_FLIP} timestamp. */ + private static final String FLIP_TRIGGER_TS = "00:00:10"; + + /** A build-side key that never matches any probe row. */ + private static final String FLIP_TRIGGER_KEY = "__flip_trigger__"; + + private static final String[] PROBE_SCHEMA = { + "pk STRING", "pv INT", "pts TIMESTAMP(3)", "WATERMARK FOR pts AS pts" + }; + + private static final String[] BUILD_SCHEMA = { + "bk STRING", "bv INT", "bts TIMESTAMP(3)", "WATERMARK FOR bts AS bts" + }; + + // ------------------------------------------------------------------------------------------ + // Core join semantics + // ------------------------------------------------------------------------------------------ + + public static final TableTestProgram INNER_JOIN = + TableTestProgram.of("lateral-snapshot-inner-join", "LATERAL SNAPSHOT inner join") + // Throttle the probe so the fast build flips first; the probes are then joined + // live in the JOIN phase (the build-side is finalized, so buffered-vs-live does + // not change the result). + .setupTableSource(throttledProbe(defaultProbe(), 40L)) + .setupTableSource(appendBuild(withFlipTrigger(defaultBuild()))) + .setupTableSink( + keyValueSink() + .consumedValues( + "+I[a, 100, a, 10]", + "+I[a, 100, a, 11]", + "+I[b, 200, b, 20]") + .build()) + .runSql( + innerJoin( + "probe.pk, probe.pv, s.bk, s.bv", MID_FLIP, "probe.pk = s.bk")) + .build(); + + public static final TableTestProgram LEFT_JOIN = + TableTestProgram.of("lateral-snapshot-left-join", "LATERAL SNAPSHOT left join") + .setupTableSource(throttledProbe(defaultProbe(), 40L)) + .setupTableSource(appendBuild(withFlipTrigger(defaultBuild()))) + .setupTableSink( + keyValueSink() + .consumedValues( + "+I[a, 100, a, 10]", + "+I[a, 100, a, 11]", + "+I[b, 200, b, 20]", + "+I[c, 300, null, null]") + .build()) + .runSql(leftJoin("probe.pk, probe.pv, s.bk, s.bv", MID_FLIP, "probe.pk = s.bk")) + .build(); + + public static final TableTestProgram SELECT_STAR = + TableTestProgram.of( + "lateral-snapshot-select-star", + "SELECT * materializes the build-side rowtime as a regular TIMESTAMP") + .setupTableSource(probe(List.of(Row.of("a", 100, ts("00:01:00"))))) + .setupTableSource( + appendBuild(withFlipTrigger(List.of(Row.of("a", 10, ts("00:00:01")))))) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema( + "pk STRING", + "pv INT", + "pts TIMESTAMP(3)", + "bk STRING", + "bv INT", + "bts TIMESTAMP(3)") + .testMaterializedData() + .consumedValues( + "+I[a, 100, 2020-01-01T00:01, a, 10, 2020-01-01T00:00:01]") + .build()) + .runSql( + "INSERT INTO sink SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_FLIP + + ")) AS s ON probe.pk = s.bk") + .build(); + + public static final TableTestProgram COMPOSITE_KEYS = + TableTestProgram.of("lateral-snapshot-composite-keys", "join on composite keys") + .setupTableSource( + probe( + Arrays.asList( + Row.of("a", 10, ts("00:01:00")), + Row.of("a", 11, ts("00:01:01")), + Row.of("z", 10, ts("00:01:01")), + Row.of("b", 20, ts("00:01:02"))))) + .setupTableSource( + appendBuild( + withFlipTrigger( + Arrays.asList( + Row.of("a", 10, ts("00:00:01")), + Row.of("a", 99, ts("00:00:02")), + Row.of("b", 20, ts("00:00:03")))))) + .setupTableSink( + keyValueSink() + .consumedValues("+I[a, 10, a, 10]", "+I[b, 20, b, 20]") + .build()) + .runSql( + innerJoin( + "probe.pk, probe.pv, s.bk, s.bv", + MID_FLIP, + "probe.pk = s.bk AND probe.pv = s.bv")) + .build(); + + public static final TableTestProgram NON_EQUI = + TableTestProgram.of("lateral-snapshot-non-equi", "join with a non-equi condition") + .setupTableSource( + probe( + Arrays.asList( + Row.of("a", 15, ts("00:01:00")), + Row.of("a", 5, ts("00:01:01"))))) + .setupTableSource( + appendBuild( + withFlipTrigger( + Arrays.asList( + Row.of("a", 10, ts("00:00:01")), + Row.of("a", 20, ts("00:00:02")))))) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("pk STRING", "pv INT", "bv INT") + .testMaterializedData() + .consumedValues("+I[a, 15, 10]") + .build()) + .runSql( + innerJoin( + "probe.pk, probe.pv, s.bv", + MID_FLIP, + "probe.pk = s.bk AND probe.pv > s.bv")) + .build(); + + public static final TableTestProgram EMPTY_BUILD_INNER = + TableTestProgram.of( + "lateral-snapshot-empty-build-inner", + "inner join over an empty build side yields no rows") + .setupTableSource(probe(defaultProbe())) + .setupTableSource(appendBuild(withFlipTrigger(List.of()))) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("pk STRING", "bk STRING") + .testMaterializedData() + .consumedValues(new String[0]) + .build()) + .runSql(innerJoin("probe.pk, s.bk", MID_FLIP, "probe.pk = s.bk")) + .build(); + + public static final TableTestProgram EMPTY_BUILD_LEFT = + TableTestProgram.of( + "lateral-snapshot-empty-build-left", + "left join over an empty build side preserves the probe rows") + .setupTableSource(probe(defaultProbe())) + .setupTableSource(appendBuild(withFlipTrigger(List.of()))) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("pk STRING", "bk STRING") + .testMaterializedData() + .consumedValues("+I[a, null]", "+I[b, null]", "+I[c, null]") + .build()) + .runSql(leftJoin("probe.pk, s.bk", MID_FLIP, "probe.pk = s.bk")) + .build(); + + // ------------------------------------------------------------------------------------------ + // Flip timing + // ------------------------------------------------------------------------------------------ + + public static final TableTestProgram FLIP_AT_END = + TableTestProgram.of( + "lateral-snapshot-flip-at-end", + "far-future flip condition: the operator flips only at end-of-input") + .setupTableSource(probe(defaultProbe())) + .setupTableSource(appendBuild(withFlipTrigger(defaultBuild()))) + .setupTableSink( + keyValueSink() + .consumedValues( + "+I[a, 100, a, 10]", + "+I[a, 100, a, 11]", + "+I[b, 200, b, 20]") + .build()) + .runSql( + innerJoin( + "probe.pk, probe.pv, s.bk, s.bv", END_FLIP, "probe.pk = s.bk")) + .build(); + + public static final TableTestProgram DEFAULT_COMPILE_TIME = + TableTestProgram.of( + "lateral-snapshot-default-compile-time", + "default 'compile_time' condition flips at end-of-input for 2020 data") + .setupTableSource(probe(defaultProbe())) + .setupTableSource(appendBuild(withFlipTrigger(defaultBuild()))) + .setupTableSink( + keyValueSink() + .consumedValues( + "+I[a, 100, a, 10]", + "+I[a, 100, a, 11]", + "+I[b, 200, b, 20]") + .build()) + // No options: 'load_completed_condition' defaults to 'compile_time'. + .runSql( + "INSERT INTO sink SELECT probe.pk, probe.pv, s.bk, s.bv " + + "FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b" + + ")) AS s ON probe.pk = s.bk") + .build(); + + public static final TableTestProgram LIVE_JOIN = + TableTestProgram.of( + "lateral-snapshot-live-join", + "probes are joined live per-record in the JOIN phase") + // Fast build flips almost immediately; the throttled probe stream is consumed + // entirely in the JOIN phase (including non-matching keys c). The snapshot is + // static after the flip, so the result is deterministic. + .setupTableSource( + throttledProbe( + Arrays.asList( + Row.of("a", 1, ts("00:00:01")), + Row.of("b", 2, ts("00:00:02")), + Row.of("c", 3, ts("00:00:03")), + Row.of("a", 4, ts("00:00:04")), + Row.of("b", 5, ts("00:00:05")), + Row.of("c", 6, ts("00:00:06"))), + 40L)) + .setupTableSource( + appendBuild( + withFlipTrigger( + Arrays.asList( + Row.of("a", 10, ts("00:00:01")), + Row.of("b", 20, ts("00:00:02")))))) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("pv INT", "pk STRING", "bv INT") + .testMaterializedData() + .consumedValues( + "+I[1, a, 10]", + "+I[4, a, 10]", + "+I[2, b, 20]", + "+I[5, b, 20]") + .build()) + .runSql(innerJoin("probe.pv, probe.pk, s.bv", MID_FLIP, "probe.pk = s.bk")) + .build(); + + public static final TableTestProgram BUFFERED_THEN_DRAINED = + TableTestProgram.of( + "lateral-snapshot-buffered-then-drained", + "probes buffered during LOAD are drained against the final version") + // Un-throttled: buffered within a few ms, long before the flip. + .setupTableSource(probe(manyProbes(20))) + // Throttled build flips at ~240 ms, by which time the probe stream is fully + // buffered in LOAD; the flip drains all buffered probes against version 3. + .setupTableSource( + throttledUpsertBuild( + Arrays.asList( + Row.ofKind(RowKind.INSERT, "k", 1, ts("00:00:01")), + Row.ofKind( + RowKind.UPDATE_AFTER, "k", 2, ts("00:00:02")), + Row.ofKind( + RowKind.UPDATE_AFTER, "k", 3, ts("00:00:05")), + Row.ofKind( + RowKind.INSERT, + FLIP_TRIGGER_KEY, + 0, + ts(FLIP_TRIGGER_TS))), + 80L)) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema("pv INT", "bv INT") + .testMaterializedData() + .consumedValues( + IntStream.rangeClosed(1, 20) + .mapToObj(i -> String.format("+I[%d, 3]", i)) + .toArray(String[]::new)) + .build()) + .runSql(innerJoin("probe.pv, s.bv", MID_FLIP, "probe.pk = s.bk")) + .build(); + + // ------------------------------------------------------------------------------------------ + // Helpers + // ------------------------------------------------------------------------------------------ + + private static String innerJoin(String projection, String flip, String condition) { + return join("JOIN", projection, flip, condition); + } + + private static String leftJoin(String projection, String flip, String condition) { + return join("LEFT JOIN", projection, flip, condition); + } + + private static String join(String joinType, String projection, String flip, String condition) { + return "INSERT INTO sink SELECT " + + projection + + " FROM probe " + + joinType + + " LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + flip + + ")) AS s ON " + + condition; + } + + private static List<Row> defaultProbe() { + return Arrays.asList( + Row.of("a", 100, ts("00:01:00")), + Row.of("b", 200, ts("00:01:01")), + Row.of("c", 300, ts("00:01:02"))); + } + + private static List<Row> defaultBuild() { + return Arrays.asList( + Row.of("a", 10, ts("00:00:01")), + Row.of("b", 20, ts("00:00:02")), + Row.of("a", 11, ts("00:00:03"))); + } + + private static List<Row> manyProbes(int count) { + return IntStream.rangeClosed(1, count) + .mapToObj(i -> Row.of("k", i, ts(String.format("00:00:%02d", i)))) + .collect(Collectors.toList()); + } + + /** + * Appends a non-matching build row at the {@link #FLIP_TRIGGER_TS} timestamp; its watermark + * flips the operator to the JOIN phase mid-stream (all real build rows are earlier). + */ + private static List<Row> withFlipTrigger(List<Row> data) { + final List<Row> withTrigger = new ArrayList<>(data); + withTrigger.add(Row.of(FLIP_TRIGGER_KEY, 0, ts(FLIP_TRIGGER_TS))); + return withTrigger; + } + + private static SourceTestStep probe(List<Row> data) { + return SourceTestStep.newBuilder("probe") + .addSchema(PROBE_SCHEMA) + .producedValues(data.toArray(new Row[0])) + .build(); + } + + private static SourceTestStep throttledProbe(List<Row> data, long sleepMillis) { + return SourceTestStep.newBuilder("probe") + .addSchema(PROBE_SCHEMA) + .addOptions(throttleOptions(sleepMillis)) + .producedValues(data.toArray(new Row[0])) + .build(); + } + + private static SourceTestStep appendBuild(List<Row> data) { + return SourceTestStep.newBuilder("b") + .addSchema(BUILD_SCHEMA) + .producedValues(data.toArray(new Row[0])) + .build(); + } + + private static SourceTestStep throttledUpsertBuild(List<Row> data, long sleepMillis) { + return SourceTestStep.newBuilder("b") + .addSchema( + "bk STRING", + "bv INT", + "bts TIMESTAMP(3)", + "WATERMARK FOR bts AS bts", + "PRIMARY KEY (bk) NOT ENFORCED") + .addOption("changelog-mode", "I,UA,D") + .addOptions(throttleOptions(sleepMillis)) + .producedValues(data.toArray(new Row[0])) + .build(); + } + + private static SinkTestStep.Builder keyValueSink() { + return SinkTestStep.newBuilder("sink") + .addSchema("pk STRING", "pv INT", "bk STRING", "bv INT") + .testMaterializedData(); + } + + private static Map<String, String> throttleOptions(long sleepMillis) { + final Map<String, String> options = new HashMap<>(); + options.put("source.sleep-after-elements", "1"); + options.put("source.sleep-time", sleepMillis + "ms"); + return options; + } + + private static LocalDateTime ts(String time) { + return LocalDateTime.parse("2020-01-01T" + time); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java new file mode 100644 index 00000000000..6536f34a591 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java @@ -0,0 +1,43 @@ +/* + * 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.SemanticTestBase; +import org.apache.flink.table.test.program.TableTestProgram; + +import java.util.List; + +/** Semantic tests for {@link StreamExecLateralSnapshotJoin}. */ +public class LateralSnapshotJoinSemanticTests extends SemanticTestBase { + @Override + public List<TableTestProgram> programs() { + return List.of( + LateralSnapshotJoinSemanticTestPrograms.INNER_JOIN, + LateralSnapshotJoinSemanticTestPrograms.LEFT_JOIN, + LateralSnapshotJoinSemanticTestPrograms.SELECT_STAR, + LateralSnapshotJoinSemanticTestPrograms.COMPOSITE_KEYS, + LateralSnapshotJoinSemanticTestPrograms.NON_EQUI, + LateralSnapshotJoinSemanticTestPrograms.EMPTY_BUILD_INNER, + LateralSnapshotJoinSemanticTestPrograms.EMPTY_BUILD_LEFT, + LateralSnapshotJoinSemanticTestPrograms.FLIP_AT_END, + LateralSnapshotJoinSemanticTestPrograms.DEFAULT_COMPILE_TIME, + LateralSnapshotJoinSemanticTestPrograms.LIVE_JOIN, + LateralSnapshotJoinSemanticTestPrograms.BUFFERED_THEN_DRAINED); + } +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java new file mode 100644 index 00000000000..b1c58be353b --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinTestPrograms.java @@ -0,0 +1,176 @@ +/* + * 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.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>The programs cover a savepoint taken in each of the operator's two phases; the {@code + * 'user_time'} gate is at {@code 00:00:03} in both. + * + * <ul> + * <li>{@link #LATERAL_SNAPSHOT_JOIN_PHASE_LOAD}: the transition to JOIN is not triggered before + * the savepoint, so the operator is still in LOAD; the savepoint captures the partial build + * multi-set and the buffered probe row. The flip to JOIN phase happens only after restore. + * <li>{@link #LATERAL_SNAPSHOT_JOIN_PHASE_JOIN}: the whole build side is loaded before the + * savepoint (the last build-side row triggers the flip), so the operator has flipped to JOIN + * and its build-side snapshot is materialized and frozen when the stop-with-savepoint fires. + * </ul> + * + * <p>Both verify that the build state and the LOAD/JOIN phase (union operator state) survive the + * savepoint: after restore the "after restore" probe rows join the restored snapshot. + */ +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") + }; + + // LOAD-phase restore scenario: none of these build rows reaches the 00:00:03 gate, so the + // operator is still in LOAD when the savepoint is taken. The number of rows is the count the + // savepoint trigger waits for (see LateralSnapshotJoinRestoreTest#awaitSavepointReady). + static final Row[] LOAD_BUILD_BEFORE_DATA = { + Row.of("a", 10, "2020-01-01 00:00:01"), Row.of("a", 11, "2020-01-01 00:00:02") + }; + + // The gate row (00:00:03) arrives only after restore and flips the operator to JOIN, draining + // the probe rows buffered before the savepoint. + static final Row[] LOAD_BUILD_AFTER_DATA = {Row.of("a", 12, "2020-01-01 00:00:03")}; + + // Buffered during LOAD (LOAD emits nothing), so it survives the savepoint in probe state. + static final Row[] LOAD_PROBE_BEFORE_DATA = {Row.of("a", 100, "2020-01-01 00:00:06")}; + + static final Row[] LOAD_PROBE_AFTER_DATA = {Row.of("a", 101, "2020-01-01 00:00:10")}; + + 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"; + + // Restore taken while the operator is in LOAD phase: the savepoint captures the partial build + // multi-set and the buffered probe row, the LOAD phase is recorded in union operator state, and + // nothing has been emitted yet. After restore the 00:00:03 build row completes the load, flips + // to JOIN, and the buffered + after-restore probe rows join the full snapshot {10, 11, 12} for + // key 'a'. + public static final TableTestProgram LATERAL_SNAPSHOT_JOIN_PHASE_LOAD = + TableTestProgram.of( + "lateral-snapshot-join-load-phase", + "validates a LATERAL SNAPSHOT inner join restored from a LOAD-phase savepoint") + .setupTableSource( + SourceTestStep.newBuilder("probe") + .addSchema(PROBE_SCHEMA) + .producedBeforeRestore(LOAD_PROBE_BEFORE_DATA) + .producedAfterRestore(LOAD_PROBE_AFTER_DATA) + .build()) + .setupTableSource( + SourceTestStep.newBuilder("b") + .addSchema(BUILD_SCHEMA) + .producedBeforeRestore(LOAD_BUILD_BEFORE_DATA) + .producedAfterRestore(LOAD_BUILD_AFTER_DATA) + .build()) + .setupTableSink( + SinkTestStep.newBuilder("sink") + .addSchema(SINK_SCHEMA) + // Empty before restore: LOAD phase emits nothing before the + // savepoint. The explicit String[] picks the String overload + // and marks this a SINK_WITH_RESTORE_DATA step. + .consumedBeforeRestore(new String[0]) + .consumedAfterRestore( + "+I[a, 100, a, 10]", + "+I[a, 100, a, 11]", + "+I[a, 100, a, 12]", + "+I[a, 101, a, 10]", + "+I[a, 101, a, 11]", + "+I[a, 101, a, 12]") + .build()) + .runSql( + "INSERT INTO sink SELECT probe.pk, probe.pv, s.bk, s.bv " + + "FROM probe JOIN " + + SNAPSHOT_BUILD) + .build(); + + // Restore taken while the operator is in JOIN phase + public static final TableTestProgram LATERAL_SNAPSHOT_JOIN_PHASE_JOIN = + TableTestProgram.of( + "lateral-snapshot-join-inner", + "validates a LATERAL SNAPSHOT inner join across a restore") + .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(); +} diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/CommonSemanticTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/CommonSemanticTestBase.java index f406cd972e3..2a4d9e67d36 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/CommonSemanticTestBase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/CommonSemanticTestBase.java @@ -22,6 +22,7 @@ import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.api.TableEnvironment; import org.apache.flink.table.api.config.OptimizerConfigOptions; +import org.apache.flink.table.api.config.TableConfigOptions; import org.apache.flink.table.planner.factories.TestValuesModelFactory; import org.apache.flink.table.planner.factories.TestValuesTableFactory; import org.apache.flink.table.test.program.ConfigOptionTestStep; @@ -192,6 +193,7 @@ public abstract class CommonSemanticTestBase implements TableTestProgramRunner { config.set( OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY, OptimizerConfigOptions.NonDeterministicUpdateStrategy.TRY_RESOLVE); + config.set(TableConfigOptions.LOCAL_TIME_ZONE, "UTC"); } private Map<String, String> createSourceOptions(SourceTestStep sourceTestStep, String id) { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java index 83f910762dc..dc61bbe030b 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java @@ -352,6 +352,7 @@ public abstract class RestoreTestBase implements TableTestProgramRunner { .set( TableConfigOptions.PLAN_COMPILE_CATALOG_OBJECTS, TableConfigOptions.CatalogPlanCompilation.SCHEMA); + tEnv.getConfig().set(TableConfigOptions.LOCAL_TIME_ZONE, "UTC"); for (SourceTestStep sourceTestStep : program.getSetupSourceTestSteps()) { final String id = TestValuesTableFactory.registerData(sourceTestStep.dataBeforeRestore); @@ -427,6 +428,7 @@ public abstract class RestoreTestBase implements TableTestProgramRunner { .set( TableConfigOptions.PLAN_RESTORE_CATALOG_OBJECTS, TableConfigOptions.CatalogPlanRestore.IDENTIFIER); + tEnv.getConfig().set(TableConfigOptions.LOCAL_TIME_ZONE, "UTC"); program.getSetupConfigOptionTestSteps().forEach(s -> s.apply(tEnv)); diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestCompleteness.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestCompleteness.java index 29d9f0db3b9..ea183bf3dba 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestCompleteness.java +++ b/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); diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java new file mode 100644 index 00000000000..4972d4cb28c --- /dev/null +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java @@ -0,0 +1,290 @@ +/* + * 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.stream.sql.join; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableDescriptor; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase; +import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension; +import org.apache.flink.testutils.junit.extensions.parameterized.Parameters; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; +import org.apache.flink.util.CollectionUtil; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThatList; + +/** + * Result tests for the {@code LATERAL SNAPSHOT} processing-time temporal join that are + * non-deterministic or benefit from dual-backend (HEAP/ROCKSDB) coverage. Deterministic, + * backend-agnostic cases live in {@code LateralSnapshotJoinSemanticTests}; time-driven behavior + * (idle-timeout flip, state-TTL eviction) in {@code LateralSnapshotJoinOperatorTest}. + * + * <p>To stabilize results, each build source appends a non-matching "flip-trigger" row at {@link + * #MID_FLIP} (after all real rows); its per-record watermark flips the operator to the JOIN phase. + */ +@ExtendWith(ParameterizedTestExtension.class) +public class LateralSnapshotJoinITCase extends StreamingWithStateTestBase { + + /** The {@code 'user_time'} condition reached mid-stream by the build-side flip-trigger row. */ + private static final String MID_FLIP = + "load_completed_condition => 'user_time', " + + "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3))"; + + /** Event time of the flip-trigger row; equal to the {@link #MID_FLIP} timestamp. */ + private static final String FLIP_TRIGGER_TS = "00:00:10"; + + /** A build-side key that never matches any probe row. */ + private static final String FLIP_TRIGGER_KEY = "__flip_trigger__"; + + public LateralSnapshotJoinITCase(StateBackendMode state) { + super(state); + } + + @BeforeEach + @Override + public void before() { + super.before(); + env().setParallelism(1); + tEnv().getConfig().setLocalTimeZone(ZoneId.of("UTC")); + } + + @Parameters(name = "StateBackend={0}") + public static Collection<Object[]> parameters() { + return Arrays.asList( + new Object[][] { + {StreamingWithStateTestBase.HEAP_BACKEND()}, + {StreamingWithStateTestBase.ROCKSDB_BACKEND()} + }); + } + + // ------------------------------------------------------------------------------------------ + // Build-side changelog consolidation (all changes have rowtime < the flip ts) + // ------------------------------------------------------------------------------------------ + + @TestTemplate + void testRetractingBuildChangelog() { + createProbe( + Arrays.asList(Row.of("a", 100, ts("00:01:00")), Row.of("b", 200, ts("00:01:01")))); + // Key a is updated 10 -> 11; key b is inserted then deleted. + createChangelogBuild( + Arrays.asList( + Row.ofKind(RowKind.INSERT, "a", 10, ts("00:00:01")), + Row.ofKind(RowKind.INSERT, "b", 20, ts("00:00:03")), + Row.ofKind(RowKind.UPDATE_BEFORE, "a", 10, ts("00:00:01")), + Row.ofKind(RowKind.UPDATE_AFTER, "a", 11, ts("00:00:02")), + Row.ofKind(RowKind.DELETE, "b", 20, ts("00:00:03")))); + + final List<Row> actual = + collect( + "SELECT probe.pk, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_FLIP + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual).containsExactlyInAnyOrder(Row.of("a", 11)); + } + + @TestTemplate + void testUpsertBuildSource() { + createProbe( + Arrays.asList(Row.of("a", 100, ts("00:01:00")), Row.of("b", 200, ts("00:01:01")))); + // Upsert source (I,UA,D with PK): a is upserted 10 -> 11, b is deleted. + createUpsertBuild( + Arrays.asList( + Row.ofKind(RowKind.INSERT, "a", 10, ts("00:00:01")), + Row.ofKind(RowKind.UPDATE_AFTER, "a", 11, ts("00:00:02")), + Row.ofKind(RowKind.INSERT, "b", 20, ts("00:00:03")), + Row.ofKind(RowKind.DELETE, "b", 20, ts("00:00:04")))); + + final List<Row> actual = + collect( + "SELECT probe.pk, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_FLIP + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual).containsExactlyInAnyOrder(Row.of("a", 11)); + } + + // ------------------------------------------------------------------------------------------ + // Non-deterministic behavior (asserted with tolerant checks) + // ------------------------------------------------------------------------------------------ + + @TestTemplate + void testProbesObserveProgressiveBuildVersions() { + // Throttled build applies v2, v3, v4 progressively after the flip while a throttled probe + // stream spans the progression. Observed versions are non-decreasing; the first probe sees + // v1 and the last sees v4. Exact per-probe versions are not asserted (source timing and + // post-flip visibility latency are not precise enough across environments). + final int probeCount = 8; + final List<Row> probes = + IntStream.range(0, probeCount) + .mapToObj(i -> Row.of("k", i, ts(String.format("00:00:%02d", i)))) + .toList(); + createProbe(probes, 200L); // last probe at ~1400 ms, well past the last post-flip update + createUpsertBuild( + Arrays.asList( + Row.ofKind(RowKind.INSERT, "k", 1, ts("00:00:01")), + Row.ofKind(RowKind.INSERT, FLIP_TRIGGER_KEY, 0, ts(FLIP_TRIGGER_TS)), + Row.ofKind(RowKind.UPDATE_AFTER, "k", 2, ts("00:00:20")), + Row.ofKind(RowKind.UPDATE_AFTER, "k", 3, ts("00:00:30")), + Row.ofKind(RowKind.UPDATE_AFTER, "k", 4, ts("00:00:40"))), + 50L); + + final List<Row> byProbeId = + sortedByProbeId( + collect( + "SELECT probe.pv, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_FLIP + + ")) AS s ON probe.pk = s.bk")); + + assertThat(byProbeId).hasSize(probeCount); + assertMonotonicVersions(byProbeId); + assertThat((Integer) byProbeId.get(0).getField(1)).isEqualTo(1); + assertThat((Integer) byProbeId.get(probeCount - 1).getField(1)).isEqualTo(4); + } + + // ------------------------------------------------------------------------------------------ + // Helpers + // ------------------------------------------------------------------------------------------ + + private List<Row> collect(String query) { + return CollectionUtil.iteratorToList(tEnv().executeSql(query).collect()); + } + + /** + * Appends a non-matching build row at the {@link #MID_FLIP} timestamp; its watermark flips the + * operator to the JOIN phase mid-stream (all real build rows are earlier). + */ + private List<Row> withFlipTrigger(List<Row> data) { + final List<Row> withTrigger = new ArrayList<>(data); + withTrigger.add(Row.of(FLIP_TRIGGER_KEY, 0, ts(FLIP_TRIGGER_TS))); + return withTrigger; + } + + private void createProbe(List<Row> data) { + createProbe(data, 0L); + } + + private void createProbe(List<Row> data, long sleepMillis) { + final String id = TestValuesTableFactory.registerData(data); + final Schema schema = + Schema.newBuilder() + .column("pk", DataTypes.STRING()) + .column("pv", DataTypes.INT()) + .column("pts", DataTypes.TIMESTAMP(3)) + .watermark("pts", "pts") + .build(); + tEnv().createTable("probe", valuesDescriptor(schema, id, sleepMillis).build()); + } + + private void createChangelogBuild(List<Row> data) { + final String id = TestValuesTableFactory.registerData(withFlipTrigger(data)); + final Schema schema = + Schema.newBuilder() + .column("bk", DataTypes.STRING()) + .column("bv", DataTypes.INT()) + .column("bts", DataTypes.TIMESTAMP(3)) + .watermark("bts", "bts") + .build(); + tEnv().createTable( + "b", + valuesDescriptor(schema, id, 0L) + .option("changelog-mode", "I,UB,UA,D") + .build()); + } + + private void createUpsertBuild(List<Row> data) { + createUpsertBuild(withFlipTrigger(data), 0L); + } + + private void createUpsertBuild(List<Row> data, long sleepMillis) { + final String id = TestValuesTableFactory.registerData(data); + final Schema schema = + Schema.newBuilder() + .column("bk", DataTypes.STRING().notNull()) + .column("bv", DataTypes.INT()) + .column("bts", DataTypes.TIMESTAMP(3)) + .watermark("bts", "bts") + .primaryKey("bk") + .build(); + tEnv().createTable( + "b", + valuesDescriptor(schema, id, sleepMillis) + .option("changelog-mode", "I,UA,D") + .build()); + } + + private static TableDescriptor.Builder valuesDescriptor( + Schema schema, String id, long sleepMillis) { + final TableDescriptor.Builder descriptor = + TableDescriptor.forConnector("values") + .schema(schema) + .option("bounded", "false") + .option("disable-lookup", "true") + .option("enable-watermark-push-down", "true") + .option("scan.watermark.emit.strategy", "on-event") + .option("data-id", id); + if (sleepMillis > 0) { + descriptor + .option("source.sleep-after-elements", "1") + .option("source.sleep-time", sleepMillis + "ms"); + } + return descriptor; + } + + /** Sorts join results (pv, bv) by the probe id pv, i.e. into probe processing order. */ + private static List<Row> sortedByProbeId(List<Row> results) { + return results.stream() + .sorted(Comparator.comparingInt(r -> r.<Integer>getFieldAs(0))) + .collect(Collectors.toList()); + } + + /** Asserts the build version (field 1) is non-decreasing across the probe-ordered results. */ + private static void assertMonotonicVersions(List<Row> byProbeId) { + int previous = Integer.MIN_VALUE; + for (Row r : byProbeId) { + final int version = r.<Integer>getFieldAs(1); + assertThat(version).isGreaterThanOrEqualTo(previous); + previous = version; + } + } + + private static LocalDateTime ts(String time) { + return LocalDateTime.parse("2020-01-01T" + time); + } +} diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-inner/plan/lateral-snapshot-join-inner.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-inner/plan/lateral-snapshot-join-inner.json new file mode 100644 index 00000000000..eea074a1535 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-inner/plan/lateral-snapshot-join-inner.json @@ -0,0 +1,469 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`probe`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "pk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "pv", + "dataType" : "INT" + }, { + "name" : "pts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "pts", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`pts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "pts", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`pts`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `pts_str` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `pts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, probe, watermark=[TO_TIMESTAMP(pts_str)], watermarkEmitStrategy=[on-event]]], fields=[pk, pv, pts_str])" + }, { + "id" : 2, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "INT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT>", + "description" : "Calc(select=[pk, pv])" + }, { + "id" : 3, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT>", + "description" : "Exchange(distribution=[hash[pk]])" + }, { + "id" : 4, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`b`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "bk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "dataType" : "INT" + }, { + "name" : "bts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "bts", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`bts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "bts", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`bts`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`bk` VARCHAR(2147483647), `bv` INT, `bts_str` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`bk` VARCHAR(2147483647), `bv` INT, `bts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, b, watermark=[TO_TIMESTAMP(bts_str)], watermarkEmitStrategy=[on-event]]], fields=[bk, bv, bts_str])" + }, { + "id" : 5, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "INT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "bk", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "fieldType" : "INT" + }, { + "name" : "bts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[bk, bv, Reinterpret(TO_TIMESTAMP(bts_str)) AS bts])" + }, { + "id" : 6, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "bk", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "fieldType" : "INT" + }, { + "name" : "bts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[bk]])" + }, { + "id" : 7, + "type" : "stream-exec-lateral-snapshot-join_1", + "joinSpec" : { + "joinType" : "INNER", + "leftKeys" : [ 0 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "rightTimeAttributeIndex" : 2, + "loadCompletedCondition" : "user_time", + "loadCompletedTime" : 1577836803000, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `bk` VARCHAR(2147483647), `bv` INT, `bts` TIMESTAMP(3)>", + "description" : "LateralSnapshotJoin(joinType=[InnerJoin], where=[(pk = bk)], select=[pk, pv, bk, bv, bts], loadCompletedCondition=[user_time], loadCompletedTime=[1577836803000])" + }, { + "id" : 8, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "INT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `bk` VARCHAR(2147483647), `bv` INT>", + "description" : "Calc(select=[pk, pv, bk, bv])" + }, { + "id" : 9, + "type" : "stream-exec-sink_2", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`sink`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "pk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "pv", + "dataType" : "INT" + }, { + "name" : "bk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "dataType" : "INT" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "upsertMaterializeStrategy" : "VALUE", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `bk` VARCHAR(2147483647), `bv` INT>", + "description" : "Sink(table=[default_catalog.default_database.sink], fields=[pk, pv, bk, bv])" + } ], + "edges" : [ { + "source" : 1, + "target" : 2, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 2, + "target" : 3, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 4, + "target" : 5, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 5, + "target" : 6, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 3, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 6, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 7, + "target" : 8, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 8, + "target" : 9, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + } ] +} \ No newline at end of file diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-inner/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-inner/savepoint/_metadata new file mode 100644 index 00000000000..ea9e93db4c1 Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-inner/savepoint/_metadata differ diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-load-phase/plan/lateral-snapshot-join-load-phase.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-load-phase/plan/lateral-snapshot-join-load-phase.json new file mode 100644 index 00000000000..d6799f7ab90 --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-load-phase/plan/lateral-snapshot-join-load-phase.json @@ -0,0 +1,468 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 1, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`probe`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "pk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "pv", + "dataType" : "INT" + }, { + "name" : "pts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "pts", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`pts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "pts", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`pts`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `pts_str` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `pts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, probe, watermark=[TO_TIMESTAMP(pts_str)], watermarkEmitStrategy=[on-event]]], fields=[pk, pv, pts_str])" + }, { + "id" : 2, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "INT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT>", + "description" : "Calc(select=[pk, pv])" + }, { + "id" : 3, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT>", + "description" : "Exchange(distribution=[hash[pk]])" + }, { + "id" : 4, + "type" : "stream-exec-table-source-scan_2", + "scanTableSource" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`b`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "bk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "dataType" : "INT" + }, { + "name" : "bts_str", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "bts", + "kind" : "COMPUTED", + "expression" : { + "rexNode" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "TO_TIMESTAMP(`bts_str`)" + } + } ], + "watermarkSpecs" : [ { + "rowtimeAttribute" : "bts", + "expression" : { + "rexNode" : { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "TIMESTAMP(3)" + }, + "serializableString" : "`bts`" + } + } ] + } + } + }, + "abilities" : [ { + "type" : "WatermarkPushDown", + "watermarkExpr" : { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + }, + "rowtimeExpr" : { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + }, + "idleTimeoutMillis" : -1, + "producedType" : "ROW<`bk` VARCHAR(2147483647), `bv` INT, `bts_str` VARCHAR(2147483647)> NOT NULL", + "watermarkParams" : { + "emitStrategy" : "ON_EVENT", + "alignGroupName" : null, + "alignMaxDrift" : "PT0S", + "alignUpdateInterval" : "PT1S", + "sourceIdleTimeout" : -1 + } + } ] + }, + "outputType" : "ROW<`bk` VARCHAR(2147483647), `bv` INT, `bts_str` VARCHAR(2147483647)>", + "description" : "TableSourceScan(table=[[default_catalog, default_database, b, watermark=[TO_TIMESTAMP(bts_str)], watermarkEmitStrategy=[on-event]]], fields=[bk, bv, bts_str])" + }, { + "id" : 5, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "INT" + }, { + "kind" : "CALL", + "syntax" : "SPECIAL", + "internalName" : "$REINTERPRET$1", + "operands" : [ { + "kind" : "CALL", + "internalName" : "$TO_TIMESTAMP$1", + "operands" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + } ], + "type" : "TIMESTAMP(3)" + } ], + "type" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "bk", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "fieldType" : "INT" + }, { + "name" : "bts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Calc(select=[bk, bv, Reinterpret(TO_TIMESTAMP(bts_str)) AS bts])" + }, { + "id" : 6, + "type" : "stream-exec-exchange_1", + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "HASH", + "keys" : [ 0 ] + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : { + "type" : "ROW", + "fields" : [ { + "name" : "bk", + "fieldType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "fieldType" : "INT" + }, { + "name" : "bts", + "fieldType" : { + "type" : "TIMESTAMP_WITHOUT_TIME_ZONE", + "precision" : 3, + "kind" : "ROWTIME" + } + } ] + }, + "description" : "Exchange(distribution=[hash[bk]])" + }, { + "id" : 7, + "type" : "stream-exec-lateral-snapshot-join_1", + "joinSpec" : { + "joinType" : "INNER", + "leftKeys" : [ 0 ], + "rightKeys" : [ 0 ], + "filterNulls" : [ true ], + "nonEquiCondition" : null + }, + "rightTimeAttributeIndex" : 2, + "loadCompletedCondition" : "user_time", + "loadCompletedTime" : 1577836803000, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + }, { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `bk` VARCHAR(2147483647), `bv` INT, `bts` TIMESTAMP(3)>", + "description" : "LateralSnapshotJoin(joinType=[InnerJoin], where=[(pk = bk)], select=[pk, pv, bk, bv, bts], loadCompletedCondition=[user_time], loadCompletedTime=[1577836803000])" + }, { + "id" : 8, + "type" : "stream-exec-calc_1", + "projection" : [ { + "kind" : "INPUT_REF", + "inputIndex" : 0, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 1, + "type" : "INT" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 2, + "type" : "VARCHAR(2147483647)" + }, { + "kind" : "INPUT_REF", + "inputIndex" : 3, + "type" : "INT" + } ], + "condition" : null, + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `bk` VARCHAR(2147483647), `bv` INT>", + "description" : "Calc(select=[pk, pv, bk, bv])" + }, { + "id" : 9, + "type" : "stream-exec-sink_2", + "configuration" : { + "table.exec.sink.keyed-shuffle" : "AUTO", + "table.exec.sink.not-null-enforcer" : "ERROR", + "table.exec.sink.rowtime-inserter" : "ENABLED", + "table.exec.sink.type-length-enforcer" : "IGNORE", + "table.exec.sink.upsert-materialize" : "AUTO" + }, + "dynamicTableSink" : { + "table" : { + "identifier" : "`default_catalog`.`default_database`.`sink`", + "resolvedTable" : { + "schema" : { + "columns" : [ { + "name" : "pk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "pv", + "dataType" : "INT" + }, { + "name" : "bk", + "dataType" : "VARCHAR(2147483647)" + }, { + "name" : "bv", + "dataType" : "INT" + } ] + } + } + } + }, + "inputChangelogMode" : [ "INSERT" ], + "inputProperties" : [ { + "requiredDistribution" : { + "type" : "UNKNOWN" + }, + "damBehavior" : "PIPELINED", + "priority" : 0 + } ], + "outputType" : "ROW<`pk` VARCHAR(2147483647), `pv` INT, `bk` VARCHAR(2147483647), `bv` INT>", + "description" : "Sink(table=[default_catalog.default_database.sink], fields=[pk, pv, bk, bv])" + } ], + "edges" : [ { + "source" : 1, + "target" : 2, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 2, + "target" : 3, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 4, + "target" : 5, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 5, + "target" : 6, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 3, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 6, + "target" : 7, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 7, + "target" : 8, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 8, + "target" : 9, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + } ] +} \ No newline at end of file diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-load-phase/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-load-phase/savepoint/_metadata new file mode 100644 index 00000000000..cf486d2151f Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-load-phase/savepoint/_metadata differ
