diff --git a/api/src/main/java/org/apache/iceberg/RewriteFiles.java b/api/src/main/java/org/apache/iceberg/RewriteFiles.java index 62ba58e828d1..0d87f7319e7e 100644 --- a/api/src/main/java/org/apache/iceberg/RewriteFiles.java +++ b/api/src/main/java/org/apache/iceberg/RewriteFiles.java @@ -73,4 +73,12 @@ RewriteFiles rewriteFiles(Set dataFilesToReplace, Set dele * @return this for method chaining */ RewriteFiles validateFromSnapshot(long snapshotId); + + /** + * Set one older sequenceNumber for rewriteaction snapshot in manifest-list file. + * + * @param sequenceNumber a sequenceNumber + * @return this for method chaining + */ + 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..932490d5665a 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java @@ -23,9 +23,13 @@ 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; + private Long replaceSequenceNumber = null; BaseRewriteFiles(String tableName, TableOperations ops) { super(tableName, ops); @@ -44,6 +48,11 @@ protected String operation() { return DataOperations.REPLACE; } + @Override + protected Long sequenceNumberOverride() { + return replaceSequenceNumber; + } + private void verifyInputAndOutputFiles(Set dataFilesToDelete, Set deleteFilesToDelete, Set dataFilesToAdd, Set deleteFilesToAdd) { Preconditions.checkNotNull(dataFilesToDelete, "Data files to delete can not be null"); @@ -97,8 +106,17 @@ 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 (!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/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index 7825500bc7cb..3042b35c9c33 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 sequenceNumberOverride() { + return null; + } + /** * Validate the current metadata. *

@@ -167,7 +176,10 @@ 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) && 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/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; } 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..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,7 +202,9 @@ public RewriteDataFilesActionResult execute() { return RewriteDataFilesActionResult.empty(); } - long startingSnapshotId = table.currentSnapshot().snapshotId(); + Snapshot currentSnapshot = table.currentSnapshot(); + long startingSnapshotId = currentSnapshot.snapshotId(); + long sequenceNumber = currentSnapshot.sequenceNumber(); try { fileScanTasks = table.newScan() .useSnapshot(startingSnapshotId) @@ -247,7 +250,7 @@ public RewriteDataFilesActionResult execute() { 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 +272,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) { 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/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java index 19b97fad91e3..332a439eb609 100644 --- a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java +++ b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java @@ -616,6 +616,10 @@ public void testAlreadyDeletedFile() { public void testNewDeleteFile() { Assume.assumeTrue("Delete files are only supported in v2", formatVersion > 1); + table.updateProperties() + .set(TableProperties.COMPACT_NEW_DELETE_FILE_VALIDATE, "true") + .commit(); + table.newAppend() .appendFile(FILE_A) .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());