rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4405983515
##########
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:
Added `testRightInsertMatchesLeftRecordWithNegativeCount` to cover this.
It'll write -1 through the state view and then insert a corresponding, matching
right row (and fail when the check goes back to 0).
--
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]
