You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2021/11/17 23:26:47 UTC
[iceberg] branch master updated: Core: Support rewriting data files at a sequence number (#3480)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new d6cbca0 Core: Support rewriting data files at a sequence number (#3480)
d6cbca0 is described below
commit d6cbca0c2fb3d10083b304bcbdf16bf51e3b586b
Author: Jack Ye <yz...@amazon.com>
AuthorDate: Wed Nov 17 15:26:36 2021 -0800
Core: Support rewriting data files at a sequence number (#3480)
---
.../main/java/org/apache/iceberg/RewriteFiles.java | 11 +++
.../apache/iceberg/actions/RewriteDataFiles.java | 11 +++
.../java/org/apache/iceberg/BaseRewriteFiles.java | 7 ++
.../org/apache/iceberg/GenericManifestEntry.java | 6 +-
.../java/org/apache/iceberg/ManifestWriter.java | 13 ++++
.../apache/iceberg/MergingSnapshotProducer.java | 47 +++++++++++-
.../actions/RewriteDataFilesCommitManager.java | 17 ++++-
.../java/org/apache/iceberg/TableTestBase.java | 11 +++
.../org/apache/iceberg/TestManifestWriter.java | 21 ++++++
.../java/org/apache/iceberg/TestRewriteFiles.java | 67 ++++++++++++++++-
.../test/java/org/apache/iceberg/TestRowDelta.java | 85 ++++++++++++++++++++++
.../actions/BaseRewriteDataFilesSparkAction.java | 10 ++-
.../actions/TestNewRewriteDataFilesAction.java | 63 ++++++++++++++++
13 files changed, 358 insertions(+), 11 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/RewriteFiles.java b/api/src/main/java/org/apache/iceberg/RewriteFiles.java
index 62ba58e..1e13a53 100644
--- a/api/src/main/java/org/apache/iceberg/RewriteFiles.java
+++ b/api/src/main/java/org/apache/iceberg/RewriteFiles.java
@@ -52,6 +52,17 @@ public interface RewriteFiles extends SnapshotUpdate<RewriteFiles> {
}
/**
+ * Add a rewrite that replaces one set of data files with another set that contains the same data.
+ * The sequence number provided will be used for all the data files added.
+ *
+ * @param filesToDelete files that will be replaced (deleted), cannot be null or empty.
+ * @param filesToAdd files that will be added, cannot be null or empty.
+ * @param sequenceNumber sequence number to use for all data files added
+ * @return this for method chaining
+ */
+ RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> filesToAdd, long sequenceNumber);
+
+ /**
* Add a rewrite that replaces one set of files with another set that contains the same data.
*
* @param dataFilesToReplace data files that will be replaced (deleted).
diff --git a/api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java b/api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java
index ce587d1..aa5d49c 100644
--- a/api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java
+++ b/api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java
@@ -78,6 +78,17 @@ public interface RewriteDataFiles extends SnapshotUpdate<RewriteDataFiles, Rewri
String TARGET_FILE_SIZE_BYTES = "target-file-size-bytes";
/**
+ * If the compaction should use the sequence number of the snapshot at compaction start time for new data files,
+ * instead of using the sequence number of the newly produced snapshot.
+ * <p>
+ * This avoids commit conflicts with updates that add newer equality deletes at a higher sequence number.
+ * <p>
+ * Defaults to true.
+ */
+ String USE_STARTING_SEQUENCE_NUMBER = "use-starting-sequence-number";
+ boolean USE_STARTING_SEQUENCE_NUMBER_DEFAULT = true;
+
+ /**
* Choose BINPACK as a strategy for this rewrite operation
* @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 f94bcd1..3cb71fa 100644
--- a/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java
+++ b/core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java
@@ -21,6 +21,7 @@ package org.apache.iceberg;
import java.util.Set;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
class BaseRewriteFiles extends MergingSnapshotProducer<RewriteFiles> implements RewriteFiles {
@@ -67,6 +68,12 @@ class BaseRewriteFiles extends MergingSnapshotProducer<RewriteFiles> implements
}
@Override
+ public RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> filesToAdd, long sequenceNumber) {
+ setNewFilesSequenceNumber(sequenceNumber);
+ return rewriteFiles(filesToDelete, ImmutableSet.of(), filesToAdd, ImmutableSet.of());
+ }
+
+ @Override
public RewriteFiles rewriteFiles(Set<DataFile> dataFilesToReplace, Set<DeleteFile> deleteFilesToReplace,
Set<DataFile> dataFilesToAdd, Set<DeleteFile> deleteFilesToAdd) {
verifyInputAndOutputFiles(dataFilesToReplace, deleteFilesToReplace, dataFilesToAdd, deleteFilesToAdd);
diff --git a/core/src/main/java/org/apache/iceberg/GenericManifestEntry.java b/core/src/main/java/org/apache/iceberg/GenericManifestEntry.java
index dfc9475..0a44be2 100644
--- a/core/src/main/java/org/apache/iceberg/GenericManifestEntry.java
+++ b/core/src/main/java/org/apache/iceberg/GenericManifestEntry.java
@@ -62,9 +62,13 @@ class GenericManifestEntry<F extends ContentFile<F>>
}
ManifestEntry<F> wrapAppend(Long newSnapshotId, F newFile) {
+ return wrapAppend(newSnapshotId, null, newFile);
+ }
+
+ ManifestEntry<F> wrapAppend(Long newSnapshotId, Long newSequenceNumber, F newFile) {
this.status = Status.ADDED;
this.snapshotId = newSnapshotId;
- this.sequenceNumber = null;
+ this.sequenceNumber = newSequenceNumber;
this.file = newFile;
return this;
}
diff --git a/core/src/main/java/org/apache/iceberg/ManifestWriter.java b/core/src/main/java/org/apache/iceberg/ManifestWriter.java
index a03d18a..50b045f 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestWriter.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestWriter.java
@@ -103,6 +103,19 @@ public abstract class ManifestWriter<F extends ContentFile<F>> implements FileAp
addEntry(reused.wrapAppend(snapshotId, addedFile));
}
+ /**
+ * Add an added entry for a file with a specific sequence number.
+ * <p>
+ * The entry's snapshot ID will be this manifest's snapshot ID.
+ * The entry's sequence number will be the provided sequence number.
+ *
+ * @param addedFile a data file
+ * @param sequenceNumber sequence number for the data file
+ */
+ public void add(F addedFile, long sequenceNumber) {
+ addEntry(reused.wrapAppend(snapshotId, sequenceNumber, addedFile));
+ }
+
void add(ManifestEntry<F> entry) {
addEntry(reused.wrapAppend(snapshotId, entry.file()));
}
diff --git a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
index f325edf..2a5a65c 100644
--- a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
+++ b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
@@ -21,6 +21,7 @@ package org.apache.iceberg;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.util.Arrays;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
@@ -80,6 +81,7 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
// update data
private final List<DataFile> newFiles = Lists.newArrayList();
+ private Long newFilesSequenceNumber;
private final Map<Integer, List<DeleteFile>> newDeleteFilesBySpec = Maps.newHashMap();
private final List<ManifestFile> appendManifests = Lists.newArrayList();
private final List<ManifestFile> rewrittenAppendManifests = Lists.newArrayList();
@@ -297,7 +299,8 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
*/
protected void validateNoNewDeletesForDataFiles(TableMetadata base, Long startingSnapshotId,
Iterable<DataFile> dataFiles) {
- validateNoNewDeletesForDataFiles(base, startingSnapshotId, null, dataFiles, true);
+ validateNoNewDeletesForDataFiles(base, startingSnapshotId, null, dataFiles, true,
+ newFilesSequenceNumber != null);
}
/**
@@ -313,6 +316,28 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
protected void validateNoNewDeletesForDataFiles(TableMetadata base, Long startingSnapshotId,
Expression dataFilter, Iterable<DataFile> dataFiles,
boolean caseSensitive) {
+ validateNoNewDeletesForDataFiles(base, startingSnapshotId, dataFilter, dataFiles, caseSensitive, false);
+ }
+
+ /**
+ * Validates that no new delete files that must be applied to the given data files have been added to the table since
+ * a starting snapshot, with the option to ignore equality deletes during the validation.
+ * <p>
+ * For example, in the case of rewriting data files, if the added data files have the same sequence number as the
+ * replaced data files, equality deletes added at a higher sequence number are still effective against the added
+ * data files, so there is no risk of commit conflict between RewriteFiles and RowDelta. In cases like this,
+ * validation against equality delete files can be omitted.
+ *
+ * @param base table metadata to validate
+ * @param startingSnapshotId id of the snapshot current at the start of the operation
+ * @param dataFilter a data filter
+ * @param dataFiles data files to validate have no new row deletes
+ * @param caseSensitive whether expression binding should be case-sensitive
+ * @param ignoreEqualityDeletes whether equality deletes should be ignored in validation
+ */
+ private void validateNoNewDeletesForDataFiles(TableMetadata base, Long startingSnapshotId,
+ Expression dataFilter, Iterable<DataFile> dataFiles,
+ boolean caseSensitive, boolean ignoreEqualityDeletes) {
// if there is no current table state, no files have been added
if (base.currentSnapshot() == null || base.formatVersion() < 2) {
return;
@@ -327,8 +352,14 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
for (DataFile dataFile : dataFiles) {
// if any delete is found that applies to files written in or before the starting snapshot, fail
- if (deletes.forDataFile(startingSequenceNumber, dataFile).length > 0) {
- throw new ValidationException("Cannot commit, found new delete for replaced data file: %s", dataFile);
+ DeleteFile[] deleteFiles = deletes.forDataFile(startingSequenceNumber, dataFile);
+ if (ignoreEqualityDeletes) {
+ ValidationException.check(
+ Arrays.stream(deleteFiles).noneMatch(deleteFile -> deleteFile.content() == FileContent.POSITION_DELETES),
+ "Cannot commit, found new position delete for replaced data file: %s", dataFile);
+ } else {
+ ValidationException.check(deleteFiles.length == 0,
+ "Cannot commit, found new delete for replaced data file: %s", dataFile);
}
}
}
@@ -360,6 +391,10 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
dataFilter, Iterables.transform(deletes.referencedDeleteFiles(), ContentFile::path));
}
+ protected void setNewFilesSequenceNumber(long sequenceNumber) {
+ this.newFilesSequenceNumber = sequenceNumber;
+ }
+
private long startingSequenceNumber(TableMetadata metadata, Long staringSnapshotId) {
if (staringSnapshotId != null && metadata.snapshot(staringSnapshotId) != null) {
Snapshot startingSnapshot = metadata.snapshot(staringSnapshotId);
@@ -591,7 +626,11 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
try {
ManifestWriter<DataFile> writer = newManifestWriter(dataSpec());
try {
- writer.addAll(newFiles);
+ if (newFilesSequenceNumber == null) {
+ writer.addAll(newFiles);
+ } else {
+ newFiles.forEach(f -> writer.add(f, newFilesSequenceNumber));
+ }
} finally {
writer.close();
}
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 05bfddc..9a5cc4c 100644
--- a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java
+++ b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java
@@ -48,6 +48,7 @@ public class RewriteDataFilesCommitManager {
private final Table table;
private final long startingSnapshotId;
+ private final boolean useStartingSequenceNumber;
// constructor used for testing
public RewriteDataFilesCommitManager(Table table) {
@@ -55,8 +56,13 @@ public class RewriteDataFilesCommitManager {
}
public RewriteDataFilesCommitManager(Table table, long startingSnapshotId) {
+ this(table, startingSnapshotId, RewriteDataFiles.USE_STARTING_SEQUENCE_NUMBER_DEFAULT);
+ }
+
+ public RewriteDataFilesCommitManager(Table table, long startingSnapshotId, boolean useStartingSequenceNumber) {
this.table = table;
this.startingSnapshotId = startingSnapshotId;
+ this.useStartingSequenceNumber = useStartingSequenceNumber;
}
/**
@@ -72,9 +78,14 @@ public class RewriteDataFilesCommitManager {
addedDataFiles = Sets.union(addedDataFiles, group.addedFiles());
}
- RewriteFiles rewrite = table.newRewrite()
- .validateFromSnapshot(startingSnapshotId)
- .rewriteFiles(rewrittenDataFiles, addedDataFiles);
+ RewriteFiles rewrite = table.newRewrite().validateFromSnapshot(startingSnapshotId);
+ if (useStartingSequenceNumber) {
+ long sequenceNumber = table.snapshot(startingSnapshotId).sequenceNumber();
+ rewrite.rewriteFiles(rewrittenDataFiles, addedDataFiles, sequenceNumber);
+ } else {
+ rewrite.rewriteFiles(rewrittenDataFiles, addedDataFiles);
+ }
+
rewrite.commit();
}
diff --git a/core/src/test/java/org/apache/iceberg/TableTestBase.java b/core/src/test/java/org/apache/iceberg/TableTestBase.java
index 1a74232..1b8a3ca 100644
--- a/core/src/test/java/org/apache/iceberg/TableTestBase.java
+++ b/core/src/test/java/org/apache/iceberg/TableTestBase.java
@@ -469,6 +469,17 @@ public class TableTestBase {
.build();
}
+ protected DeleteFile newEqualityDeleteFile(int specId, String partitionPath, int... fieldIds) {
+ PartitionSpec spec = table.specs().get(specId);
+ return FileMetadata.deleteFileBuilder(spec)
+ .ofEqualityDeletes(fieldIds)
+ .withPath("/path/to/delete-" + UUID.randomUUID() + ".parquet")
+ .withFileSizeInBytes(10)
+ .withPartitionPath(partitionPath)
+ .withRecordCount(1)
+ .build();
+ }
+
protected <T> PositionDelete<T> positionDelete(CharSequence path, long pos, T row) {
PositionDelete<T> positionDelete = PositionDelete.create();
return positionDelete.set(path, pos, row);
diff --git a/core/src/test/java/org/apache/iceberg/TestManifestWriter.java b/core/src/test/java/org/apache/iceberg/TestManifestWriter.java
index 84db670..a686646 100644
--- a/core/src/test/java/org/apache/iceberg/TestManifestWriter.java
+++ b/core/src/test/java/org/apache/iceberg/TestManifestWriter.java
@@ -19,13 +19,16 @@
package org.apache.iceberg;
+import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.UUID;
import org.apache.iceberg.ManifestEntry.Status;
+import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -87,6 +90,24 @@ public class TestManifestWriter extends TableTestBase {
Conversions.fromByteBuffer(Types.IntegerType.get(), partitionFieldSummary.upperBound()));
}
+ @Test
+ public void testWriteManifestWithSequenceNumber() throws IOException {
+ Assume.assumeTrue("sequence number is only valid for format version > 1", formatVersion > 1);
+ File manifestFile = temp.newFile("manifest.avro");
+ Assert.assertTrue(manifestFile.delete());
+ OutputFile outputFile = table.ops().io().newOutputFile(manifestFile.getCanonicalPath());
+ ManifestWriter<DataFile> writer = ManifestFiles.write(formatVersion, table.spec(), outputFile, 1L);
+ writer.add(newFile(10, TestHelpers.Row.of(1)), 1000L);
+ writer.close();
+ ManifestFile manifest = writer.toManifestFile();
+ Assert.assertEquals("Manifest should have no sequence number", -1L, manifest.sequenceNumber());
+ ManifestReader<DataFile> manifestReader = ManifestFiles.read(manifest, table.io());
+ for (ManifestEntry<DataFile> entry : manifestReader.entries()) {
+ Assert.assertEquals("Custom sequence number should be used for all manifest entries",
+ 1000L, (long) entry.sequenceNumber());
+ }
+ }
+
private DataFile newFile(long recordCount) {
return newFile(recordCount, null);
}
diff --git a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java
index 19b97fa..b117a3e 100644
--- a/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java
+++ b/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java
@@ -264,11 +264,76 @@ public class TestRewriteFiles extends TableTestBase {
files(FILE_A_DELETES, FILE_B_DELETES),
statuses(DELETED, EXISTING));
- // We should only get the 3 manifests that this test is expected to add.
+ // We should only get the 5 manifests that this test is expected to add.
Assert.assertEquals("Only 5 manifests should exist", 5, listManifestFiles().size());
}
@Test
+ public void testRewriteDataAndAssignOldSequenceNumber() {
+ Assume.assumeTrue("Sequence number is only supported in iceberg format v2. ", formatVersion > 1);
+ Assert.assertEquals("Table should start empty", 0, listManifestFiles().size());
+
+ table.newRowDelta()
+ .addRows(FILE_A)
+ .addRows(FILE_B)
+ .addRows(FILE_C)
+ .addDeletes(FILE_A_DELETES)
+ .addDeletes(FILE_B_DELETES)
+ .commit();
+
+ TableMetadata base = readMetadata();
+ Snapshot baseSnap = base.currentSnapshot();
+ long baseSnapshotId = baseSnap.snapshotId();
+ Assert.assertEquals("Should create 2 manifests for initial write", 2, baseSnap.allManifests().size());
+ List<ManifestFile> initialManifests = baseSnap.allManifests();
+
+ validateManifestEntries(initialManifests.get(0),
+ ids(baseSnapshotId, baseSnapshotId, baseSnapshotId),
+ files(FILE_A, FILE_B, FILE_C),
+ statuses(ADDED, ADDED, ADDED));
+ validateDeleteManifest(initialManifests.get(1),
+ seqs(1, 1),
+ ids(baseSnapshotId, baseSnapshotId),
+ files(FILE_A_DELETES, FILE_B_DELETES),
+ statuses(ADDED, ADDED));
+
+ // Rewrite the files.
+ long oldSequenceNumber = table.currentSnapshot().sequenceNumber();
+ Snapshot pending = table.newRewrite()
+ .validateFromSnapshot(table.currentSnapshot().snapshotId())
+ .rewriteFiles(ImmutableSet.of(FILE_A), ImmutableSet.of(FILE_D), oldSequenceNumber)
+ .apply();
+
+ Assert.assertEquals("Should contain 3 manifest", 3, pending.allManifests().size());
+ Assert.assertFalse("Should not contain data manifest from initial write",
+ pending.dataManifests().stream().anyMatch(initialManifests::contains));
+
+ long pendingId = pending.snapshotId();
+ ManifestFile newManifest = pending.allManifests().get(0);
+ validateManifestEntries(newManifest, ids(pendingId), files(FILE_D), statuses(ADDED));
+ for (ManifestEntry<DataFile> entry : ManifestFiles.read(newManifest, FILE_IO).entries()) {
+ Assert.assertEquals("Should have old sequence number for manifest entries",
+ oldSequenceNumber, (long) entry.sequenceNumber());
+ }
+ Assert.assertEquals("Should use new sequence number for the manifest file",
+ oldSequenceNumber + 1, newManifest.sequenceNumber());
+
+ validateManifestEntries(pending.allManifests().get(1),
+ ids(pendingId, baseSnapshotId, baseSnapshotId),
+ files(FILE_A, FILE_B, FILE_C),
+ statuses(DELETED, EXISTING, EXISTING));
+
+ validateDeleteManifest(pending.allManifests().get(2),
+ seqs(1, 1),
+ ids(baseSnapshotId, baseSnapshotId),
+ files(FILE_A_DELETES, FILE_B_DELETES),
+ statuses(ADDED, ADDED));
+
+ // We should only get the 4 manifests that this test is expected to add.
+ Assert.assertEquals("Only 4 manifests should exist", 4, listManifestFiles().size());
+ }
+
+ @Test
public void testFailure() {
table.newAppend()
.appendFile(FILE_A)
diff --git a/core/src/test/java/org/apache/iceberg/TestRowDelta.java b/core/src/test/java/org/apache/iceberg/TestRowDelta.java
index 11b745e..348206b 100644
--- a/core/src/test/java/org/apache/iceberg/TestRowDelta.java
+++ b/core/src/test/java/org/apache/iceberg/TestRowDelta.java
@@ -26,6 +26,7 @@ import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.junit.Assert;
import org.junit.Test;
@@ -1134,4 +1135,88 @@ public class TestRowDelta extends V2TableTestBase {
validateTableDeleteFiles(table, deleteFile1, deleteFile2);
}
+
+ @Test
+ public void testConcurrentNonConflictingRowDeltaAndRewriteFilesWithSequenceNumber() {
+ // change the spec to be partitioned by data
+ table.updateSpec()
+ .removeField(Expressions.bucket("data", 16))
+ .addField(Expressions.ref("data"))
+ .commit();
+
+ // add a data file to partition A
+ DataFile dataFile1 = newDataFile("data=a");
+
+ table.newAppend()
+ .appendFile(dataFile1)
+ .commit();
+
+ Snapshot baseSnapshot = table.currentSnapshot();
+
+ // add an equality delete file
+ DeleteFile deleteFile1 = newEqualityDeleteFile(table.spec().specId(), "data=a",
+ table.schema().asStruct().fields().get(0).fieldId());
+
+ // mock a DELETE operation with serializable isolation
+ RowDelta rowDelta = table.newRowDelta()
+ .addDeletes(deleteFile1)
+ .validateFromSnapshot(baseSnapshot.snapshotId())
+ .validateNoConflictingDataFiles()
+ .validateNoConflictingDeleteFiles();
+
+ // mock a REWRITE operation with serializable isolation
+ DataFile dataFile2 = newDataFile("data=a");
+
+ RewriteFiles rewriteFiles = table.newRewrite()
+ .rewriteFiles(ImmutableSet.of(dataFile1), ImmutableSet.of(dataFile2), baseSnapshot.sequenceNumber())
+ .validateFromSnapshot(baseSnapshot.snapshotId());
+
+ rowDelta.commit();
+ rewriteFiles.commit();
+
+ validateTableDeleteFiles(table, deleteFile1);
+ validateTableFiles(table, dataFile2);
+ }
+
+ @Test
+ public void testConcurrentConflictingRowDeltaAndRewriteFilesWithSequenceNumber() {
+ // change the spec to be partitioned by data
+ table.updateSpec()
+ .removeField(Expressions.bucket("data", 16))
+ .addField(Expressions.ref("data"))
+ .commit();
+
+ // add a data file to partition A
+ DataFile dataFile1 = newDataFile("data=a");
+
+ table.newAppend()
+ .appendFile(dataFile1)
+ .commit();
+
+ Snapshot baseSnapshot = table.currentSnapshot();
+
+ // add an position delete file
+ DeleteFile deleteFile1 = newDeleteFile(table.spec().specId(), "data=a");
+
+ // mock a DELETE operation with serializable isolation
+ RowDelta rowDelta = table.newRowDelta()
+ .addDeletes(deleteFile1)
+ .validateFromSnapshot(baseSnapshot.snapshotId())
+ .validateNoConflictingDataFiles()
+ .validateNoConflictingDeleteFiles();
+
+ // mock a REWRITE operation with serializable isolation
+ DataFile dataFile2 = newDataFile("data=a");
+
+ RewriteFiles rewriteFiles = table.newRewrite()
+ .rewriteFiles(ImmutableSet.of(dataFile1), ImmutableSet.of(dataFile2), baseSnapshot.sequenceNumber())
+ .validateFromSnapshot(baseSnapshot.snapshotId());
+
+ rowDelta.commit();
+
+ AssertHelpers.assertThrows("Should not allow any new position delete associated with the data file",
+ ValidationException.class,
+ "Cannot commit, found new position delete for replaced data file",
+ rewriteFiles::commit);
+ }
}
diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java
index 406b2bb..c038474 100644
--- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java
+++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseRewriteDataFilesSparkAction.java
@@ -79,7 +79,8 @@ abstract class BaseRewriteDataFilesSparkAction
MAX_FILE_GROUP_SIZE_BYTES,
PARTIAL_PROGRESS_ENABLED,
PARTIAL_PROGRESS_MAX_COMMITS,
- TARGET_FILE_SIZE_BYTES
+ TARGET_FILE_SIZE_BYTES,
+ USE_STARTING_SEQUENCE_NUMBER
);
private final Table table;
@@ -88,6 +89,7 @@ abstract class BaseRewriteDataFilesSparkAction
private int maxConcurrentFileGroupRewrites;
private int maxCommits;
private boolean partialProgressEnabled;
+ private boolean useStartingSequenceNumber;
private RewriteStrategy strategy = null;
protected BaseRewriteDataFilesSparkAction(SparkSession spark, Table table) {
@@ -245,7 +247,7 @@ abstract class BaseRewriteDataFilesSparkAction
@VisibleForTesting
RewriteDataFilesCommitManager commitManager(long startingSnapshotId) {
- return new RewriteDataFilesCommitManager(table, startingSnapshotId);
+ return new RewriteDataFilesCommitManager(table, startingSnapshotId, useStartingSequenceNumber);
}
private Result doExecute(RewriteExecutionContext ctx, Stream<RewriteFileGroup> groupStream,
@@ -375,6 +377,10 @@ abstract class BaseRewriteDataFilesSparkAction
PARTIAL_PROGRESS_ENABLED,
PARTIAL_PROGRESS_ENABLED_DEFAULT);
+ useStartingSequenceNumber = PropertyUtil.propertyAsBoolean(options(),
+ USE_STARTING_SEQUENCE_NUMBER,
+ USE_STARTING_SEQUENCE_NUMBER_DEFAULT);
+
Preconditions.checkArgument(maxConcurrentFileGroupRewrites >= 1,
"Cannot set %s to %s, the value must be positive.",
MAX_CONCURRENT_FILE_GROUP_REWRITES, maxConcurrentFileGroupRewrites);
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java
index a8b7fac..92c4ba9 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestNewRewriteDataFilesAction.java
@@ -36,6 +36,7 @@ import org.apache.iceberg.ContentFile;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MetadataTableType;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RowDelta;
import org.apache.iceberg.Schema;
@@ -68,6 +69,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
import org.apache.iceberg.spark.FileRewriteCoordinator;
import org.apache.iceberg.spark.FileScanTaskSetManager;
+import org.apache.iceberg.spark.SparkTableUtil;
import org.apache.iceberg.spark.SparkTestBase;
import org.apache.iceberg.spark.source.ThreeColumnRecord;
import org.apache.iceberg.types.Comparators;
@@ -241,6 +243,67 @@ public class TestNewRewriteDataFilesAction extends SparkTestBase {
}
@Test
+ public void testBinPackWithStartingSequenceNumber() {
+ Table table = createTablePartitioned(4, 2);
+ shouldHaveFiles(table, 8);
+ List<Object[]> expectedRecords = currentData();
+ table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit();
+ table.refresh();
+ long oldSequenceNumber = table.currentSnapshot().sequenceNumber();
+
+ Result result = basicRewrite(table)
+ .option(RewriteDataFiles.USE_STARTING_SEQUENCE_NUMBER, "true")
+ .execute();
+ Assert.assertEquals("Action should rewrite 8 data files", 8, result.rewrittenDataFilesCount());
+ Assert.assertEquals("Action should add 4 data file", 4, result.addedDataFilesCount());
+
+ shouldHaveFiles(table, 4);
+ List<Object[]> actualRecords = currentData();
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+
+ table.refresh();
+ Assert.assertTrue("Table sequence number should be incremented",
+ oldSequenceNumber < table.currentSnapshot().sequenceNumber());
+
+ Dataset<Row> rows = SparkTableUtil.loadMetadataTable(spark, table, MetadataTableType.ENTRIES);
+ for (Row row : rows.collectAsList()) {
+ if (row.getInt(0) == 1) {
+ Assert.assertEquals("Expect old sequence number for added entries", oldSequenceNumber, row.getLong(2));
+ }
+ }
+ }
+
+ @Test
+ public void testBinPackWithStartingSequenceNumberV1Compatibility() {
+ Table table = createTablePartitioned(4, 2);
+ shouldHaveFiles(table, 8);
+ List<Object[]> expectedRecords = currentData();
+ table.refresh();
+ long oldSequenceNumber = table.currentSnapshot().sequenceNumber();
+ Assert.assertEquals("Table sequence number should be 0", 0, oldSequenceNumber);
+
+ Result result = basicRewrite(table)
+ .option(RewriteDataFiles.USE_STARTING_SEQUENCE_NUMBER, "true")
+ .execute();
+ Assert.assertEquals("Action should rewrite 8 data files", 8, result.rewrittenDataFilesCount());
+ Assert.assertEquals("Action should add 4 data file", 4, result.addedDataFilesCount());
+
+ shouldHaveFiles(table, 4);
+ List<Object[]> actualRecords = currentData();
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+
+ table.refresh();
+ Assert.assertEquals("Table sequence number should still be 0",
+ oldSequenceNumber, table.currentSnapshot().sequenceNumber());
+
+ Dataset<Row> rows = SparkTableUtil.loadMetadataTable(spark, table, MetadataTableType.ENTRIES);
+ for (Row row : rows.collectAsList()) {
+ Assert.assertEquals("Expect sequence number 0 for all entries",
+ oldSequenceNumber, row.getLong(2));
+ }
+ }
+
+ @Test
public void testRewriteLargeTableHasResiduals() {
PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).build();
Map<String, String> options = Maps.newHashMap();