rionmonster commented on code in PR #29583:
URL: https://github.com/apache/flink/pull/29583#discussion_r4486633194
##########
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?
You're right here as I think this was a bit of the recent NULL-key-related
PR redundancies bleeding into this separate issue. We can save the storage
optimization related to that for a follow-up once all of those issues have been
resolved.
--
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]
