gustavodemorais commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4404033244
##########
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:
Plans compiled before 2.4 have no changelog mode, so retract inputs now also
skip DELETEs of rows that are not stored. With state TTL, the DELETE of an
expired row is lost:
```
-- 2.3 plan, TTL 1s: SELECT * FROM orders o WHERE o.line_order_id IN (SELECT
line_order_id FROM lines)
t=1 R +I(LineOrd#1, AIR)
t=500 L +I(Ord#1, LineOrd#1, ADDR) -- +I[Ord#1]
t=1200 R -D(LineOrd#1, AIR) -- expected -D[Ord#1], nothing emitted
```
Same for the regular join and the semi/anti left side. Could we skip these
DELETEs only when the mode is known, keep the old fallback for null modes, and
add a test for both?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java:
##########
@@ -177,52 +219,81 @@ public void processElement2(StreamRecord<RowData>
element) throws Exception {
RowKind inputRowKind = input.getRowKind();
input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
- Iterator<OuterRecord> associatedRecords =
- AbstractStreamingJoinOperator.iterator(
- input, false, leftRecordStateView, joinCondition);
if (isAccumulateMsg) { // record is accumulate
+ final RowData replacedRecord = replacedRecord(input,
rightRecordStateView, false, true);
// do not store a record that can never match
if (!joinCondition.currentKeyCannotMatch()) {
rightRecordStateView.addRecord(input);
}
- while (associatedRecords.hasNext()) {
- OuterRecord outerRecord = associatedRecords.next();
- RowData other = outerRecord.record;
- if (outerRecord.numOfAssociations == 0) {
- if (isAntiJoin) {
- // send -D[other]
- other.setRowKind(RowKind.DELETE);
- } else {
- // send +I/+U[other] (using input RowKind)
- other.setRowKind(inputRowKind);
+ // a single pass for the matches that the record gains and the
replaced record loses
+ for (Tuple2<RowData, Integer> left :
+ leftRecordStateView.getRecordsAndNumOfAssociations()) {
+ final RowData other = left.f0;
+ final int numOfAssociations = left.f1;
+ final boolean replacedRecordMatches =
+ replacedRecord != null && matches(replacedRecord,
other, false);
+ if (matches(input, other, false)) {
+ if (numOfAssociations <= 0) {
+ // send -D[other] if anti join, +I/+U[other] (using
input RowKind) if semi
+ collect(other, isAntiJoin ? RowKind.DELETE :
inputRowKind);
+ leftRecordStateView.updateNumOfAssociations(other, 1);
+ } else if (!replacedRecordMatches &&
!joinKeyContainsUniqueKey(false)) {
+ // with at most one record per join key, the existing
match is the
+ // replaced record
+ leftRecordStateView.updateNumOfAssociations(other,
numOfAssociations + 1);
}
- collector.collect(other);
- // set header back to INSERT, because we will update the
other row to state
- other.setRowKind(RowKind.INSERT);
- } // ignore when number > 0
- leftRecordStateView.updateNumOfAssociations(
- other, outerRecord.numOfAssociations + 1);
+ } else if (replacedRecordMatches) {
+ removeMatch(other, numOfAssociations, RowKind.DELETE);
+ }
}
} else { // retract input
- rightRecordStateView.retractRecord(input);
+ final RowData retractedRecord =
+ inputRowKind == RowKind.DELETE &&
mayDeleteRecordsThatAreNotStored(false)
+ ? rightRecordStateView.getRecord(input)
+ : retractedRecord(input, inputRowKind,
rightRecordStateView, false);
+ if (retractedRecord == null) {
+ // nothing is stored, so no left record counts it as a match
+ return;
+ }
+ rightRecordStateView.retractRecord(retractedRecord);
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
+ retractedRecord, false, leftRecordStateView,
joinCondition);
while (associatedRecords.hasNext()) {
- OuterRecord outerRecord = associatedRecords.next();
- RowData other = outerRecord.record;
- if (outerRecord.numOfAssociations == 1) {
- if (!isAntiJoin) {
- // send -D/-U[other] (using input RowKind)
- other.setRowKind(inputRowKind);
- } else {
- // send +I[other]
- other.setRowKind(RowKind.INSERT);
- }
- collector.collect(other);
- // set RowKind back, because we will update the other row
to state
- other.setRowKind(RowKind.INSERT);
- } // ignore when number > 0
- leftRecordStateView.updateNumOfAssociations(
- other, outerRecord.numOfAssociations - 1);
+ final OuterRecord outerRecord = associatedRecords.next();
+ removeMatch(outerRecord.record, outerRecord.numOfAssociations,
inputRowKind);
}
}
}
+
+ /**
+ * Removes a match of the left record. If it was the last one, a semi join
retracts the record
+ * with the given RowKind and an anti join sends it.
+ */
+ private void removeMatch(RowData other, int numOfAssociations, RowKind
retractionKind)
+ throws Exception {
+ if (numOfAssociations <= 0) {
+ // the retracted record was not counted, e.g. it had expired when
the left record was
+ // stored
+ return;
+ }
+ if (numOfAssociations == 1) {
Review Comment:
Old savepoints hold inflated counts (2, 5, ...) when the right PK is the
join key, and a DELETE only takes 1 off, so the left row stays. With at most
one right row per key here, could we write 0 for any positive count when
`joinKeyContainsUniqueKey(false)`?
##########
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperatorTest.java:
##########
@@ -327,6 +335,355 @@ private void assertSkipsNullKeys(boolean isAntiJoin)
throws Exception {
assertThat(testHarness.numKeyedStateEntries()).isOne();
}
+ /** A right row replaced without UPDATE_BEFORE is counted once, so its
DELETE removes it. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightUpsertIsCountedOnce(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec, joinCondition,
false);
+ assertRightUpsertIsCountedOnce(isAntiJoin, "TRUCK");
+ }
+
+ /** The same holds if the unique key of the right side contains more than
the join key. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightUpsertWithUniqueKeyIsCountedOnce(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightHasUniqueKeySpec(),
joinCondition, false);
+ assertRightUpsertIsCountedOnce(isAntiJoin, "AIR");
+ }
+
+ private void assertRightUpsertIsCountedOnce(boolean isAntiJoin, String
newShipMode)
+ throws Exception {
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1",
newShipMode));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", newShipMode));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /** A right row replaced without UPDATE_BEFORE that no longer matches
removes its match. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightUpsertRemovesLostMatch(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1", UNKNOWN));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /** A DELETE of only the key of a right row is joined with the stored row.
*/
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightKeyOnlyDeleteRemovesMatch(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * A left row replaced without UPDATE_BEFORE retracts the old version if
it was emitted but the
+ * new version is not.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftUpsertRetractsOldVersion(boolean isAntiJoin) throws Exception
{
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ null));
+
+ testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1",
UNKNOWN));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
UNKNOWN)));
+
+ testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1",
NEW_ADDRESS));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
NEW_ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
UNKNOWN)));
+ }
+
+ /**
+ * A DELETE of only the key of a left row is joined with the stored row,
so it is emitted with
+ * all fields if the row was emitted, and not at all otherwise.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftKeyOnlyDelete(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
knownValuesNonEquiCondition(), true);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement1(insertRecord("Ord#2", "LineOrd#2",
ADDRESS));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#2", "LineOrd#2",
ADDRESS)));
+
+ testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ null));
+
+ testHarness.processElement1(deleteRecord("Ord#2", "LineOrd#2", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ null,
+ rowOfKind(RowKind.DELETE, "Ord#2", "LineOrd#2",
ADDRESS)));
+ }
+
+ /** A DELETE of a right row that was never inserted does not remove a
match. */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightDeleteBeforeInsert(boolean isAntiJoin) throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec, joinCondition,
false);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", null));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS)));
+
+ testHarness.processElement2(updateAfterRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * A DELETE of a right row that was never inserted does not remove the
match of another right
+ * row with the same join key.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightDeleteOfRecordThatIsNotStored(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightHasUniqueKeySpec(),
joinCondition, false);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", "AIR"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * A retraction of a right row that the left row did not count, because
the right row had
+ * expired when the left row was stored, does not make the count negative.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testRightRetractionOfExpiredRecord(boolean isAntiJoin) throws
Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin,
+ rightInputSpec,
+ joinCondition,
+ false,
+ RuntimeChangelogMode.serialize(ChangelogMode.all()),
+ 4000L,
+ 1000L);
+ testHarness.setStateTtlProcessingTime(1);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.setStateTtlProcessingTime(2001);
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement2(deleteRecord("LineOrd#1", "AIR"));
+ assertor.shouldEmitNothing(testHarness);
+
+ testHarness.processElement2(insertRecord("LineOrd#1", "SHIP"));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.INSERT, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS)));
+ }
+
+ /**
+ * Whether the replaced left row was emitted is decided by its stored
number of matches, so it
+ * is retracted even if its matches expired.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftUpsertRetractsOldVersionAfterMatchExpired(boolean isAntiJoin)
throws Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin,
+ rightInputSpec,
+ knownValuesNonEquiCondition(),
+ true,
+ UPSERT_WITH_KEY_ONLY_DELETES,
+ 4000L,
+ 1000L);
+ testHarness.setStateTtlProcessingTime(1);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.processElement1(insertRecord("Ord#1", "LineOrd#1",
ADDRESS));
+ testHarness.getOutput().clear();
+
+ testHarness.setStateTtlProcessingTime(2001);
+ testHarness.processElement1(updateAfterRecord("Ord#1", "LineOrd#1",
UNKNOWN));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(
+ isAntiJoin,
+ rowOfKind(RowKind.DELETE, "Ord#1", "LineOrd#1",
ADDRESS),
+ rowOfKind(RowKind.UPDATE_AFTER, "Ord#1", "LineOrd#1",
UNKNOWN)));
+ }
+
+ /**
+ * A DELETE of only the key of a left row that is not stored matches
nothing, even if its null
+ * fields would match.
+ */
+ @ParameterizedTest(name = "isAntiJoin = {0}")
+ @ValueSource(booleans = {false, true})
+ void testLeftKeyOnlyDeleteOfRecordThatIsNotStored(boolean isAntiJoin)
throws Exception {
+ useSemiAntiJoinOperator(isAntiJoin, rightInputSpec,
nullTolerantNonEquiCondition(), true);
+ testHarness.processElement2(insertRecord("LineOrd#1", "AIR"));
+ testHarness.getOutput().clear();
+
+ testHarness.processElement1(deleteRecord("Ord#1", "LineOrd#1", null));
+ assertor.shouldEmit(
+ testHarness,
+ emitted(isAntiJoin, null, rowOfKind(RowKind.DELETE, "Ord#1",
"LineOrd#1", null)));
+ }
+
+ private void useSemiAntiJoinOperator(
+ boolean isAntiJoin,
+ JoinInputSideSpec rightSpec,
+ GeneratedJoinCondition condition,
+ boolean hasNonEquiCondition)
+ throws Exception {
+ useSemiAntiJoinOperator(
+ isAntiJoin,
+ rightSpec,
+ condition,
+ hasNonEquiCondition,
+ UPSERT_WITH_KEY_ONLY_DELETES,
+ 0L,
+ 0L);
+ }
+
+ private void useSemiAntiJoinOperator(
+ boolean isAntiJoin,
+ JoinInputSideSpec rightSpec,
+ GeneratedJoinCondition condition,
+ boolean hasNonEquiCondition,
+ RuntimeChangelogMode rightChangelogMode,
+ long leftStateRetentionTime,
+ long rightStateRetentionTime)
+ throws Exception {
+ testHarness.close();
+ testHarness =
+ new KeyedTwoInputStreamOperatorTestHarness<>(
+ new StreamingSemiAntiJoinOperator(
+ isAntiJoin,
+ leftTypeInfo,
+ rightTypeInfo,
+ condition,
+ leftInputSpec,
Review Comment:
All tests use `leftInputSpec` (unique key = join key), so
`InputSideHasUniqueKey#getRecordAndNumOfAssociations` never runs. Could we add
a left spec with unique key `(order_id, line_order_id)` and two left rows per
join key?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingJoinOperator.java:
##########
@@ -365,16 +350,26 @@ protected void processElement(
// skip when there is no matched rows on the other side
}
} else { // input record is retract
- final RowData retractedRecord =
- retractedRecord(
- input, inputRowKind, inputSideStateView,
inputIsLeft, isSuppress);
+ // the retraction of a mini-batch pair is used as it is
+ final RowData fullRecord =
+ isSuppress
+ ? input
+ : retractedRecord(input, inputRowKind,
inputSideStateView, inputIsLeft);
+ // a key-only DELETE of a record that is not stored matches nothing
Review Comment:
This changes the regular join (FLINK-40858) under the semi/anti ticket.
Could we keep `StreamingJoinOperator` as before and move this to the separate
ticket?
##########
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:
When the join key contains the right unique key, the clamp in `removeMatch`
already covers this. Could we add `&& !joinKeyContainsUniqueKey(isLeft)`, as in
your reply? It saves a state read per right DELETE in the most common shape.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java:
##########
@@ -177,52 +219,81 @@ public void processElement2(StreamRecord<RowData>
element) throws Exception {
RowKind inputRowKind = input.getRowKind();
input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
- Iterator<OuterRecord> associatedRecords =
- AbstractStreamingJoinOperator.iterator(
- input, false, leftRecordStateView, joinCondition);
if (isAccumulateMsg) { // record is accumulate
+ final RowData replacedRecord = replacedRecord(input,
rightRecordStateView, false, true);
// do not store a record that can never match
if (!joinCondition.currentKeyCannotMatch()) {
rightRecordStateView.addRecord(input);
}
- while (associatedRecords.hasNext()) {
- OuterRecord outerRecord = associatedRecords.next();
- RowData other = outerRecord.record;
- if (outerRecord.numOfAssociations == 0) {
- if (isAntiJoin) {
- // send -D[other]
- other.setRowKind(RowKind.DELETE);
- } else {
- // send +I/+U[other] (using input RowKind)
- other.setRowKind(inputRowKind);
+ // a single pass for the matches that the record gains and the
replaced record loses
+ for (Tuple2<RowData, Integer> left :
+ leftRecordStateView.getRecordsAndNumOfAssociations()) {
+ final RowData other = left.f0;
+ final int numOfAssociations = left.f1;
+ final boolean replacedRecordMatches =
+ replacedRecord != null && matches(replacedRecord,
other, false);
+ if (matches(input, other, false)) {
+ if (numOfAssociations <= 0) {
Review Comment:
Only old state has negative counts here, and no test reaches this (reverting
to `== 0` keeps all tests green). Could we add a test that writes a left row
with count -1 and then inserts a matching right row?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java:
##########
@@ -109,29 +124,47 @@ public void open() throws Exception {
@Override
public void processElement1(StreamRecord<RowData> element) throws
Exception {
RowData input = element.getValue();
- Iterator<OuterRecord> associatedRecords =
- AbstractStreamingJoinOperator.iterator(
- input, true, rightRecordStateView, joinCondition);
- if (!associatedRecords.hasNext()) {
- if (isAntiJoin) {
- collector.collect(input);
- }
- } else { // there are matched rows on the other side
- if (!isAntiJoin) {
- collector.collect(input);
+ boolean isAccumulateMsg = RowDataUtil.isAccumulateMsg(input);
+ RowKind inputRowKind = input.getRowKind();
+ input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
+
+ if (isAccumulateMsg) {
+ final int numOfAssociations =
+ Iterators.size(
+ AbstractStreamingJoinOperator.iterator(
+ input, true, rightRecordStateView,
joinCondition));
+ if (isSent(numOfAssociations)) {
+ collect(input, inputRowKind);
+ } else if (needsReplacedRecord(true, false)) {
+ final Tuple2<RowData, Integer> replaced =
+
leftRecordStateView.getRecordAndNumOfAssociations(input);
+ if (replaced != null && isSent(replaced.f1)) {
+ // send -D[replaced]
+ collect(replaced.f0, RowKind.DELETE);
+ }
}
- }
- if (RowDataUtil.isAccumulateMsg(input)) {
- // erase RowKind for state updating
- input.setRowKind(RowKind.INSERT);
// do not store a record that can never match
if (!joinCondition.currentKeyCannotMatch()) {
- leftRecordStateView.addRecord(input,
Iterators.size(associatedRecords));
+ leftRecordStateView.addRecord(input, numOfAssociations);
}
- } else { // input is retract
- // erase RowKind for state updating
- input.setRowKind(RowKind.INSERT);
- leftRecordStateView.retractRecord(input);
+ } else {
+ final RowData retractedRecord =
+ retractedRecord(input, inputRowKind, leftRecordStateView,
true);
+ if (retractedRecord == null) {
+ // nothing is stored for the key-only DELETE, which matches
nothing
+ if (isAntiJoin) {
+ collect(input, inputRowKind);
+ }
+ return;
+ }
+ final boolean hasMatches =
Review Comment:
Suggestion: a left DELETE still scans the right state. If the stored count
disagrees (expired match, inflated count), the semi join doesn't retract and
the anti join sends a second -D. Could we use the stored count here too, like
the upsert branch?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java:
##########
@@ -565,6 +566,57 @@ public class JoinTestPrograms {
+ "FROM orders o JOIN lines l "
+ "ON o.order_id = l.order_id AND
l.status <> o.excluded_status")
.build();
+
+ // the line of order 1 is updated without UPDATE_BEFORE and deleted by
key, the full image
+ // of the deleted line is only in the restored state; order 2 stops
matching its line
+ SEMI_JOIN_UPSERT_INPUTS_NON_EQUI =
Review Comment:
Could we add a 2.3 plan variant of this program, like
`LEFT_JOIN_UPSERT_INPUT_NON_EQUI_PRE_2_4`, that updates and deletes restored
rows? The existing pre-2.4 semi/anti programs only insert after the restore, so
they never reach the new lookups.
##########
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:
nit: only `StreamingJoinOperator` uses `isNewMatch` now. Could we keep it
private there?
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/JoinTestPrograms.java:
##########
@@ -565,6 +566,57 @@ public class JoinTestPrograms {
+ "FROM orders o JOIN lines l "
+ "ON o.order_id = l.order_id AND
l.status <> o.excluded_status")
.build();
+
+ // the line of order 1 is updated without UPDATE_BEFORE and deleted by
key, the full image
+ // of the deleted line is only in the restored state; order 2 stops
matching its line
Review Comment:
nit: the deleted line's full image comes from the UPDATE_AFTER after the
restore; only the replaced version is in the restored state.
```suggestion
// the line of order 1 is updated without UPDATE_BEFORE and deleted
by key, the replaced
// version of the line is only in the restored state; order 2 stops
matching its line
```
--
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]
