dramaticlly commented on code in PR #18477:
URL: https://github.com/apache/iceberg/pull/18477#discussion_r4277913902
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -407,14 +416,21 @@ public long totalFilesCount() {
private static class ReadManifest implements
FlatMapFunction<ManifestFileBean, FileInfo> {
private final Broadcast<Table> table;
+ private final CollectionAccumulator<String> missingManifests;
- ReadManifest(Broadcast<Table> table) {
+ ReadManifest(Broadcast<Table> table, CollectionAccumulator<String>
missingManifests) {
this.table = table;
+ this.missingManifests = missingManifests;
}
@Override
public Iterator<FileInfo> call(ManifestFileBean manifest) {
- return new ClosingIterator<>(entries(manifest));
+ if (missingManifests == null) {
+ return new ClosingIterator<>(entries(manifest));
+ }
+
+ return new BestEffortManifestIterator(
+ () -> entries(manifest), manifest.path(), missingManifests);
Review Comment:
Do we need a new BestEffortManifestIterator ? I think we can try and catch
instead. In the catch of NotFoundException, we can then evaluate the
missingManifests and conditionally apply throw or add to set accumulator? I
think it shall simplify a lot
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -222,27 +234,93 @@ private String jobDesc() {
}
private ExpireSnapshots.Result doExecute() {
+ boolean bestEffort = bestEffort();
+ Dataset<FileInfo> expiredFiles = expireFiles();
+ if (bestEffort) {
+ missingManifests.reset();
+ }
+
+ ExpireSnapshots.Result result;
if (streamResults()) {
- return deleteFiles(expireFiles().toLocalIterator());
+ result = deleteFiles(expiredFiles.toLocalIterator());
} else {
- return deleteFiles(expireFiles().collectAsList().iterator());
+ result = deleteFiles(expiredFiles.collectAsList().iterator());
+ }
+
+ if (bestEffort) {
+ failIfMissingFiles();
+ }
+
+ return result;
+ }
Review Comment:
I think we can just move `bestEffort` entirely within deleteCandidateFileDS
and the logic of failIfMissingFiles can solely rely on value presence in
`missingManifestListLocations`
```suggestion
private ExpireSnapshots.Result doExecute() {
Dataset<FileInfo> expiredFiles = expireFiles();
Iterator<FileInfo> filesToDelete = streamResults() ?
expiredFiles.toLocalIterator() : expiredFiles.collectAsList().iterator();
ExpireSnapshots.Result result = deleteFiles(filesToDelete);
failIfMissingMetadata();
}
```
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java:
##########
@@ -149,6 +152,11 @@ protected Dataset<FileInfo> contentFileDS(Table table) {
}
protected Dataset<FileInfo> contentFileDS(Table table, Set<Long>
snapshotIds) {
+ return contentFileDS(table, snapshotIds, null);
+ }
+
+ protected Dataset<FileInfo> contentFileDS(
+ Table table, Set<Long> snapshotIds, CollectionAccumulator<String>
missingManifests) {
Review Comment:
I think we can reuse existing
org.apache.iceberg.spark.actions.SetAccumulator instead, as we don't really
need the ordering, just need to figure out which files are missing.
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -222,27 +234,93 @@ private String jobDesc() {
}
private ExpireSnapshots.Result doExecute() {
+ boolean bestEffort = bestEffort();
+ Dataset<FileInfo> expiredFiles = expireFiles();
+ if (bestEffort) {
+ missingManifests.reset();
+ }
+
+ ExpireSnapshots.Result result;
if (streamResults()) {
- return deleteFiles(expireFiles().toLocalIterator());
+ result = deleteFiles(expiredFiles.toLocalIterator());
} else {
- return deleteFiles(expireFiles().collectAsList().iterator());
+ result = deleteFiles(expiredFiles.collectAsList().iterator());
+ }
+
+ if (bestEffort) {
+ failIfMissingFiles();
+ }
+
+ return result;
+ }
+
+ private void failIfMissingFiles() {
+ Set<String> missingFiles = Sets.newTreeSet(missingManifestListLocations);
+ missingFiles.addAll(missingManifests.value());
+
+ if (!missingFiles.isEmpty()) {
+ throw new NotFoundException(
+ "Expired snapshots and deleted all fully-resolved files, but %s
metadata file(s) "
+ + "reachable only from expired snapshots are missing from
storage and were skipped. "
+ + "Missing files: %s",
+ missingFiles.size(), COMMA_JOINER.join(missingFiles));
}
}
private boolean streamResults() {
return PropertyUtil.propertyAsBoolean(options(), STREAM_RESULTS,
STREAM_RESULTS_DEFAULT);
}
+ private boolean bestEffort() {
+ return PropertyUtil.propertyAsBoolean(options(), BEST_EFFORT,
BEST_EFFORT_DEFAULT);
+ }
+
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;
+ if (bestEffort()) {
+ this.missingManifestListLocations = Sets.newConcurrentHashSet();
+ this.missingManifests =
+
spark().sparkContext().collectionAccumulator("expire-snapshots-missing-manifests");
+
+ Set<Long> resolvable = Sets.newConcurrentHashSet();
+ Tasks.foreach(
+
deletedSnapshotIds.stream().map(metadata::snapshot).collect(Collectors.toList()))
+ .executeWith(ThreadPools.getWorkerPool())
+ .run(
+ snapshot -> {
+ long snapshotId = snapshot.snapshotId();
+ String manifestListLocation = snapshot.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);
+ }
+ });
+ resolvableSnapshotIds = resolvable;
+ } else {
+ resolvableSnapshotIds = deletedSnapshotIds;
Review Comment:
can we move the logic of resolving snapshot ids to a separate method? I
think it help keep the methods more organized as this purely focused on fileSet
based on the value of bestEffort()?
--
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]
