chenjian2664 commented on code in PR #18408:
URL: https://github.com/apache/iceberg/pull/18408#discussion_r4259528179
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -111,6 +113,7 @@ public void cleanFiles(
// find manifests to clean up that are still referenced by a valid
snapshot, but written by an
// expired snapshot
Set<String> validManifests = ConcurrentHashMap.newKeySet();
+ Map<String, ManifestFile> liveDeleteManifests = Maps.newConcurrentMap();
Review Comment:
Nit: Use a `Set<ManifestFile>` for `liveDeleteManifests`, like
`manifestsToScan`; `GenericManifestFile` equality is by path.
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -302,14 +314,21 @@ private Set<String> findFilesToDelete(
if (entry.status() == ManifestEntry.Status.DELETED
&& !validIds.contains(entry.snapshotId())) {
// use toString to ensure the path will not change (Utf8
is reused)
- filesToDelete.add(entry.file().location());
+ String location = entry.file().location();
Review Comment:
Nit: Drop this comment: no `toString` call has existed here since
`location()` replaced `path().toString()`。
```suggestion
String location = entry.file().location();
```
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -335,4 +354,45 @@ private Set<String> findFilesToDelete(
return filesToDelete;
}
+
+ // A Puffin file can be shared by multiple deletion vectors, so keep it if
any live one remains
+ private Set<String> unreferencedDVs(
+ Set<String> dvFileRemovalCandidates,
+ Collection<ManifestFile> liveDeleteManifests,
+ Map<Integer, PartitionSpec> specsById) {
+ if (dvFileRemovalCandidates.isEmpty()) {
+ return dvFileRemovalCandidates;
+ }
+
+ Set<String> unreferenced = ConcurrentHashMap.newKeySet();
+ unreferenced.addAll(dvFileRemovalCandidates);
+
+ try {
+ Tasks.foreach(liveDeleteManifests)
+ .retry(3)
+ .stopOnFailure()
+ .throwFailureWhenFinished()
+ .executeWith(planExecutorService)
+ .onFailure(
+ (item, exc) ->
+ LOG.warn(
+ "Failed to determine live DVs in manifest {}. Retrying",
item.path(), exc))
Review Comment:
Nit: Drop "Retrying" from the onFailure message
Drop `. Retrying`: `Tasks` calls `onFailure` only after the last attempt,
and `stopOnFailure()` then ends the run.
```suggestion
LOG.warn("Failed to determine live DVs in manifest {}",
item.path(), exc))
```
##########
core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java:
##########
@@ -335,4 +354,45 @@ private Set<String> findFilesToDelete(
return filesToDelete;
}
+
+ // A Puffin file can be shared by multiple deletion vectors, so keep it if
any live one remains
+ private Set<String> unreferencedDVs(
+ Set<String> dvFileRemovalCandidates,
+ Collection<ManifestFile> liveDeleteManifests,
+ Map<Integer, PartitionSpec> specsById) {
+ if (dvFileRemovalCandidates.isEmpty()) {
+ return dvFileRemovalCandidates;
+ }
+
+ Set<String> unreferenced = ConcurrentHashMap.newKeySet();
+ unreferenced.addAll(dvFileRemovalCandidates);
+
+ try {
+ Tasks.foreach(liveDeleteManifests)
+ .retry(3)
+ .stopOnFailure()
+ .throwFailureWhenFinished()
+ .executeWith(planExecutorService)
+ .onFailure(
+ (item, exc) ->
+ LOG.warn(
+ "Failed to determine live DVs in manifest {}. Retrying",
item.path(), exc))
+ .run(
+ manifest -> {
+ try (CloseableIterable<String> paths =
+ ManifestFiles.readPaths(manifest, fileIO, specsById)) {
Review Comment:
Nit: Return early once no candidate remains
```suggestion
manifest -> {
if (unreferenced.isEmpty()) {
return;
}
try (CloseableIterable<String> paths =
ManifestFiles.readPaths(manifest, fileIO, specsById)) {
```
##########
core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java:
##########
@@ -2144,6 +2144,89 @@ public void testCleanupLevelNullValidation() {
.hasMessageContaining("Invalid cleanup level: null");
}
+ @TestTemplate
+ public void expireDoesNotDeletePuffinFileSharedByLiveDeletionVectors()
throws Exception {
Review Comment:
Nit: Drop the leftover `throws Exception` and `public`
```suggestion
void expireDoesNotDeletePuffinFileSharedByLiveDeletionVectors() {
```
--
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]
