mxm commented on code in PR #18717:
URL: https://github.com/apache/iceberg/pull/18717#discussion_r4513096358
##########
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.
Review Comment:
We can fix it by doing two passes over the snapshots / manifests, which are
necessary if we don't want to materialize the data file metadata like the
original PR version did. The cost is pretty high for an edge case.
I would prefer to restore the original code, which didn't have this
limitation:
https://github.com/apache/iceberg/pull/18717/changes/3de38d1e5b4d60a246dd2fd983915d04dcc6a45d#diff-abdfc041859e39f2253465f611658eb1b251d7d5540a55b6479d9361c95a4735R482
Here is an experiment I ran:
```
┌──────────────────────────────────────────────┬─────────────────────────────────────────┬────────────────┐
│ What │ Held by
│ Bytes per file │
├──────────────────────────────────────────────┼─────────────────────────────────────────┼────────────────┤
│ DataFile with stats │ SnapshotChanges, added
files │ 23,184 │
├──────────────────────────────────────────────┼─────────────────────────────────────────┼────────────────┤
│ DataFile without stats, in a list │ SnapshotChanges, removed
files │ 500 │
├──────────────────────────────────────────────┼─────────────────────────────────────────┼────────────────┤
│ DataFile without stats, in a HashMap by path │ a materializing pass
│ 550 │
├──────────────────────────────────────────────┼─────────────────────────────────────────┼────────────────┤
│ Minimal record, in a HashMap by path │ a materializing pass with
a custom type │ 397 │
├──────────────────────────────────────────────┼─────────────────────────────────────────┼────────────────┤
│ Path, in a HashSet │ the current pass
│ 229 │
└──────────────────────────────────────────────┴─────────────────────────────────────────┴────────────────┘
Example: a compaction in one delta removes 300,000 files and adds 10,000, so
310,000 paths are touched.
current pass (paths): 310,000 x 229 B = 71.0 MB
materializing, no-stats copy: 310,000 x 550 B = 170.5 MB
materializing, minimal record: 310,000 x 397 B = 123.1 MB (28% less
than the copies)
```
--
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]
