rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4405951826
##########
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:
I've gone ahead and reverted this back to what it was previously (apart from
one of the comments re: `isNewMatch`). I've created FLINK-41258 to track this
separately (feel free to assign) and linked to FLINK-40858 (and one of the
other related issues).
--
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]
