rionmonster commented on code in PR #29583:
URL: https://github.com/apache/flink/pull/29583#discussion_r4497534810
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/StreamingMultiJoinOperator.java:
##########
@@ -406,10 +432,93 @@ 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 (!isUpsertInput[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
+ || !hasKeyOnlyDeletes[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: the joined rows and null
padding of the old
+ * version are retracted, then the new version is joined and stored.
+ */
+ private void replaceRecord(RowData replacedRecord, RowData record, int
inputId)
+ throws Exception {
+ // set the RowKind back, because the record may be the one in state
+ replacedRecord.setRowKind(RowKind.DELETE);
+ performMultiJoin(replacedRecord, inputId);
+ replacedRecord.setRowKind(RowKind.INSERT);
+ // the old version must not count as a match of the other inputs for
the new version
+ stateHandlers
+ .get(inputId)
+ .retractRecord(keyExtractor.getJoinKey(replacedRecord,
inputId), replacedRecord);
+ performMultiJoin(record, inputId);
+ addRecordToState(record, inputId);
+ }
+
+ private boolean conditionFieldsEqual(RowData storedRecord, RowData record,
int inputId) {
+ for (int field : conditionFields[inputId]) {
+ final RowData.FieldGetter getter =
inputFieldGetters[inputId][field];
+ if (!Objects.equals(
+ getter.getFieldOrNull(storedRecord),
getter.getFieldOrNull(record))) {
Review Comment:
Yup, makes sense. I've updated things and the planner now generates one for
the condition fields to compare the projected views (along with an associated
restore tests to check for these types of scenarios, e.g., emitted -D/+I
instead of +U).
--
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]
