Flink: Integrate ConvertEqualityDeletes with IcebergSink#17142
Open
mxm wants to merge 3 commits into
Open
Conversation
This adds the option to run `ConvertEqualityDeletes` as a post-commit maintenance task on `IcebergSink`, similarly to the RewriteDataFiles / ExpireSnapshots / DeleteOrphanFiles post-commit integration. The sink's write branch becomes the converter's staging branch. The target branch defaults to the same branch (in-place conversion). This is also configurable via options on the builder, as well as via properties (see ConvertEqualityDeletesConfig). Equality field columns are derived from the resolved equality field ids of the sink, to ensure they match with the sink's equality fields.
pvary
reviewed
Jul 9, 2026
| | flink-maintenance.expire-snapshots.enabled | false | Expire snapshots after commit | | ||
| | flink-maintenance.delete-orphan-files.enabled | false | Delete orphan files after commit | | ||
| | flink-maintenance.convert-equality-deletes.enabled | false | Convert equality deletes to deletion vectors after commit (requires equality fields and format version >= 3) | | ||
| | flink-maintenance.convert-equality-deletes.target-branch | Write branch | Branch the converted DVs are committed to; defaults to the write branch (in-place conversion) | |
Contributor
There was a problem hiding this comment.
Do we want this here?
Maybe it is enough to keep the detailed config on the flink-maintenance.md
ferenc-csaky
reviewed
Jul 9, 2026
ferenc-csaky
left a comment
ferenc-csaky
left a comment
There was a problem hiding this comment.
Added 2 nitty comments, the logic itself LGTM
| | flink-maintenance.expire-snapshots.enabled | false | Expire snapshots after commit | | ||
| | flink-maintenance.delete-orphan-files.enabled | false | Delete orphan files after commit | | ||
| | flink-maintenance.convert-equality-deletes.enabled | false | Convert equality deletes to deletion vectors after commit (requires equality fields and format version >= 3) | | ||
| | flink-maintenance.convert-equality-deletes.target-branch | Write branch | Branch the converted DVs are committed to; defaults to the write branch (in-place conversion) | |
There was a problem hiding this comment.
nit: The config option itself has no default value. Maybe it would be less confusing to keep this table in sync with that and put null to the Default table col (based on the read option table) and rephrase the description that something like "if not specified, the write branch will be used (in-place conversion)"
|
|
||
| JobClient jobClient = env.executeAsync("Test Convert Equality Deletes E2E"); | ||
| try { | ||
| Awaitility.await() |
There was a problem hiding this comment.
nit: static import for await(), as it's a widely used lib and a common practice
ferenc-csaky
reviewed
Jul 9, 2026
| flinkConf.put("flink-maintenance.delete-orphan-files.min-age-seconds", "259200"); | ||
|
|
||
| // Configure convert equality deletes. Converts in place on the write branch by default; | ||
| // set a target branch to instead promote the converted deletion vectors there. |
There was a problem hiding this comment.
nit: weird sentence, possible rewording:
// set a target branch to promote the converted deletion vectors there instead.
| - The `RANGE` distribution mode is not yet available for the `IcebergSink` | ||
| - When using `IcebergSink` use `uidSuffix` instead of the `uidPrefix` | ||
|
|
||
| ### Post-commit table maintenance |
Contributor
There was a problem hiding this comment.
Should we only make this part only in flink-maintenance.md?
| ); | ||
| ``` | ||
|
|
||
| ### IcebergSink Maintenance Configuration (SQL) |
Contributor
There was a problem hiding this comment.
Should we add related config in this part？
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This adds the option to run
ConvertEqualityDeletesas a post-commit maintenance task onIcebergSink, similarly to the RewriteDataFiles / ExpireSnapshots / DeleteOrphanFiles post-commit integration. The sink's write branch becomes the converter's staging branch. The target branch defaults to the same branch (in-place conversion). This is also configurable via options on the builder, as well as via properties (see ConvertEqualityDeletesConfig). Equality field columns are derived from the resolved equality field ids of the sink, to ensure they match with the sink's equality fields.Documentation changes are in separate commits. Ideally, they should be preserved to make backporting of the code changes more straight-forward.