This is an automated email from the ASF dual-hosted git repository. fhueske pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8460639e2b402cd63f4eaf7dbb482221c3f480ea Author: Fabian Hueske <[email protected]> AuthorDate: Wed Sep 16 09:50:37 2026 +0200 [FLINK-40666][table] Derive default SNAPSHOT load_completed_time from query-start epoch Reuses the query-start epoch time that is set once before the compilation, instead of calling System.currentTimeMillis() per rule match, so multiple compile_time LATERAL SNAPSHOT joins in the same query agree on load_completed_time. Generated-By: Claude Sonnet 5 --- .../LogicalJoinToLateralSnapshotJoinRule.java | 4 ++- .../plan/utils/LateralSnapshotJoinUtil.java | 17 +++++++++++ .../stream/sql/join/LateralSnapshotJoinTest.java | 33 ++++++++++++++++++++ .../plan/utils/LateralSnapshotJoinUtilTest.java | 35 ++++++++++++++++++++++ 4 files changed, 88 insertions(+), 1 deletion(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java index 0b48c9920ea..6edb403f934 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java @@ -173,7 +173,9 @@ public class LogicalJoinToLateralSnapshotJoinRule loadCompletedTime = loadCompletedTimeLiteral.getValueAs(Long.class); loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_USER_TIME; } else { - loadCompletedTime = System.currentTimeMillis(); + loadCompletedTime = + LateralSnapshotJoinUtil.resolveDefaultLoadCompletedTime( + ShortcutUtils.unwrapTableConfig(call)); loadCompletedCondition = LateralSnapshotJoinUtil.LOAD_COMPLETED_CONDITION_COMPILE_TIME; } final Long loadCompletedIdleTimeoutMs = diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java index 61fa3471c2b..764dca00c37 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtil.java @@ -19,12 +19,14 @@ package org.apache.flink.table.planner.plan.utils; import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.functions.BuiltInFunctionDefinition; import org.apache.flink.table.functions.BuiltInFunctionDefinitions; import org.apache.flink.table.functions.FunctionDefinition; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction; import org.apache.flink.table.planner.plan.schema.TimeIndicatorRelDataType; +import org.apache.flink.table.planner.utils.InternalConfigOptions; import org.apache.flink.table.types.logical.LocalZonedTimestampType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.TimestampType; @@ -59,6 +61,21 @@ public final class LateralSnapshotJoinUtil { */ public static final String LOAD_COMPLETED_CONDITION_USER_TIME = "user_time"; + /** + * Resolves the {@code load_completed_time} used by a {@code compile_time} LATERAL SNAPSHOT + * join, i.e. one where the user did not provide the argument. + * + * <p>Uses {@link InternalConfigOptions#TABLE_QUERY_START_EPOCH_TIME}: {@link + * org.apache.flink.table.planner.delegation.PlannerBase#beforeTranslation} records it once + * before optimization begins, so every SNAPSHOT call reduced during that optimization run + * resolves to the same value. If the config is absent, we fall back to wall clock time. + */ + public static long resolveDefaultLoadCompletedTime(TableConfig tableConfig) { + return tableConfig + .getOptional(InternalConfigOptions.TABLE_QUERY_START_EPOCH_TIME) + .orElseGet(System::currentTimeMillis); + } + /** * {@code true} when {@code definition} is the {@link BuiltInFunctionDefinitions#SNAPSHOT} * built-in. diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java index 16a9aa322c8..81ac5c8ed80 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java @@ -35,7 +35,11 @@ import org.junit.jupiter.api.Test; import java.time.Duration; import java.time.ZoneId; +import java.util.HashSet; import java.util.List; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -291,6 +295,35 @@ public class LateralSnapshotJoinTest extends TableTestBase { .contains("where=[=(pk, bk)]"); } + @Test + void testMultipleDefaultCompileTimeSnapshotJoinsShareTheSameLoadCompletedTime() { + util.tableEnv() + .executeSql( + "CREATE TABLE b2 (" + + " bk STRING," + + " bv INT," + + " bts TIMESTAMP(3)," + + " WATERMARK FOR bts AS bts" + + ") WITH ('connector' = 'values', 'bounded' = 'false')"); + // Two LATERAL SNAPSHOT joins in the same query, neither providing load_completed_time: + // both must resolve to the same wall-clock compile-time timestamp. + final String sql = + "SELECT probe.pk, s1.bv, s2.bv FROM probe " + + "JOIN LATERAL SNAPSHOT(input => TABLE b, on_time => DESCRIPTOR(bts)) AS s1 " + + "ON probe.pk = s1.bk " + + "JOIN LATERAL SNAPSHOT(input => TABLE b2, on_time => DESCRIPTOR(bts)) AS s2 " + + "ON probe.pk = s2.bk"; + final String plan = util.tableEnv().explainSql(sql); + + // assert that both LateralSnapshotJoins use the same loadCompletedTime + final Matcher matcher = Pattern.compile("loadCompletedTime=\\[(\\d+)]").matcher(plan); + final Set<String> loadCompletedTimes = new HashSet<>(); + while (matcher.find()) { + loadCompletedTimes.add(matcher.group(1)); + } + assertThat(loadCompletedTimes).as("plan:%n%s", plan).hasSize(1); + } + // ------------------------------------------------------------------------------------------ // Validation: rejection paths // ------------------------------------------------------------------------------------------ diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java index fb1dd964918..f7aa9d3e31e 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/LateralSnapshotJoinUtilTest.java @@ -18,9 +18,11 @@ package org.apache.flink.table.planner.plan.utils; +import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.calcite.FlinkTypeSystem; import org.apache.flink.table.planner.plan.schema.TimeIndicatorRelDataType; +import org.apache.flink.table.planner.utils.InternalConfigOptions; import org.apache.calcite.rel.core.JoinRelType; import org.apache.calcite.rel.type.RelDataType; @@ -28,6 +30,7 @@ import org.apache.calcite.rel.type.RelDataTypeField; import org.apache.calcite.sql.type.BasicSqlType; import org.apache.calcite.sql.type.SqlTypeName; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; @@ -159,6 +162,38 @@ class LateralSnapshotJoinUtilTest { .isEqualTo(nullableTimeCol); } + /** + * When the planner has recorded a query-start epoch time (as {@link + * org.apache.flink.table.planner.delegation.PlannerBase#beforeTranslation} does before + * optimization begins), every SNAPSHOT call in the query must resolve its default {@code + * load_completed_time} to that same value. + */ + @Test + void testResolveDefaultLoadCompletedTimeUsesQueryStartEpochTimeWhenSet() { + final TableConfig tableConfig = TableConfig.getDefault(); + tableConfig.set(InternalConfigOptions.TABLE_QUERY_START_EPOCH_TIME, 1_700_000_000_000L); + + assertThat(LateralSnapshotJoinUtil.resolveDefaultLoadCompletedTime(tableConfig)) + .isEqualTo(1_700_000_000_000L); + } + + /** + * Outside of query translation (e.g. {@code TableTestUtil#verifyRelPlan} in tests) no + * query-start epoch time is recorded; the resolution must still fall back to the wall clock + * rather than fail. + */ + @Test + void testResolveDefaultLoadCompletedTimeFallsBackToWallClockWhenUnset() { + final TableConfig tableConfig = TableConfig.getDefault(); + + final long before = System.currentTimeMillis(); + final long loadCompletedTime = + LateralSnapshotJoinUtil.resolveDefaultLoadCompletedTime(tableConfig); + final long after = System.currentTimeMillis(); + + assertThat(loadCompletedTime).isBetween(before, after); + } + private RelDataType varchar(boolean nullable) { return typeFactory.createTypeWithNullability( typeFactory.createSqlType(SqlTypeName.VARCHAR), nullable);
