[spark] Commit a streaming micro-batch through the stream write API

[zhuxiangyi]

Purpose

Closes #10010. This is the implementation #9667 was asked to be compared against, so #9667 stays
open and untouched while the two are weighed; whichever lands, the other is closed.

A Spark Structured Streaming write goes through the batch write API: every micro-batch builds its
own write builder and commits through a one-shot committer that it closes again. Such a committer
cannot recognise a micro-batch a previous run already committed, so a query that fails between the
sink returning from addBatch and Spark recording that batch in its commit log writes the whole
batch a second time on restart (#9666).

This gives the sink the shape a Flink job has:

• one commit user per query, stable across its runs, derived from the query id Spark persists
  in the checkpoint — new when a checkpoint is recreated, unchanged when a query resumes from one —
  or set explicitly with the new write.stream.commit-user option. It is read only from sources
  scoped to one query (the writer's options, the session conf), never from a table property that
  every writer of the table would share. commit.user-prefix is read from the table as usual and
  prefixes the derived user, as it does the random user of a batch write, so a job upgrading to
  this keeps the name its table was configured with.
• one StreamTableCommit per run, created with the first micro-batch and closed when the query
  terminates, so the maintenance a commit starts (tag creation, partition and snapshot expiration)
  is not cancelled by a close after every batch.
• micro-batch n committed under identifier n + 1, the way Flink numbers its checkpoints,
  which is also what a compacted-full scan recognises a scheduled full compaction by. The
  executors write it with prepareCommit(waitCompaction, identifier).
• the first micro-batch of a run goes through filterAndCommit, the ones after it through
  commit — the division of work a Flink committer makes between restoring and its steady state.
• a write to a postpone bucket table writes the postpone bucket, as FlinkSinkBuilder does for
  a stream, instead of the fixed-bucket paths meant for a batch job that ends with its commit. Those
  rows become readable once a compaction has sorted them into real buckets. This also closes the
  hole where such a write went through the staged committer, which cannot skip a replay at all.

Relation to #9667

#9667 keeps the one-shot batch committer and makes it drive a filtered commit, which needed three
switches on InnerTableCommit (checkFilesExistence, checkAppendFiles, inlineMaintenance) plus
filterCommittedIgnoresLastSafeSnapshot and a withCommitUser on two write builders. None of that
exists here: a long-lived stream committer needs no switch, because it does not file-list what it
just wrote on a steady-state commit, its maintenance has a next commit to report to, and it never
recomputes a last-safe-snapshot bound.

The only core change left is the chain-table overwrite callback (first commit), which is
independent of either approach: an overwrite of a chain table publishes its delta snapshot and only
then clears the snapshot-branch files it supersedes; when that cleanup fails after the snapshot is
published, the retried commit is recognised as a replay and its retry has to redo the cleanup for
exactly the files that overwrite superseded.

Tests

PaimonSinkIdempotencyTest (15 cases) and PaimonSinkTest, plus ChainTableFileStoreTableTest and
SimpleTableTestBase in core. Most of the sink cases were written for #9667 against externally
visible behaviour, so they carry over unchanged; what changed:

• complete mode replay on a postpone bucket table ... now asserts the rows wait in bucket -2 and
  reads them after sys.compact;
• one committer commits every micro-batch of a query and closes with it (new) counts committers
  through a commit.callbacks implementation: one per query, closed when the query stops;
• maintenance of a micro-batch runs while the query is alive replaces the case that asserted
  maintenance had to finish before a per-batch committer closed;
• in core, testFilterAndCommitAcrossRunsOfOneCommitUser replaces the two cases covering the
  removed switches: one committer commits two identifiers, a restarted run replays the last one and
  commits the next.

Run locally: paimon-core in full, paimon-spark-ut in full on Spark 3.5, and the sink suites on
Spark 3.2 / 3.4 / 3.5 / 4.0 / 4.1. Mutation checks: closing the committer after every micro-batch,
never closing it on termination, and never filtering a possible replay each fail the cases that
cover them (3, 1 and 9 cases).

Postpone write performance (asked for in #9667), local, one JVM, 5 micro-batches of 50k rows
into a primary-key postpone table with postpone.default-bucket-num = 4, two runs:

	per micro-batch (steady state)	compaction	read after
streaming to bucket -2 (this PR)	188-286 ms	2.1-2.3 s	64-65 ms
fixed-bucket write (before)	293-670 ms	-	156-213 ms

Committing a micro-batch gets about twice as cheap; the bucketing it no longer does is what the
compaction then does, as for a Flink streaming job.

API and Format

No format change. New Spark connector option write.stream.commit-user. Behaviour change: a
streaming write to a postpone bucket table now lands in the postpone bucket and becomes readable
after a compaction, as with Flink.

A query that upgrades to this keeps its Spark checkpoint: the sink stores nothing in it. Its commit
user changes from a random one per micro-batch to the stable one, which has no history yet, so the
first micro-batch after the upgrade is committed as usual, neither filtered nor duplicated.

Documentation

docs/docs/spark/structured-streaming.md: an "Exactly-once" section covering the commit user, the
commit identifier, the committer's lifetime and what a postpone bucket table does.