mxm commented on code in PR #18477:
URL: https://github.com/apache/iceberg/pull/18477#discussion_r4414861459
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -234,15 +256,46 @@ 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, resolvableSnapshotIds));
Review Comment:
Statistics files are listed from table metadata, not from the manifest list,
so we can still delete them when the manifest list is missing. Should this be:
```suggestion
.union(statisticsFileDS(staticTable, deletedSnapshotIds));
```
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java:
##########
@@ -234,15 +256,46 @@ 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, resolvableSnapshotIds));
+ }
+
+ private Set<Long> resolvableSnapshotIds(TableMetadata metadata, Set<Long>
deletedSnapshotIds) {
+ 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();
Review Comment:
NIT: We can iterate over the IDs directly. Should this be:
```suggestion
Tasks.foreach(deletedSnapshotIds)
.executeWith(ThreadPools.getWorkerPool())
.run(
snapshotId -> {
String manifestListLocation =
metadata.snapshot(snapshotId).manifestListLocation();
```
##########
spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestExpireSnapshotsAction.java:
##########
@@ -1416,4 +1418,94 @@ public void testCleanExpiredMetadata() {
.contains(FILE_A.location())
.doesNotContain(fileInNewSpec.location());
}
+
+ @TestTemplate
+ public void
testExpireWithMissingManifestListReachableOnlyFromExpiredSnapshots() {
+ assertExpireSkipsMissingMetadata(Snapshot::manifestListLocation);
+ }
+
+ @TestTemplate
+ public void testExpireWithMissingManifestReachableOnlyFromExpiredSnapshots()
{
+ assertExpireSkipsMissingMetadata(
+ snapshot ->
Iterables.getOnlyElement(snapshot.allManifests(table.io())).path());
+ }
+
+ private void assertExpireSkipsMissingMetadata(
+ Function<Snapshot, String> missingLocationResolver) {
+ table.newAppend().appendFile(FILE_A).commit();
+ String missingLocation =
missingLocationResolver.apply(table.currentSnapshot());
+
+ table.newOverwrite().deleteFile(FILE_A).addFile(FILE_B).commit();
+ table.newOverwrite().deleteFile(FILE_B).addFile(FILE_C).commit();
+ table.newAppend().appendFile(FILE_D).commit();
+
+ long end = rightAfterSnapshot();
+
+ table.io().deleteFile(missingLocation);
+
+ Set<String> deletedFiles = ConcurrentHashMap.newKeySet();
+
+ assertThatThrownBy(
+ () ->
+ SparkActions.get()
+ .expireSnapshots(table)
+ .expireOlderThan(end)
+ .deleteWith(deletedFiles::add)
+ .execute())
+ .isInstanceOf(NotFoundException.class)
+ .hasMessageContaining(missingLocation);
+
+ assertThat(table.snapshots()).as("Expiration should still
commit").hasSize(1);
+ assertThat(deletedFiles)
+ .as("Should delete files resolved from other expired snapshots")
+ .contains(FILE_B.location());
+ assertThat(deletedFiles)
+ .as("Should leave files reachable only through the missing metadata as
orphans")
+ .doesNotContain(FILE_A.location());
+ }
+
+ @TestTemplate
+ public void
testExpireFailsFastWhenMissingManifestReachableFromRetainedSnapshot() {
+
assertExpireFailsFastWhenMissingManifestReachableFromRetainedSnapshot(false);
+ }
+
+ @TestTemplate
+ public void
testExpireFailsFastWhenMissingManifestReachableFromRetainedSnapshotStreaming() {
+
assertExpireFailsFastWhenMissingManifestReachableFromRetainedSnapshot(true);
+ }
+
+ private void
assertExpireFailsFastWhenMissingManifestReachableFromRetainedSnapshot(
+ boolean streamResults) {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long firstSnapshotId = table.currentSnapshot().snapshotId();
+ String sharedManifest =
+
Iterables.getOnlyElement(table.snapshot(firstSnapshotId).allManifests(table.io())).path();
+
+ table.newFastAppend().appendFile(FILE_B).commit();
+ table.newFastAppend().appendFile(FILE_C).commit();
+
+ long end = rightAfterSnapshot();
+
+ table.io().deleteFile(sharedManifest);
+
+ Set<String> deletedFiles = ConcurrentHashMap.newKeySet();
+
+ ExpireSnapshots action =
+ SparkActions.get()
+ .expireSnapshots(table)
+ .expireOlderThan(end)
+ .retainLast(1)
+ .deleteWith(deletedFiles::add);
+ if (streamResults) {
+ action.option("stream-results", "true");
+ }
+
+ assertThatThrownBy(action::execute)
+ .isInstanceOf(Exception.class)
Review Comment:
NIT: Could we assert `SparkException` here?
--
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]
