singhpk234 commented on code in PR #17113:
URL: https://github.com/apache/iceberg/pull/17113#discussion_r3530732337
##########
docs/docs/flink-maintenance.md:
##########
@@ -82,6 +82,44 @@ Used to remove files which are not referenced in any
metadata files of an Iceber
.deleteBatchSize(1000))
```
+#### ConvertEqualityDeletes
+Converts equality delete files staged on a source branch into deletion vectors
(DVs) on the target branch. Requires table format version >= 3 (which
introduces DVs). This is typically used together with the Flink `IcebergSink`,
which writes new data files and equality deletes to a staging branch; the
converter then resolves the equality deletes into row-position DVs and commits
the data files together with the DVs to the target branch.
+
+The pipeline runs a single conversion cycle for every trigger event.
Internally it is split into parallel stages (planner → reader → PK index → DV
writer → committer), and it uses the same maintenance-framework lock as other
tasks to guarantee mutual exclusion with concurrent maintenance operations
(e.g. compaction) on the same table.
+
+```java
+.add(ConvertEqualityDeletes.builder()
+ .stagingBranch("staging")
+ .equalityFieldColumns(ImmutableList.of("id"))
+ .scheduleOnEqDeleteFileCount(10)
+ .parallelism(4))
+```
+
+Notes:
+
+**Prerequisites and staging-branch layout**
+
+- Table format version must be `>= 3` (the version that introduces deletion
vectors). This is validated at job startup so a wrongly-configured job fails
fast instead of silently doing nothing.
+- The staging branch is expected to be produced by the Flink `IcebergSink` in
V3 mode: it may contain new data files, equality delete files, and deletion
vectors, but **must not** contain V2 positional delete files. The planner
throws if it encounters any V2 positional delete on the staging branch.
+- The staging branch **must not remove data files**, i.e. do not run
compaction / rewrite on the staging branch. If a staging snapshot removes data
files, the cycle fails; run `RewriteDataFiles` on the target branch instead —
the shared maintenance lock keeps it safe alongside this task.
+- `equalityFieldColumns` is required, must be non-empty, and must match the
equality field columns the writer used to produce the staged equality delete
files (see `IcebergSink.Builder#equalityFieldColumns`). Every staged equality
delete file's `equalityFieldIds` must exactly equal this set; a mismatch fails
the cycle. The partition source columns of every staged equality delete's spec
must also be a subset of these equality columns.
+
+**Branch semantics**
+
+- When `stagingBranch` and `targetBranch` are different, the converter reads
eq-deletes from the staging branch and commits the resolved data files and DVs
to the target branch. The staging branch itself is not modified by this task.
+- When `stagingBranch == targetBranch`, the converter operates in-place: it
acts as an equality-delete-to-DV compaction on that single branch. On cold
start it walks the entire branch history so eq-deletes committed before the
converter started are still picked up. Pure-insert snapshots on the shared
branch are skipped without running a full cycle.
+- Only one **oldest unprocessed** staging snapshot is processed per trigger.
If the writer produces staging snapshots faster than the converter can process
them, they queue up on the staging branch. Configure `scheduleOnInterval` /
`scheduleOnEqDeleteFileCount` / `scheduleOnCommitCount` (and
`TableMaintenance.rateLimit`) accordingly to keep up with the ingestion rate.
+
+**State and restart**
+
+- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state. Its size scales with the number of live rows in the table for the
configured equality columns; size Flink state backend, checkpoint storage, and
TaskManager memory accordingly.
Review Comment:
[for my understanding] is this done in an incremental manner from the last
check-pointed state ?
##########
docs/docs/flink-maintenance.md:
##########
@@ -82,6 +82,44 @@ Used to remove files which are not referenced in any
metadata files of an Iceber
.deleteBatchSize(1000))
```
+#### ConvertEqualityDeletes
+Converts equality delete files staged on a source branch into deletion vectors
(DVs) on the target branch. Requires table format version >= 3 (which
introduces DVs). This is typically used together with the Flink `IcebergSink`,
which writes new data files and equality deletes to a staging branch; the
converter then resolves the equality deletes into row-position DVs and commits
the data files together with the DVs to the target branch.
+
+The pipeline runs a single conversion cycle for every trigger event.
Internally it is split into parallel stages (planner → reader → PK index → DV
writer → committer), and it uses the same maintenance-framework lock as other
tasks to guarantee mutual exclusion with concurrent maintenance operations
(e.g. compaction) on the same table.
+
+```java
+.add(ConvertEqualityDeletes.builder()
+ .stagingBranch("staging")
+ .equalityFieldColumns(ImmutableList.of("id"))
+ .scheduleOnEqDeleteFileCount(10)
+ .parallelism(4))
+```
+
+Notes:
+
+**Prerequisites and staging-branch layout**
+
+- Table format version must be `>= 3` (the version that introduces deletion
vectors). This is validated at job startup so a wrongly-configured job fails
fast instead of silently doing nothing.
+- The staging branch is expected to be produced by the Flink `IcebergSink` in
V3 mode: it may contain new data files, equality delete files, and deletion
vectors, but **must not** contain V2 positional delete files. The planner
throws if it encounters any V2 positional delete on the staging branch.
+- The staging branch **must not remove data files**, i.e. do not run
compaction / rewrite on the staging branch. If a staging snapshot removes data
files, the cycle fails; run `RewriteDataFiles` on the target branch instead —
the shared maintenance lock keeps it safe alongside this task.
+- `equalityFieldColumns` is required, must be non-empty, and must match the
equality field columns the writer used to produce the staged equality delete
files (see `IcebergSink.Builder#equalityFieldColumns`). Every staged equality
delete file's `equalityFieldIds` must exactly equal this set; a mismatch fails
the cycle. The partition source columns of every staged equality delete's spec
must also be a subset of these equality columns.
+
+**Branch semantics**
+
+- When `stagingBranch` and `targetBranch` are different, the converter reads
eq-deletes from the staging branch and commits the resolved data files and DVs
to the target branch. The staging branch itself is not modified by this task.
+- When `stagingBranch == targetBranch`, the converter operates in-place: it
acts as an equality-delete-to-DV compaction on that single branch. On cold
start it walks the entire branch history so eq-deletes committed before the
converter started are still picked up. Pure-insert snapshots on the shared
branch are skipped without running a full cycle.
+- Only one **oldest unprocessed** staging snapshot is processed per trigger.
If the writer produces staging snapshots faster than the converter can process
them, they queue up on the staging branch. Configure `scheduleOnInterval` /
`scheduleOnEqDeleteFileCount` / `scheduleOnCommitCount` (and
`TableMaintenance.rateLimit`) accordingly to keep up with the ingestion rate.
+
+**State and restart**
+
+- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state. Its size scales with the number of live rows in the table for the
configured equality columns; size Flink state backend, checkpoint storage, and
TaskManager memory accordingly.
+- The `equalityFieldColumns` set is persisted in operator state. Reconfiguring
it across a restart from savepoint is **not supported** and the job will fail
fast on restore. Restart from a clean state (no savepoint) if the equality
columns change.
+- The committer is intentionally stateless. On restart, the planner
rediscovers its position by walking the target branch for the marker property
`equality-convert-staging-snapshot` written by the committer. If that marker is
no longer reachable on the target branch (target was rolled back,
replace-main'ed, or the marker snapshot was expired) the planner fails and
requires manual intervention.
+
+**Concurrent commits on the target branch**
+
+- External commits to the target branch (e.g. compaction, direct writes) are
detected by `RowDelta.validateFromSnapshot`. A conflicting cycle fails at
commit; the next trigger reindexes the worker's PK index from the new target
head and retries.
Review Comment:
Do we wanna put a recommendation to avoid concurrent writes to the tables ?
i understand we can rebase / re-index and then retry but i wonder how practical
it is for the next attempt to succeed if there is another writer writing to the
table continuously
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
