xiaoxuandev commented on code in PR #17764:
URL: https://github.com/apache/iceberg/pull/17764#discussion_r3909837909
##########
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java:
##########
@@ -528,87 +528,117 @@ private void validateNoNewDeletesForDataFiles(
return;
}
- DeleteFileIndex deletes = addedDeleteFiles(base, startingSnapshotId,
dataFilter, null, parent);
+ List<DeleteFileIndex> deleteIndexes =
+ addedDeleteFileIndexes(base, startingSnapshotId, dataFilter, null,
parent);
long startingSequenceNumber = startingSequenceNumber(base,
startingSnapshotId);
for (DataFile dataFile : dataFiles) {
- // if any delete is found that applies to files written in or before the
starting snapshot,
- // fail
- DeleteFile[] deleteFiles = deletes.forDataFile(startingSequenceNumber,
dataFile);
- if (ignoreEqualityDeletes) {
- ValidationException.check(
- Arrays.stream(deleteFiles)
- .noneMatch(deleteFile -> deleteFile.content() ==
FileContent.POSITION_DELETES),
- "Cannot commit, found new position delete for replaced data file:
%s",
- dataFile);
- } else {
- ValidationException.check(
- deleteFiles.length == 0,
- "Cannot commit, found new delete for replaced data file: %s",
- dataFile);
+ for (DeleteFileIndex deletes : deleteIndexes) {
+ // if any delete is found that applies to files written in or before
the starting snapshot,
+ // fail
+ DeleteFile[] deleteFiles = deletes.forDataFile(startingSequenceNumber,
dataFile);
+ if (ignoreEqualityDeletes) {
+ ValidationException.check(
+ !containsPositionDeletes(deleteFiles),
+ "Cannot commit, found new position delete for replaced data
file: %s",
+ dataFile);
+ } else {
+ ValidationException.check(
+ deleteFiles.length == 0,
+ "Cannot commit, found new delete for replaced data file: %s",
+ dataFile);
+ }
+ }
+ }
+ }
+
+ private static boolean containsPositionDeletes(DeleteFile[] deleteFiles) {
+ for (DeleteFile deleteFile : deleteFiles) {
+ if (deleteFile.content() == FileContent.POSITION_DELETES) {
+ return true;
}
}
+
+ return false;
}
/**
* Validates that no delete files matching a filter have been added to the
table since a starting
* snapshot.
*
+ * <p>On failure, the exception lists the matching delete files added since
that snapshot rather
+ * than those currently live in the table, so it can name a file that was
later removed.
+ *
* @param base table metadata to validate
* @param startingSnapshotId id of the snapshot current at the start of the
operation
* @param dataFilter an expression used to find new conflicting delete files
* @param parent ending snapshot on the branch being validated
*/
protected void validateNoNewDeleteFiles(
TableMetadata base, Long startingSnapshotId, Expression dataFilter,
Snapshot parent) {
- DeleteFileIndex deletes = addedDeleteFiles(base, startingSnapshotId,
dataFilter, null, parent);
+ List<DeleteFileIndex> deleteIndexes =
+ addedDeleteFileIndexes(base, startingSnapshotId, dataFilter, null,
parent);
ValidationException.check(
- deletes.isEmpty(),
+ deleteIndexes.stream().allMatch(DeleteFileIndex::isEmpty),
"Found new conflicting delete files that can apply to records matching
%s: %s",
dataFilter,
- Iterables.transform(deletes.referencedDeleteFiles(),
ContentFile::location));
+ referencedDeleteFileLocations(deleteIndexes));
}
/**
* Validates that no delete files matching a partition set have been added
to the table since a
* starting snapshot.
*
+ * <p>On failure, the exception lists the matching delete files added since
that snapshot rather
+ * than those currently live in the table, so it can name a file that was
later removed.
+ *
* @param base table metadata to validate
* @param startingSnapshotId id of the snapshot current at the start of the
operation
* @param partitionSet a partition set used to find new conflicting delete
files
* @param parent ending snapshot on the branch being validated
*/
protected void validateNoNewDeleteFiles(
TableMetadata base, Long startingSnapshotId, PartitionSet partitionSet,
Snapshot parent) {
- DeleteFileIndex deletes =
- addedDeleteFiles(base, startingSnapshotId, null, partitionSet, parent);
+ List<DeleteFileIndex> deleteIndexes =
+ addedDeleteFileIndexes(base, startingSnapshotId, null, partitionSet,
parent);
ValidationException.check(
- deletes.isEmpty(),
+ deleteIndexes.stream().allMatch(DeleteFileIndex::isEmpty),
"Found new conflicting delete files that can apply to records matching
%s: %s",
partitionSet,
- Iterables.transform(deletes.referencedDeleteFiles(),
ContentFile::location));
+ referencedDeleteFileLocations(deleteIndexes));
+ }
+
+ private static Iterable<String> referencedDeleteFileLocations(
+ List<DeleteFileIndex> deleteIndexes) {
+ return Iterables.transform(
+ Iterables.concat(
+ Iterables.transform(deleteIndexes,
DeleteFileIndex::referencedDeleteFiles)),
+ ContentFile::location);
}
/**
- * Returns matching delete files have been added to the table since a
starting snapshot.
+ * Returns one index of matching delete files per snapshot added to the
table since a starting
+ * snapshot.
+ *
+ * <p>Callers must check every index. A single index cannot hold them all
because it allows at
+ * most one deletion vector per data file. That holds within a snapshot but
not across them: a
+ * data file can be referenced by a replacement deletion vector in each
snapshot in the window.
*
* @param base table metadata to validate
* @param startingSnapshotId id of the snapshot current at the start of the
operation
* @param dataFilter an expression used to find delete files
* @param partitionSet a partition set used to find delete files
* @param parent parent snapshot of the branch
*/
- protected DeleteFileIndex addedDeleteFiles(
+ private List<DeleteFileIndex> addedDeleteFileIndexes(
Review Comment:
Renamed to `addedDeleteFilesIndexedPerSnapshot`, I kept the return type as a
List though: none of the three callers reads the snapshot ID, so a Map hands
them a key they'd each have to drop with .values(). Same reason for PerSnapshot
over BySnapshot.
It also wouldn't save work, if that was the motivation: the map we build
holds manifests, not indexes, so the loop after it isn't a map-to-list
conversion, it's building one index per group. That loop only goes away if
construction moves into the collector (groupingBy + collectingAndThen), which
buries buildDeleteFileIndex and its manifest reads inside a nested collector.
Happy to switch if you'd still rather have the grouping stated in the type.
--
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]
