From 2e3a1ecf8cd5de29f3032a72f230ec191a9380ee Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Wed, 29 Sep 2021 16:01:22 +0800 Subject: [PATCH 01/10] Handle the case that RewriteFiles and RowDelta commit the transaction at the same time --- .../java/org/apache/iceberg/RewriteFiles.java | 7 +++++++ .../org/apache/iceberg/BaseRewriteFiles.java | 16 ++++++++++++---- .../org/apache/iceberg/SnapshotProducer.java | 11 ++++++++++- .../actions/BaseRewriteDataFilesAction.java | 6 ++++-- 4 files changed, 33 insertions(+), 7 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/RewriteFiles.java b/api/src/main/java/org/apache/iceberg/RewriteFiles.java index 62ba58e828d1..39579e0d72a7 100644 --- a/api/src/main/java/org/apache/iceberg/RewriteFiles.java +++ b/api/src/main/java/org/apache/iceberg/RewriteFiles.java @@ -73,4 +73,11 @@ RewriteFiles rewriteFiles(Set dataFilesToReplace, Set dele * @return this for method chaining */ RewriteFiles validateFromSnapshot(long snapshotId); + + /** + * Set the sequenceNumber write in manifest-list file. + * @param sequenceNumber + * @return + */ + RewriteFiles setSequenceNumber(long sequenceNumber); } diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java index f94bcd157d0c..fc9de8510100 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java @@ -26,6 +26,7 @@ class BaseRewriteFiles extends MergingSnapshotProducer implements RewriteFiles { private final Set replacedDataFiles = Sets.newHashSet(); private Long startingSnapshotId = null; + private Long sequenceNumber = null; BaseRewriteFiles(String tableName, TableOperations ops) { super(tableName, ops); @@ -44,6 +45,11 @@ protected String operation() { return DataOperations.REPLACE; } + @Override + protected Long sequenceNumber() { + return sequenceNumber; + } + private void verifyInputAndOutputFiles(Set dataFilesToDelete, Set deleteFilesToDelete, Set dataFilesToAdd, Set deleteFilesToAdd) { Preconditions.checkNotNull(dataFilesToDelete, "Data files to delete can not be null"); @@ -97,11 +103,13 @@ public RewriteFiles validateFromSnapshot(long snapshotId) { return this; } + @Override + public RewriteFiles setSequenceNumber(long sequenceNumber) { + this.sequenceNumber = 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); - } } } diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index 7825500bc7cb..6cf57eceb3fa 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -133,6 +133,15 @@ public ThisT deleteWith(Consumer deleteCallback) { */ protected abstract String operation(); + /** + * A Long that write sequenceNumber in manifest-list file. + * + * @return a string operation + */ + protected Long sequenceNumber() { + return null; + } + /** * Validate the current metadata. *

@@ -167,7 +176,7 @@ 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() : sequenceNumber)) { // keep track of the manifest lists created manifestLists.add(manifestList.location()); diff --git a/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java b/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java index 5c0cf7a6b1bf..225517cd6ddb 100644 --- a/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java +++ b/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java @@ -243,11 +243,12 @@ public RewriteDataFilesActionResult execute() { return RewriteDataFilesActionResult.empty(); } + long sequenceNumber = table.currentSnapshot().sequenceNumber(); List addedDataFiles = rewriteDataForTasks(combinedScanTasks); List 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); } @@ -269,10 +270,11 @@ private Map> groupTasksByPartition( } private void replaceDataFiles(Iterable deletedDataFiles, Iterable 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) { From 39b5d42a2395f6d30520a9ff1d1f79709bb5e665 Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Wed, 29 Sep 2021 16:14:50 +0800 Subject: [PATCH 02/10] fix test NPE --- core/src/main/java/org/apache/iceberg/SnapshotProducer.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index 6cf57eceb3fa..6dc5af8a72a3 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -176,7 +176,8 @@ public Snapshot apply() { OutputFile manifestList = manifestListPath(); try (ManifestListWriter writer = ManifestLists.write( - ops.current().formatVersion(), manifestList, snapshotId(), parentSnapshotId, operation().equals(DataOperations.REPLACE) ? sequenceNumber() : sequenceNumber)) { + ops.current().formatVersion(), manifestList, snapshotId(), parentSnapshotId, + operation().equals(DataOperations.REPLACE) && sequenceNumber() != null ? sequenceNumber() : sequenceNumber)) { // keep track of the manifest lists created manifestLists.add(manifestList.location()); From f68a5e552f6579eda667cad4b64a5c0a8c76f50b Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Wed, 29 Sep 2021 16:38:20 +0800 Subject: [PATCH 03/10] fix test and code style --- .../java/org/apache/iceberg/RewriteFiles.java | 5 ++-- .../org/apache/iceberg/TestRewriteFiles.java | 29 ------------------- 2 files changed, 3 insertions(+), 31 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/RewriteFiles.java b/api/src/main/java/org/apache/iceberg/RewriteFiles.java index 39579e0d72a7..21edb56c3430 100644 --- a/api/src/main/java/org/apache/iceberg/RewriteFiles.java +++ b/api/src/main/java/org/apache/iceberg/RewriteFiles.java @@ -76,8 +76,9 @@ RewriteFiles rewriteFiles(Set dataFilesToReplace, Set dele /** * Set the sequenceNumber write in manifest-list file. - * @param sequenceNumber - * @return + * + * @param sequenceNumber a sequenceNumber + * @return this for method chaining */ RewriteFiles setSequenceNumber(long sequenceNumber); } diff --git a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java index 19b97fad91e3..b07b614caa91 100644 --- a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java +++ b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java @@ -612,33 +612,4 @@ public void testAlreadyDeletedFile() { Assert.assertEquals("Only 3 manifests should exist", 3, listManifestFiles().size()); } - @Test - public void testNewDeleteFile() { - 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(); - } } From 191cccb8e41cdee798265aeb15fcd972b4e5c150 Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Wed, 29 Sep 2021 17:27:08 +0800 Subject: [PATCH 04/10] fix test and code style --- core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java index fc9de8510100..6ee7b628c3ca 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java @@ -26,7 +26,7 @@ class BaseRewriteFiles extends MergingSnapshotProducer implements RewriteFiles { private final Set replacedDataFiles = Sets.newHashSet(); private Long startingSnapshotId = null; - private Long sequenceNumber = null; + private Long replaceSequenceNumber = null; BaseRewriteFiles(String tableName, TableOperations ops) { super(tableName, ops); @@ -47,7 +47,7 @@ protected String operation() { @Override protected Long sequenceNumber() { - return sequenceNumber; + return replaceSequenceNumber; } private void verifyInputAndOutputFiles(Set dataFilesToDelete, Set deleteFilesToDelete, @@ -105,7 +105,7 @@ public RewriteFiles validateFromSnapshot(long snapshotId) { @Override public RewriteFiles setSequenceNumber(long sequenceNumber) { - this.sequenceNumber = sequenceNumber; + this.replaceSequenceNumber = sequenceNumber; return this; } From ad660e5f3f8b8d23ae04c0627c4431b91051f3af Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Mon, 11 Oct 2021 11:27:29 +0800 Subject: [PATCH 05/10] some fix --- .../java/org/apache/iceberg/RewriteFiles.java | 2 +- .../org/apache/iceberg/BaseRewriteFiles.java | 6 +++- .../org/apache/iceberg/SnapshotProducer.java | 4 +-- .../org/apache/iceberg/TestRewriteFiles.java | 29 +++++++++++++++++++ 4 files changed, 37 insertions(+), 4 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/RewriteFiles.java b/api/src/main/java/org/apache/iceberg/RewriteFiles.java index 21edb56c3430..0d87f7319e7e 100644 --- a/api/src/main/java/org/apache/iceberg/RewriteFiles.java +++ b/api/src/main/java/org/apache/iceberg/RewriteFiles.java @@ -75,7 +75,7 @@ RewriteFiles rewriteFiles(Set dataFilesToReplace, Set dele RewriteFiles validateFromSnapshot(long snapshotId); /** - * Set the sequenceNumber write in manifest-list file. + * Set one older sequenceNumber for rewriteaction snapshot in manifest-list file. * * @param sequenceNumber a sequenceNumber * @return this for method chaining diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java index 6ee7b628c3ca..d549eda527f8 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java @@ -46,7 +46,7 @@ protected String operation() { } @Override - protected Long sequenceNumber() { + protected Long sequenceNumberOverride() { return replaceSequenceNumber; } @@ -111,5 +111,9 @@ public RewriteFiles setSequenceNumber(long sequenceNumber) { @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); + } } } diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index 6dc5af8a72a3..b86f1cb345d0 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -138,7 +138,7 @@ public ThisT deleteWith(Consumer deleteCallback) { * * @return a string operation */ - protected Long sequenceNumber() { + protected Long sequenceNumberOverride() { return null; } @@ -177,7 +177,7 @@ public Snapshot apply() { try (ManifestListWriter writer = ManifestLists.write( ops.current().formatVersion(), manifestList, snapshotId(), parentSnapshotId, - operation().equals(DataOperations.REPLACE) && sequenceNumber() != null ? sequenceNumber() : sequenceNumber)) { + operation().equals(DataOperations.REPLACE) && sequenceNumberOverride() != null ? sequenceNumberOverride() : sequenceNumber)) { // keep track of the manifest lists created manifestLists.add(manifestList.location()); diff --git a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java index b07b614caa91..a14527cb2e7f 100644 --- a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java +++ b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java @@ -612,4 +612,33 @@ public void testAlreadyDeletedFile() { Assert.assertEquals("Only 3 manifests should exist", 3, listManifestFiles().size()); } + @Test + public void testNewDeleteFile() { + 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(); + } } From d1d3da434c1a15dab24fe0005b0d9ebb8b177cd5 Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Mon, 11 Oct 2021 11:45:15 +0800 Subject: [PATCH 06/10] some fix --- core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java | 6 ++++++ core/src/main/java/org/apache/iceberg/TableProperties.java | 3 +++ 2 files changed, 9 insertions(+) diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java index d549eda527f8..932490d5665a 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java @@ -23,6 +23,9 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import static org.apache.iceberg.TableProperties.COMPACT_NEW_DELETE_FILE_VALIDATE; +import static org.apache.iceberg.TableProperties.COMPACT_NEW_DELETE_FILE_VALIDATE_DEFAULT; + class BaseRewriteFiles extends MergingSnapshotProducer implements RewriteFiles { private final Set replacedDataFiles = Sets.newHashSet(); private Long startingSnapshotId = null; @@ -111,6 +114,9 @@ public RewriteFiles setSequenceNumber(long sequenceNumber) { @Override protected void validate(TableMetadata base) { + if (!base.propertyAsBoolean(COMPACT_NEW_DELETE_FILE_VALIDATE, COMPACT_NEW_DELETE_FILE_VALIDATE_DEFAULT)) { + return; + } 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); diff --git a/core/src/main/java/org/apache/iceberg/TableProperties.java b/core/src/main/java/org/apache/iceberg/TableProperties.java index da28543ad18e..6374f2a0ed53 100644 --- a/core/src/main/java/org/apache/iceberg/TableProperties.java +++ b/core/src/main/java/org/apache/iceberg/TableProperties.java @@ -244,4 +244,7 @@ private TableProperties() { public static final String UPSERT_MODE_ENABLE = "write.upsert.enable"; public static final boolean UPSERT_MODE_ENABLE_DEFAULT = false; + + public static final String COMPACT_NEW_DELETE_FILE_VALIDATE = "compact.new.delete.file.validate"; + public static final boolean COMPACT_NEW_DELETE_FILE_VALIDATE_DEFAULT = false; } From 28aae83e0f8704d909989af69a8bd90d3b6135c2 Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Mon, 11 Oct 2021 13:48:50 +0800 Subject: [PATCH 07/10] some fix --- core/src/main/java/org/apache/iceberg/SnapshotProducer.java | 4 +++- .../apache/iceberg/actions/BaseRewriteDataFilesAction.java | 2 +- core/src/test/java/org/apache/iceberg/TestRewriteFiles.java | 2 ++ 3 files changed, 6 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index b86f1cb345d0..3042b35c9c33 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -177,7 +177,9 @@ public Snapshot apply() { try (ManifestListWriter writer = ManifestLists.write( ops.current().formatVersion(), manifestList, snapshotId(), parentSnapshotId, - operation().equals(DataOperations.REPLACE) && sequenceNumberOverride() != null ? sequenceNumberOverride() : sequenceNumber)) { + operation().equals(DataOperations.REPLACE) && sequenceNumberOverride() != null ? + sequenceNumberOverride() : + sequenceNumber)) { // keep track of the manifest lists created manifestLists.add(manifestList.location()); diff --git a/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java b/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java index 225517cd6ddb..cd6a08a6dda2 100644 --- a/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java +++ b/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java @@ -202,6 +202,7 @@ public RewriteDataFilesActionResult execute() { } long startingSnapshotId = table.currentSnapshot().snapshotId(); + long sequenceNumber = table.currentSnapshot().sequenceNumber(); try { fileScanTasks = table.newScan() .useSnapshot(startingSnapshotId) @@ -243,7 +244,6 @@ public RewriteDataFilesActionResult execute() { return RewriteDataFilesActionResult.empty(); } - long sequenceNumber = table.currentSnapshot().sequenceNumber(); List addedDataFiles = rewriteDataForTasks(combinedScanTasks); List currentDataFiles = combinedScanTasks.stream() .flatMap(tasks -> tasks.files().stream().map(FileScanTask::file)) diff --git a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java index a14527cb2e7f..30e547730785 100644 --- a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java +++ b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java @@ -616,6 +616,8 @@ public void testAlreadyDeletedFile() { public void testNewDeleteFile() { Assume.assumeTrue("Delete files are only supported in v2", formatVersion > 1); + table.properties().put("compact.new.delete.file.validate", "true"); + table.newAppend() .appendFile(FILE_A) .commit(); From f9d5bbb9a9cc013bce21df7d042a6cd5c996c8d6 Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Mon, 11 Oct 2021 14:42:10 +0800 Subject: [PATCH 08/10] fix test --- .../org/apache/iceberg/TestRewriteFiles.java | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java index 30e547730785..332a439eb609 100644 --- a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java +++ b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java @@ -614,33 +614,35 @@ public void testAlreadyDeletedFile() { @Test public void testNewDeleteFile() { - Assume.assumeTrue("Delete files are only supported in v2", formatVersion > 1); + Assume.assumeTrue("Delete files are only supported in v2", formatVersion > 1); - table.properties().put("compact.new.delete.file.validate", "true"); + table.updateProperties() + .set(TableProperties.COMPACT_NEW_DELETE_FILE_VALIDATE, "true") + .commit(); table.newAppend() .appendFile(FILE_A) .commit(); - long snapshotBeforeDeletes = table.currentSnapshot().snapshotId(); + long snapshotBeforeDeletes = table.currentSnapshot().snapshotId(); table.newRowDelta() .addDeletes(FILE_A_DELETES) .commit(); - long snapshotAfterDeletes = table.currentSnapshot().snapshotId(); + 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) + 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()); + .apply()); - // the rewrite should be valid when validating from the snapshot after the deletes + // 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(); + .apply(); } } From 79d6712dc8c21129b02e9d0cbc655cc8da896fad Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Wed, 20 Oct 2021 10:14:50 +0800 Subject: [PATCH 09/10] fix new rewriteaction --- .../iceberg/actions/RewriteDataFilesCommitManager.java | 7 +++++-- .../spark/actions/BaseRewriteDataFilesSparkAction.java | 7 ++++--- .../spark/actions/TestNewRewriteDataFilesAction.java | 6 +++--- 3 files changed, 12 insertions(+), 8 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java index 05bfddc64c5e..7cd596ba2e5d 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java @@ -48,15 +48,17 @@ public class RewriteDataFilesCommitManager { private final Table table; private final long startingSnapshotId; + private final long replaceSeqNumber; // constructor used for testing public RewriteDataFilesCommitManager(Table table) { - this(table, table.currentSnapshot().snapshotId()); + this(table, table.currentSnapshot().snapshotId(), table.currentSnapshot().sequenceNumber()); } - public RewriteDataFilesCommitManager(Table table, long startingSnapshotId) { + public RewriteDataFilesCommitManager(Table table, long startingSnapshotId, long replaceSeqNumber) { this.table = table; this.startingSnapshotId = startingSnapshotId; + this.replaceSeqNumber = replaceSeqNumber; } /** @@ -74,6 +76,7 @@ public void commitFileGroups(Set fileGroups) { RewriteFiles rewrite = table.newRewrite() .validateFromSnapshot(startingSnapshotId) + .setSequenceNumber(replaceSeqNumber) .rewriteFiles(rewrittenDataFiles, addedDataFiles); rewrite.commit(); } diff --git a/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java b/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java index 7a6a7a9b1c28..b548ce236280 100644 --- a/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java +++ b/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java @@ -119,6 +119,7 @@ public RewriteDataFiles.Result execute() { } long startingSnapshotId = table.currentSnapshot().snapshotId(); + long replaceSequenceNumber = table.currentSnapshot().sequenceNumber(); validateAndInitOptions(); strategy = strategy.options(options()); @@ -133,7 +134,7 @@ public RewriteDataFiles.Result execute() { Stream groupStream = toGroupStream(ctx, fileGroupsByPartition); - RewriteDataFilesCommitManager commitManager = commitManager(startingSnapshotId); + RewriteDataFilesCommitManager commitManager = commitManager(startingSnapshotId, replaceSequenceNumber); if (partialProgressEnabled) { return doExecuteWithPartialProgress(ctx, groupStream, commitManager); } else { @@ -195,8 +196,8 @@ private ExecutorService rewriteService() { } @VisibleForTesting - RewriteDataFilesCommitManager commitManager(long startingSnapshotId) { - return new RewriteDataFilesCommitManager(table, startingSnapshotId); + RewriteDataFilesCommitManager commitManager(long startingSnapshotId, long replaceSequenceNumber) { + return new RewriteDataFilesCommitManager(table, startingSnapshotId, replaceSequenceNumber); } private Result doExecute(RewriteExecutionContext ctx, Stream groupStream, diff --git a/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java b/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java index 85000659c682..e46db149fb90 100644 --- a/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java +++ b/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java @@ -395,7 +395,7 @@ public void testSingleCommitWithCommitFailure() { doReturn(util) .when(spyRewrite) - .commitManager(table.currentSnapshot().snapshotId()); + .commitManager(table.currentSnapshot().snapshotId(), table.currentSnapshot().sequenceNumber()); AssertHelpers.assertThrows("Should fail entire rewrite if commit fails", RuntimeException.class, () -> spyRewrite.execute()); @@ -548,7 +548,7 @@ public void testParallelPartialProgressWithCommitFailure() { doReturn(util) .when(spyRewrite) - .commitManager(table.currentSnapshot().snapshotId()); + .commitManager(table.currentSnapshot().snapshotId(), table.currentSnapshot().sequenceNumber()); RewriteDataFiles.Result result = spyRewrite.execute(); @@ -608,7 +608,7 @@ public void testCommitStateUnknownException() { doReturn(util) .when(spyAction) - .commitManager(table.currentSnapshot().snapshotId()); + .commitManager(table.currentSnapshot().snapshotId(), table.currentSnapshot().sequenceNumber()); AssertHelpers.assertThrows("Should propagate CommitStateUnknown Exception", CommitStateUnknownException.class, () -> spyAction.execute()); From 3502646fdc7b790b76b8228365fa83d87aa696a9 Mon Sep 17 00:00:00 2001 From: hameizi <1249369293@qq.com> Date: Thu, 28 Oct 2021 14:00:41 +0800 Subject: [PATCH 10/10] guarantee rewrite snapshot's snapshotId and seqnumber from the same snapshot --- .../apache/iceberg/actions/BaseRewriteDataFilesAction.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java b/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java index cd6a08a6dda2..51bbd4d748fe 100644 --- a/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java +++ b/core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java @@ -29,6 +29,7 @@ import org.apache.iceberg.FileScanTask; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.RewriteFiles; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.encryption.EncryptionManager; @@ -201,8 +202,9 @@ public RewriteDataFilesActionResult execute() { return RewriteDataFilesActionResult.empty(); } - long startingSnapshotId = table.currentSnapshot().snapshotId(); - long sequenceNumber = table.currentSnapshot().sequenceNumber(); + Snapshot currentSnapshot = table.currentSnapshot(); + long startingSnapshotId = currentSnapshot.snapshotId(); + long sequenceNumber = currentSnapshot.sequenceNumber(); try { fileScanTasks = table.newScan() .useSnapshot(startingSnapshotId)