rionmonster commented on code in PR #29583:
URL: https://github.com/apache/flink/pull/29583#discussion_r4486597595
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java:
##########
@@ -406,10 +444,152 @@ public void processElement(int inputId,
StreamRecord<RowData> element) throws Ex
return;
}
+ if (isUpsert(input)) {
+ final RowData replacedRecord = replacedRecord(input, inputId);
+ if (replacedRecord != null) {
+ replaceRecord(replacedRecord, input, inputId);
+ return;
+ }
+ } else {
+ input = retractedRecord(input, inputId);
+ }
performMultiJoin(input, inputId);
addRecordToState(input, inputId);
}
+ /**
+ * Returns the stored record that the given record replaces, or null if
there is none or if the
+ * join conditions match both versions the same way. An upsert input
updates a record by sending
+ * only the new version, without retracting the old one.
+ */
+ private @Nullable RowData replacedRecord(RowData record, int inputId)
throws Exception {
+ if (!isUpsertOrUnknown[inputId]
+ || !inputSpecs.get(inputId).hasUniqueKey()
+ || conditionFields[inputId].length == 0) {
+ return null;
+ }
+ final RowData storedRecord =
+ stateHandlers
+ .get(inputId)
+ .getRecord(keyExtractor.getJoinKey(record, inputId),
record);
+ if (storedRecord == null || conditionFieldsEqual(storedRecord, record,
inputId)) {
+ // the new version produces the same joined rows, which update the
old ones
+ return null;
+ }
+ return storedRecord;
+ }
+
+ /**
+ * Returns the full record to retract. A DELETE may contain only the
unique key, e.g. from an
+ * upsert source, so the stored record with the same unique key is
returned if there is one.
+ */
+ private RowData retractedRecord(RowData retraction, int inputId) throws
Exception {
+ if (retraction.getRowKind() != RowKind.DELETE
+ || !hasKeyOnlyDeletesOrUnknown[inputId]
+ || !inputSpecs.get(inputId).hasUniqueKey()
+ || conditionFields[inputId].length == 0) {
+ return retraction;
+ }
+ final RowData storedRecord =
+ stateHandlers
+ .get(inputId)
+ .getRecord(keyExtractor.getJoinKey(retraction,
inputId), retraction);
+ if (storedRecord == null) {
+ return retraction;
+ }
+ // the stored record is removed from state with this retraction
+ storedRecord.setRowKind(RowKind.DELETE);
+ return storedRecord;
+ }
+
+ /**
+ * Replaces a stored record with its new version. Only the joined rows and
null padding that the
+ * old version produces are retracted, the others are updated by the
joined rows of the new
+ * version.
+ */
+ private void replaceRecord(RowData replacedRecord, RowData record, int
inputId)
+ throws Exception {
+ bufferedRows = new ArrayList<>();
Review Comment:
> Buffering all joined rows of both versions is very risky: we don't know
how many rows a key matches, this can lead to OOM and a job that can never
progress.
Yup, we are trying to solve a similar issue in streaming bootstrap at
present. We've seen some statements with low cardinality JOIN keys that don't
spill to disk and just build on the heap until we OOM at scale. Haven't quite
solved that puzzle yet unfortunately.
> Could we emit the retractions and the new rows directly here so we don't
run into this?
I'm good with that. I'll go ahead and drop the
`bufferedRow`/`emitDifference` bits and just emit directly (e.g. retract the
old, join the new).
--
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]
