rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4379927010
##########
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);
+ 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);
+ removeMatch(associatedRecords.next(), inputRowKind);
}
}
}
+
+ /** Removes the matches of the left records that only the replaced right
record matched. */
+ private void removeLostMatches(RowData record, RowData replacedRecord)
throws Exception {
+ final Iterator<OuterRecord> matchesOfReplacedRecord =
+ AbstractStreamingJoinOperator.iterator(
+ replacedRecord, false, leftRecordStateView,
joinCondition);
+ while (matchesOfReplacedRecord.hasNext()) {
+ final OuterRecord other = matchesOfReplacedRecord.next();
+ if (!matches(record, other.record, false)) {
+ removeMatch(other, RowKind.DELETE);
+ }
+ }
+ }
+
+ /**
+ * 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(OuterRecord other, RowKind retractionKind) throws
Exception {
+ if (other.numOfAssociations == 1) {
+ // send -D/-U[other] if semi join, +I[other] if anti join
+ collect(other.record, isAntiJoin ? RowKind.INSERT :
retractionKind);
+ } // ignore when number > 1
+ leftRecordStateView.updateNumOfAssociations(other.record,
other.numOfAssociations - 1);
Review Comment:
I think you're right on this one. A left row's count is recomputed from the
right state whenever the left row is stored, so a retraction that finds a count
of 0 is for a row that was never counted (e.g. one that had expired when the
left row was stored). The `removeMatch` call now returns early.
--
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]
