You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fhueske (via GitHub)" <gi...@apache.org> on 2026/10/05 10:07:20 UTC
[PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
fhueske opened a new pull request, #29380:
URL: https://github.com/apache/flink/pull/29380
## What is the purpose of the change
Handle LATERAL SNAPSHOT join correctly during TRY_RESOLVE pass.
The LSJ plan node extends CommonJoin, but handles retractions on its build side differently (always full row comparison).
Hence all build-side input rows must be fully deterministic.
This PR ensures that this requirement is passed down to the build-side input.
## Brief change log
* add special handling of LSJ to StreamNonDeterministicUpdatePlanVisitor
* add unit tests to assert correct handling
## Verifying this change
* run new unit tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? n/a
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 4.8)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on code in PR #29380:
URL: https://github.com/apache/flink/pull/29380#discussion_r4184312638
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -630,6 +631,10 @@ private StreamPhysicalRel visitJoin(
throwNonDeterministicConditionError(
Review Comment:
Good finding! Thanks
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29380:
URL: https://github.com/apache/flink/pull/29380#issuecomment-5992478112
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "6d8d6785a4bdb803f1a392bebe9794a1f7f609f8",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6d8d6785a4bdb803f1a392bebe9794a1f7f609f8",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6d8d6785a4bdb803f1a392bebe9794a1f7f609f8 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on code in PR #29380:
URL: https://github.com/apache/flink/pull/29380#discussion_r4184557830
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -630,6 +631,10 @@ private StreamPhysicalRel visitJoin(
throwNonDeterministicConditionError(
Review Comment:
I've created https://issues.apache.org/jira/browse/FLINK-40915 to follow up on this.
The change itself is tiny but the tests are a bit more involved and I'd like to get this PR in first.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #29380:
URL: https://github.com/apache/flink/pull/29380#discussion_r4183708391
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -630,6 +631,10 @@ private StreamPhysicalRel visitJoin(
throwNonDeterministicConditionError(
Review Comment:
nit: LSJ still goes through the generic non-deterministic condition check. That rejects a non-deterministic join condition whenever the build side updates or the join is a LEFT join, even with insert-only inputs. For LSJ that's stricter than needed: the output is append-only and the condition is only evaluated when a probe row arrives, never when build rows are retracted. Up to you if we want to change this in this ticket or follow up ticket
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske merged PR #29380:
URL: https://github.com/apache/flink/pull/29380
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #29380:
URL: https://github.com/apache/flink/pull/29380#discussion_r4184935668
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -630,6 +631,10 @@ private StreamPhysicalRel visitJoin(
throwNonDeterministicConditionError(
Review Comment:
Sounds resaonable
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40890][table] Require deterministic build columns for LATERAL SNAPSHOT join [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #29380:
URL: https://github.com/apache/flink/pull/29380#discussion_r4183708391
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -630,6 +631,10 @@ private StreamPhysicalRel visitJoin(
throwNonDeterministicConditionError(
Review Comment:
nit: LSJ still goes through the generic non-deterministic condition check. That rejects a non-deterministic join condition whenever the build side updates or the join is a LEFT join, even with insert-only inputs. For LSJ that's stricter than needed: the output is append-only and the condition is only evaluated when a probe row arrives, never when build rows are retracted. Fine to keep it as is for now, but maybe worth a follow-up. A dedicated `visitLateralSnapshotJoin` could relax this, and would also avoid the flag in `visitJoinChild`.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -259,6 +263,71 @@ void testInnerJoinWithUpsertBuildSourceMaterializesRetractions() {
assertThat(plan).doesNotContain("DropUpdateBefore");
}
+ @Test
+ void testRejectNonDeterministicBuildColumnWithPrimaryKey() {
+ // 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 testAcceptDeterministicBuildColumnWithPrimaryKey() {
+ // 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();
+ }
+
+ private void enableTryResolve() {
Review Comment:
nit: could we move `enableTryResolve()` and `createUpdatingPkBuildSource()` down to the other private helpers at the bottom of the file?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java:
##########
@@ -259,6 +263,71 @@ void testInnerJoinWithUpsertBuildSourceMaterializesRetractions() {
assertThat(plan).doesNotContain("DropUpdateBefore");
}
+ @Test
+ void testRejectNonDeterministicBuildColumnWithPrimaryKey() {
Review Comment:
nit: what's being tested is whether the build side is updating, not the primary key itself. Something like `testRejectNonDeterministicBuildColumnWithUpdatingBuildSide` (same for the accept test at line 290) might read clearer.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org