Skip to content

Handle the case that RewriteFiles and RowDelta commit the transaction… #3204

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
hameizi wants to merge 10 commits into from
Closed

Handle the case that RewriteFiles and RowDelta commit the transaction… #3204

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
2e3a1ec
Handle the case that RewriteFiles and RowDelta commit the transaction…
hameizi Sep 29, 2021
39b5d42
fix test NPE
hameizi Sep 29, 2021
f68a5e5
fix test and code style
hameizi Sep 29, 2021
191cccb
fix test and code style
hameizi Sep 29, 2021
ad660e5
some fix
hameizi Oct 11, 2021
d1d3da4
some fix
hameizi Oct 11, 2021
28aae83
some fix
hameizi Oct 11, 2021
f9d5bbb
fix test
hameizi Oct 11, 2021
79d6712
fix new rewriteaction
hameizi Oct 20, 2021
3502646
guarantee rewrite snapshot's snapshotId and seqnumber from the same s…
hameizi Oct 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions api/src/main/java/org/apache/iceberg/RewriteFiles.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,4 +73,12 @@ RewriteFiles rewriteFiles(Set<DataFile> dataFilesToReplace, Set<DeleteFile> dele
* @return this for method chaining
*/
RewriteFiles validateFromSnapshot(long snapshotId);

/**
* Set the sequenceNumber write in manifest-list file.
*
* @param sequenceNumber a sequenceNumber
* @return this for method chaining
*/
RewriteFiles setSequenceNumber(long sequenceNumber);

@rdblue rdblue Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we need better documentation for what this method does and when you would call it, since this affects correctness.

}
16 changes: 12 additions & 4 deletions core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
class BaseRewriteFiles extends MergingSnapshotProducer<RewriteFiles> implements RewriteFiles {
private final Set<DataFile> replacedDataFiles = Sets.newHashSet();
private Long startingSnapshotId = null;
private Long replaceSequenceNumber = null;

BaseRewriteFiles(String tableName, TableOperations ops) {
super(tableName, ops);
Expand All @@ -44,6 +45,11 @@ protected String operation() {
return DataOperations.REPLACE;
}

@Override
protected Long sequenceNumber() {
return replaceSequenceNumber;

@rdblue rdblue Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With this approach, I think we need a validation that none of the data or delete files that are being replaced have sequence numbers newer than the override sequence number.

@hameizi hameizi Oct 11, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The override sequence number is init when compact action generate fileScanTasks in https://github.com/apache/iceberg/pull/3204/files#:~:text=long%20sequenceNumber%20%3D%20table.currentSnapshot().sequenceNumber()%3B, so it can guarantee that none of the data or delete files that are being replaced have sequence numbers newer than the override sequence number.

}

private void verifyInputAndOutputFiles(Set<DataFile> dataFilesToDelete, Set<DeleteFile> deleteFilesToDelete,
Set<DataFile> dataFilesToAdd, Set<DeleteFile> deleteFilesToAdd) {
Preconditions.checkNotNull(dataFilesToDelete, "Data files to delete can not be null");
Expand Down Expand Up @@ -97,11 +103,13 @@ public RewriteFiles validateFromSnapshot(long snapshotId) {
return this;
}

@Override
public RewriteFiles setSequenceNumber(long sequenceNumber) {
this.replaceSequenceNumber = sequenceNumber;
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);
}

@rdblue rdblue Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This should not be removed. There is a valid use case for the operation to check whether there are conflicts instead of re-sequencing data files. If you want, you can add a configuration method to enable/disable the validation.

}
}
12 changes: 11 additions & 1 deletion core/src/main/java/org/apache/iceberg/SnapshotProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,15 @@ public ThisT deleteWith(Consumer<String> deleteCallback) {
*/
protected abstract String operation();

/**
* A Long that write sequenceNumber in manifest-list file.
*
* @return a string operation
*/
protected Long sequenceNumber() {

@rdblue rdblue Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this needs a more specific name, like sequenceNumberOverride

return null;
}

/**
* Validate the current metadata.
* <p>
Expand Down Expand Up @@ -167,7 +176,8 @@ public Snapshot apply() {
OutputFile manifestList = manifestListPath();

try (ManifestListWriter writer = ManifestLists.write(
ops.current().formatVersion(), manifestList, snapshotId(), parentSnapshotId, sequenceNumber)) {
ops.current().formatVersion(), manifestList, snapshotId(), parentSnapshotId,
operation().equals(DataOperations.REPLACE) && sequenceNumber() != null ? sequenceNumber() : sequenceNumber)) {

@rdblue rdblue Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is a clever way to fix the problem. What we had considered before is setting the sequence number of individual files rather than the sequence number used for the manifest list. This is an easier way to set the sequence number for all new data files in the snapshot.

I need to think about whether this is the right approach a bit more. As long as we have a static sequence number, using inheritance is only a convenience. We could set that static sequence number on the individual data or delete files that we add in the commit. I tend to lean toward that solution because it minimizes the places that use the overridden sequence number -- so we know that the manifest list and manifests all have the latest sequence number that is assigned to the snapshot rather than also being re-sequenced.

I'll think about the trade-off some more.


// keep track of the manifest lists created
manifestLists.add(manifestList.location());
Expand Down
6 changes: 4 additions & 2 deletions core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,11 +243,12 @@ public RewriteDataFilesActionResult execute() {
return RewriteDataFilesActionResult.empty();
}

long sequenceNumber = table.currentSnapshot().sequenceNumber();
List<DataFile> addedDataFiles = rewriteDataForTasks(combinedScanTasks);
List<DataFile> currentDataFiles = combinedScanTasks.stream()
.flatMap(tasks -> tasks.files().stream().map(FileScanTask::file))
.collect(Collectors.toList());
replaceDataFiles(currentDataFiles, addedDataFiles, startingSnapshotId);
replaceDataFiles(currentDataFiles, addedDataFiles, startingSnapshotId, sequenceNumber);

return new RewriteDataFilesActionResult(currentDataFiles, addedDataFiles);
}
Expand All @@ -269,10 +270,11 @@ private Map<StructLikeWrapper, Collection<FileScanTask>> groupTasksByPartition(
}

private void replaceDataFiles(Iterable<DataFile> deletedDataFiles, Iterable<DataFile> addedDataFiles,
long startingSnapshotId) {
long startingSnapshotId, long sequenceNumber) {
try {
RewriteFiles rewriteFiles = table.newRewrite()
.validateFromSnapshot(startingSnapshotId)
.setSequenceNumber(sequenceNumber)
.rewriteFiles(Sets.newHashSet(deletedDataFiles), Sets.newHashSet(addedDataFiles));
commit(rewriteFiles);
} catch (Exception e) {
Expand Down
29 changes: 0 additions & 29 deletions core/src/test/java/org/apache/iceberg/TestRewriteFiles.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -612,33 +612,4 @@ public void testAlreadyDeletedFile() {
Assert.assertEquals("Only 3 manifests should exist", 3, listManifestFiles().size());
}

@Test
public void testNewDeleteFile() {

@rdblue rdblue Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are you removing tests? This is probably incorrect.

@rdblue rdblue Sep 30, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see that this is related to removing the validation in BaseRewriteFiles. Like I noted there, we want to keep that validation and use it in certain cases.

Assume.assumeTrue("Delete files are only supported in v2", formatVersion > 1);

table.newAppend()
.appendFile(FILE_A)
.commit();

long snapshotBeforeDeletes = table.currentSnapshot().snapshotId();

table.newRowDelta()
.addDeletes(FILE_A_DELETES)
.commit();

long snapshotAfterDeletes = table.currentSnapshot().snapshotId();

AssertHelpers.assertThrows("Should fail because deletes were added after the starting snapshot",
ValidationException.class, "Cannot commit, found new delete for replaced data file",
() -> table.newRewrite()
.validateFromSnapshot(snapshotBeforeDeletes)
.rewriteFiles(Sets.newSet(FILE_A), Sets.newSet(FILE_A2))
.apply());

// the rewrite should be valid when validating from the snapshot after the deletes
table.newRewrite()
.validateFromSnapshot(snapshotAfterDeletes)
.rewriteFiles(Sets.newSet(FILE_A), Sets.newSet(FILE_A2))
.apply();
}
}