amogh-jahagirdar commented on code in PR #17764:
URL: https://github.com/apache/iceberg/pull/17764#discussion_r3904638440
##########
core/src/test/java/org/apache/iceberg/TestRewriteFiles.java:
##########
@@ -813,4 +813,254 @@ public void removingDataFileAlsoRemovesDV() {
files(fileADeletes(), fileBDeletes()),
statuses(ManifestEntry.Status.DELETED, ManifestEntry.Status.EXISTING));
}
+
+ /**
+ * A rewrite of a data file that acquired a new deletion vector after the
starting snapshot is a
+ * genuine conflict and must be reported as one, even when concurrent
writers superseded that
+ * deletion vector several times within the validation window. This is the
path compaction takes
+ * on merge-on-read tables where writers replace deletion vectors faster
than a rewrite commits.
+ *
+ * <p>See: https://github.com/apache/iceberg/issues/17206
+ */
Review Comment:
Can we remove these top level comments from the tests in this and the other
files? It's a bit noisy and not adding too much value, especially when the test
method names are pretty clear.
##########
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:
`deleteFileIndicesBySnapshot`? And I think it should return a Map<Long,
DeleteFileIndex>, I think that would avoid the extra work we do in the method
to take the map we already build and make a list out of it.
##########
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:
Maybe not required in this PR because I don't think it's a huge deal in
practice, but honestly after staring at all this validation code, I feel like
we don't even need a DeleteFileIndex for at least the validations we do
currently.
e.g.
validateNoNewDeleteFiles needs to just look for any delete files matching
the partition predicate or data filter since the starting snapshotID. etc
validateNoNewDeletesForDataFiles is a bit more complex with equality deletes
matching that DeleteFileIndex does take care of for us but I think it's
possible (maybe a bit duplicative).
But either way where my comment is largely coming from is that
DeleteFileIndex allocates a few maps and has some state that largely is
irrelevant for the kinds of checks we do currently in validation, so that's why
it feels a bit heavy. But like I said, I don't think it matters too much. For
the batch case it's basically nothing. For the streaming upsert case where
there's a bunch of DV churn and many concurrent snapshots added, then there's
probably GC churn as maps/state get created/torn down eetc. But for those cases
the structure,lookups should be fast because the contents of what's being added
is fairly small in those cases.
##########
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
Review Comment:
Think we should remove this, not adding too much value and can just be a
comment that churns as the implementation changes.
##########
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
Review Comment:
Same as the other comments, don't think we need this.
##########
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:
I'd just keep track of Set<String> referencedDeleteFiles.
```
protected void validateNoNewDeleteFiles(
TableMetadata base, Long startingSnapshotId, Expression dataFilter,
Snapshot parent) {
List<DeleteFileIndex> deleteIndices =
addedDeleteFilesIndexedBySnapshot(base, startingSnapshotId,
dataFilter, null, parent);
Set<String> deleteFileLocations = Sets.newLinkedHashSet();
for (DeleteFileIndex deletes : deleteIndices) {
if (!deletes.isEmpty()) {
for (DeleteFile deleteFile : deletes.referencedDeleteFiles()) {
deleteFileLocations.add(deleteFile.location());
}
}
}
ValidationException.check(
deleteFileLocations.isEmpty(),
"Found new conflicting delete files that can apply to records matching
%s: %s",
dataFilter,
deleteFileLocations);
}
```
This also avoids the minor issue in the exception message currently where
we'll include duplicate paths in the error message because we aggregate across
all indices. To include all the conflicting delete file paths in case of a
conflict exception we can't really short-circuit as soon as a single
DeleteFileIndex fails anyways,.
##########
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:
Or addedDeleteFilesIndexedBySnapshot? Bit wordy but I feel like
addedDeleteFileIndexes reads kind of weird (no indices were really added, it's
an in memory construct we're leveraging)
--
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]
