anand-h-codes opened a new pull request, #29264:
URL: https://github.com/apache/flink/pull/29264
## What is the purpose of the change
Outer, semi and anti joins keep a `numOfAssociations` counter per
preserved-side record, and use its transitions through zero to decide when to
emit or retract the null-padded row (outer) or the left row (semi/anti). When
the other side is fed upsert input — `UPDATE_AFTER` without a preceding
`UPDATE_BEFORE`, which the planner allows whenever no downstream operator
requires `UPDATE_BEFORE` — the record replaces the one already in state, but
the operator still counted it as a *new* association. The counter then outgrew
the number of records actually in state, the transition back to zero never
happened, and the row that should have reappeared was lost.
Concretely, for `orders LEFT JOIN line_items`, an `UPDATE_AFTER` on a line
item followed by its deletion leaves the order missing from the sink entirely,
instead of reappearing null-padded.
The reproducing tests in this PR are the work of @MartijnVisser,
cherry-picked from his `FLINK-40681-reproducer` branch with their original
authorship; two small corrections to them are in a separate commit.
## Brief change log
- `JoinRecordStateView` / `JoinRecordAsyncStateView` gain
`containsRecord(RowData)`, implemented by all six state views, telling the
operator whether `addRecord` would replace an existing record
- `StreamingJoinOperator` and `JoinHelper` (shared with the async-state
operator) grow the counter only for a genuinely new association; a counter of 0
always grows, since the padded row was just retracted
- `StreamingSemiAntiJoinOperator` gets the same treatment for its
left-side counter
- No state access is made when the join key contains the unique key, since
that side holds at most one record per join key, nor when the other side keeps
no counter
- `AsyncStateStreamingJoinOperator` issues the check before `processJoin`
writes the record, and combines it with the matched-records future
## Verifying this change [WIP]
This change added tests and can be verified as follows:
- `StreamingJoinOperatorTest`: LEFT, RIGHT and FULL OUTER cases where the
inner side sends `UPDATE_AFTER` and is then deleted, asserting the null-padded
row returns. These run in both sync and async-state modes.
`JOIN_TYPE_EXTRACTOR` had no `FullOuterJoin` case, so the full outer tests
previously built a right outer join; that is fixed
## Known follow-up, not addressed here [WIP]
- This fix would not cover the problem with the
`StreamingMiniBatchJoinOperator`.
## 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** — one
additional state read per accumulate record, and only when the other side keeps
a counter andinput side has a unique key not contained in the join key. The
`JoinKeyConcided without touching state
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: **no** — no plan or
state-format change, so existing savepoints restore
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no** — bug fix, no
user-facing configuration
- 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 (Opus 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]
