rionmonster opened a new pull request, #29567: URL: https://github.com/apache/flink/pull/29567
## What is the purpose of the change FLINK-40858 made the streaming join look up the stored record for a DELETE that only carries the unique key (e.g. from an upsert source), so a non-equi condition is evaluated on the full row. In mini-batch mode, a retraction followed by an accumulation of the same unique key within a bundle is processed as a suppressed pair, and that lookup was skipped for the pair. Both halves used the key-only DELETE as it arrived, i.e. with NULL non-key fields. With a non-equi condition on those fields, the retraction matched nothing, so the joined rows of the old version were never retracted, and the null-padding change of an outer join was decided on the partial row. This is a sub-task of FLINK-40681. ## Brief change log - `StreamingJoinOperator#pairedRetraction` resolves the retraction of a suppressed pair to the stored record, using the same conditions as `retractedRecord` (non-equi condition, unique key, key-only or unknown deletes). The lookup happens once per pair, before the accumulation replaces the stored record. - `MiniBatchStreamingJoinOperator#processElementWithSuppress` passes the resolved record to both halves of the pair: the retraction half retracts the rows the old version produced, and the accumulation half decides the null-padding change against it. - The RowKind of the stored record is set to the retraction's kind for the pair. The record may be the object held by heap state, but `processElement` sets it back to INSERT before it is stored again, the same pattern as `StreamingSemiAntiJoinOperator`, and `CopyOnWriteStateMap` copies entries that a running snapshot still references. ## Verifying this change This change added tests to `StreamingMiniBatchJoinOperatorTest`: - `testLeftJoinJoinKeyContainsUniqueKeyNonEquiKeyOnlyDeletesOfRightWithinBundle`: key-only DELETE + INSERT pairs on the inner side of a LEFT join, across bundles, covering a match turning into a non-match, a non-match turning into a match, a match staying a match, and a lone key-only DELETE. - `testLeftJoinJoinKeyContainsUniqueKeyNonEquiKeyOnlyDeleteOfLeftWithinBundle`: the same pair on the outer side. - `testLeftJoinHasUniqueKeyNonEquiKeyOnlyDeleteOfRightWithinBundle`: the pair when the unique key is not contained in the join key. All three fail without the fix. Passing the raw retraction to the accumulation half only fails the "stays a match" case, so that half is covered too. I also reproduced the bug end to end from SQL (upsert source with key-only deletes, non-equi condition on both sides, upsert sink, mini-batch enabled) on the heap and RocksDB backends: it emits the wrong rows without the fix and the correct ones with it. That test isn't included, since the harness tests cover the same paths. The existing runtime join tests and the planner join tests (`JoinITCase`, `JoinRestoreTest`, `SemiAntiJoinStreamITCase`, `JoinTest`) pass. No new restore test is needed, because neither the plan nor the state layout changes. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): yes. A suppressed pair now does one state lookup when its retraction is a DELETE and the join has a non-equi condition, a unique key, and key-only or unknown deletes. Plans compiled before 2.4 have no changelog modes and are treated as unknown, so they pay the lookup too. Other pairs are unchanged. - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Claude Opus 5.5) -- 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]
