You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "anand-h-codes (via GitHub)" <gi...@apache.org> on 2026/09/21 09:48:41 UTC
[PR] [FLINK-40681][table-runtime] Do not count an UPDATE_AFTER that replaces a record as a new association [flink]
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40681][table-runtime] Do not count an UPDATE_AFTER that replaces a record as a new association [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29264:
URL: https://github.com/apache/flink/pull/29264#issuecomment-5758752158
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fcdd9772dfe81c5310ce4c08a5355bfdf5063e64",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fcdd9772dfe81c5310ce4c08a5355bfdf5063e64",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fcdd9772dfe81c5310ce4c08a5355bfdf5063e64 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40681][table-runtime] Do not count an UPDATE_AFTER that replaces a record as a new association [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #29264:
URL: https://github.com/apache/flink/pull/29264#issuecomment-5758565451
@anand-h-codes Isn't this just a duplicate of https://github.com/apache/flink/pull/29204 ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40681][table-runtime] Do not count an UPDATE_AFTER that replaces a record as a new association [flink]
Posted by "anand-h-codes (via GitHub)" <gi...@apache.org>.
anand-h-codes commented on PR #29264:
URL: https://github.com/apache/flink/pull/29264#issuecomment-5759092085
@MartijnVisser How do you wish to proceed? Happy to go whichever way you prefer.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40681][table-runtime] Do not count an UPDATE_AFTER that replaces a record as a new association [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on PR #29264:
URL: https://github.com/apache/flink/pull/29264#issuecomment-5759569131
I'm working offline with some of the committers on the project to verify that the issue is indeed something that's wrong; when that's confirmed, we'll update the Jira and decide on next steps
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40681][table-runtime] Do not count an UPDATE_AFTER that replaces a record as a new association [flink]
Posted by "anand-h-codes (via GitHub)" <gi...@apache.org>.
anand-h-codes commented on PR #29264:
URL: https://github.com/apache/flink/pull/29264#issuecomment-5758733972
@MartijnVisser I was under the impression that goal of the said PR was just to reproduce the issues . Apologies for the confusion. I shall delete my draft.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org