fishfishfishfishaa opened a new issue, #9236: URL: https://github.com/apache/paimon/issues/9236
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. ### Paimon version 1.1.1 ### Compute Engine Flink1.19 ### Minimal reproduce step ## Problem `Long.MAX_VALUE` is used as the END_INPUT commit identifier. Different writer subtasks may reach `endInput()` at different times, so a checkpoint can contain only a subset of the final END_INPUT committables. Two correctness issues exist around such pending END_INPUT state. ### 1. Partial END_INPUT committable may be committed during recovery and cause data loss Consider a job with multiple writer subtasks: ```text writer-0 reaches endInput -> emits END_INPUT committable(MAX) checkpoint N -> committer state contains writer-0's MAX writer-1 has not reached endInput yet failover ``` During recovery, the existing implementation immediately calls `filterAndCommit` for restored committables. Therefore, the partial MAX containing only writer-0's data can be committed before writer-1 produces its END_INPUT committable. Once that partial commit succeeds, the latest Paimon snapshot already has: ```text commitIdentifier = Long.MAX_VALUE ``` `filterAndCommit` filters by commit identifier rather than comparing the actual payload. Later, when writer-1 reaches `endInput()`, its END_INPUT committable also has identifier `Long.MAX_VALUE`, so it is filtered out and never committed. This results in data loss. The same completeness invariant must also hold at runtime. Remaining END_INPUT committables may already have reached `CommitterOperator.processElement()` but still be buffered in `inputs`. If `endInput=true` becomes visible before `pollInputs()` materializes them into the MAX bucket, a delayed checkpoint-complete notification may again commit only a partial MAX. The required invariant is: ```text Any Long.MAX_VALUE committable that is allowed to commit must represent the complete logical END_INPUT of all writer subtasks. ``` ### 2. Pending END_INPUT watermark may become stale after recovery When an existing END_INPUT committable is merged with later committables, its watermark is not updated. For example: ```text partial MAX(watermark=1024) -> restore -> runtime watermark advances to 2048 -> merge remaining MAX -> checkpoint -> restore again ``` The restored END_INPUT committable may still carry watermark `1024` instead of `2048`, because the merge path appends the payload but does not persist the newer watermark. Configured END_INPUT watermark has a similar issue when the MAX committable already exists and there are no new buffered inputs to trigger a merge. ## Proposed direction Handle the two problems independently: 1. Prevent incomplete END_INPUT committables from being committed during recovery or runtime finalization. 2. Preserve and correctly update END_INPUT watermark across merge, checkpoint, and recovery. ### What doesn't meet your expectations? none ### Anything else? _No response_ ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
