gustavodemorais commented on code in PR #29570:
URL: https://github.com/apache/flink/pull/29570#discussion_r4379179508
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java:
##########
@@ -102,6 +132,102 @@ public void close() throws Exception {
}
}
+ /**
+ * Returns the stored record that the given record replaces, or null if
there is none. An upsert
+ * input updates a record by sending only the new version, without
retracting the old one.
+ *
+ * <p>The lookup only happens if it can change the result.
+ */
+ protected @Nullable RowData replacedRecord(
+ RowData record, JoinRecordStateView stateView, boolean isLeft,
boolean otherIsOuter)
+ throws Exception {
+ final boolean inputIsUpsertOrUnknown =
+ isLeft ? leftIsUpsertOrUnknown : rightIsUpsertOrUnknown;
+ // with at most one record per join key, the count of the other side
is right without it
+ final boolean mayCountTwice = otherIsOuter &&
!joinKeyContainsUniqueKey(isLeft);
+ final boolean mayMatchDifferently = hasNonEquiCondition;
+ if (!inputIsUpsertOrUnknown || !(mayCountTwice ||
mayMatchDifferently)) {
+ return null;
+ }
+ return stateView.getRecord(record);
+ }
+
+ /**
+ * Returns the full record to retract.
+ *
+ * <p>A DELETE may contain only the unique key, e.g. from an upsert
source, but a non-equi
+ * condition needs all fields. In that case, the stored record with the
same unique key is
+ * returned, or the given retraction if nothing is stored.
+ */
+ protected RowData retractedRecord(
+ RowData retraction,
+ RowKind retractionKind,
+ JoinRecordStateView stateView,
+ boolean isLeft)
+ throws Exception {
+ final boolean inputHasKeyOnlyDeletesOrUnknown =
+ isLeft ? leftHasKeyOnlyDeletesOrUnknown :
rightHasKeyOnlyDeletesOrUnknown;
+ final boolean mayBeKeyOnlyDelete =
+ retractionKind == RowKind.DELETE
+ && inputHasKeyOnlyDeletesOrUnknown
+ && hasUniqueKey(isLeft);
+ if (!mayBeKeyOnlyDelete || !hasNonEquiCondition) {
+ return retraction;
+ }
+ final RowData storedRecord = stateView.getRecord(retraction);
+ return storedRecord != null ? storedRecord : retraction;
Review Comment:
When nothing is stored, a key-only DELETE falls back to the retraction with
NULL non-key fields. With a NULL-tolerant condition like `r.k = l.k AND (r.v IS
NULL OR r.v = l.v)` and a right PK `(k, id)`, a key-only `-D(k1, r9)` that was
never stored matches every left row of `k1`, so the semi join sends -D (anti:
+I) although the real match `(k1, r1, A)` is still in state. Could we return
null here and skip the other side in that case?
##########
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:
A retraction can only remove a match that was counted, so could we return
early when `other.numOfAssociations <= 0` instead of writing a negative count?
(not sure)
##########
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 {
Review Comment:
nit: with a non-equi condition a right upsert now iterates all left rows of
the key twice, once here and once in the accumulate loop. Could we do it in one
pass, evaluating both `matches(replaced, l)` and `matches(input, l)` per left
row?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/AbstractStreamingJoinOperator.java:
##########
@@ -55,6 +60,21 @@ public abstract class AbstractStreamingJoinOperator extends
AbstractStreamOperat
protected final JoinInputSideSpec leftInputSideSpec;
protected final JoinInputSideSpec rightInputSideSpec;
+ // changelog mode of the left input, null for compiled plans before Flink
2.4
+ @Nullable protected final RuntimeChangelogMode leftInputChangelogMode;
Review Comment:
nit: `leftInputChangelogMode`/`rightInputChangelogMode` are never read, only
the booleans derived from them. Could we drop the fields?
##########
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);
Review Comment:
A DELETE for a right row that is not in state still decrements the left
count to -1, and since `isNewMatch` never increments a non-zero count when the
join key contains the right unique key, the left row never comes back. With
`lines` PK `order_id` (upsert, key-only deletes) and `SELECT * FROM orders o
WHERE EXISTS (SELECT 1 FROM lines l WHERE l.order_id = o.order_id)`:
```
L +I(Ord#1, LineOrd#1, ADDR)
R -D(LineOrd#1) -- never inserted
R +I(LineOrd#1, AIR) -- expected +I[Ord#1], nothing emitted
R +U(LineOrd#1, SHIP)
R -D(LineOrd#1)
R +I(LineOrd#1, AIR) -- still nothing
```
The anti join is stuck the same way. Before the change the next upsert
brought the count back. Could we skip the retraction when the right side has a
unique key and nothing is stored for it, and add a harness test with a DELETE
before the first INSERT for semi and anti?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingSemiAntiJoinOperator.java:
##########
@@ -109,29 +122,37 @@ public void open() throws Exception {
@Override
public void processElement1(StreamRecord<RowData> element) throws
Exception {
RowData input = element.getValue();
+ boolean isAccumulateMsg = RowDataUtil.isAccumulateMsg(input);
+ RowKind inputRowKind = input.getRowKind();
+ input.setRowKind(RowKind.INSERT); // erase RowKind for later state
updating
+
+ final RowData record =
+ isAccumulateMsg
+ ? input
+ : retractedRecord(input, inputRowKind,
leftRecordStateView, true);
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);
+ record, true, rightRecordStateView, joinCondition);
+ final boolean recordIsSent = isAntiJoin != associatedRecords.hasNext();
+ // count before the state of the other side is read again, which may
reuse the iterable
+ final int numOfAssociations = isAccumulateMsg ?
Iterators.size(associatedRecords) : 0;
+ if (isAccumulateMsg && !recordIsSent) {
+ final RowData replacedRecord = replacedRecord(input,
leftRecordStateView, true, false);
+ if (replacedRecord != null && isSent(replacedRecord)) {
Review Comment:
Suggestion: the left state already stores the replaced record's
`numOfAssociations`, so could we decide whether it was sent from that count
(`isAntiJoin ? count == 0 : count > 0`) and drop `isSent`? It saves a second
scan of the right state, and it also retracts correctly when the right state
changed without retractions (TTL, or counts restored from before the fix).
--
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]
