rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4405934192
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java:
##########
@@ -102,6 +126,114 @@ 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 {
+ return needsReplacedRecord(isLeft, otherIsOuter) ?
stateView.getRecord(record) : null;
+ }
+
+ /**
+ * Returns whether looking up the record that an input record replaces can
change the result.
+ */
+ protected boolean needsReplacedRecord(boolean isLeft, boolean
otherIsOuter) {
+ 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;
+ return inputIsUpsertOrUnknown && (mayCountTwice ||
mayMatchDifferently);
+ }
+
+ /**
+ * Returns the full record to retract, or null if it is not stored and
matches nothing.
+ *
+ * <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. If nothing is stored, e.g. because the record was never
inserted, its null fields
+ * must not match the other side.
+ */
+ protected @Nullable 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;
+ }
+ return stateView.getRecord(retraction);
+ }
+
+ /**
+ * Returns whether a DELETE of the input may be for a record that was
never inserted, e.g. an
+ * upsert source deletes a key that it never inserted.
+ */
+ protected boolean mayDeleteRecordsThatAreNotStored(boolean isLeft) {
+ return (isLeft ? leftIsUpsertOrUnknown : rightIsUpsertOrUnknown) &&
hasUniqueKey(isLeft);
Review Comment:
I've addressed this but _slightly_ differently at the right-DELETE call site
rather than in `mayDeleteRecordsThatAreNotStored`. Iterated a bit with Claude
on it and In the method it would also apply to the left side, where a DELETE of
a row that isn't stored should still match nothing when the left join key
contains its unique key.
--
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]
