diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/ForbidSnapshotOutsideLateralRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/ForbidSnapshotOutsideLateralRule.java new file mode 100644 index 00000000000000..95db7221528f9b --- /dev/null +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/ForbidSnapshotOutsideLateralRule.java @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.planner.plan.rules.logical; + +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalTableFunctionScan; +import org.apache.flink.table.planner.plan.utils.LateralSnapshotJoinUtil; + +import org.apache.calcite.plan.RelOptRuleCall; +import org.apache.calcite.plan.RelRule; +import org.apache.calcite.rex.RexCall; +import org.immutables.value.Value; + +/** + * Rejects any {@link FlinkLogicalTableFunctionScan} that is still backed by the built-in {@code + * SNAPSHOT} function, with a clear error message. + * + *

{@code SNAPSHOT} is a planner placeholder that is only valid as the build side of a {@code + * LATERAL} join, where {@link LogicalJoinToLateralSnapshotJoinRule} rewrites the surrounding join + * into a dedicated {@link + * org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalLateralSnapshotJoin} and removes + * the SNAPSHOT scan. This rule must therefore run after that rewrite (see {@code + * FlinkStreamRuleSets.LOGICAL_REWRITE}). + */ +@Value.Enclosing +public class ForbidSnapshotOutsideLateralRule + extends RelRule { + + public static final ForbidSnapshotOutsideLateralRule INSTANCE = + ForbidSnapshotOutsideLateralRule.ForbidSnapshotOutsideLateralRuleConfig.DEFAULT + .toRule(); + + private ForbidSnapshotOutsideLateralRule(ForbidSnapshotOutsideLateralRuleConfig config) { + super(config); + } + + @Override + public boolean matches(RelOptRuleCall call) { + final FlinkLogicalTableFunctionScan scan = call.rel(0); + return scan.getCall() instanceof RexCall + && LateralSnapshotJoinUtil.isSnapshotCall((RexCall) scan.getCall()); + } + + @Override + public void onMatch(RelOptRuleCall call) { + throw new ValidationException( + "The SNAPSHOT function can only be used as the build side (right-hand side) of a " + + "LATERAL join. It cannot be used as a standalone table function or " + + "outside of a LATERAL context."); + } + + /** Rule configuration. */ + @Value.Immutable(singleton = false) + public interface ForbidSnapshotOutsideLateralRuleConfig extends RelRule.Config { + + ForbidSnapshotOutsideLateralRule.ForbidSnapshotOutsideLateralRuleConfig DEFAULT = + ImmutableForbidSnapshotOutsideLateralRule.ForbidSnapshotOutsideLateralRuleConfig + .builder() + .build() + .withOperandSupplier( + b0 -> b0.operand(FlinkLogicalTableFunctionScan.class).anyInputs()) + .withDescription("ForbidSnapshotOutsideLateralRule"); + + @Override + default ForbidSnapshotOutsideLateralRule toRule() { + return new ForbidSnapshotOutsideLateralRule(this); + } + } +} diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala index 65e6adbf772049..284b1dbc056dbb 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala @@ -405,6 +405,9 @@ object FlinkStreamRuleSets { // Rewrites a join over a SNAPSHOT table function call into a dedicated // FlinkLogicalLateralSnapshotJoin for the LATERAL SNAPSHOT operator. LogicalJoinToLateralSnapshotJoinRule.INSTANCE, + // Rejects SNAPSHOT scans that survived the rewrite above, i.e. SNAPSHOT calls used outside a + // LATERAL context. Must run after LogicalJoinToLateralSnapshotJoinRule. + ForbidSnapshotOutsideLateralRule.INSTANCE, // Avoids accessing a field from the result (condition). PythonCalcSplitRule.SPLIT_CONDITION_REX_FIELD, // Avoids accessing a field from the result (projection). diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java index df439bd2c9e7f3..851275d4ef804a 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesRuntimeFunctions.java +++ b/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 serializer, Iterable elements, WatermarkStrategy 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) { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java index 3387f522232e44..4e260a3e75fc6c 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java +++ b/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 watermarkStrategy = WatermarkStrategy.noWatermarks(); @@ -1771,7 +1775,9 @@ private TestValuesScanTableSourceWithWatermarkPushDown( List> allPartitions, Map 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; 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 00000000000000..5a4fdf644c49ce --- /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,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 programs() { + return Arrays.asList( + LateralSnapshotJoinTestPrograms.LATERAL_SNAPSHOT_JOIN_INNER, + LateralSnapshotJoinTestPrograms.LATERAL_SNAPSHOT_JOIN_LEFT); + } +} 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 00000000000000..9af4e9049d1d27 --- /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,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}. + * + *

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(); +} 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 29d9f0db3b9c0e..ea183bf3dbad39 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>> SKIP_EXEC_NODES = new HashSet>>() { { - // 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/plan/stream/sql/SnapshotTableFunctionTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java index 12511c4ae49361..6d254b220b86f8 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/SnapshotTableFunctionTest.java @@ -19,6 +19,7 @@ package org.apache.flink.table.planner.plan.stream.sql; 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; @@ -153,13 +154,14 @@ void testPartitionByNotAllowed() { @Test void testFromContextRejectedOutsideLateral() { - // SNAPSHOT used outside a LATERAL context is not rewritten by the LATERAL SNAPSHOT rule and - // reaches the regular PTF physical rule. Because SNAPSHOT disables system arguments, that - // rule rejects it. FLINK-39784 will replace this with a clearer message - // ("SNAPSHOT can only be used inside a LATERAL clause"). + // SNAPSHOT used outside a LATERAL context is not rewritten by the LATERAL SNAPSHOT rule. + // ForbidSnapshotOutsideLateralRule intercepts the surviving SNAPSHOT scan and rejects it + // with a clear message before it reaches the generic PTF translation. assertThatThrownBy(() -> util.verifyRelPlan("SELECT * FROM SNAPSHOT(input => TABLE Rates)")) .satisfies( anyCauseMatches( - "Disabling system arguments is not supported for user-defined PTF.")); + ValidationException.class, + "The SNAPSHOT function can only be used as the build side " + + "(right-hand side) of a LATERAL join")); } } 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 00000000000000..1aa873271a3f46 --- /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,606 @@ +/* + * 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.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.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThatList; + +/** + * End-to-end result tests for the {@code LATERAL SNAPSHOT} processing-time temporal table join. + * + *

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 gate timestamp ({@link + * #MID_GATE}) while all real build rows are earlier. With per-record ({@code on-event}) watermarks + * the operator flips to the JOIN phase. Since no more build-records are received, all probe-side + * records see the same build-side table and the append-only output can be asserted as an unordered + * set. + * + *

Some tests throttle source emission ({@code source.sleep-*}) to place probes deterministically + * around the flip. Purely processing-time-driven behavior (idle-timeout flip, state-TTL eviction) + * is covered deterministically by {@code LateralSnapshotJoinOperatorTest}. + */ +@ExtendWith(ParameterizedTestExtension.class) +public class LateralSnapshotJoinITCase extends StreamingWithStateTestBase { + + /** The {@code 'user_time'} gate reached mid-stream by the build-side flip-trigger row. */ + private static final String MID_GATE = + "load_completed_condition => 'user_time', " + + "load_completed_time => CAST(TIMESTAMP '2020-01-01 00:00:10' AS TIMESTAMP_LTZ(3))"; + + /** A far-future gate: the flip happens only at end of all input. */ + private static final String END_GATE = + "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_GATE} 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 parameters() { + return Arrays.asList( + new Object[][] { + {StreamingWithStateTestBase.HEAP_BACKEND()}, + {StreamingWithStateTestBase.ROCKSDB_BACKEND()} + }); + } + + // ------------------------------------------------------------------------------------------ + // Core join semantics + // ------------------------------------------------------------------------------------------ + + @TestTemplate + void testInnerJoin() { + // Throttle the probe so the fast build flips first; the probes are then joined live in the + // JOIN phase (the snapshot is frozen, so buffered-vs-live does not change the result). + createProbe(defaultProbe(), 40L); + createAppendBuild(defaultBuild()); + + final List actual = + collect( + "SELECT probe.pk, probe.pv, s.bk, s.bv " + + "FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual) + .containsExactlyInAnyOrder( + Row.of("a", 100, "a", 10), + Row.of("a", 100, "a", 11), + Row.of("b", 200, "b", 20)); + } + + @TestTemplate + void testLeftJoin() { + createProbe(defaultProbe(), 40L); + createAppendBuild(defaultBuild()); + + final List actual = + collect( + "SELECT probe.pk, probe.pv, s.bk, s.bv " + + "FROM probe LEFT JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual) + .containsExactlyInAnyOrder( + Row.of("a", 100, "a", 10), + Row.of("a", 100, "a", 11), + Row.of("b", 200, "b", 20), + Row.of("c", 300, null, null)); + } + + @TestTemplate + void testSelectStarMaterializesBuildRowtime() { + createProbe(Arrays.asList(Row.of("a", 100, ts("00:01:00")))); + createAppendBuild(Arrays.asList(Row.of("a", 10, ts("00:00:01")))); + + // SELECT * yields probe columns then build columns; the build rowtime is materialized as a + // regular TIMESTAMP(3), not a time attribute. + final List actual = + collect( + "SELECT * FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual) + .containsExactlyInAnyOrder( + Row.of( + "a", + 100, + LocalDateTime.parse("2020-01-01T00:01:00"), + "a", + 10, + LocalDateTime.parse("2020-01-01T00:00:01"))); + } + + @TestTemplate + void testCompositeKeys() { + createProbe( + Arrays.asList( + Row.of("a", 10, ts("00:01:00")), + Row.of("a", 11, ts("00:01:01")), + Row.of("b", 20, ts("00:01:02")))); + createAppendBuild( + 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")))); + + final List actual = + collect( + "SELECT probe.pk, probe.pv, s.bk, s.bv " + + "FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk AND probe.pv = s.bv"); + + assertThatList(actual) + .containsExactlyInAnyOrder(Row.of("a", 10, "a", 10), Row.of("b", 20, "b", 20)); + } + + @TestTemplate + void testNonEquiCondition() { + createProbe(Arrays.asList(Row.of("a", 15, ts("00:01:00")), Row.of("a", 5, ts("00:01:01")))); + createAppendBuild( + Arrays.asList(Row.of("a", 10, ts("00:00:01")), Row.of("a", 20, ts("00:00:02")))); + + final List actual = + collect( + "SELECT probe.pk, probe.pv, s.bv " + + "FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk AND probe.pv > s.bv"); + + assertThatList(actual).containsExactlyInAnyOrder(Row.of("a", 15, 10)); + } + + @TestTemplate + void testEmptyBuildSide() { + createProbe(defaultProbe()); + createAppendBuild(Collections.emptyList()); + + final List inner = + collect( + "SELECT probe.pk, s.bk FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + assertThatList(inner).isEmpty(); + + final List left = + collect( + "SELECT probe.pk, s.bk FROM probe LEFT JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + assertThatList(left) + .containsExactlyInAnyOrder(Row.of("a", null), Row.of("b", null), Row.of("c", null)); + } + + // ------------------------------------------------------------------------------------------ + // Build-side changelog consolidation (all changes have rowtime < the gate) + // ------------------------------------------------------------------------------------------ + + @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 actual = + collect( + "SELECT probe.pk, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) 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 actual = + collect( + "SELECT probe.pk, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual).containsExactlyInAnyOrder(Row.of("a", 11)); + } + + // ------------------------------------------------------------------------------------------ + // Flip timing (when LOAD flips to JOIN, and probe placement around the flip) + // ------------------------------------------------------------------------------------------ + + @TestTemplate + void testInnerJoinFlipsAtEndOfInput() { + // Far-future gate: the operator stays in LOAD and flips only at end-of-input, buffering and + // draining every probe. The frozen snapshot and result match the mid-flip case. + createProbe(defaultProbe()); + createAppendBuild(defaultBuild()); + + final List actual = + collect( + "SELECT probe.pk, probe.pv, s.bk, s.bv " + + "FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + END_GATE + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual) + .containsExactlyInAnyOrder( + Row.of("a", 100, "a", 10), + Row.of("a", 100, "a", 11), + Row.of("b", 200, "b", 20)); + } + + @TestTemplate + void testDefaultCompileTimeCondition() { + // No options: 'load_completed_condition' defaults to 'compile_time', so the gate is the + // wall-clock time at planning. The 2020 build watermarks never reach it, so the flip + // happens at end-of-input. + createProbe(defaultProbe()); + createAppendBuild(defaultBuild()); + + final List actual = + collect( + "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"); + + assertThatList(actual) + .containsExactlyInAnyOrder( + Row.of("a", 100, "a", 10), + Row.of("a", 100, "a", 11), + Row.of("b", 200, "b", 20)); + } + + @TestTemplate + void testProbesJoinedLiveInJoinPhase() { + // Fast build flips almost immediately; the throttled probe stream is consumed entirely in + // the JOIN phase, exercising the live per-record path (including a non-matching key). The + // snapshot is static after the flip, so the result is deterministic. + final List probes = + 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")), // non-matching + Row.of("a", 4, ts("00:00:04")), + Row.of("b", 5, ts("00:00:05")), + Row.of("c", 6, ts("00:00:06"))); // non-matching + createProbe(probes, 40L); + createAppendBuild( + Arrays.asList(Row.of("a", 10, ts("00:00:01")), Row.of("b", 20, ts("00:00:02")))); + + final List actual = + collect( + "SELECT probe.pv, probe.pk, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + + assertThatList(actual) + .containsExactlyInAnyOrder( + Row.of(1, "a", 10), + Row.of(4, "a", 10), + Row.of(2, "b", 20), + Row.of(5, "b", 20)); + } + + @TestTemplate + void testManyProbesBufferedDuringLoadThenDrained() { + // Throttled build flips at ~240 ms, by which time the un-throttled probe stream is fully + // buffered in LOAD. The flip drains all buffered probes against the final loaded version; + // all must match and none may be lost. + final int probeCount = 20; + final List probes = new ArrayList<>(); + for (int i = 1; i <= probeCount; i++) { + probes.add(Row.of("k", i, ts(String.format("00:00:%02d", i)))); + } + createProbe(probes); // un-throttled: buffered within a few ms, long before the flip + createUpsertBuildVerbatim( + 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); // ~240 ms until the gate-crossing trigger row is emitted + + final List actual = + collect( + "SELECT probe.pv, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk"); + + assertThat(actual).hasSize(probeCount); + assertThat(actual).allSatisfy(r -> assertThat((int) r.getField(1)).isEqualTo(3)); + } + + @TestTemplate + void testProbesStraddlingFlip() { + // Throttled build loads v1, flips at ~100 ms, then applies v2 at ~150 ms. The throttled + // probe stream straddles both: the first probe is buffered then drained (sees v1) while the + // last is joined live after v2 (sees v2). Per-probe versions vary with timing, but the + // endpoints and non-decreasing order are deterministic. + final int probeCount = 8; + final List probes = new ArrayList<>(); + for (int i = 1; i <= probeCount; i++) { + probes.add(Row.of("k", i, ts(String.format("00:00:%02d", i)))); + } + createProbe(probes, 200L); // last probe at ~1400 ms, well past the post-flip update + createUpsertBuildVerbatim( + 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"))), + 50L); + + final List byProbeId = + sortedByProbeId( + collect( + "SELECT probe.pv, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) AS s ON probe.pk = s.bk")); + + assertThat(byProbeId).hasSize(probeCount); + assertMonotonicVersions(byProbeId); + assertThat((int) byProbeId.get(0).getField(1)).isEqualTo(1); // buffered -> v1 + assertThat((int) byProbeId.get(probeCount - 1).getField(1)).isEqualTo(2); // live -> v2 + } + + @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 probes = new ArrayList<>(); + for (int i = 1; i <= probeCount; i++) { + probes.add(Row.of("k", i, ts(String.format("00:00:%02d", i)))); + } + createProbe(probes, 200L); // last probe at ~1400 ms, well past the last post-flip update + createUpsertBuildVerbatim( + 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 byProbeId = + sortedByProbeId( + collect( + "SELECT probe.pv, s.bv FROM probe JOIN LATERAL TABLE(SNAPSHOT(" + + "input => TABLE b, " + + MID_GATE + + ")) 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 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 List 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 List collect(String query) { + return CollectionUtil.iteratorToList(tEnv().executeSql(query).collect()); + } + + /** + * Appends a non-matching build row at the {@link #MID_GATE} timestamp; its watermark flips the + * operator to the JOIN phase mid-stream (all real build rows are earlier). + */ + private List withFlipTrigger(List data) { + final List withTrigger = new ArrayList<>(data); + withTrigger.add(Row.of(FLIP_TRIGGER_KEY, 0, ts(FLIP_TRIGGER_TS))); + return withTrigger; + } + + private void createProbe(List data) { + createProbe(data, 0L); + } + + /** + * Creates the append-only probe table {@code probe}. {@code sleepMillis > 0} throttles emission + * to control when probe rows are consumed relative to the build-side flip. + */ + private void createProbe(List data, long sleepMillis) { + final String id = TestValuesTableFactory.registerData(data); + tEnv().executeSql( + "CREATE TABLE probe (" + + " pk STRING," + + " pv INT," + + " pts TIMESTAMP(3)," + + " WATERMARK FOR pts AS pts" + + ") WITH (" + + valuesOptions(id, sleepMillis) + + ")"); + } + + private void createAppendBuild(List data) { + final String id = TestValuesTableFactory.registerData(withFlipTrigger(data)); + tEnv().executeSql( + "CREATE TABLE b (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts" + + ") WITH (" + + valuesOptions(id, 0L) + + ")"); + } + + private void createChangelogBuild(List data) { + final String id = TestValuesTableFactory.registerData(withFlipTrigger(data)); + tEnv().executeSql( + "CREATE TABLE b (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts" + + ") WITH (" + + valuesOptions(id, 0L, "'changelog-mode' = 'I,UB,UA,D'") + + ")"); + } + + private void createUpsertBuild(List data) { + createUpsertBuildVerbatim(withFlipTrigger(data), 0L); + } + + /** + * Creates the upsert build table {@code b} from the given rows verbatim -- the caller places + * the flip-trigger row explicitly, so the flip and any post-flip updates can be ordered + * precisely. + */ + private void createUpsertBuildVerbatim(List data, long sleepMillis) { + final String id = TestValuesTableFactory.registerData(data); + tEnv().executeSql( + "CREATE TABLE b (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts," + + " PRIMARY KEY (bk) NOT ENFORCED" + + ") WITH (" + + valuesOptions(id, sleepMillis, "'changelog-mode' = 'I,UA,D'") + + ")"); + } + + private static String valuesOptions(String id, long sleepMillis, String... extraOptions) { + final StringBuilder options = + new StringBuilder() + .append(" 'connector' = 'values',") + .append(" 'bounded' = 'false',") + .append(" 'disable-lookup' = 'true',") + .append(" 'enable-watermark-push-down' = 'true',") + .append(" 'scan.watermark.emit.strategy' = 'on-event',"); + for (String extra : extraOptions) { + options.append(" ").append(extra).append(","); + } + if (sleepMillis > 0) { + options.append(" 'source.sleep-after-elements' = '1',") + .append(" 'source.sleep-time' = '") + .append(sleepMillis) + .append("ms',"); + } + return options.append(" 'data-id' = '").append(id).append("'").toString(); + } + + /** Sorts join results (pv, bv) by the probe id pv, i.e. into probe processing order. */ + private static List sortedByProbeId(List results) { + return results.stream() + .sorted(Comparator.comparingInt(r -> (int) r.getField(0))) + .collect(Collectors.toList()); + } + + /** Asserts the build version (field 1) is non-decreasing across the probe-ordered results. */ + private static void assertMonotonicVersions(List byProbeId) { + int previous = Integer.MIN_VALUE; + for (Row r : byProbeId) { + final int version = (int) r.getField(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 00000000000000..eea074a15352b3 --- /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 00000000000000..ea9e93db4c1e8b 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-left/plan/lateral-snapshot-join-left.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-left/plan/lateral-snapshot-join-left.json new file mode 100644 index 00000000000000..fa475e5b1c68ab --- /dev/null +++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-left/plan/lateral-snapshot-join-left.json @@ -0,0 +1,468 @@ +{ + "flinkVersion" : "2.4", + "nodes" : [ { + "id" : 10, + "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" : 11, + "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" : 12, + "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" : 13, + "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" : 14, + "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" : 15, + "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" : 16, + "type" : "stream-exec-lateral-snapshot-join_1", + "joinSpec" : { + "joinType" : "LEFT", + "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=[LeftOuterJoin], where=[(pk = bk)], select=[pk, pv, bk, bv, bts], loadCompletedCondition=[user_time], loadCompletedTime=[1577836803000])" + }, { + "id" : 17, + "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" : 18, + "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" : 10, + "target" : 11, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 11, + "target" : 12, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 13, + "target" : 14, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 14, + "target" : 15, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 12, + "target" : 16, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 15, + "target" : 16, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 16, + "target" : 17, + "shuffle" : { + "type" : "FORWARD" + }, + "shuffleMode" : "PIPELINED" + }, { + "source" : 17, + "target" : 18, + "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-left/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-lateral-snapshot-join_1/lateral-snapshot-join-left/savepoint/_metadata new file mode 100644 index 00000000000000..5b03d9a1259680 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-left/savepoint/_metadata differ