Core: Add validation for row-level deletes with rewrites #2865
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | |||||||
|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -21,8 +21,12 @@ | ||||||||
|
|
|||||||||
| import java.util.Set; | |||||||||
| import org.apache.iceberg.relocated.com.google.common.base.Preconditions; | |||||||||
| import org.apache.iceberg.relocated.com.google.common.collect.Sets; | |||||||||
|
|
|||||||||
| class BaseRewriteFiles extends MergingSnapshotProducer<RewriteFiles> implements RewriteFiles { | |||||||||
| private final Set<DataFile> replacedDataFiles = Sets.newHashSet(); | |||||||||
| private Long startingSnapshotId = null; | |||||||||
|
|
|||||||||
| BaseRewriteFiles(String tableName, TableOperations ops) { | |||||||||
| super(tableName, ops); | |||||||||
|
|
|||||||||
|
|
@@ -63,15 +67,16 @@ private void verifyInputAndOutputFiles(Set<DataFile> dataFilesToDelete, Set<Dele | ||||||||
| } | |||||||||
|
|
|||||||||
| @Override | |||||||||
| public RewriteFiles rewriteFiles(Set<DataFile> dataFilesToReplace, Set<DeleteFile> deleteFilesToReplace, | |||||||||
| Set<DataFile> dataFilesToAdd, Set<DeleteFile> deleteFilesToAdd) { | |||||||||
| verifyInputAndOutputFiles(dataFilesToReplace, deleteFilesToReplace, dataFilesToAdd, deleteFilesToAdd); | |||||||||
|
Member
There was a problem hiding this comment. As we've already changed the
Contributor
Author
There was a problem hiding this comment. @openinx, I had a version where I did that and fixed the error messages, but it ended up being too many unrelated test changes so I'll do that in a follow-up. |
|||||||||
| replacedDataFiles.addAll(dataFilesToReplace); | |||||||||
|
|
|||||||||
| for (DataFile dataFile : dataFilesToReplace) { | |||||||||
| delete(dataFile); | |||||||||
| } | |||||||||
|
|
|||||||||
| for (DeleteFile deleteFile : deleteFilesToReplace) { | |||||||||
| delete(deleteFile); | |||||||||
| } | |||||||||
|
|
|||||||||
|
|
@@ -85,4 +90,18 @@ public RewriteFiles rewriteFiles(Set<DataFile> dataFilesToDelete, Set<DeleteFile | ||||||||
|
|
|||||||||
| return this; | |||||||||
| } | |||||||||
|
|
|||||||||
| @Override | |||||||||
| public RewriteFiles validateFromSnapshot(long snapshotId) { | |||||||||
| this.startingSnapshotId = snapshotId; | |||||||||
| return this; | |||||||||
| } | |||||||||
|
|
|||||||||
| @Override | |||||||||
| protected void validate(TableMetadata base) { | |||||||||
| if (replacedDataFiles.size() > 0) { | |||||||||
| // if there are replaced data files, there cannot be any new row-level deletes for those data files | |||||||||
| validateNoNewDeletesForDataFiles(base, startingSnapshotId, replacedDataFiles); | |||||||||
|
Member
There was a problem hiding this comment. I think we should also handle the case :
At timestamp Do we plan to add this validation in the following PR, or did not consider this case because we currently don't have introduced any RewriteFiles action that converting equality delete files into positional delete files ?
Member
There was a problem hiding this comment. In my thought, the data conflicts between REPLACE operation and APPEND/OVERWRITE/DELETE operations is because of the dependency relationship from positional delete files to data files. Then there will be two cases:
As a final solution, I think we should handle both the cases perfectly, so that we won't encounter any unexpected interruption when reading the iceberg table.
Contributor
Author
There was a problem hiding this comment. @openinx, I opened a draft PR, #2892, for to fix this, but I'm not sure that we actually want to merge it. Originally, I copied the Then I realized that not all of those cases apply. For example, if I delete a file that has row deltas, that's perfectly okay. The operation actually deletes all rows that weren't already deleted and any dangling positional deletes will just never get applied. Similarly, an overwrite from, for example, a MERGE INTO operation that replaces a file will need to first read that file and then build its replacement. The read must merge the existing deletes, so the rows will already be removed from the replacement file. Then the deletes in that case are also fine to leave dangling even if they are rewritten into a new file. I ended up leaving a validation that the data files are not rewritten, but I also think that may not be necessary because I don't think it would be reasonable to rewrite a data file without also applying the deletes. If we build a compaction operation we will probably apply deletes during compaction. We could also build one that compacts just data files, but in order for that operation to be correct it would also need to rewrite delete files (that are pointing to the compacted data files) or use an older sequence number with equality deletes. If it rewrote delete files, then the delete rewrite would fail. If it changed sequence numbers, then we would have a lot more validation to do when building that plan. After thinking about it, I don't think that we actually need to handle any cases where a compaction and delete file rewrite are concurrent, at least not from the delete file rewrite side. Any compaction should handle deletes as part of the compaction.
Member
There was a problem hiding this comment.
You mean people will need to know whether the data-files-only compaction will break the existing data set ? This puts too much burden on users ? I think we still need to provide the validation because we developers are sure that those cases will break the data correctness ...
Contributor
Author
There was a problem hiding this comment.
The contract of the
I think that is a reasonable requirement for the operation. Compacting only data files without accounting for deletes would change the table data, even without a concurrent delete file rewrite. For example, if I have
Compacting We have to trust the That means either the caller must do one of the following:
If the caller uses option 1, then it doesn't matter to concurrent operations that are rewriting the delete. It becomes a dangling delete that doesn't need to be applied. If the caller uses option 2, then there is a new file tracking the delete that would not be part of a concurrent rewrite operation; a concurrent rewrite may contain a dangling delete. The only case we would need to worry about is option 3, and only if a concurrent delete rewrites an equality delete as a positional delete. To be careful, we could merge #2892 to be able to validate this case. But I'm skeptical that altering the sequence number of the compacted file is a good idea. It may not even be possible if the sequence number of a delete that must be applied is between the sequence numbers of the starting data files. As I said, I'm okay not merging the validation for option 3 because I think it is unlikely that this is going to be a viable strategy. But I'm fine merging #2892 if you think it is needed. |
|||||||||
| } | |||||||||
| } | |||||||||
| } | |||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I see all the
snapshotIdare set to beops.current().snapshotId()except the test cases ? I'm thinking is there neccesary to introduce a method that validating since a older history snaphsot id.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I was wondering if this needs to be an open api? Seems like we always want to set it to table.currentSnapshot.snapshotID?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is necessary because the current snapshot's ID may not be the one that was used for the operation. This
RewriteFilesoperation may be created significantly later than the original planning that is done for compaction. For example, in the new Spark compaction code, the rewrite is created when a group is ready to commit. The table state may have changed from that time so using the current snapshot ID could easily skip delta commits that were interleaved, leaving the same issue that we are fixing here.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You can also see this in the test case, where we actually create commits and then run the rewrite differently to simulate concurrent operations: https://github.com/apache/iceberg/pull/2865/files#diff-b914d99ace31a09150792c10e3b0f1e40331edb1c24233ee6912d4608cfde0c4R615-R643
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
OK, that make sense !