You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "gustavodemorais (via GitHub)" <gi...@apache.org> on 2026/09/29 16:13:08 UTC
[PR] [FLINK-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
gustavodemorais opened a new pull request, #29339:
URL: https://github.com/apache/flink/pull/29339
To be rebased after https://github.com/apache/flink/pull/29296 is merged.
## What is the purpose of the change
With upsert inputs, a record that replaces a stored record with the same unique key may match different rows than the replaced one if the join condition has a non-equi part. The streaming join never retracted the joined rows and null padding that only the replaced record produced, which led to stale rows and wrong null padding.
## Brief change log
- Join cases with non-equi join keys also work now
The summary of the price to pay for this correctness
1. **Extra state read.** One lookup per accumulate.
- Equi: has-unique-key side (upsert key contains the join key), other side outer.
- Non-equi: any side with unique key.
- Retract/append pay it too (returns null). Extra cost for a case that already works.
- Can drop for retract/append: planner flag "input is upsert". Needs optional plan field.
2. **TTL refresh.** Upsert on R that keeps same match no longer rewrites L count -> L TTL not refreshed -> L may expire earlier.
- It'd possible to change this and always write count (`count + (isNewMatch ? 1 : 0)`). This would be more write for this path.
## Verifying this change
- StreamingJoinOperatorTest
- StreamingMiniBatchJoinOperatorTest
## 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): yes - one extra state lookup per accumulate record for joins with a non-equi condition
- 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? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes (please specify the tool below)
2.1.282 (Claude Code) with Opus 5.5
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29339:
URL: https://github.com/apache/flink/pull/29339#discussion_r4164605412
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java:
##########
@@ -501,5 +506,112 @@ public class JoinTestPrograms {
+ " WHEN a NOT IN (SELECT CAST(j AS INTEGER) FROM source_t3) THEN 2 ELSE 3 END)"
+ " NOT IN (SELECT d FROM source_t2 WHERE source_t1.c = source_t2.f)")
.build();
+
+ LEFT_JOIN_WITH_UPSERT_INPUT_AND_NON_EQUI_CONDITION =
Review Comment:
```suggestion
LEFT_JOIN_UPSERT_INPUT_NON_EQUI =
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java:
##########
@@ -501,5 +506,112 @@ public class JoinTestPrograms {
+ " WHEN a NOT IN (SELECT CAST(j AS INTEGER) FROM source_t3) THEN 2 ELSE 3 END)"
+ " NOT IN (SELECT d FROM source_t2 WHERE source_t1.c = source_t2.f)")
.build();
+
+ LEFT_JOIN_WITH_UPSERT_INPUT_AND_NON_EQUI_CONDITION =
+ leftJoinWithUpsertInputAndNonEquiCondition(
+ "join-left-join-with-upsert-input-and-non-equi-condition",
Review Comment:
```suggestion
"join-left-join-upsert-input-non-equi",
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java:
##########
@@ -501,5 +506,112 @@ public class JoinTestPrograms {
+ " WHEN a NOT IN (SELECT CAST(j AS INTEGER) FROM source_t3) THEN 2 ELSE 3 END)"
+ " NOT IN (SELECT d FROM source_t2 WHERE source_t1.c = source_t2.f)")
.build();
+
+ LEFT_JOIN_WITH_UPSERT_INPUT_AND_NON_EQUI_CONDITION =
+ leftJoinWithUpsertInputAndNonEquiCondition(
+ "join-left-join-with-upsert-input-and-non-equi-condition",
+ "left join with an upsert input and a non-equi condition");
+
+ // leftInputChangelogMode and rightInputChangelogMode are removed from its stored plan,
+ // which
+ // makes it a plan of a previous version
+ LEFT_JOIN_WITH_UPSERT_INPUT_AND_NON_EQUI_CONDITION_PREVIOUS_PLAN =
+ leftJoinWithUpsertInputAndNonEquiCondition(
+ "join-left-join-with-upsert-input-and-non-equi-condition-previous-plan",
+ "left join with an upsert input and a non-equi condition restored from a plan of a previous version");
+
+ // the full image of the deleted line is only in the restored state
+ INNER_JOIN_WITH_KEY_ONLY_DELETES_AND_NON_EQUI_CONDITION =
Review Comment:
```suggestion
INNER_JOIN_KEY_ONLY_DELETES_NON_EQUI =
```
##########
flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-join_1/join-left-join-with-upsert-input-and-non-equi-condition-previous-plan/plan/join-left-join-with-upsert-input-and-non-equi-condition-previous-plan.json:
##########
@@ -0,0 +1,252 @@
+{
+ "flinkVersion" : "2.4",
Review Comment:
compile a 2.3 plan
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java:
##########
@@ -43,6 +47,20 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator {
protected final boolean leftIsOuter;
// whether right side is outer side, e.g. right is outer but left is not when RIGHT OUTER JOIN
protected final boolean rightIsOuter;
+ // changelog mode of the left input, null for compiled plans of previous versions
Review Comment:
concrete Flink version number here
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java:
##########
@@ -501,5 +506,112 @@ public class JoinTestPrograms {
+ " WHEN a NOT IN (SELECT CAST(j AS INTEGER) FROM source_t3) THEN 2 ELSE 3 END)"
+ " NOT IN (SELECT d FROM source_t2 WHERE source_t1.c = source_t2.f)")
.build();
+
+ LEFT_JOIN_WITH_UPSERT_INPUT_AND_NON_EQUI_CONDITION =
+ leftJoinWithUpsertInputAndNonEquiCondition(
+ "join-left-join-with-upsert-input-and-non-equi-condition",
+ "left join with an upsert input and a non-equi condition");
+
+ // leftInputChangelogMode and rightInputChangelogMode are removed from its stored plan,
+ // which
+ // makes it a plan of a previous version
+ LEFT_JOIN_WITH_UPSERT_INPUT_AND_NON_EQUI_CONDITION_PREVIOUS_PLAN =
Review Comment:
```suggestion
LEFT_JOIN_UPSERT_INPUT_NON_EQUI_PRE_2_4 =
```
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais merged PR #29339:
URL: https://github.com/apache/flink/pull/29339
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #29339:
URL: https://github.com/apache/flink/pull/29339#discussion_r4155618888
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java:
##########
@@ -100,6 +104,16 @@ public class StreamExecJoin extends ExecNodeBase<RowData>
@JsonInclude(JsonInclude.Include.NON_DEFAULT)
private final List<int[]> rightUpsertKeys;
+ @Nullable
+ @JsonProperty(FIELD_NAME_LEFT_INPUT_CHANGELOG_MODE)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private final ChangelogMode leftInputChangelogMode;
Review Comment:
done
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29339:
URL: https://github.com/apache/flink/pull/29339#issuecomment-5894456938
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "483bbb1a26b1e76a2a00e1302bdcbfe8f658cb47",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "483bbb1a26b1e76a2a00e1302bdcbfe8f658cb47",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 483bbb1a26b1e76a2a00e1302bdcbfe8f658cb47 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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #29339:
URL: https://github.com/apache/flink/pull/29339#issuecomment-5931992302
Hey @twalthr, thanks for the review and going with me through these. I've updated the PR with restore test, adjusted the new changelogMode field and moved the RuntimeChangelogMode util and also extended the PR to fix one edge case which is non-equi joins with partial deletes.
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #29339:
URL: https://github.com/apache/flink/pull/29339#discussion_r4156602577
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java:
##########
@@ -287,4 +319,12 @@ protected Transformation<RowData> translateToPlanInternal(
transform.setStateKeyType(leftSelect.getProducedType());
return transform;
}
+
+ /** Compiled plans of previous versions do not contain the changelog modes of the inputs. */
+ private static @Nullable RuntimeChangelogMode toRuntimeChangelogMode(
Review Comment:
> Just to double check: leftUpsertKeys is also set in retract?
Yes
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29339:
URL: https://github.com/apache/flink/pull/29339#discussion_r4144792962
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java:
##########
@@ -43,6 +45,8 @@ public class StreamingJoinOperator extends AbstractStreamingJoinOperator {
protected final boolean leftIsOuter;
// whether right side is outer side, e.g. right is outer but left is not when RIGHT OUTER JOIN
protected final boolean rightIsOuter;
+ // whether the join condition has a non-equi part, e.g. l.v > r.v
Review Comment:
add a better example:
```
A JOIN B ON A.k = B.k AND A.v > B.v
```
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #29339:
URL: https://github.com/apache/flink/pull/29339#issuecomment-5950624487
Thanks! Renamed the programs as suggested, the pre-2.4 plan and savepoint are now compiled with Flink 2.3 (release-2.3), and the comments name the concrete version (Flink 2.4).
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #29339:
URL: https://github.com/apache/flink/pull/29339#issuecomment-5915437942
I've added an additional optimization to avoid the additional state look up for retract and append for new statements https://github.com/apache/flink/pull/29339/commits/ae5a7630ae612f0cd9528da5f2f9ee1f181a7f50. Take a look
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "twalthr (via GitHub)" <gi...@apache.org>.
twalthr commented on code in PR #29339:
URL: https://github.com/apache/flink/pull/29339#discussion_r4153077041
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java:
##########
@@ -287,4 +319,12 @@ protected Transformation<RowData> translateToPlanInternal(
transform.setStateKeyType(leftSelect.getProducedType());
return transform;
}
+
+ /** Compiled plans of previous versions do not contain the changelog modes of the inputs. */
+ private static @Nullable RuntimeChangelogMode toRuntimeChangelogMode(
Review Comment:
`RuntimeChangelogMode` comes from PTFs. We should only store minimal information in CompiledPlan. As far as I see it, only the "isUpsert" information is required? Let's only add this boolean flag. Just to double check: leftUpsertKeys is also set in retract?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecJoin.java:
##########
@@ -100,6 +104,16 @@ public class StreamExecJoin extends ExecNodeBase<RowData>
@JsonInclude(JsonInclude.Include.NON_DEFAULT)
private final List<int[]> rightUpsertKeys;
+ @Nullable
+ @JsonProperty(FIELD_NAME_LEFT_INPUT_CHANGELOG_MODE)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private final ChangelogMode leftInputChangelogMode;
Review Comment:
Since this is a CompiledPlan change, make sure that one RestoreTest covers the old plan representation and one the new one.
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #29339:
URL: https://github.com/apache/flink/pull/29339#issuecomment-5915426089
Hey Timo, thanks for looking into the PR and the review
1. Tests on leftHasUniqueKeySpec:
The planner drops UPDATE_BEFORE when the upsert key contains the join key (StreamPhysicalJoin#inputUniqueKeyContainsJoinKey), so this spec does actually get bare upserts.
A join input only receives upsert if input's upsert key contains the join key (StreamPhysicalJoin#inputUniqueKeyContainsJoinKey). That covers two specs: the unique key equals the join key (JoinKeyContainsUniqueKey), or it has additional columns (HasUniqueKey), e.g. PK (order_id, line_id) joined on order_id. In both cases a replacement keeps its join key, so the tests model a real shape. Added a comment on the spec
2. Pseudocode drift
Synced
3. Delete-by-key + non-equi
Confirmed. It's indeed an issue for this case. I've created the ticket https://issues.apache.org/jira/browse/FLINK-40858
--
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-40841][table] Non-equi joins fix: retract the matches that a replaced record loses [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on PR #29339:
URL: https://github.com/apache/flink/pull/29339#issuecomment-5935459104
One more ticket came out of the fix and research around this ticket: https://issues.apache.org/jira/browse/FLINK-40875. A possible optimization to make some pipelines drop ChangelogNormalize when the join can provide full deletes, if required.
--
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