dramaticlly commented on code in PR #18477:
URL: https://github.com/apache/iceberg/pull/18477#discussion_r4428937164
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -149,6 +150,11 @@ protected Dataset<FileInfo> contentFileDS(Table table) {
}
protected Dataset<FileInfo> contentFileDS(Table table, Set<Long>
snapshotIds) {
+ return contentFileDS(table, snapshotIds, null);
Review Comment:
the 2 argument method of `contentFileDS` seem now useless as the snapshotIds
is always null among all caller. Can probably drop it
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -234,15 +256,45 @@ private boolean streamResults() {
}
private Dataset<FileInfo> fileDS(TableMetadata metadata) {
- return fileDS(metadata, null);
+ Table staticTable = newStaticTable(metadata, table.io());
+ return contentFileDS(staticTable)
+ .union(manifestDS(staticTable))
+ .union(manifestListDS(staticTable))
+ .union(statisticsFileDS(staticTable, null));
}
- private Dataset<FileInfo> fileDS(TableMetadata metadata, Set<Long>
snapshotIds) {
+ private Dataset<FileInfo> deleteCandidateFileDS(
+ TableMetadata metadata, Set<Long> deletedSnapshotIds) {
Table staticTable = newStaticTable(metadata, table.io());
- return contentFileDS(staticTable, snapshotIds)
- .union(manifestDS(staticTable, snapshotIds))
- .union(manifestListDS(staticTable, snapshotIds))
- .union(statisticsFileDS(staticTable, snapshotIds));
+
+ Set<Long> resolvableSnapshotIds = resolvableSnapshotIds(metadata,
deletedSnapshotIds);
+
+ return contentFileDS(staticTable, resolvableSnapshotIds, missingManifests)
+ .union(manifestDS(staticTable, resolvableSnapshotIds))
+ .union(manifestListDS(staticTable, resolvableSnapshotIds))
+ .union(statisticsFileDS(staticTable, deletedSnapshotIds));
+ }
+
+ private Set<Long> resolvableSnapshotIds(TableMetadata metadata, Set<Long>
deletedSnapshotIds) {
+ Set<Long> resolvable = Sets.newConcurrentHashSet();
+ Tasks.foreach(deletedSnapshotIds)
+ .executeWith(ThreadPools.getWorkerPool())
+ .run(
+ snapshotId -> {
+ String manifestListLocation =
metadata.snapshot(snapshotId).manifestListLocation();
+ if (manifestListLocation != null
+ && !table.io().newInputFile(manifestListLocation).exists()) {
+ LOG.warn(
+ "Manifest list {} of expired snapshot {} is missing;
skipping its files",
+ manifestListLocation,
+ snapshotId);
+ missingManifestListLocations.add(manifestListLocation);
+ } else {
+ resolvable.add(snapshotId);
+ }
+ });
Review Comment:
I think the snapshot resolution now happens on driver, so I would add retry
and suppressFailure like in IncrementalFileCleanup.java.
```suggestion
Set<Long> resolvable = Sets.newConcurrentHashSet(deletedSnapshotIds);
Tasks.foreach(deletedSnapshotIds)
.retry(3)
.suppressFailureWhenFinished()
.executeWith(ThreadPools.getWorkerPool())
.onFailure((snapshotId, exc) ->
LOG.warn("Failed to check manifest list of expired snapshot {}",
snapshotId, exc))
.run(snapshotId -> {
String location = metadata.snapshot(snapshotId).manifestListLocation();
if (location != null && !table.io().newInputFile(location).exists()) {
LOG.warn("Manifest list {} of expired snapshot {} is missing;
skipping its files", location, snapshotId);
missingManifestListLocations.add(location);
resolvable.remove(snapshotId);
}
});
return resolvable;
```
--
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]
