wombatu-kun commented on code in PR #18603:
URL: https://github.com/apache/iceberg/pull/18603#discussion_r4375571806
##########
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java:
##########
@@ -605,11 +605,71 @@ protected void validateNoNewDeleteFiles(
locations);
}
+ /**
+ * Validates that no delete files that may apply to the given data files,
among those matching a
+ * filter, have been added to the table since a starting snapshot.
+ *
+ * <p>Position deletes and deletion vectors that apply to a single data file
not in {@code
+ * dataFiles} are ignored. Equality deletes and position deletes that may
apply to more than one
+ * data file are always considered conflicting.
+ *
+ * @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 dataFiles locations of the data files to validate have no new row
deletes
+ * @param parent ending snapshot on the branch being validated
+ */
+ @SuppressWarnings("CollectionUndefinedEquality")
+ protected void validateNoNewDeleteFiles(
+ TableMetadata base,
+ Long startingSnapshotId,
+ Expression dataFilter,
+ CharSequenceSet dataFiles,
+ Snapshot parent) {
+ Set<String> locations =
+ referencedDeleteFileLocations(
+ addedDeleteFilesIndexedPerSnapshot(base, startingSnapshotId,
dataFilter, null, parent),
+ deleteFile -> {
+ CharSequence referencedDataFile =
ContentFileUtil.referencedDataFile(deleteFile);
+ return referencedDataFile == null ||
dataFiles.contains(referencedDataFile);
+ });
+ ValidationException.check(
+ locations.isEmpty(),
+ "Found new conflicting delete files that can apply to records matching
%s: %s",
+ dataFilter,
+ locations);
+ }
+
+ /**
+ * Returns the locations of the data files that the delete files added by
this operation apply to,
+ * or null if one of them may apply to more than one data file.
+ */
+ protected CharSequenceSet referencedDataFilesOfAddedDeletes() {
Review Comment:
referencedDataFilesOfAddedDeletes and the new validateNoNewDeleteFiles
overload are protected, so revapi tracks them on every public
MergingSnapshotProducer subclass. Can they be package-private, since
BaseRowDelta is their only caller and lives in the same package?
##########
spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java:
##########
@@ -285,7 +285,11 @@ public void commit(WriterCommitMessage[] messages,
WriteSummary summary) {
rowDelta.validateFromSnapshot(scan.snapshotId());
}
- if (command == UPDATE || command == MERGE) {
+ if (command == UPDATE) {
+ // data files that the UPDATE does not modify have no rows matching
its condition
+ rowDelta.validateDeletedFiles();
+ rowDelta.validateNoConflictingDeleteFilesForModifiedDataFiles();
Review Comment:
No Spark test covers this branch: reverting it to
validateNoConflictingDeleteFiles, or moving MERGE to the new method, leaves
every test green. Can TestMergeOnReadUpdate add a case where a DELETE of
another file in the same partition commits between the UPDATE's scan and commit
(a spied RowDelta commit, as in
TestMergeOnReadDelete.testCommitUnknownException, can inject it), asserting the
UPDATE succeeds?
##########
api/src/main/java/org/apache/iceberg/RowDelta.java:
##########
@@ -161,4 +161,31 @@ default RowDelta removeDeletes(DeleteFile deletes) {
* @return this for method chaining
*/
RowDelta validateNoConflictingDeleteFiles();
+
+ /**
+ * Enables validation that delete files added concurrently do not conflict
with this commit's
+ * operation, like {@link #validateNoConflictingDeleteFiles()} but limited
to the data files this
+ * commit modifies: those its position deletes and deletion vectors apply
to, and those it
+ * removes.
+ *
+ * <p>Concurrent position deletes and deletion vectors that apply to a
single other data file do
+ * not conflict. Concurrent equality deletes and position deletes that may
apply to more than one
+ * data file still conflict, and so does every concurrent delete file if
this commit adds such a
+ * delete file itself.
+ *
+ * <p>This method can be called instead of {@link
#validateNoConflictingDeleteFiles()} for UPDATE
Review Comment:
validateNoConflictingDeleteFiles still says it must be called for UPDATE
operations, which this paragraph now contradicts. Can that javadoc point UPDATE
callers to this method, and can this one state that a concurrent delete of a
row the UPDATE only reads through a subquery can go undetected?
##########
core/src/test/java/org/apache/iceberg/TestRowDelta.java:
##########
@@ -1324,6 +1324,163 @@ public void testConcurrentConflictingRowDelta() {
.hasMessageStartingWith("Found new conflicting delete files");
}
+ private RowDelta updateOfFileA(long startingSnapshotId, DeleteFile
fileADeletes) {
+ return table
+ .newRowDelta()
+ .toBranch(branch)
+ .addRows(FILE_B)
+ .addDeletes(fileADeletes)
+ .validateFromSnapshot(startingSnapshotId)
+ .conflictDetectionFilter(Expressions.alwaysTrue())
+ .validateDataFilesExist(ImmutableList.of(FILE_A.location()))
Review Comment:
updateOfFileA always passes FILE_A to validateDataFilesExist, so FILE_A
reaches the modified set even when referencedDataFilesOfAddedDeletes
contributes nothing, and none of these tests fail if the commit's own DV or
position-delete targets are left out. Can
concurrentDeleteForModifiedDataFileConflicts (or the helper) skip
validateDataFilesExist so it depends on those targets?
--
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]
