xiaoxuandev commented on code in PR #17764:
URL: https://github.com/apache/iceberg/pull/17764#discussion_r3909891561
##########
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),
Review Comment:
Good catch on the duplicates, updated, thanks!
--
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]
