[Bug] Spark Structured Streaming write commits a replayed micro-batch twice

Comienza con PaimonSink.addBatch y rastrea cómo table.newBatchWriteBuilder(), BatchWriteBuilderImpl y StreamTableCommit gestionan los usuarios y los identificadores de commit; compáralo con StreamWriteBuilder y la ruta del sink de Flink. Ejecuta el escenario de replay proporcionado y añade cobertura de regresión que demuestre que reproducir un batch no duplica filas ni snapshots, incluidas las tablas de agregación.

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

master (reproduced on e7db7cf85).

Compute Engine

Spark Structured Streaming (reproduced on Spark 3.5; the sink code path is shared by
Spark 3.2–4.1).

Minimal reproduce step

Structured Streaming delivers exactly-once only if the sink is idempotent for a repeated batch id.
When a query fails after the sink returns from addBatch but before Spark records the batch as
completed, the offset log still contains the batch while the commit log does not, and the restarted
query replays that micro-batch with its original batch id.

The state a failure leaves behind can be reproduced exactly by deleting the commit log entry of the
last batch:

// An append-only table.
spark.sql("CREATE TABLE T (a INT, b STRING)")
val location = /* table location */

val inputData = MemoryStream[(Int, String)]
val df = inputData.toDS().toDF("a", "b")
inputData.addData((1, "a"), (2, "b"), (3, "c"))

def start() = df.writeStream
  .option("checkpointLocation", checkpointPath)
  .format("paimon")
  .start(location)

val q = start()
q.processAllAvailable()
q.stop()
// 3 rows, 1 snapshot -- as expected.

// Simulate the driver dying in the window described above.
new File(s"$checkpointPath/commits/0").delete()
new File(s"$checkpointPath/commits/.0.crc").delete()

val restarted = start()          // Spark replays batch 0 with the same batch id
restarted.processAllAvailable()
restarted.stop()

Result:

	expected	actual
rows in T	3	6
snapshots	1	2

The same is visible without any streaming machinery, which shows it is the sink and not the engine:
calling PaimonSink.addBatch(0L, batch) twice with the same batch id writes the data twice
(2 rows become 4).

A primary-key table is not generally safe either. With merge-engine = aggregation the replayed
batch is aggregated a second time:

CREATE TABLE AGG (k INT, v BIGINT) TBLPROPERTIES (
  'primary-key' = 'k', 'bucket' = '1',
  'merge-engine' = 'aggregation', 'fields.v.aggregate-function' = 'sum');

Writing (1, 10) once and then replaying that micro-batch yields v = 20
(verified: afterFirst=10 afterReplay=20).

What doesn't meet your expectations?

A replayed micro-batch should be recognised as already committed and skipped, so that a driver
failure cannot change the table contents. Instead the batch is committed a second time:

• append-only tables get every row of the batch duplicated;
• aggregation merge-engine tables silently produce wrong values;
• deduplicate primary-key tables happen to be masked by key overwrite, which is luck rather than
  a guarantee.

Nothing fails and nothing is logged; the corruption is discovered only by comparing row counts
downstream.

Anything else?

Root cause. PaimonSink.addBatch(batchId, data) receives the batch id but uses it only to pace
full compaction (DataWrite), and commits through table.newBatchWriteBuilder(), whose commit user
is a fresh random UUID per builder (BatchWriteBuilderImpl) and whose commit identifier is always
BatchWriteBuilder.COMMIT_IDENTIFIER = Long.MAX_VALUE. Neither of the two dimensions Paimon
deduplicates on can therefore identify a replay.

The machinery already exists in core and is what the Flink sink uses: StreamWriteBuilder with a
stable commit user plus StreamTableCommit#filterAndCommit, which drops a committable whose
identifier the commit user has already committed. So this is a connector that took the batch write
path, not a missing capability.

Documentation currently presents Spark streaming write without an at-least-once caveat, so users
have no reason to expect duplicates.

Are you willing to submit a PR?

• I'm willing to submit a PR!