twalthr commented on code in PR #28713: URL: https://github.com/apache/flink/pull/28713#discussion_r3613125318
########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java: ########## @@ -0,0 +1,451 @@ +/* + * 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; +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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") Review Comment: feel free to move this setup config to the restore test bases and semantic test bases, all tests should run under this config by default. ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java: ########## @@ -491,6 +504,31 @@ void testRestore(TableTestProgram program, Path planPath, String savepointPath) } } + /** + * Triggers a stop-with-savepoint, retrying while the job is not yet fully running. A savepoint + * gated on source emission (see {@link #awaitSavepointReady}) can be requested before all + * downstream tasks have reached RUNNING, which aborts with "Not all required tasks are + * currently running". Sink-gated triggers never hit this (sink output implies a running + * pipeline), so they succeed on the first attempt. + */ + private String stopWithSavepointWhenRunning(JobClient jobClient) throws Exception { + final long deadline = System.currentTimeMillis() + RESULT_AWAIT_TIMEOUT_MILLIS; + while (true) { + try { + return jobClient + .stopWithSavepoint(false, tmpDir.toString(), SavepointFormatType.DEFAULT) + .get(); + } catch (ExecutionException e) { + final boolean notAllRunning = + e.getMessage() != null && e.getMessage().contains("Not all required tasks"); + if (!notAllRunning || System.currentTimeMillis() >= deadline) { + throw e; + } Review Comment: why was this never required before? I hope this won't cause flaky tests. did you run restore tests a couple of times? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java: ########## @@ -0,0 +1,451 @@ +/* + * 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; +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))"; Review Comment: side question: isn't the PTF logic smart enough nowadays to insert this cast implicitly? ```suggestion + "load_completed_time => TIMESTAMP '2020-01-01 00:00:10'"; ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/join/LateralSnapshotJoinITCase.java: ########## @@ -0,0 +1,282 @@ +/* + * 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.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); + tEnv().executeSql( + "CREATE TABLE probe (" + + " pk STRING," + + " pv INT," + + " pts TIMESTAMP(3)," + + " WATERMARK FOR pts AS pts" + + ") WITH (" + + valuesOptions(id, sleepMillis) + + ")"); + } + + private void createChangelogBuild(List<Row> 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<Row> data) { + createUpsertBuild(withFlipTrigger(data), 0L); + } + + private void createUpsertBuild(List<Row> 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 = Review Comment: use TableDescriptor API instead and avoid excessive string concats? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java: ########## @@ -0,0 +1,451 @@ +/* + * 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; +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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + // 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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + .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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + // 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") + .setupConfig(TableConfigOptions.LOCAL_TIME_ZONE, "UTC") + // Un-throttled: buffered within a few ms, long before the flip. + .setupTableSource(probe(manyProbes(20))) Review Comment: very nicely readable test! -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
