zhengxiang378928908-code opened a new pull request, #12937:
URL: https://github.com/apache/seatunnel/pull/12937
### Purpose of this pull request
Closes #10623. Takes over #10755 from @chl-wxp, who is kept as co-author.
The takeover was proposed in #10755 on 2026-10-15, with no reply from the
author since.
**Root cause.** With a primary key (upsert mode, CDC, or
`iceberg.table.primary-keys`), the sink writes through Iceberg's
`BaseEqualityDeltaWriter`. It keeps `insertedRowMap`, a copy of every key it
has written, so that a later change of the same key can become a position
delete, and it buffers those position deletes in
`SortingPositionOnlyDeleteWriter`. Both are released only when the task writer
completes, which today happens only at a checkpoint. A 100 million row upsert
load with a long checkpoint interval, or a batch job that checkpoints only at
the end, keeps every key in memory until it OOMs. Data files already roll by
`write.target-file-size-bytes`, so they are not the problem.
**Change.**
1. `flush_row_size` (optional, `> 0`, unset by default = current behavior):
`IcebergRecordWriter` completes the task writer after that many rows and starts
a new one. Each completed task writer becomes its own `WriteResult`, in order.
2. Ordered commits. As @DanielLeens pointed out in #10755, rotating alone is
not correct in upsert mode: an equality delete written after the rotation lands
in the same snapshot as the row it targets, and Iceberg applies equality
deletes only to data files with a **smaller** sequence number, so the old row
stays. `IcebergAggregatedCommitter.planCommits` groups the i-th results of all
writers into round i; a round that carries equality deletes starts a new group,
and the groups are committed as separate snapshots in order. Flink's
`IcebergFilesCommitter.commitDeltaTxn` commits pending results one by one for
the same reason.
- Append-only writes, and checkpoints without a rotation, still produce
exactly one snapshot, unchanged.
- Snapshots of a split checkpoint carry `seatunnel.checkpoint-id` plus
`seatunnel.commit-group` / `seatunnel.commit-group-count`. `restoreCommit`
commits only the groups that are missing. A snapshot with the checkpoint id but
without `seatunnel.commit-group` (single-snapshot commit, including ones
written by older versions) counts as the whole checkpoint. With checkpoint id
0, each group is checked by its file paths as before.
On "rotate only where in-flight keys cannot cross": a sink cannot know when
a key will not appear again, so the PR keeps rotation in all modes and makes
the commit order carry the semantics, instead of restricting rotation to
append-only, which would not help the upsert case reported in #10623.
**Same bug on `dev` today, without this option.** `applySchemaUpdate`
already completes the task writer in the middle of a checkpoint. In upsert
mode, an UPDATE of a row written before an `ADD COLUMN` in the same checkpoint
leaves the old row (`schemaChangeInsideCheckpointKeepsUpsertCorrect` below
fails on `dev` with `[1=a, 1=a2]`). The ordered commit fixes this too.
Addresses the open review items on #10755: `> 0` validation, EN/ZH docs,
focused tests including INSERT → rotate → DELETE/UPDATE.
### Does this PR introduce _any_ user-facing change?
Yes.
- New option `flush_row_size` for the Iceberg sink (docs en/zh, with when to
use it and the trade-off: more and smaller files, and more snapshots for
upsert/CDC).
- Behavior change, recorded in `incompatible-changes.md` (en/zh): an upsert
or CDC checkpoint in which the task writer was completed mid-checkpoint (by
`flush_row_size` or by a schema change) is committed as several ordered
snapshots instead of one, and readers can see them one by one. Each is a prefix
of the checkpoint's changes. Before, such a checkpoint was one snapshot but
could keep deleted or replaced rows.
### How was this patch tested?
Unit tests against real Iceberg format v2 tables (`InMemoryCatalog`),
written through `IcebergWriterFactory` and committed through
`IcebergAggregatedCommitter`, read back with `IcebergGenerics`
(`IcebergWriterRotationTest`, 9 cases):
- append-only with rotation: 2 results, 1 snapshot, no group property
- upsert, unpartitioned and `bucket(id, 4)`: INSERT 1, 2 → rotate → DELETE
1, UPDATE_AFTER 2 → INSERT 3 gives `[2=b2, 3=c]` in 3 snapshots
- CDC (primary key, upsert off): INSERT → rotate →
UPDATE_BEFORE/UPDATE_AFTER gives only the new row
- two parallel writers with different rotation counts
- restore after only the first group was committed: commits the rest once;
restoring again adds nothing
- restore of a checkpoint committed as a single snapshot (no group
property): nothing re-committed
- schema change inside a checkpoint in upsert mode (the `dev` bug above)
- control case: the same rotated files committed as one `RowDelta` (the old
committer) give `[1=a, 2=b, 2=b2]`, which is the stale-row problem
Mutation check: with the round split disabled (all rounds in one group), 6
of the 9 cases fail with stale rows.
`IcebergSinkConfigTest`: unset by default; `0` and `-1` rejected with
`OptionValidationException`. Existing `IcebergAggregatedCommitterTest` /
`IcebergFilesCommitterTest` pass unchanged. Whole `connector-iceberg` module:
144 tests pass; `ConnectorOptionCheckTest` passes.
Heap, local run (JDK 11, `-Xmx512m`, upsert, unpartitioned, 4,000,000
distinct keys written without a checkpoint; not part of the PR):
| `flush_row_size` | used heap after GC | result |
|---|---|---|
| unset | 222 MB at 0.5M rows, 403 MB at 1M rows | `OutOfMemoryError` before
1.5M rows |
| 200000 | 37 to 79 MB through 4M rows | done, 20 results |
### Check list
* [ ] If any new Jar binary package adding in your PR, please add License
Notice according
[New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
* [x] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs
* [x] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR.
* [ ] If you are contributing the connector code, please check that the
following files are updated:
1. Update
[plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties)
and add new connector information in it
2. Update the pom file of
[seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml)
3. Add ci label in
[label-scope-conf](https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml)
4. Add e2e testcase in
[seatunnel-e2e](https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/)
5. Update connector
[plugin_config](https://github.com/apache/seatunnel/blob/dev/config/plugin_config)
--
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]
