zhuxiangyi opened a new issue, #10010: URL: https://github.com/apache/paimon/issues/10010
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. ### Motivation #9667 makes a Spark Structured Streaming write idempotent for a replayed micro-batch, but does so on top of the batch write path: every micro-batch creates a batch committer, commits it through `filterAndCommit`, and closes it. That mismatch — a one-shot committer driving a streaming commit protocol — is what the three switches added to `InnerTableCommit` in #9667 patch over: - `inlineMaintenance`, because a committer closed right after the batch cannot leave maintenance to an executor that is about to be shut down; - `checkFilesExistence(false)`, because `filterAndCommit` otherwise lists the files of every batch; - `filterCommittedIgnoresLastSafeSnapshot`, because the direct postpone path recomputes `commit.last-safe-snapshot` on every attempt, which hides the previous run's commit from the history search. It also leaves a gap: an append into a postpone bucket table that already has real buckets goes through the staged committer, which cannot recognise a replay and only warns. Flink has none of this. Its committer lives across checkpoints, uses `filterAndCommit` only on recovery and `commit` in steady state, lets maintenance run asynchronously, and its streaming postpone path (`PostponeBucketSink`) writes to bucket -2 without setting a history bound at all. ### Solution A dedicated Spark streaming write path that reuses `StreamWriteBuilder` / `StreamTableCommit`: - `PaimonSink` keeps one `StreamTableCommit` for the lifetime of the query, created with the stable commit user of #9667; `addBatch` commits under `batchId + 1`, with `filterAndCommit` on the first batch of a run and `commit` afterwards, the way Flink's committer does; - the executors call `prepareCommit(waitCompaction, identifier)`; - a postpone bucket table is written to bucket -2, as by Flink, instead of through the fixed-bucket direct or staged committers, so no history bound is recomputed and the staged path gap closes; - the three switches above, and their tests, are removed. To preserve, as listed in the review of #9667: complete-mode overwrite semantics, commit callback retries, and maintenance behaviour (asynchronous again, with a failure reported on the next commit like Flink). The performance impact of writing postpone tables to bucket -2 needs validating, since today the sink rewrites them to fixed buckets synchronously. ### Anything else? The 14 cases of `PaimonSinkIdempotencyTest` and the full-compaction case of `PaimonSinkTest` from #9667 are the regression baseline; they are written against external behaviour and should pass unchanged. To start after #9667 lands. ### 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]
