mxm commented on code in PR #18717:
URL: https://github.com/apache/iceberg/pull/18717#discussion_r4523402670
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java:
##########
@@ -366,41 +368,212 @@ 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 reads it once.
+ 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;
+ }
+
+ 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. */
+ private void emitIncrementalReindex(Snapshot mainSnapshot, List<Snapshot>
externalCommits) {
+ int addedFileNum = 0;
+ int removedFileNum = 0;
+ Set<String> removedLater = Sets.newHashSet();
+
+ // Traverse from latest to oldest to figure out which files have been
added-then-deleted.
+ // We don't have a defined ordering in the parallel reader tasks, which is
why we need to
+ // prevent the same file being added and removed. We can only guarantee
correctness if we
+ // either add or remove a file, not both.
+ for (Snapshot snapshot : externalCommits) {
+ if (!hasFileChanges(
+ snapshot,
+ manifest -> manifest.hasAddedFiles() || manifest.hasDeletedFiles(),
+ SnapshotSummary.ADDED_FILES_PROP,
+ SnapshotSummary.ADDED_DELETE_FILES_PROP,
+ SnapshotSummary.DELETED_FILES_PROP)) {
+ // No data added or removed, e.g. a manifest rewrite, for which
SnapshotChanges would read
+ // every manifest to find no file change. Skip.
+ continue;
+ }
+
+ SnapshotChanges changes =
SnapshotChanges.builderFor(table).snapshot(snapshot).build();
+ validateTargetDeleteFiles(snapshot, changes);
+
+ for (DataFile dataFile : changes.removedDataFiles()) {
+ removedLater.add(dataFile.location());
+ // We need to emit the removed file even if an earlier snapshot in the
range added the file.
+ // Otherwise, we would have to traverse the snapshot chain twice.
Removing a file which is
+ // never added will not change the index.
+ emitRemovedDataFile(mainSnapshot, dataFile);
+ removedFileNum++;
+ }
+
+ for (DataFile dataFile : changes.addedDataFiles()) {
+ // Make sure to not index a file which has already been removed.
+ if (!removedLater.remove(dataFile.location())) {
+ emitAddedDataFile(
+ mainSnapshot, // Column stats cost memory and the reader doesn't
need them.
+ dataFile.copyWithoutStats());
+ addedFileNum++;
+ }
+ }
+ }
+
+ if (addedFileNum == 0 && removedFileNum == 0) {
+ return;
+ }
+
+ incrementalReindexCounter.inc();
+ LOG.info(
+ "Incrementally reindexing worker index to main snapshot {} for field
IDs {}: {} added and "
+ + "{} removed data files.",
+ mainSnapshot.snapshotId(),
+ eqFieldIds,
+ addedFileNum,
+ removedFileNum);
+ advancePhase();
+ }
+
+ private void emitRemovedDataFile(Snapshot mainSnapshot, DataFile dataFile) {
+ output.collect(
+ new StreamRecord<>(
+ ReadCommand.removedDataFile(
+ reindexTask(dataFile),
+ mainSnapshot.snapshotId(),
+ indexGeneration,
+ dataSequenceNumber(dataFile)),
+ nextPhaseTs));
+ }
+
+ private void emitAddedDataFile(Snapshot mainSnapshot, DataFile dataFile) {
+ output.collect(
+ new StreamRecord<>(
+ ReadCommand.dataFile(
+ reindexTask(dataFile),
+ mainSnapshot.snapshotId(),
+ indexGeneration,
+ dataSequenceNumber(dataFile)),
+ nextPhaseTs));
+ }
+
+ // Mirrors the reader's check on the delete files of a rebuild.
+ private void validateTargetDeleteFiles(Snapshot snapshot, SnapshotChanges
changes) {
+ if (stagingOnTargetBranch) {
+ // On a shared branch, equality deletes on the target are the
converter's input.
+ return;
+ }
+
+ for (DeleteFile deleteFile : changes.addedDeleteFiles()) {
+ Preconditions.checkState(
Review Comment:
Yes, we will fail in the EqualityConvertReader when we load the data file.
--
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]
