[Feature] Dedicated Spark Structured Streaming write path on StreamWriteBuilder / StreamTableCommit

[zhuxiangyi]

Search before asking

• I searched in the 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 [core][spark] Deduplicate a replayed Structured Streaming micro-batch #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?

• I'm willing to submit a PR!