raunaqmorarka commented on code in PR #18074:
URL: https://github.com/apache/iceberg/pull/18074#discussion_r4478149604
##########
core/src/main/java/org/apache/iceberg/SnapshotProducer.java:
##########
@@ -108,7 +109,9 @@ public void accept(String file) {
private final String commitUUID = UUID.randomUUID().toString();
private final AtomicInteger manifestCount = new AtomicInteger(0);
private final AtomicInteger attempt = new AtomicInteger(0);
- private final List<String> manifestLists = Lists.newArrayList();
+
+ private final Map<String, Set<String>> manifestsByManifestList =
Maps.newHashMap();
Review Comment:
Done.
##########
core/src/main/java/org/apache/iceberg/BaseTransaction.java:
##########
@@ -455,17 +455,33 @@ private void applyUpdates(TableOperations underlyingOps) {
}
// returns the manifest lists and manifests referenced by the given
committed snapshots
- private static Set<String> committedFiles(FileIO io, Set<Snapshot>
snapshots) {
+ private Set<String> committedFiles(Set<Snapshot> snapshots) {
Set<String> committedFiles = Sets.newHashSet();
for (Snapshot snap : snapshots) {
committedFiles.add(snap.manifestListLocation());
- snap.allManifests(io).forEach(manifest ->
committedFiles.add(manifest.path()));
+ committedFiles.addAll(committedManifestPaths(snap));
}
return committedFiles;
}
+ // uses the manifest paths held by the update that wrote the snapshot's
manifest list, if any
Review Comment:
Done.
##########
core/src/main/java/org/apache/iceberg/BaseTransaction.java:
##########
@@ -455,17 +455,33 @@ private void applyUpdates(TableOperations underlyingOps) {
}
// returns the manifest lists and manifests referenced by the given
committed snapshots
- private static Set<String> committedFiles(FileIO io, Set<Snapshot>
snapshots) {
+ private Set<String> committedFiles(Set<Snapshot> snapshots) {
Set<String> committedFiles = Sets.newHashSet();
for (Snapshot snap : snapshots) {
committedFiles.add(snap.manifestListLocation());
- snap.allManifests(io).forEach(manifest ->
committedFiles.add(manifest.path()));
+ committedFiles.addAll(committedManifestPaths(snap));
}
return committedFiles;
}
+ // uses the manifest paths held by the update that wrote the snapshot's
manifest list, if any
+ private Set<String> committedManifestPaths(Snapshot snapshot) {
+ for (PendingUpdate update : updates) {
+ if (update instanceof SnapshotProducer<?> producer) {
+ Set<String> paths =
producer.writtenManifestPaths(snapshot.manifestListLocation());
+ if (paths != null) {
+ return paths;
+ }
+ }
+ }
+
+ return snapshot.allManifests(ops.io()).stream()
Review Comment:
Added `transactionCleanupReadsManifestListOfExternalSnapshot`, which commits
a snapshot through the transaction's table operations so no update wrote it,
and checks that cleanup reads its manifest list.
##########
core/src/main/java/org/apache/iceberg/SnapshotProducer.java:
##########
@@ -610,13 +615,36 @@ private void notifyListeners() {
}
protected void cleanAll() {
- for (String manifestList : manifestLists) {
+ for (String manifestList : manifestsByManifestList.keySet()) {
deleteFile(manifestList);
}
- manifestLists.clear();
+ manifestsByManifestList.clear();
cleanUncommitted(EMPTY_SET);
}
+ /**
+ * Returns the manifest paths of the committed snapshot without reading its
manifest list when
+ * this producer wrote that manifest list.
+ */
+ private Set<String> committedManifestPaths(Snapshot saved) {
+ Set<String> paths = writtenManifestPaths(saved.manifestListLocation());
+ if (paths != null) {
+ return paths;
+ }
+
+ return saved.allManifests(ops.io()).stream()
Review Comment:
Done.
--
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]
