[Bug] Spark Structured Streaming write commits a replayed micro-batch twice
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### 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!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with PaimonSink.addBatch and trace how table.newBatchWriteBuilder(), BatchWriteBuilderImpl, and StreamTableCommit handle commit users and identifiers; compare this with StreamWriteBuilder and the Flink sink path. Run the supplied replay scenario and add regression coverage showing that replaying one batch does not duplicate rows or snapshots, including aggregation tables.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, scala
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 55/100