zhuxiangyi opened a new issue, #9666: URL: https://github.com/apache/paimon/issues/9666
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/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: ```scala // 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: ```sql 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? - [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]
