nastra commented on code in PR #16686:
URL: https://github.com/apache/iceberg/pull/16686#discussion_r3440832727
##########
core/src/main/java/org/apache/iceberg/ManifestFilterManager.java:
##########
@@ -279,14 +281,16 @@ SnapshotSummary.Builder
buildSummary(Iterable<ManifestFile> manifests) {
private void validateRequiredDeletes(ManifestFile... manifests) {
if (failMissingDeletePaths) {
Set<F> deletedFiles = deletedFiles(manifests);
- ValidationException.check(
- deletedFiles.containsAll(deleteFiles),
- "Missing required files to delete: %s",
- COMMA.join(
- deleteFiles.stream()
- .filter(f -> !deletedFiles.contains(f))
- .map(ContentFile::location)
- .collect(Collectors.toList())));
+ synchronized (deleteFiles) {
Review Comment:
it would be good to avoid the synchronized block. An alternative approach
that avoids this would be
```
diff --git
a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
index 7d146d9246..66ef96a658 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
@@ -22,7 +22,9 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
@@ -82,18 +84,22 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
private long minSequenceNumber = 0;
private boolean failAnyDelete = false;
private boolean failMissingDeletePaths = false;
- private int duplicateDeleteCount = 0;
private boolean caseSensitive = true;
private boolean allDeletesReferenceManifests = true;
// this is only being used for the DeleteManifestFilterManager to detect
orphaned DVs for removed
// data file paths
private Set<String> removedDataFilePaths = Sets.newHashSet();
+ // thread-safe queue for files discovered via expression matching during
parallel filtering,
+ // drained into deleteFiles after the parallel phase completes
+ private final Queue<F> pendingDeletes = new ConcurrentLinkedQueue<>();
+
// cache filtered manifests to avoid extra work when commits fail.
private final Map<ManifestFile, ManifestFile> filteredManifests =
Maps.newConcurrentMap();
- // tracking where files were deleted to validate retries quickly
- private final Map<ManifestFile, Iterable<F>>
filteredManifestToDeletedFiles =
+ // cached per-manifest filter results (deleted files and duplicate count)
for summary building
+ // and retry correctness
+ private final Map<ManifestFile, FilterResult<F>>
filteredManifestToDeletedFiles =
Maps.newConcurrentMap();
private final Supplier<ExecutorService> workerPoolSupplier;
@@ -229,6 +235,11 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
filtered[index] = manifest;
});
+ // merge expression-matched files discovered during parallel filtering
into deleteFiles
+ for (F file = pendingDeletes.poll(); file != null; file =
pendingDeletes.poll()) {
+ deleteFiles.add(file);
+ }
+
validateRequiredDeletes(filtered);
return Arrays.asList(filtered);
@@ -256,16 +267,16 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
for (ManifestFile manifest : manifests) {
PartitionSpec manifestSpec =
specsById.get(manifest.partitionSpecId());
- Iterable<F> manifestDeletes =
filteredManifestToDeletedFiles.get(manifest);
- if (manifestDeletes != null) {
- for (F file : manifestDeletes) {
+ FilterResult<F> result = filteredManifestToDeletedFiles.get(manifest);
+ if (result != null) {
+ for (F file : result.deletedFiles()) {
summaryBuilder.deletedFile(manifestSpec, file);
}
+
+
summaryBuilder.incrementDuplicateDeletes(result.duplicateDeleteCount());
}
}
- summaryBuilder.incrementDuplicateDeletes(duplicateDeleteCount);
-
return summaryBuilder;
}
@@ -305,9 +316,9 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
if (manifests != null) {
for (ManifestFile manifest : manifests) {
- Iterable<F> manifestDeletes =
filteredManifestToDeletedFiles.get(manifest);
- if (manifestDeletes != null) {
- for (F file : manifestDeletes) {
+ FilterResult<F> result =
filteredManifestToDeletedFiles.get(manifest);
+ if (result != null) {
+ for (F file : result.deletedFiles()) {
deletedFiles.add(file);
}
}
@@ -351,8 +362,9 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
replacedManifestsCount.decrementAndGet();
}
- // remove the entry from the cache
+ // remove the entries from the caches
filteredManifests.remove(manifest);
+ filteredManifestToDeletedFiles.remove(filtered);
}
}
}
@@ -501,6 +513,7 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
// when this point is reached, there is at least one file that will be
deleted in the
// manifest. produce a copy of the manifest with all deleted files
removed.
Set<F> deletedFiles = newFileSet();
+ AtomicInteger duplicateDeleteCount = new AtomicInteger(0);
try {
ManifestWriter<F> writer = newManifestWriter(reader.spec());
@@ -533,16 +546,17 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
if (allRowsMatch) {
writer.delete(entry);
F fileCopy = file.copyWithoutStats();
- // add the file here in case it was deleted using an
expression. The
- // DeleteManifestFilterManager will then remove its
matching DV
- deleteFiles.add(fileCopy);
+ // add the file to the pending queue so it can be
merged into deleteFiles
+ // after the parallel phase. The
DeleteManifestFilterManager will then
+ // remove its matching DV
+ pendingDeletes.add(fileCopy);
if (deletedFiles.contains(file)) {
LOG.warn(
"Deleting a duplicate path from manifest {}:
{}",
manifest.path(),
file.location());
- duplicateDeleteCount += 1;
+ duplicateDeleteCount.incrementAndGet();
} else {
// only add the file to deletes if it is a new
delete
// this keeps the snapshot summary accurate for
non-duplicate data
@@ -565,7 +579,8 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
// update caches
filteredManifests.put(manifest, filtered);
- filteredManifestToDeletedFiles.put(filtered, deletedFiles);
+ filteredManifestToDeletedFiles.put(
+ filtered, new FilterResult<>(deletedFiles,
duplicateDeleteCount.get()));
return filtered;
@@ -574,6 +589,8 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
}
}
+ private record FilterResult<F>(Iterable<F> deletedFiles, int
duplicateDeleteCount) {}
+
// an evaluator that checks whether rows in a file may/must match a given
expression
// this class first partially evaluates the provided expression using the
partition tuple
// and then checks the remaining part of the expression using metrics
evaluators
```
--
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]
