amogh-jahagirdar commented on code in PR #16686:
URL: https://github.com/apache/iceberg/pull/16686#discussion_r3446852410
##########
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:
Huge +1 to getting rid of this synchronized.
I pasted a diff above as well, but I think we can consolidate the duplicate
and deletedFiles tracking, and and essentially not even populate deleteFiles in
a multi-threaded manner and avoid this issue entirely. The usage of parallelism
on that path is largely for I/O anyways.
I also think we don't even need the concurrent linked queue state for
tracking the files deleted by expresssion. I think once we figure out the
manifests that need to be rewritten, we already have all of the deleted files,
we can add those back to the global deleteFiles after the parallel read.
##########
core/src/main/java/org/apache/iceberg/ManifestFilterManager.java:
##########
@@ -95,6 +95,8 @@ public String partition() {
// tracking where files were deleted to validate retries quickly
private final Map<ManifestFile, Iterable<F>> filteredManifestToDeletedFiles =
Maps.newConcurrentMap();
+ private final Map<ManifestFile, Integer>
filteredManifestToDuplicateDeleteCounts =
+ Maps.newConcurrentMap();
Review Comment:
```
--- a/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestFilterManager.java
@@ -82,7 +82,6 @@ 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
@@ -93,8 +92,7 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
private final Map<ManifestFile, ManifestFile> filteredManifests =
Maps.newConcurrentMap();
// tracking where files were deleted to validate retries quickly
- private final Map<ManifestFile, Iterable<F>>
filteredManifestToDeletedFiles =
- Maps.newConcurrentMap();
+ private final Map<ManifestFile, FilterResult<F>> filteredManifestResults
= Maps.newConcurrentMap();
private final Supplier<ExecutorService> workerPoolSupplier;
@@ -229,6 +227,15 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
filtered[index] = manifest;
});
+ // merge per-manifest deletes into deleteFiles after the parallel
phase, avoiding concurrent
+ // mutation of deleteFiles during filtering
+ for (ManifestFile manifest : filtered) {
+ FilterResult<F> result = filteredManifestResults.get(manifest);
+ if (result != null) {
+ result.deletedFiles().forEach(deleteFiles::add);
+ }
+ }
+
validateRequiredDeletes(filtered);
return Arrays.asList(filtered);
@@ -256,16 +263,15 @@ 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 = filteredManifestResults.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 +311,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 = filteredManifestResults.get(manifest);
+ if (result != null) {
+ for (F file : result.deletedFiles()) {
deletedFiles.add(file);
}
}
@@ -353,6 +359,7 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
// remove the entry from the cache
filteredManifests.remove(manifest);
+ filteredManifestResults.remove(filtered);
}
}
}
@@ -501,6 +508,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 +541,13 @@ 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);
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 +570,8 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
// update caches
filteredManifests.put(manifest, filtered);
- filteredManifestToDeletedFiles.put(filtered, deletedFiles);
+ filteredManifestResults.put(
+ filtered, new FilterResult<>(deletedFiles,
duplicateDeleteCount.get()));
return filtered;
@@ -574,6 +580,8 @@ abstract class ManifestFilterManager<F extends
ContentFile<F>> {
}
}
+ private record FilterResult<T>(Iterable<T> 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
```
here's the diff I came up with that I think addresses this issue as well
##########
core/src/main/java/org/apache/iceberg/ManifestFilterManager.java:
##########
@@ -95,6 +95,8 @@ public String partition() {
// tracking where files were deleted to validate retries quickly
private final Map<ManifestFile, Iterable<F>> filteredManifestToDeletedFiles =
Maps.newConcurrentMap();
+ private final Map<ManifestFile, Integer>
filteredManifestToDuplicateDeleteCounts =
+ Maps.newConcurrentMap();
Review Comment:
Instead of 2 separate maps I think we can have a bit more simplified of a
state by having a single map and the value is a pair or record type with the
Iterable and duplicateDelete counts.
##########
core/src/main/java/org/apache/iceberg/ManifestFilterManager.java:
##########
@@ -95,6 +95,8 @@ public String partition() {
// tracking where files were deleted to validate retries quickly
private final Map<ManifestFile, Iterable<F>> filteredManifestToDeletedFiles =
Maps.newConcurrentMap();
+ private final Map<ManifestFile, Integer>
filteredManifestToDuplicateDeleteCounts =
+ Maps.newConcurrentMap();
Review Comment:
Maybe use a class instead of the record type, this seems like something we'd
want to backport and older versions would need to be JDK 11 compliant etc.
--
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]
