mxm commented on code in PR #18674:
URL: https://github.com/apache/iceberg/pull/18674#discussion_r4443011859
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java:
##########
@@ -209,6 +251,16 @@ private void commitIfNeeded() {
return;
}
+ if (planResult.indexGeneration() == failedIndexGeneration) {
+ deleteUncommittedDVs();
+ throw new IllegalStateException(
Review Comment:
Agreed. I opted for a mere warn message.
We already have metrics for errors and the error will be logged where it
occurs.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java:
##########
@@ -71,22 +74,23 @@ public boolean noOp() {
/** No-op cycle: empty file lists, sentinel staging snapshot id. */
public static EqualityConvertPlan noOp(
- Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) {
+ Long mainSnapshotId, long triggerTimestamp, long doneTimestamp, long
indexGeneration) {
return new EqualityConvertPlan(
ImmutableList.of(),
ImmutableList.of(),
ImmutableList.of(),
NO_OP_STAGING_SNAPSHOT_ID,
mainSnapshotId,
triggerTimestamp,
- doneTimestamp);
+ doneTimestamp,
+ indexGeneration);
}
/**
* Cycle-start marker: a no-op plan whose done timestamp is never reached,
so consumers hold it
* until the cycle's actual plan replaces it.
*/
- public static EqualityConvertPlan cycleStart(long triggerTimestamp) {
- return noOp(null, triggerTimestamp, Long.MAX_VALUE);
+ public static EqualityConvertPlan cycleStart(long triggerTimestamp, long
indexGeneration) {
+ return noOp(null, triggerTimestamp, Long.MAX_VALUE, indexGeneration);
Review Comment:
We don't need it. Removed.
##########
docs/docs/flink-maintenance.md:
##########
@@ -114,9 +114,9 @@ Notes:
**State and restart**
- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state and **maintains it incrementally across checkpoints**: each trigger
cycle only applies the commits added on the target branch since the last
indexed snapshot, and the resulting index is persisted with the next Flink
checkpoint. On failover, the worker resumes from the most recent checkpointed
index rather than rebuilding from scratch. 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. **RocksDB is
the preferred state backend** because the PK index can grow beyond the
available heap and RocksDB spills to local disk instead of failing with an
`OutOfMemoryError`.
-- Full (re)indexing from the target branch head is only triggered in a few
cases: cold start with no checkpoint, external commits that advance the target
past the currently-indexed snapshot (see below), when the planner detects that
the target branch has diverged from the indexed snapshot (rollback /
replace-main / expired marker), or when a staging snapshot is planned again
because the cycle that planned it did not commit.
+- Full (re)indexing from the target branch head is triggered when the index
does not reflect the head of the target branch: cold start with no checkpoint,
any commit on the target other than the conversion commit of the staging
snapshot planned last (external commits as described below, rollback /
replace-main, or a restore from a checkpoint taken before a conversion commit),
a planned staging snapshot whose conversion commit did not land, after an index
update failed, or on the first trigger after a restore from state written by an
older version. A failed update on a trigger with nothing to convert keeps the
next conversion from committing, so that conversion is planned again on a
rebuilt index, unless another rebuild finished without a failure in between.
Review Comment:
Fair, much easier to read.
##########
docs/docs/flink-maintenance.md:
##########
@@ -114,9 +114,9 @@ Notes:
**State and restart**
- The PK-index worker keeps a keyed row-position index of the target branch in
Flink state and **maintains it incrementally across checkpoints**: each trigger
cycle only applies the commits added on the target branch since the last
indexed snapshot, and the resulting index is persisted with the next Flink
checkpoint. On failover, the worker resumes from the most recent checkpointed
index rather than rebuilding from scratch. 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. **RocksDB is
the preferred state backend** because the PK index can grow beyond the
available heap and RocksDB spills to local disk instead of failing with an
`OutOfMemoryError`.
-- Full (re)indexing from the target branch head is only triggered in a few
cases: cold start with no checkpoint, external commits that advance the target
past the currently-indexed snapshot (see below), when the planner detects that
the target branch has diverged from the indexed snapshot (rollback /
replace-main / expired marker), or when a staging snapshot is planned again
because the cycle that planned it did not commit.
+- Full (re)indexing from the target branch head is triggered when the index
does not reflect the head of the target branch: cold start with no checkpoint,
any commit on the target other than the conversion commit of the staging
snapshot planned last (external commits as described below, rollback /
replace-main, or a restore from a checkpoint taken before a conversion commit),
a planned staging snapshot whose conversion commit did not land, after an index
update failed, or on the first trigger after a restore from state written by an
older version. A failed update on a trigger with nothing to convert keeps the
next conversion from committing, so that conversion is planned again on a
rebuilt index, unless another rebuild finished without a failure in between.
- 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.
+- The committer keeps no state besides the index generation of the last failed
cycle, which keeps the next conversion commit on that generation from
committing. 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.
Review Comment:
Sounds good. Updated.
--
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]
