rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4405924368
##########
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(
+ OuterRecord other, @Nullable RowData replacedRecord, boolean
inputIsLeft) {
+ if (other.numOfAssociations == 0) {
+ return true;
+ }
+ // with at most one record per join key, the existing match is the
replaced record
+ if (joinKeyContainsUniqueKey(inputIsLeft)) {
+ return false;
+ }
+ return replacedRecord == null || !matches(replacedRecord,
other.record, inputIsLeft);
+ }
+
+ protected boolean matches(RowData inputRecord, RowData otherRecord,
boolean inputIsLeft) {
+ return inputIsLeft
+ ? joinCondition.apply(inputRecord, otherRecord)
+ : joinCondition.apply(otherRecord, inputRecord);
+ }
+
+ protected boolean joinKeyContainsUniqueKey(boolean isLeft) {
+ return (isLeft ? leftInputSideSpec :
rightInputSideSpec).joinKeyContainsUniqueKey();
+ }
+
+ private boolean hasUniqueKey(boolean isLeft) {
+ return (isLeft ? leftInputSideSpec :
rightInputSideSpec).hasUniqueKey();
+ }
+
+ /** 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
Review Comment:
Okay, that's good to know. Only an input whose changelog mode is known to be
upsert now skips the DELETE of a record that is not stored (and plans without
go back to the old handling). I've gone ahead and added tests to cover these as
well.
--
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]
