rionmonster commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4405974899
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java:
##########
@@ -109,29 +124,47 @@ public void open() throws Exception {
@Override
public void processElement1(StreamRecord<RowData> element) throws
Exception {
RowData input = element.getValue();
- Iterator<OuterRecord> associatedRecords =
- AbstractStreamingJoinOperator.iterator(
- input, true, rightRecordStateView, joinCondition);
- if (!associatedRecords.hasNext()) {
- if (isAntiJoin) {
- collector.collect(input);
- }
- } else { // there are matched rows on the other side
- if (!isAntiJoin) {
- collector.collect(input);
+ boolean isAccumulateMsg = RowDataUtil.isAccumulateMsg(input);
+ RowKind inputRowKind = input.getRowKind();
+ input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
+
+ if (isAccumulateMsg) {
+ final int numOfAssociations =
+ Iterators.size(
+ AbstractStreamingJoinOperator.iterator(
+ input, true, rightRecordStateView,
joinCondition));
+ if (isSent(numOfAssociations)) {
+ collect(input, inputRowKind);
+ } else if (needsReplacedRecord(true, false)) {
+ final Tuple2<RowData, Integer> replaced =
+
leftRecordStateView.getRecordAndNumOfAssociations(input);
+ if (replaced != null && isSent(replaced.f1)) {
+ // send -D[replaced]
+ collect(replaced.f0, RowKind.DELETE);
+ }
}
- }
- if (RowDataUtil.isAccumulateMsg(input)) {
- // erase RowKind for state updating
- input.setRowKind(RowKind.INSERT);
// do not store a record that can never match
if (!joinCondition.currentKeyCannotMatch()) {
- leftRecordStateView.addRecord(input,
Iterators.size(associatedRecords));
+ leftRecordStateView.addRecord(input, numOfAssociations);
}
- } else { // input is retract
- // erase RowKind for state updating
- input.setRowKind(RowKind.INSERT);
- leftRecordStateView.retractRecord(input);
+ } else {
+ final RowData retractedRecord =
+ retractedRecord(input, inputRowKind, leftRecordStateView,
true);
+ if (retractedRecord == null) {
+ // nothing is stored for the key-only DELETE, which matches
nothing
+ if (isAntiJoin) {
+ collect(input, inputRowKind);
+ }
+ return;
+ }
+ final boolean hasMatches =
Review Comment:
Sounds good! Adjusted so the left side has a unique key, a retraction uses
the stored record and its count (in cases without a unique key it still scans).
--
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]
