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
The following commit(s) were added to refs/heads/master by this push:
new 8c479701c2e [FLINK-40890][table] Require deterministic build columns
for LATERAL SNAPSHOT join (#29380)
8c479701c2e is described below
commit 8c479701c2e4a57f2e2a6787595c821051d83e12
Author: Fabian Hueske <[email protected]>
AuthorDate: Mon Oct 5 16:39:05 2026 +0200
[FLINK-40890][table] Require deterministic build columns for LATERAL
SNAPSHOT join (#29380)
Generated-by: Claude Code (Claude Opus 4.8)
---
.../StreamNonDeterministicUpdatePlanVisitor.java | 20 +++++--
.../stream/sql/join/LateralSnapshotJoinTest.java | 69 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 4 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
index 344352b958f..8a7c1e04cdf 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
@@ -40,6 +40,7 @@ import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalD
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalExchange;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalExpand;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalGroupAggregateBase;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLateralSnapshotJoin;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLegacySink;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLegacyTableSourceScan;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLimit;
@@ -630,6 +631,10 @@ public class StreamNonDeterministicUpdatePlanVisitor {
throwNonDeterministicConditionError(
ndCall.get(), join.getCondition(), (StreamPhysicalRel)
join);
}
+ // The LATERAL SNAPSHOT join keys its build (right) state by the
complete build row, so
+ // build retractions are matched by exact row equality. A unique key
offers no protection
+ // here: every build column must be deterministic whenever the build
side can update.
+ boolean forceRightWholeRowDeterminism = join instanceof
StreamPhysicalLateralSnapshotJoin;
int leftFieldCnt = leftRel.getRowType().getFieldCount();
StreamPhysicalRel newLeft =
visitJoinChild(
@@ -641,7 +646,8 @@ public class StreamNonDeterministicUpdatePlanVisitor {
join.joinSpec().getLeftKeys(),
// TODO remove this conversion when scala-free was
total done.
scala.collection.JavaConverters.seqAsJavaList(
- join.getUpsertKeys(leftRel,
join.joinSpec().getLeftKeys())));
+ join.getUpsertKeys(leftRel,
join.joinSpec().getLeftKeys())),
+ false);
StreamPhysicalRel newRight =
visitJoinChild(
requireDeterminism,
@@ -652,7 +658,8 @@ public class StreamNonDeterministicUpdatePlanVisitor {
join.joinSpec().getRightKeys(),
// TODO remove this conversion when scala-free was
total done.
scala.collection.JavaConverters.seqAsJavaList(
- join.getUpsertKeys(rightRel,
join.joinSpec().getRightKeys())));
+ join.getUpsertKeys(rightRel,
join.joinSpec().getRightKeys())),
+ forceRightWholeRowDeterminism);
return (StreamPhysicalRel)
join.copy(
@@ -930,7 +937,8 @@ public class StreamNonDeterministicUpdatePlanVisitor {
final int leftFieldCnt,
final boolean isLeft,
final int[] joinKeys,
- final List<int[]> inputUniqueKeys) {
+ final List<int[]> inputUniqueKeys,
+ final boolean forceWholeRowDeterminism) {
JoinInputSideSpec joinInputSideSpec =
JoinUtil.analyzeJoinInput(
ShortcutUtils.unwrapClassLoader(rel),
@@ -939,7 +947,11 @@ public class StreamNonDeterministicUpdatePlanVisitor {
inputUniqueKeys);
ImmutableBitSet inputRequireDeterminism;
if (inputHasUpdate) {
- if (joinInputSideSpec.hasUniqueKey() ||
joinInputSideSpec.joinKeyContainsUniqueKey()) {
+ if (forceWholeRowDeterminism) {
+ // the operator retracts by the whole input row regardless of
any unique key
+ inputRequireDeterminism =
ImmutableBitSet.range(rel.getRowType().getFieldCount());
+ } else if (joinInputSideSpec.hasUniqueKey()
+ || joinInputSideSpec.joinKeyContainsUniqueKey()) {
// join hasUniqueKey or joinKeyContainsUniqueKey, then
transmit corresponding
// requirement to input
if (isLeft) {
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 81ac5c8ed80..aa5ee760490 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
@@ -23,8 +23,11 @@ import
org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
import org.apache.flink.streaming.api.transformations.TwoInputTransformation;
import org.apache.flink.table.api.CompiledPlan;
import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import
org.apache.flink.table.api.config.OptimizerConfigOptions.NonDeterministicUpdateStrategy;
import org.apache.flink.table.api.internal.CompiledPlanUtils;
import org.apache.flink.table.planner.utils.TableTestBase;
import org.apache.flink.table.planner.utils.TableTestUtil;
@@ -42,6 +45,7 @@ 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.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Plan tests for the {@code LATERAL SNAPSHOT} processing-time temporal table
join. */
@@ -259,6 +263,47 @@ public class LateralSnapshotJoinTest extends TableTestBase
{
assertThat(plan).doesNotContain("DropUpdateBefore");
}
+ @Test
+ void testRejectNonDeterministicUpdatingBuildColumnWithPrimaryKey() {
+ // The LSJ operator keys its state by the whole build row, so
retractions are matched by
+ // exact row equality. A non-deterministic build column therefore
breaks retraction even
+ // when the build side has a unique key (FLINK-40890). TRY_RESOLVE
must reject it.
+ enableTryResolve();
+ createUpdatingPkBuildSource();
+ util.tableEnv()
+ .executeSql(
+ "CREATE VIEW b_view AS "
+ + "SELECT bk, CAST(NOW() AS STRING) AS bn, bts
FROM b_pk");
+
+ final String sql =
+ "SELECT probe.pk, s.bk, s.bn FROM probe JOIN LATERAL SNAPSHOT("
+ + "input => TABLE b_view, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
+ + ") AS s ON probe.pk = s.bk";
+
+ assertThatThrownBy(() -> util.tableEnv().explainSql(sql))
+ .isInstanceOf(TableException.class)
+ .hasMessageContaining("can not satisfy the determinism
requirement");
+ }
+
+ @Test
+ void testAcceptDeterministicUpdatingBuildColumnWithPrimaryKey() {
+ // Counterpart to the rejection test: a deterministic build column
over the same
+ // primary-keyed, updating source must still pass TRY_RESOLVE (no
over-rejection).
+ enableTryResolve();
+ createUpdatingPkBuildSource();
+ util.tableEnv()
+ .executeSql("CREATE VIEW b_view AS SELECT bk, UPPER(bk) AS bn,
bts FROM b_pk");
+
+ final String sql =
+ "SELECT probe.pk, s.bk, s.bn FROM probe JOIN LATERAL SNAPSHOT("
+ + "input => TABLE b_view, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
+ + ") AS s ON probe.pk = s.bk";
+
+ assertThatCode(() ->
util.tableEnv().explainSql(sql)).doesNotThrowAnyException();
+ }
+
@Test
void testNonEquiConditionCompilesEndToEnd() {
final String sql =
@@ -681,4 +726,28 @@ public class LateralSnapshotJoinTest extends TableTestBase
{
new AssertionError(
"No LateralSnapshotJoinOperator found
in the plan."));
}
+
+ private void enableTryResolve() {
+ util.tableEnv()
+ .getConfig()
+ .set(
+
OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
+ NonDeterministicUpdateStrategy.TRY_RESOLVE);
+ }
+
+ private void createUpdatingPkBuildSource() {
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE b_pk ("
+ + " bk STRING,"
+ + " bv INT,"
+ + " bts TIMESTAMP(3),"
+ + " WATERMARK FOR bts AS bts,"
+ + " PRIMARY KEY (bk) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'bounded' = 'false',"
+ + " 'changelog-mode' = 'I,UA,D'"
+ + ")");
+ }
}
