pvary commented on code in PR #18717:
URL: https://github.com/apache/iceberg/pull/18717#discussion_r4492640646
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -366,34 +374,209 @@ private void ensureIndexCurrent(Snapshot mainSnapshot) {
}
/**
- * Returns whether the index reflects {@code mainSnapshot}.
- *
- * <p>Planning a staging snapshot changes the index before its conversion
commit exists: its eq
- * deletes consume the entries they match and its data files are added. So
with a snapshot
- * planned, the index reflects the head only if that snapshot's conversion
commit sits directly on
- * top of the indexed snapshot. Without one, the head must be the indexed
snapshot. Any other head
- * holds commits the index never saw (external commits, rollbacks, a restore
from before a
- * conversion commit) or follows a cycle that did not commit.
+ * Returns the commits after the indexed snapshot that this converter did
not make, newest first,
+ * or {@code null} if the index needs a rebuild. The conversion commit of
the planned staging
+ * snapshot is not among them, as its cycle already updated the index.
*
- * <p>After a restore, the conversion commit can come from a retry of the
planned cycle. A cycle
- * that only failed to commit leaves the same index as its retry. A rebuild
that fails part way
+ * <p>After a restore, the conversion commit can come from a retry that ran
on an index rebuilt at
+ * a later head. The restored index plus the delta then also holds the rows
the external commits
+ * added and the retry's eq deletes matched. The conversion commit deleted
those rows, so
+ * resolving them again only adds a position the DV already holds. A rebuild
that fails part way
* clears {@link #indexSnapshotId}, and the committer does not commit a
conversion on a generation
* whose update failed downstream.
*/
- private boolean indexReflects(Snapshot mainSnapshot) {
+ private List<Snapshot> externalCommitsSinceIndex(Snapshot mainSnapshot) {
if (indexSnapshotId == null) {
+ return null;
+ }
+
+ List<Snapshot> externalCommits = Lists.newArrayList();
+ boolean plannedSnapshotCommitted = false;
+ Snapshot current = mainSnapshot;
+ while (current != null && current.snapshotId() != indexSnapshotId) {
+ String committedStaging =
+
current.summary().get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
+ if (committedStaging == null) {
+ externalCommits.add(current);
+ } else if (pendingStagingSnapshotId != null
+ && Long.parseLong(committedStaging) == pendingStagingSnapshotId) {
+ plannedSnapshotCommitted = true;
+ } else {
+ return null;
+ }
+
+ current = parentOf(current);
+ }
+
+ // A null current means the walk passed the root, or an expired parent,
without reaching the
+ // indexed snapshot.
+ if (current == null || (pendingStagingSnapshotId != null &&
!plannedSnapshotCommitted)) {
+ return null;
+ }
+
+ return externalCommits;
+ }
+
+ private boolean deltaReadsMoreThanRebuild(Snapshot mainSnapshot,
List<Snapshot> externalCommits) {
+ // The manifest list counts bound the delta before any manifest is read or
any DataFile is held.
+ // A file added and removed within the range counts twice, although the
delta skips it.
+ long deltaRows = 0L;
+ for (Snapshot snapshot : externalCommits) {
+ for (ManifestFile manifest : snapshot.dataManifests(table.io())) {
+ if (Objects.equals(manifest.snapshotId(), snapshot.snapshotId())) {
+ if (!hasRowCounts(manifest)) {
+ return true;
+ }
+
+ deltaRows += manifest.addedRowsCount() + manifest.deletedRowsCount();
+ }
+ }
+ }
+
+ if (deltaRows == 0L) {
return false;
}
- if (pendingStagingSnapshotId == null) {
- return mainSnapshot.snapshotId() == indexSnapshotId;
+ long rebuildRows = 0L;
+ for (ManifestFile manifest : mainSnapshot.dataManifests(table.io())) {
+ if (!hasRowCounts(manifest)) {
+ return true;
+ }
+
+ rebuildRows += manifest.addedRowsCount() + manifest.existingRowsCount();
+ }
+
+ if (deltaRows <= rebuildRows) {
+ return false;
}
- String committedStaging =
-
mainSnapshot.summary().get(EqualityConvertCommitter.COMMITTED_STAGING_SNAPSHOT_PROPERTY);
- return committedStaging != null
- && Long.parseLong(committedStaging) == pendingStagingSnapshotId
- && Objects.equals(mainSnapshot.parentId(), indexSnapshotId);
+ LOG.info(
+ "The external commits since the indexed snapshot change {} rows, more
than the {} rows of "
+ + "main snapshot {}. Rebuilding the worker index instead of
applying them.",
+ deltaRows,
+ rebuildRows,
+ mainSnapshot.snapshotId());
+ return true;
+ }
+
+ // The row counts are required by the spec, but a writer may leave them out.
+ private static boolean hasRowCounts(ManifestFile manifest) {
+ return manifest.addedRowsCount() != null
+ && manifest.existingRowsCount() != null
+ && manifest.deletedRowsCount() != null;
+ }
+
+ /**
+ * Applies the data-file changes of {@code externalCommits} to the worker
index in one phase.
+ *
+ * <p>Clears {@link #indexSnapshotId} before the first command. The caller
records the head only
+ * after this returns, so a delta that fails or is emitted in part is not
trusted.
+ */
+ private void emitIncrementalReindex(Snapshot mainSnapshot, List<Snapshot>
externalCommits) {
+ Map<String, DataFile> added = Maps.newLinkedHashMap();
+ Map<String, DataFile> removed = Maps.newLinkedHashMap();
+ for (Snapshot snapshot : externalCommits) {
+ collectDataFileChanges(snapshot, added, removed);
+ }
Review Comment:
Maybe keep a separate list with the locations
--
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]
