xiaoxuandev commented on code in PR #17764:
URL: https://github.com/apache/iceberg/pull/17764#discussion_r3909818095
##########
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);
Review Comment:
Yeah, agreed on the abstraction, an existence check doesn't care about the
one-DV-per-data-file invariant, so dropping the index in both
`validateNoNewDeleteFiles` overloads would also drop the per-snapshot grouping
they inherit but never need. `validateAddedDVs` already scans delete manifests
this way without an index.
But worth noting it probably won't move the GC churn case you're describing,
though. N there is the validation window, which scales with how stale the
committer's starting snapshot is, not with commit rate — so the large-N
committer is RewriteFiles, which only goes through
validateNoNewDeletesForDataFiles and would keep the index either way. RowDelta
commits far more often but usually reads and commits within a window of a few
snapshots.
--
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]
