mxm commented on code in PR #18717:
URL: https://github.com/apache/iceberg/pull/18717#discussion_r4482647244
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -350,14 +351,17 @@ private void ensureIndexCurrent(Snapshot mainSnapshot) {
return;
}
- if (!indexReflects(mainSnapshot)) {
+ List<Snapshot> externalCommits = externalCommitsSinceIndex(mainSnapshot);
+ if (externalCommits == null || deltaReadsMoreThanRebuild(mainSnapshot,
externalCommits)) {
boolean bootstrap = indexSnapshotId == null;
LOG.info(
"{} worker index from main snapshot {} for field IDs {}.",
bootstrap ? "Bootstrapping" : "Reindexing",
mainSnapshot.snapshotId(),
eqFieldIds);
rebuildIndex(mainSnapshot, !bootstrap);
+ } else {
+ emitIncrementalReindex(mainSnapshot, externalCommits);
Review Comment:
You're partially right. I think it is worth to distinguish here between a
transient error which can happen from time to time and a non-transient error
which makes it impossible for this code to ever complete. For transient errors,
we want to retry with a delta, if possible. That's what the PR does. For a
non-transient, e.g. one of the snapshots in the chain has been expired, we want
to do a full rebuild. And that's also the case for this PR. There is a test
which covers this: `rebuildsIndexWhenIndexedSnapshotExpired`.
--
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]
