rionmonster opened a new pull request, #29570: URL: https://github.com/apache/flink/pull/29570
## What is the purpose of the change `StreamingSemiAntiJoinOperator` handled upsert inputs like retract inputs. An upsert input updates a row without retracting the old version first, and its DELETE may contain only the unique key. This is the semi/anti counterpart of FLINK-40809, FLINK-40841 and FLINK-40858 for the regular join: - A right row replaced by INSERT or UPDATE_AFTER counted its left matches again, so a later DELETE never removed the last match: a semi join never retracted the left row and an anti join never sent it again. With a non-equi condition, a match that the new version lost was not removed either. - A left row replaced by a version that is not sent left the old version in the output. - A key-only DELETE was joined with NULL non-key fields, so with a non-equi condition it matched nothing. On the right side the match was not removed. On the left side a semi join dropped the delete, and an anti join sent a delete with NULL fields for a row that it never sent. The ticket describes the right side; the left side has the same root cause, so this PR covers both. ## Brief change log - Refactoring commit, no behaviour change: the changelog modes of the inputs, the non-equi flag and the lookups of replaced and retracted records move from `StreamingJoinOperator` to `AbstractStreamingJoinOperator`. The mini-batch special cases stay in `StreamingJoinOperator`. - The semi/anti join receives the changelog modes and the non-equi flag from `StreamExecJoin`, which already serializes them. - Right side: a replaced record is not counted again, and with a non-equi condition the matches it lost are removed. A key-only DELETE is joined with the stored record. - Left side: a replaced record is retracted if it was sent but the new version is not. A key-only DELETE is joined with the stored record, so it is sent with all fields if the row was sent, and not at all otherwise. - The lookups only happen in the same cases as in the regular join: for upsert inputs, or for plans compiled before Flink 2.4, which don't contain the changelog modes. ## Verifying this change - `StreamingSemiAntiJoinOperatorTest`: five new tests, each run as semi and anti join, for both sides, with and without a non-equi condition, and with a unique key that contains more than the join key. Two existing tests asserted the double count of a right upsert and are updated. All 14 new and updated cases fail without the fix. Breaking each part of the fix on purpose fails the test written for it. - `JoinRestoreTest`: new program `join-semi-join-upsert-inputs-non-equi`. A right row is updated without UPDATE_BEFORE and then deleted by key after the restore, so its full image is only in the restored state; a left row stops matching after the restore. It fails without the fix and passed all 5 runs with it. - Existing tests pass: the runtime join tests, and the planner join and semi/anti plan, restore, semantic and ITCase tests. No other restore test is needed: neither the plan nor the state layout changes. The existing `join-semi-join` and `join-anti-join` restore tests use plans compiled before Flink 2.4, which are now handled as upsert, and they pass. State written before this change may already contain a left row whose number of matches was counted too high. The fix does not recompute it, so such a row is retracted (semi) or sent again (anti) only once the count returns to zero, or when state TTL removes it. ## Performance A/B harness benchmark of the operator before and after the fix, semi join, 10,000 join keys pre-populated on both sides, then 50,000 cycles of a right update, a left update, a right delete and a right insert. Median of 5 runs after 2 warm-up runs, alternating before and after. ns per input record: | Input | Heap before | Heap after | RocksDB before | RocksDB after | |---|---|---|---|---| | retract, equi | 446 | 428 | 4,171 | 4,210 | | upsert, equi | 367 | 384 | 5,177 | 5,199 | | upsert with full deletes, non-equi | 321 | 371 | 5,059 | 9,718 | | upsert with key-only deletes, non-equi | 262 | 414 | 5,006 | 9,982 | | plan before Flink 2.4, retract, non-equi | 328 | 328 | 4,373 | 4,728 | - Without a lookup (retract inputs, or equi conditions with the unique key in the join key) the results are within noise. - The cost is in upsert inputs with a non-equi condition. This workload consists only of updates and deletes, so it is the worst case. State reads per record go from 1.00 to 1.95, while writes stay at 1.40. The extra reads are the replaced record of an update or insert, the second pass over the left records for the matches a right update lost, and, for a left update that is not sent, the replaced record and its matches. On RocksDB, reads dominate, so the time about doubles. - Part of the "before" cost is lower because the old code skipped work it should have done, e.g. a key-only delete matched nothing and so updated no count. - The regular inner join, which does these lookups since FLINK-40841 and FLINK-40858, needs 558 ns (heap) and 14,675 ns (RocksDB) per record on the same upsert workload, so the semi/anti join stays below it. ## 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, see Performance - 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]
