apache / apache/paimon

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

Open
#9,666 0 comments 0 reactions 0 assignees View on GitHub
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.