rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4379875523
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java:
##########
@@ -177,52 +209,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);
+ if (replacedRecord != null && hasNonEquiCondition) {
+ removeLostMatches(input, replacedRecord);
+ }
// do not store a record that can never match
if (!joinCondition.currentKeyCannotMatch()) {
rightRecordStateView.addRecord(input);
}
+ Iterator<OuterRecord> associatedRecords =
+ AbstractStreamingJoinOperator.iterator(
+ input, false, leftRecordStateView, joinCondition);
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);
- }
- collector.collect(other);
- // set header back to INSERT, because we will update the
other row to state
- other.setRowKind(RowKind.INSERT);
+ // send -D[other] if anti join, +I/+U[other] (using input
RowKind) if semi join
+ collect(other, isAntiJoin ? RowKind.DELETE : inputRowKind);
} // ignore when number > 0
- leftRecordStateView.updateNumOfAssociations(
- other, outerRecord.numOfAssociations + 1);
+ if (isNewMatch(outerRecord, replacedRecord, false)) {
+ leftRecordStateView.updateNumOfAssociations(
+ other, outerRecord.numOfAssociations + 1);
+ }
}
} else { // retract input
- rightRecordStateView.retractRecord(input);
+ final RowData retractedRecord =
+ retractedRecord(input, inputRowKind, rightRecordStateView,
false);
Review Comment:
Good catch, this makes sense. I went ahead and corrected this in two ways:
- The `removeMatch` can no longer decrement below zero.
- When the right key is _wider_ than the join, a DELETE from an upsert input
now looks up the stored right row and is skipped if nothing is stored
(otherwise it would remove the match of another right row with the same key).
Added tests to cover these cases. 👍
--
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]
