rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4406009973
##########
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);
+ }
+
+ /**
+ * Returns whether the other record gains a match, so that its number of
associations increases.
+ * It does not if the replaced record matched it already.
+ */
+ protected boolean isNewMatch(
Review Comment:
Sure, it's now private in `StreamingJoinOperator` again.
--
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]
