gustavodemorais commented on code in PR #29583:
URL: https://github.com/apache/flink/pull/29583#discussion_r4482622459
##########
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. We had bugfixes in the past around this. Could we emit the
retractions and the new rows directly here so we don't run into this?
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java:
##########
@@ -788,6 +972,16 @@ private void handleInsertAfterInput(
private void addRecordToState(RowData input, int inputId) throws Exception
{
final boolean isUpsert = isUpsert(input);
RowData joinKey = keyExtractor.getJoinKey(input, inputId);
+ // A record that can never match isn't stored. Retractions still go to
state: removing a
+ // record that was never stored is a no-op, and one stored by an
earlier version is removed.
+ if (isUpsert && cannotMatch(input, inputId)) {
+ // Storing an upsert replaces the previous version of the record
with the same unique
Review Comment:
This confuses me. replaceRecord already removes the old version under its
own join key, so which path still needs this second removal?
If it's only for skipping storage of rows that can never match, I'd rather
keep storing them for upsert inputs than combine both optimizations here. Can
we only do the fix in this PR and do the optimizating for not storing null join
keys in another one?
--
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]
