gustavodemorais commented on code in PR #29583:
URL: https://github.com/apache/flink/pull/29583#discussion_r4483270871
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java:
##########
@@ -980,4 +1201,24 @@ private void initializeJoinConditions() throws Exception {
}
}
}
+
+ /** Returns whether the input may update a record without retracting it
first. */
+ private static boolean isUpsertOrUnknown(@Nullable RuntimeChangelogMode
inputChangelogMode) {
+ // compiled plans before Flink 2.4 do not contain the changelog mode,
and handling
+ // their inputs as upsert is always correct
+ if (inputChangelogMode == null) {
Review Comment:
For plans compiled before 2.4 this handles an input as upsert whenever its
unique key contains the common join key, but a unique key isn't always an
upsert key. E.g. `Users u JOIN (SELECT o.* FROM Orders o JOIN Regions r ON
o.region = r.region) o2 ON u.user_id = o2.user_id AND u.region = o2.region`:
when an order moves from EU to US, its -U and +U come from different upstream
subtasks, so the +U can arrive first and the late -U is then retracted a second
time.
Could unknown modes keep the released handling instead (return false for
null here and in `hasKeyOnlyDeletesOrUnknown`)? Then old plans behave exactly
like the version that compiled them. Or do you have other suggestions?
--
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]
