zhuxiangyi opened a new pull request, #9667:
URL: https://github.com/apache/paimon/pull/9667
### Purpose
Closes #9666.
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 restarted query replays that micro-batch with its original
batch id.
`PaimonSink` received the batch id but used it only to pace full compaction,
and committed through
`table.newBatchWriteBuilder()`, whose commit user is a fresh random UUID per
builder and whose
commit identifier is always `BatchWriteBuilder.COMMIT_IDENTIFIER =
Long.MAX_VALUE`. Neither of the
dimensions Paimon deduplicates on could identify a replay, so the batch was
committed a second
time: every row duplicated in an append-only table, and silently wrong
values in an `aggregation`
merge-engine table (writing `(1, 10)` and replaying that batch yields `v =
20`).
The machinery already exists in core and is what the Flink sink uses; the
Spark sink simply took
the batch write path.
### Tests
`PaimonSinkIdempotencyTest` (new, 7 cases), each asserting the correct
behaviour so that it fails
without the fix:
- a replayed micro-batch of an append-only table, driven end to end by
deleting the commit log
entry of the batch, which is exactly the checkpoint state a driver failure
leaves behind;
- the same when only the query id is available, i.e. the checkpoint location
never reaches the sink
options because it comes from `spark.sql.streaming.checkpointLocation`;
- a replay of a batch that is not the first one of the query;
- a replay in `complete` output mode;
- `write.stream.commit-user` as an option of the writer and as a
`spark.paimon.` session conf;
- `addBatch` called twice with the same batch id through the API directly.
Two of them assert the prefix of the commit user recorded in the snapshot,
so that the case which
is meant to exercise the query id derivation cannot pass through the
checkpoint derivation.
The full set of Spark streaming suites was run on the spark3 and spark4
profiles (Spark 3.2, 3.4,
3.5, 4.1): 34 suites, 349 tests.
### API and Format
`BatchWriteBuilderImpl` gains `withCommitUser`, and its `newCommit()` return
type is narrowed from
`BatchTableCommit` to `InnerTableCommit`. The narrowing keeps the caller in
the connector free of a
downcast that could only fail at runtime; it is source compatible, and
`BatchWriteBuilderImpl` has
no subclasses.
`InnerTableCommit` gains `checkFilesExistence(boolean)`. `filterAndCommit`
verified that every file
it is about to commit still exists, which guards a committable restored from
an engine's state that
may reference files deleted long ago. A caller filtering a committable it
has just produced knows
those files exist, so the Spark sink turns the check off; otherwise every
micro-batch would pay a
file listing proportional to the number of files it wrote. The default is
unchanged, so Flink keeps
the check.
### Documentation
`docs/docs/spark/structured-streaming.md` gains an "Exactly-once" section
covering how the commit
user is derived, the new `write.stream.commit-user` option, and the limits:
starting from a new
checkpoint location gives a query a new commit user, the data files of a
skipped replay are left to
orphan file cleaning, and a postpone bucket table committing through the
staged committer cannot
deduplicate and logs a warning per micro-batch. The generated option
reference is regenerated.
--
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]
