rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4405961579
##########
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:
Done. When the join key contains the right unique key, removeMatch now sets
the count to 0 and verified by an additional inflated-count test.
--
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]
