rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4379910763
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java:
##########
@@ -102,6 +132,102 @@ public void close() throws Exception {
}
}
+ /**
+ * Returns the stored record that the given record replaces, or null if
there is none. An upsert
+ * input updates a record by sending only the new version, without
retracting the old one.
+ *
+ * <p>The lookup only happens if it can change the result.
+ */
+ protected @Nullable RowData replacedRecord(
+ RowData record, JoinRecordStateView stateView, boolean isLeft,
boolean otherIsOuter)
+ throws Exception {
+ final boolean inputIsUpsertOrUnknown =
+ isLeft ? leftIsUpsertOrUnknown : rightIsUpsertOrUnknown;
+ // with at most one record per join key, the count of the other side
is right without it
+ final boolean mayCountTwice = otherIsOuter &&
!joinKeyContainsUniqueKey(isLeft);
+ final boolean mayMatchDifferently = hasNonEquiCondition;
+ if (!inputIsUpsertOrUnknown || !(mayCountTwice ||
mayMatchDifferently)) {
+ return null;
+ }
+ return stateView.getRecord(record);
+ }
+
+ /**
+ * Returns the full record to retract.
+ *
+ * <p>A DELETE may contain only the unique key, e.g. from an upsert
source, but a non-equi
+ * condition needs all fields. In that case, the stored record with the
same unique key is
+ * returned, or the given retraction if nothing is stored.
+ */
+ protected RowData retractedRecord(
+ RowData retraction,
+ RowKind retractionKind,
+ JoinRecordStateView stateView,
+ boolean isLeft)
+ throws Exception {
+ final boolean inputHasKeyOnlyDeletesOrUnknown =
+ isLeft ? leftHasKeyOnlyDeletesOrUnknown :
rightHasKeyOnlyDeletesOrUnknown;
+ final boolean mayBeKeyOnlyDelete =
+ retractionKind == RowKind.DELETE
+ && inputHasKeyOnlyDeletesOrUnknown
+ && hasUniqueKey(isLeft);
+ if (!mayBeKeyOnlyDelete || !hasNonEquiCondition) {
+ return retraction;
+ }
+ final RowData storedRecord = stateView.getRecord(retraction);
+ return storedRecord != null ? storedRecord : retraction;
Review Comment:
Done. The `retractedRecord` call now returns null when nothing is stored,
and the callers treat the DELETE as matching nothing, so the other side is
skipped.
--
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]
