From 836f5f7b7bb43385a6423e4a98bedaed412f1b63 Mon Sep 17 00:00:00 2001 From: Timothy Wang Date: Fri, 31 Jul 2026 08:18:59 +0000 Subject: [PATCH 1/3] Prototype explicit row ID assignment in manifest rewrites --- .../org/apache/iceberg/RewriteManifests.java | 23 ++ .../apache/iceberg/BaseRewriteManifests.java | 188 +++++++++++++- .../org/apache/iceberg/SnapshotProducer.java | 12 +- .../iceberg/TestRowLineageAssignment.java | 244 ++++++++++++++++++ .../source/TestStructuredStreamingRead3.java | 72 ++++++ 5 files changed, 536 insertions(+), 3 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/RewriteManifests.java b/api/src/main/java/org/apache/iceberg/RewriteManifests.java index 32a9011ad6f6..bc74f377010a 100644 --- a/api/src/main/java/org/apache/iceberg/RewriteManifests.java +++ b/api/src/main/java/org/apache/iceberg/RewriteManifests.java @@ -48,6 +48,29 @@ public interface RewriteManifests extends SnapshotUpdate { */ RewriteManifests clusterBy(Function func); + /** + * Rewrites live data-file entries with explicit first row IDs. + * + *

This operation is intended for assigning row IDs to files that existed before a table was + * upgraded to format version 3. It preserves each entry's original snapshot ID, data sequence + * number, and file sequence number. The active file set is not changed. + * + *

The reserved range starts at the table's current {@code next-row-id} and ends at {@code + * nextRowIdExclusive}. The range may contain gaps, but every assigned file range must be fully + * contained within it. + * + * @param firstRowIdForFile function that returns the first row ID for each live data file + * @param expectedFirstRowId expected table {@code next-row-id} before this operation + * @param nextRowIdExclusive first row ID available after this operation + * @return this for method chaining + */ + default RewriteManifests assignFirstRowIds( + Function firstRowIdForFile, + long expectedFirstRowId, + long nextRowIdExclusive) { + throw new UnsupportedOperationException("Explicit first row ID assignment is not supported"); + } + /** * Determines which existing {@link ManifestFile} for the table should be rewritten. Manifests * that do not match the predicate are kept as-is. If this is not called and no predicate is set, diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java index e98027ec4a0a..054071b9f2c8 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java @@ -24,8 +24,10 @@ import java.io.IOException; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicLong; @@ -48,6 +50,8 @@ public class BaseRewriteManifests extends SnapshotProducer implements RewriteManifests { + private static final Object ROW_ID_ASSIGNMENT_CLUSTER_KEY = new Object(); + private final String tableName; private final Map specsById; private final long manifestTargetSizeBytes; @@ -62,9 +66,14 @@ public class BaseRewriteManifests extends SnapshotProducer private final Map writers = Maps.newConcurrentMap(); private final AtomicLong entryCount = new AtomicLong(0); + private final Collection assignedRowIdRanges = new ConcurrentLinkedQueue<>(); private Function clusterByFunc; + private Function firstRowIdForFile; private Predicate predicate; + private Long expectedFirstRowId; + private Long nextRowIdExclusive; + private Long expectedSnapshotId; private final SnapshotSummary.Builder summaryBuilder = SnapshotSummary.builder(); @@ -117,20 +126,70 @@ public RewriteManifests clusterBy(Function func) { return this; } + @Override + public RewriteManifests assignFirstRowIds( + Function firstRowIdForFile, + long expectedFirstRowId, + long nextRowIdExclusive) { + Preconditions.checkArgument(firstRowIdForFile != null, "First row ID function cannot be null"); + Preconditions.checkArgument( + ops().current().formatVersion() >= 3, + "Cannot assign first row IDs to a table with format version %s", + ops().current().formatVersion()); + Preconditions.checkArgument( + expectedFirstRowId >= 0, + "Expected first row ID must be non-negative: %s", + expectedFirstRowId); + Preconditions.checkArgument( + nextRowIdExclusive >= expectedFirstRowId, + "Next row ID %s must not be less than expected first row ID %s", + nextRowIdExclusive, + expectedFirstRowId); + Preconditions.checkState( + !hasRowIdAssignment(), "First row ID assignment is already configured"); + Preconditions.checkState( + predicate == null + && deletedManifests.isEmpty() + && addedManifests.isEmpty() + && rewrittenAddedManifests.isEmpty(), + "First row ID assignment cannot be combined with manifest selection or replacement"); + + this.firstRowIdForFile = firstRowIdForFile; + this.expectedFirstRowId = expectedFirstRowId; + this.nextRowIdExclusive = nextRowIdExclusive; + Snapshot currentSnapshot = ops().current().currentSnapshot(); + this.expectedSnapshotId = currentSnapshot != null ? currentSnapshot.snapshotId() : null; + + if (clusterByFunc == null) { + this.clusterByFunc = ignored -> ROW_ID_ASSIGNMENT_CLUSTER_KEY; + } + + return this; + } + @Override public RewriteManifests rewriteIf(Predicate pred) { + Preconditions.checkState( + !hasRowIdAssignment(), + "Manifest selection cannot be combined with first row ID assignment"); this.predicate = pred; return this; } @Override public RewriteManifests deleteManifest(ManifestFile manifest) { + Preconditions.checkState( + !hasRowIdAssignment(), + "Manifest replacement cannot be combined with first row ID assignment"); deletedManifests.add(manifest); return this; } @Override public RewriteManifests addManifest(ManifestFile manifest) { + Preconditions.checkState( + !hasRowIdAssignment(), + "Manifest replacement cannot be combined with first row ID assignment"); Preconditions.checkArgument(!manifest.hasAddedFiles(), "Cannot add manifest with added files"); Preconditions.checkArgument( !manifest.hasDeletedFiles(), "Cannot add manifest with deleted files"); @@ -180,6 +239,7 @@ public List apply(TableMetadata base, Snapshot snapshot) { } validateFilesCounts(); + validateAssignedRowIds(); Iterable newManifestsWithMetadata = Iterables.transform( @@ -231,6 +291,7 @@ private void keepActiveManifests(List currentManifests) { private void reset() { deleteUncommitted(newManifests, ImmutableSet.of(), true /* clear new manifests */); entryCount.set(0); + assignedRowIdRanges.clear(); keptManifests.clear(); rewrittenManifests.clear(); writers.clear(); @@ -333,10 +394,104 @@ private void appendEntry(ManifestEntry entry, Object key, int partitio Preconditions.checkNotNull(key, "Key cannot be null"); WriterWrapper writer = getWriter(key, partitionSpecId); - writer.addEntry(entry); + if (hasRowIdAssignment()) { + writer.addEntryWithAssignedRowId( + entry, copyDataFileWithAssignedFirstRowId(entry.file(), partitionSpecId)); + } else { + writer.addEntry(entry); + } + entryCount.incrementAndGet(); } + private DataFile copyDataFileWithAssignedFirstRowId(DataFile file, int partitionSpecId) { + Long assignedFirstRowId = firstRowIdForFile.apply(file); + ValidationException.check( + assignedFirstRowId != null, "No first row ID for data file: %s", file.location()); + + long firstRowId = assignedFirstRowId; + long lastRowIdExclusive; + try { + lastRowIdExclusive = Math.addExact(firstRowId, file.recordCount()); + } catch (ArithmeticException e) { + throw new ValidationException(e, "Row ID range overflows for data file: %s", file.location()); + } + + ValidationException.check( + firstRowId >= expectedFirstRowId && lastRowIdExclusive <= nextRowIdExclusive, + "Row ID range [%s, %s) for data file %s is outside reserved range [%s, %s)", + firstRowId, + lastRowIdExclusive, + file.location(), + expectedFirstRowId, + nextRowIdExclusive); + + assignedRowIdRanges.add(new RowIdRange(firstRowId, lastRowIdExclusive, file.location())); + PartitionSpec spec = + Preconditions.checkNotNull( + specsById.get(partitionSpecId), "Cannot find partition spec: %s", partitionSpecId); + return DataFiles.builder(spec).copy(file).withFirstRowId(firstRowId).build(); + } + + private void validateAssignedRowIds() { + if (!hasRowIdAssignment()) { + return; + } + + List ranges = Lists.newArrayList(assignedRowIdRanges); + ranges.sort(Comparator.comparingLong(range -> range.firstRowId)); + + RowIdRange previous = null; + for (RowIdRange current : ranges) { + if (previous != null && current.firstRowId < previous.lastRowIdExclusive) { + throw new ValidationException( + "Overlapping row ID ranges for data files %s [%s, %s) and %s [%s, %s)", + previous.filePath, + previous.firstRowId, + previous.lastRowIdExclusive, + current.filePath, + current.firstRowId, + current.lastRowIdExclusive); + } + + if (previous == null || current.lastRowIdExclusive > previous.lastRowIdExclusive) { + previous = current; + } + } + } + + @Override + protected void validate(TableMetadata currentMetadata, Snapshot snapshot) { + if (!hasRowIdAssignment()) { + return; + } + + Long currentSnapshotId = snapshot != null ? snapshot.snapshotId() : null; + ValidationException.check( + Objects.equals(expectedSnapshotId, currentSnapshotId), + "Cannot assign first row IDs after the table snapshot changed: expected %s, found %s", + expectedSnapshotId, + currentSnapshotId); + ValidationException.check( + currentMetadata.nextRowId() == expectedFirstRowId, + "Cannot assign first row IDs after table next-row-id changed: expected %s, found %s", + expectedFirstRowId, + currentMetadata.nextRowId()); + } + + @Override + protected long assignedRows(TableMetadata base, long manifestListNextRowId) { + if (!hasRowIdAssignment()) { + return super.assignedRows(base, manifestListNextRowId); + } + + return Math.subtractExact(nextRowIdExclusive, expectedFirstRowId); + } + + private boolean hasRowIdAssignment() { + return firstRowIdForFile != null; + } + private WriterWrapper getWriter(Object key, int partitionSpecId) { return writers.computeIfAbsent( Pair.of(key, partitionSpecId), k -> new WriterWrapper(specsById.get(partitionSpecId))); @@ -363,13 +518,30 @@ class WriterWrapper { } synchronized void addEntry(ManifestEntry entry) { + prepareWriter(); + writer.existing(entry); + } + + synchronized void addEntryWithAssignedRowId(ManifestEntry entry, DataFile file) { + prepareWriter(); + writer.existing( + file, + Preconditions.checkNotNull( + entry.snapshotId(), "Missing snapshot ID for data file: %s", file.location()), + Preconditions.checkNotNull( + entry.dataSequenceNumber(), + "Missing data sequence number for data file: %s", + file.location()), + entry.fileSequenceNumber()); + } + + private void prepareWriter() { if (writer == null) { writer = newManifestWriter(spec); } else if (writer.length() >= getManifestTargetSizeBytes()) { close(); writer = newManifestWriter(spec); } - writer.existing(entry); } synchronized void close() { @@ -383,4 +555,16 @@ synchronized void close() { } } } + + private static class RowIdRange { + private final long firstRowId; + private final long lastRowIdExclusive; + private final String filePath; + + private RowIdRange(long firstRowId, long lastRowIdExclusive, String filePath) { + this.firstRowId = firstRowId; + this.lastRowIdExclusive = lastRowIdExclusive; + this.filePath = filePath; + } + } } diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index d97c63b61608..ca805f6732b6 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -334,7 +334,7 @@ public Snapshot apply() { Long assignedRows = null; if (base.formatVersion() >= 3) { nextRowId = base.nextRowId(); - assignedRows = writer.nextRowId() - base.nextRowId(); + assignedRows = assignedRows(base, writer.nextRowId()); } Map summary = summary(); @@ -368,6 +368,16 @@ public Snapshot apply() { writer.toManifestListFile().encryptionKeyID()); } + /** + * Returns the number of row IDs reserved by this snapshot. + * + *

Snapshot producers that assign explicit row IDs may override this method when the reserved + * range cannot be inferred from manifest-level row ID inheritance. + */ + protected long assignedRows(TableMetadata base, long manifestListNextRowId) { + return Math.subtractExact(manifestListNextRowId, base.nextRowId()); + } + private void runValidations(Snapshot parentSnapshot) { validate(base, parentSnapshot); diff --git a/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java b/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java index 6dab3c0dad84..4efa93079b9b 100644 --- a/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java +++ b/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java @@ -19,15 +19,20 @@ package org.apache.iceberg; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.fail; import java.io.File; import java.io.IOException; import java.io.UncheckedIOException; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; @@ -531,6 +536,245 @@ public void testTableUpgrade(@TempDir File altLocation) { checkDataFileAssignment(upgradeTable, manifests.get(1), (Long) null); } + @Test + public void testAssignExplicitFirstRowIdsAfterUpgrade(@TempDir File altLocation) { + BaseTable upgradeTable = + TestTables.create( + altLocation, + "test_upgrade", + SCHEMA, + PartitionSpec.unpartitioned(), + 2, + Map.of("random-snapshot-ids", "true")); + + upgradeTable.newAppend().appendFile(FILE_A).commit(); + upgradeTable.newAppend().appendFile(FILE_B).commit(); + upgradeTable.newRowDelta().addDeletes(FILE_A_DELETES).commit(); + + Map entriesBefore = liveDataEntriesByPath(upgradeTable); + List deleteManifestsBefore = + upgradeTable.currentSnapshot().deleteManifests(upgradeTable.io()).stream() + .map(ManifestFile::path) + .collect(Collectors.toList()); + long currentSnapshotIdBefore = upgradeTable.currentSnapshot().snapshotId(); + Set snapshotIdsBefore = new HashSet<>(); + upgradeTable.snapshots().forEach(snapshot -> snapshotIdsBefore.add(snapshot.snapshotId())); + + Map firstRowIds = Map.of(FILE_A.location(), 0L, FILE_B.location(), 1_000L); + long nextRowIdExclusive = 2_000L; + Transaction upgradeTransaction = upgradeTable.newTransaction(); + upgradeTransaction.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit(); + upgradeTransaction + .rewriteManifests() + .assignFirstRowIds(file -> firstRowIds.get(file.location()), 0L, nextRowIdExclusive) + .commit(); + upgradeTransaction.commitTransaction(); + upgradeTable.refresh(); + + Snapshot assignmentSnapshot = upgradeTable.currentSnapshot(); + assertThat(assignmentSnapshot.parentId()).isEqualTo(currentSnapshotIdBefore); + assertThat(assignmentSnapshot.operation()).isEqualTo(DataOperations.REPLACE); + assertThat(assignmentSnapshot.firstRowId()).isEqualTo(0L); + assertThat(assignmentSnapshot.addedRows()).isEqualTo(nextRowIdExclusive); + assertThat(upgradeTable.operations().current().nextRowId()).isEqualTo(nextRowIdExclusive); + + assertThat(snapshotIdsBefore) + .allSatisfy(snapshotId -> assertThat(upgradeTable.snapshot(snapshotId)).isNotNull()); + assertThat( + assignmentSnapshot.deleteManifests(upgradeTable.io()).stream() + .map(ManifestFile::path) + .collect(Collectors.toList())) + .containsExactlyElementsOf(deleteManifestsBefore); + + Map entriesAfter = liveDataEntriesByPath(upgradeTable); + assertThat(entriesAfter.keySet()).containsExactlyInAnyOrderElementsOf(entriesBefore.keySet()); + entriesBefore.forEach( + (path, before) -> { + EntryMetadata after = entriesAfter.get(path); + assertThat(after.status).isEqualTo(ManifestEntry.Status.EXISTING); + assertThat(after.snapshotId).isEqualTo(before.snapshotId); + assertThat(after.dataSequenceNumber).isEqualTo(before.dataSequenceNumber); + assertThat(after.fileSequenceNumber).isEqualTo(before.fileSequenceNumber); + assertThat(after.recordCount).isEqualTo(before.recordCount); + assertThat(after.firstRowId).isEqualTo(firstRowIds.get(path)); + }); + + upgradeTable.newAppend().appendFile(FILE_C).commit(); + assertThat(liveDataEntriesByPath(upgradeTable).get(FILE_C.location()).firstRowId) + .isEqualTo(nextRowIdExclusive); + } + + @Test + public void testRejectInvalidExplicitFirstRowIds(@TempDir File altLocation) { + BaseTable upgradeTable = upgradedTableWithTwoFiles(altLocation); + long snapshotIdBefore = upgradeTable.currentSnapshot().snapshotId(); + + assertThatThrownBy( + () -> + upgradeTable + .rewriteManifests() + .assignFirstRowIds( + file -> file.location().equals(FILE_A.location()) ? 0L : 100L, 0L, 225L) + .commit()) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Overlapping row ID ranges"); + + assertThatThrownBy( + () -> + upgradeTable + .rewriteManifests() + .assignFirstRowIds( + file -> file.location().equals(FILE_A.location()) ? 0L : 1_000L, 0L, 1_050L) + .commit()) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("outside reserved range"); + + assertThatThrownBy( + () -> + upgradeTable + .rewriteManifests() + .assignFirstRowIds( + file -> file.location().equals(FILE_A.location()) ? 0L : null, 0L, 225L) + .commit()) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("No first row ID"); + + assertThatThrownBy( + () -> + upgradeTable + .rewriteManifests() + .assignFirstRowIds( + file -> file.location().equals(FILE_A.location()) ? -1L : 125L, 0L, 225L) + .commit()) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("outside reserved range"); + + assertThatThrownBy( + () -> + upgradeTable + .rewriteManifests() + .assignFirstRowIds( + file -> + file.location().equals(FILE_A.location()) ? Long.MAX_VALUE - 100L : 0L, + 0L, + Long.MAX_VALUE) + .commit()) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Row ID range overflows"); + + assertThat(upgradeTable.currentSnapshot().snapshotId()).isEqualTo(snapshotIdBefore); + assertThat(upgradeTable.operations().current().nextRowId()).isEqualTo(0L); + } + + @Test + public void testRejectCombiningExplicitAssignmentWithOtherRewriteModes( + @TempDir File altLocation) { + BaseTable upgradeTable = upgradedTableWithTwoFiles(altLocation); + + RewriteManifests selectedRewrite = upgradeTable.rewriteManifests().rewriteIf(ignored -> true); + assertThatThrownBy( + () -> selectedRewrite.assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("cannot be combined"); + + RewriteManifests rowIdAssignment = + upgradeTable + .rewriteManifests() + .assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount() + FILE_B.recordCount()); + assertThatThrownBy(() -> rowIdAssignment.rewriteIf(ignored -> true)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("cannot be combined"); + } + + @Test + public void testRejectConcurrentSnapshotDuringExplicitAssignment(@TempDir File altLocation) { + BaseTable upgradeTable = + TestTables.create( + altLocation, + "test_upgrade", + SCHEMA, + PartitionSpec.unpartitioned(), + 2, + Map.of("random-snapshot-ids", "true")); + upgradeTable.newAppend().appendFile(FILE_A).commit(); + TestTables.upgrade(altLocation, "test_upgrade", 3); + upgradeTable.refresh(); + + RewriteManifests assignment = + upgradeTable.rewriteManifests().assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount()); + upgradeTable.newAppend().appendFile(FILE_B).commit(); + + assertThatThrownBy(assignment::commit) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("table snapshot changed"); + } + + @Test + public void testRejectExplicitAssignmentBeforeFormatV3(@TempDir File altLocation) { + BaseTable v2Table = + TestTables.create(altLocation, "test_v2", SCHEMA, PartitionSpec.unpartitioned(), 2); + v2Table.newAppend().appendFile(FILE_A).commit(); + + assertThatThrownBy( + () -> + v2Table.rewriteManifests().assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("format version 2"); + } + + private BaseTable upgradedTableWithTwoFiles(File altLocation) { + BaseTable upgradeTable = + TestTables.create( + altLocation, + "test_upgrade", + SCHEMA, + PartitionSpec.unpartitioned(), + 2, + Map.of("random-snapshot-ids", "true")); + upgradeTable.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit(); + TestTables.upgrade(altLocation, "test_upgrade", 3); + upgradeTable.refresh(); + return upgradeTable; + } + + private static Map liveDataEntriesByPath(Table table) { + Map entriesByPath = new HashMap<>(); + for (ManifestFile manifest : table.currentSnapshot().dataManifests(table.io())) { + try (ManifestReader reader = + ManifestFiles.read(manifest, table.io(), table.specs())) { + for (ManifestEntry entry : reader.liveEntries()) { + EntryMetadata previous = + entriesByPath.put(entry.file().location(), new EntryMetadata(entry)); + assertThat(previous) + .as("Active data file path must be unique: %s", entry.file().location()) + .isNull(); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + return entriesByPath; + } + + private static class EntryMetadata { + private final ManifestEntry.Status status; + private final Long snapshotId; + private final Long dataSequenceNumber; + private final Long fileSequenceNumber; + private final long recordCount; + private final Long firstRowId; + + private EntryMetadata(ManifestEntry entry) { + this.status = entry.status(); + this.snapshotId = entry.snapshotId(); + this.dataSequenceNumber = entry.dataSequenceNumber(); + this.fileSequenceNumber = entry.fileSequenceNumber(); + this.recordCount = entry.file().recordCount(); + this.firstRowId = entry.file().firstRowId(); + } + } + @Test public void testAssignmentAfterUpgrade(@TempDir File altLocation) { // data manifests: [added(FILE_C)], [existing(FILE_A), deleted(FILE_B)] diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java index 2cf728830bc8..714129ca86ad 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java @@ -26,6 +26,7 @@ import java.io.IOException; import java.time.Duration; import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.Map; import java.util.concurrent.TimeoutException; @@ -50,7 +51,9 @@ import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableOperations; +import org.apache.iceberg.TableProperties; import org.apache.iceberg.TestHelpers; +import org.apache.iceberg.Transaction; import org.apache.iceberg.data.FileHelpers; import org.apache.iceberg.data.GenericRecord; import org.apache.iceberg.data.Record; @@ -760,6 +763,75 @@ public void testResumingStreamReadFromCheckpoint() throws Exception { } } + @TestTemplate + public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { + File writerCheckpoint = temp.resolve("upgrade-checkpoint").toFile(); + File output = temp.resolve("upgrade-output").toFile(); + List beforeUpgrade = Lists.newArrayList(new SimpleRecord(1, "before")); + List afterUpgrade = Lists.newArrayList(new SimpleRecord(2, "after")); + + appendData(beforeUpgrade); + table.refresh(); + long checkpointSnapshotId = table.currentSnapshot().snapshotId(); + + DataStreamWriter querySource = + spark + .readStream() + .format("iceberg") + .load(tableName) + .writeStream() + .option("checkpointLocation", writerCheckpoint.toString()) + .format("parquet") + .queryName("v2_v3_upgrade_checkpoint_test") + .option("path", output.getPath()); + + StreamingQuery initialQuery = querySource.start(); + initialQuery.processAllAvailable(); + initialQuery.stop(); + + List activeFiles = Lists.newArrayList(); + try (CloseableIterable tasks = table.newScan().planFiles()) { + tasks.forEach(task -> activeFiles.add(task.file())); + } + activeFiles.sort(Comparator.comparing(DataFile::location)); + + Map firstRowIds = Maps.newHashMap(); + long nextRowId = 0L; + for (DataFile file : activeFiles) { + firstRowIds.put(file.location(), nextRowId); + nextRowId = Math.addExact(nextRowId, file.recordCount()); + } + long nextRowIdExclusive = nextRowId; + + Transaction upgrade = table.newTransaction(); + upgrade.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit(); + upgrade + .rewriteManifests() + .assignFirstRowIds(file -> firstRowIds.get(file.location()), 0L, nextRowIdExclusive) + .commit(); + upgrade.commitTransaction(); + table.refresh(); + + assertThat(table.snapshot(checkpointSnapshotId)).isNotNull(); + assertThat(table.currentSnapshot().parentId()).isEqualTo(checkpointSnapshotId); + assertThat(table.currentSnapshot().operation()).isEqualTo(DataOperations.REPLACE); + + appendData(afterUpgrade); + + StreamingQuery resumedQuery = querySource.start(); + resumedQuery.processAllAvailable(); + resumedQuery.stop(); + + StreamingQuery restartedQuery = querySource.start(); + restartedQuery.processAllAvailable(); + restartedQuery.stop(); + + List actual = + spark.read().load(output.getPath()).as(Encoders.bean(SimpleRecord.class)).collectAsList(); + assertThat(actual) + .containsExactlyInAnyOrderElementsOf(Iterables.concat(beforeUpgrade, afterUpgrade)); + } + @TestTemplate public void testFailReadingCheckpointInvalidSnapshot() throws IOException, TimeoutException { File writerCheckpointFolder = temp.resolve("writer-checkpoint-folder").toFile(); From 3833180b0b4754b436cf5b298d67adf7df5b89e6 Mon Sep 17 00:00:00 2001 From: Timothy Wang Date: Fri, 31 Jul 2026 09:25:30 +0000 Subject: [PATCH 2/3] Core: support distributed row ID manifest rewrites Generated-by: OpenAI Codex GPT-5 --- .../org/apache/iceberg/RewriteManifests.java | 23 ++- .../apache/iceberg/BaseRewriteManifests.java | 166 +++++------------- .../org/apache/iceberg/ManifestFiles.java | 132 ++++++++++++++ .../iceberg/TestRowLineageAssignment.java | 155 +++++++--------- .../source/TestStructuredStreamingRead3.java | 53 +++++- 5 files changed, 295 insertions(+), 234 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/RewriteManifests.java b/api/src/main/java/org/apache/iceberg/RewriteManifests.java index bc74f377010a..12daa21eca4a 100644 --- a/api/src/main/java/org/apache/iceberg/RewriteManifests.java +++ b/api/src/main/java/org/apache/iceberg/RewriteManifests.java @@ -49,26 +49,23 @@ public interface RewriteManifests extends SnapshotUpdate { RewriteManifests clusterBy(Function func); /** - * Rewrites live data-file entries with explicit first row IDs. + * Declares the row ID range assigned by replacement manifests. * - *

This operation is intended for assigning row IDs to files that existed before a table was - * upgraded to format version 3. It preserves each entry's original snapshot ID, data sequence - * number, and file sequence number. The active file set is not changed. + *

This is intended for distributed manifest rewrites during an upgrade to format version 3. + * The manifests added through {@link #addManifest(ManifestFile)} must contain the same live data + * files as the deleted manifests, with valid first row IDs in the declared range. Existing + * snapshot IDs, data sequence numbers, and file sequence numbers must be preserved. * - *

The reserved range starts at the table's current {@code next-row-id} and ends at {@code - * nextRowIdExclusive}. The range may contain gaps, but every assigned file range must be fully - * contained within it. + *

The declared range starts at the table's current {@code next-row-id} and ends at {@code + * nextRowIdExclusive}. This update reserves the range without loading all manifest entries into + * the committing process. * - * @param firstRowIdForFile function that returns the first row ID for each live data file * @param expectedFirstRowId expected table {@code next-row-id} before this operation * @param nextRowIdExclusive first row ID available after this operation * @return this for method chaining */ - default RewriteManifests assignFirstRowIds( - Function firstRowIdForFile, - long expectedFirstRowId, - long nextRowIdExclusive) { - throw new UnsupportedOperationException("Explicit first row ID assignment is not supported"); + default RewriteManifests setAssignedRowIdRange(long expectedFirstRowId, long nextRowIdExclusive) { + throw new UnsupportedOperationException("Explicit row ID ranges are not supported"); } /** diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java index 054071b9f2c8..8523b39468a4 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java @@ -24,7 +24,6 @@ import java.io.IOException; import java.util.Collection; import java.util.Collections; -import java.util.Comparator; import java.util.List; import java.util.Map; import java.util.Objects; @@ -50,8 +49,6 @@ public class BaseRewriteManifests extends SnapshotProducer implements RewriteManifests { - private static final Object ROW_ID_ASSIGNMENT_CLUSTER_KEY = new Object(); - private final String tableName; private final Map specsById; private final long manifestTargetSizeBytes; @@ -66,10 +63,8 @@ public class BaseRewriteManifests extends SnapshotProducer private final Map writers = Maps.newConcurrentMap(); private final AtomicLong entryCount = new AtomicLong(0); - private final Collection assignedRowIdRanges = new ConcurrentLinkedQueue<>(); private Function clusterByFunc; - private Function firstRowIdForFile; private Predicate predicate; private Long expectedFirstRowId; private Long nextRowIdExclusive; @@ -122,20 +117,15 @@ protected Map summary() { @Override public RewriteManifests clusterBy(Function func) { + Preconditions.checkState( + !hasAssignedRowIdRange(), + "Manifest clustering cannot be combined with an assigned row ID range"); this.clusterByFunc = func; return this; } @Override - public RewriteManifests assignFirstRowIds( - Function firstRowIdForFile, - long expectedFirstRowId, - long nextRowIdExclusive) { - Preconditions.checkArgument(firstRowIdForFile != null, "First row ID function cannot be null"); - Preconditions.checkArgument( - ops().current().formatVersion() >= 3, - "Cannot assign first row IDs to a table with format version %s", - ops().current().formatVersion()); + public RewriteManifests setAssignedRowIdRange(long expectedFirstRowId, long nextRowIdExclusive) { Preconditions.checkArgument( expectedFirstRowId >= 0, "Expected first row ID must be non-negative: %s", @@ -146,50 +136,36 @@ public RewriteManifests assignFirstRowIds( nextRowIdExclusive, expectedFirstRowId); Preconditions.checkState( - !hasRowIdAssignment(), "First row ID assignment is already configured"); + !hasAssignedRowIdRange(), "Assigned row ID range is already configured"); Preconditions.checkState( - predicate == null - && deletedManifests.isEmpty() - && addedManifests.isEmpty() - && rewrittenAddedManifests.isEmpty(), - "First row ID assignment cannot be combined with manifest selection or replacement"); + predicate == null && clusterByFunc == null, + "Assigned row ID ranges can only be used with direct manifest replacement"); - this.firstRowIdForFile = firstRowIdForFile; this.expectedFirstRowId = expectedFirstRowId; this.nextRowIdExclusive = nextRowIdExclusive; Snapshot currentSnapshot = ops().current().currentSnapshot(); this.expectedSnapshotId = currentSnapshot != null ? currentSnapshot.snapshotId() : null; - if (clusterByFunc == null) { - this.clusterByFunc = ignored -> ROW_ID_ASSIGNMENT_CLUSTER_KEY; - } - return this; } @Override public RewriteManifests rewriteIf(Predicate pred) { Preconditions.checkState( - !hasRowIdAssignment(), - "Manifest selection cannot be combined with first row ID assignment"); + !hasAssignedRowIdRange(), + "Manifest selection cannot be combined with an assigned row ID range"); this.predicate = pred; return this; } @Override public RewriteManifests deleteManifest(ManifestFile manifest) { - Preconditions.checkState( - !hasRowIdAssignment(), - "Manifest replacement cannot be combined with first row ID assignment"); deletedManifests.add(manifest); return this; } @Override public RewriteManifests addManifest(ManifestFile manifest) { - Preconditions.checkState( - !hasRowIdAssignment(), - "Manifest replacement cannot be combined with first row ID assignment"); Preconditions.checkArgument(!manifest.hasAddedFiles(), "Cannot add manifest with added files"); Preconditions.checkArgument( !manifest.hasDeletedFiles(), "Cannot add manifest with deleted files"); @@ -239,7 +215,7 @@ public List apply(TableMetadata base, Snapshot snapshot) { } validateFilesCounts(); - validateAssignedRowIds(); + validateAssignedRowIdRange(); Iterable newManifestsWithMetadata = Iterables.transform( @@ -291,7 +267,6 @@ private void keepActiveManifests(List currentManifests) { private void reset() { deleteUncommitted(newManifests, ImmutableSet.of(), true /* clear new manifests */); entryCount.set(0); - assignedRowIdRanges.clear(); keptManifests.clear(); rewrittenManifests.clear(); writers.clear(); @@ -394,78 +369,50 @@ private void appendEntry(ManifestEntry entry, Object key, int partitio Preconditions.checkNotNull(key, "Key cannot be null"); WriterWrapper writer = getWriter(key, partitionSpecId); - if (hasRowIdAssignment()) { - writer.addEntryWithAssignedRowId( - entry, copyDataFileWithAssignedFirstRowId(entry.file(), partitionSpecId)); - } else { - writer.addEntry(entry); - } - + writer.addEntry(entry); entryCount.incrementAndGet(); } - private DataFile copyDataFileWithAssignedFirstRowId(DataFile file, int partitionSpecId) { - Long assignedFirstRowId = firstRowIdForFile.apply(file); - ValidationException.check( - assignedFirstRowId != null, "No first row ID for data file: %s", file.location()); - - long firstRowId = assignedFirstRowId; - long lastRowIdExclusive; - try { - lastRowIdExclusive = Math.addExact(firstRowId, file.recordCount()); - } catch (ArithmeticException e) { - throw new ValidationException(e, "Row ID range overflows for data file: %s", file.location()); - } - - ValidationException.check( - firstRowId >= expectedFirstRowId && lastRowIdExclusive <= nextRowIdExclusive, - "Row ID range [%s, %s) for data file %s is outside reserved range [%s, %s)", - firstRowId, - lastRowIdExclusive, - file.location(), - expectedFirstRowId, - nextRowIdExclusive); - - assignedRowIdRanges.add(new RowIdRange(firstRowId, lastRowIdExclusive, file.location())); - PartitionSpec spec = - Preconditions.checkNotNull( - specsById.get(partitionSpecId), "Cannot find partition spec: %s", partitionSpecId); - return DataFiles.builder(spec).copy(file).withFirstRowId(firstRowId).build(); - } - - private void validateAssignedRowIds() { - if (!hasRowIdAssignment()) { + private void validateAssignedRowIdRange() { + if (!hasAssignedRowIdRange()) { return; } - List ranges = Lists.newArrayList(assignedRowIdRanges); - ranges.sort(Comparator.comparingLong(range -> range.firstRowId)); - - RowIdRange previous = null; - for (RowIdRange current : ranges) { - if (previous != null && current.firstRowId < previous.lastRowIdExclusive) { - throw new ValidationException( - "Overlapping row ID ranges for data files %s [%s, %s) and %s [%s, %s)", - previous.filePath, - previous.firstRowId, - previous.lastRowIdExclusive, - current.filePath, - current.firstRowId, - current.lastRowIdExclusive); - } - - if (previous == null || current.lastRowIdExclusive > previous.lastRowIdExclusive) { - previous = current; - } - } + ValidationException.check( + !deletedManifests.isEmpty() + && (!addedManifests.isEmpty() || !rewrittenAddedManifests.isEmpty()), + "Assigned row ID ranges require direct manifest replacement"); + + deletedManifests.forEach( + manifest -> + ValidationException.check( + manifest.content() == ManifestContent.DATA, + "Cannot assign row IDs by replacing a delete manifest: %s", + manifest.path())); + Iterables.concat(addedManifests, rewrittenAddedManifests) + .forEach( + manifest -> { + ValidationException.check( + manifest.content() == ManifestContent.DATA, + "Cannot assign row IDs with a replacement delete manifest: %s", + manifest.path()); + ValidationException.check( + manifest.firstRowId() != null, + "Replacement data manifest must have first row ID: %s", + manifest.path()); + }); } @Override protected void validate(TableMetadata currentMetadata, Snapshot snapshot) { - if (!hasRowIdAssignment()) { + if (!hasAssignedRowIdRange()) { return; } + ValidationException.check( + currentMetadata.formatVersion() >= 3, + "Cannot assign row IDs to a table with format version %s", + currentMetadata.formatVersion()); Long currentSnapshotId = snapshot != null ? snapshot.snapshotId() : null; ValidationException.check( Objects.equals(expectedSnapshotId, currentSnapshotId), @@ -481,15 +428,15 @@ protected void validate(TableMetadata currentMetadata, Snapshot snapshot) { @Override protected long assignedRows(TableMetadata base, long manifestListNextRowId) { - if (!hasRowIdAssignment()) { + if (!hasAssignedRowIdRange()) { return super.assignedRows(base, manifestListNextRowId); } return Math.subtractExact(nextRowIdExclusive, expectedFirstRowId); } - private boolean hasRowIdAssignment() { - return firstRowIdForFile != null; + private boolean hasAssignedRowIdRange() { + return expectedFirstRowId != null; } private WriterWrapper getWriter(Object key, int partitionSpecId) { @@ -522,19 +469,6 @@ synchronized void addEntry(ManifestEntry entry) { writer.existing(entry); } - synchronized void addEntryWithAssignedRowId(ManifestEntry entry, DataFile file) { - prepareWriter(); - writer.existing( - file, - Preconditions.checkNotNull( - entry.snapshotId(), "Missing snapshot ID for data file: %s", file.location()), - Preconditions.checkNotNull( - entry.dataSequenceNumber(), - "Missing data sequence number for data file: %s", - file.location()), - entry.fileSequenceNumber()); - } - private void prepareWriter() { if (writer == null) { writer = newManifestWriter(spec); @@ -555,16 +489,4 @@ synchronized void close() { } } } - - private static class RowIdRange { - private final long firstRowId; - private final long lastRowIdExclusive; - private final String filePath; - - private RowIdRange(long firstRowId, long lastRowIdExclusive, String filePath) { - this.firstRowId = firstRowId; - this.lastRowIdExclusive = lastRowIdExclusive; - this.filePath = filePath; - } - } } diff --git a/core/src/main/java/org/apache/iceberg/ManifestFiles.java b/core/src/main/java/org/apache/iceberg/ManifestFiles.java index dae46e5ec49e..ba69394cb957 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestFiles.java +++ b/core/src/main/java/org/apache/iceberg/ManifestFiles.java @@ -35,6 +35,7 @@ import org.apache.iceberg.encryption.EncryptedFiles; import org.apache.iceberg.encryption.EncryptedOutputFile; import org.apache.iceberg.exceptions.RuntimeIOException; +import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.ContentCache; import org.apache.iceberg.io.FileIO; @@ -295,6 +296,137 @@ public static ManifestWriter write( writerProperties); } + /** + * Rewrites the live entries in a data manifest with explicit first row IDs. + * + *

The replacement contains only {@link ManifestEntry.Status#EXISTING EXISTING} entries and + * preserves each entry's snapshot ID, data sequence number, and file sequence number. The caller + * can therefore build replacement manifests in parallel and commit them with {@link + * RewriteManifests#addManifest(ManifestFile)}. + * + * @param manifest data manifest to rewrite + * @param io file IO used to read the manifest + * @param specsById table partition specs by ID + * @param outputFile output file for the replacement manifest + * @param manifestFirstRowId non-null manifest-level first row ID + * @param firstRowIdForFile function that returns the first row ID for each live data file + * @param rangeStart inclusive start of the reserved row ID range + * @param rangeEnd exclusive end of the reserved row ID range + * @return the replacement manifest + */ + public static ManifestFile rewriteDataManifestWithFirstRowIds( + ManifestFile manifest, + FileIO io, + Map specsById, + OutputFile outputFile, + long manifestFirstRowId, + Function firstRowIdForFile, + long rangeStart, + long rangeEnd) { + return rewriteDataManifestWithFirstRowIds( + manifest, + io, + specsById, + EncryptedFiles.plainAsEncryptedOutput(outputFile), + manifestFirstRowId, + firstRowIdForFile, + rangeStart, + rangeEnd); + } + + /** + * Rewrites the live entries in a data manifest with explicit first row IDs. + * + * @see #rewriteDataManifestWithFirstRowIds(ManifestFile, FileIO, Map, OutputFile, long, Function, + * long, long) + */ + public static ManifestFile rewriteDataManifestWithFirstRowIds( + ManifestFile manifest, + FileIO io, + Map specsById, + EncryptedOutputFile outputFile, + long manifestFirstRowId, + Function firstRowIdForFile, + long rangeStart, + long rangeEnd) { + Preconditions.checkArgument( + manifest.content() == ManifestContent.DATA, + "Cannot assign row IDs in a delete manifest: %s", + manifest.path()); + Preconditions.checkArgument(firstRowIdForFile != null, "First row ID function cannot be null"); + Preconditions.checkArgument( + manifestFirstRowId >= 0, + "Manifest first row ID must be non-negative: %s", + manifestFirstRowId); + Preconditions.checkArgument( + rangeStart >= 0, "Row ID range start must be non-negative: %s", rangeStart); + Preconditions.checkArgument( + rangeEnd >= rangeStart, + "Row ID range end %s must not be less than range start %s", + rangeEnd, + rangeStart); + + PartitionSpec spec = + Preconditions.checkNotNull( + specsById.get(manifest.partitionSpecId()), + "Cannot find partition spec: %s", + manifest.partitionSpecId()); + ManifestWriter writer = newWriter(3, spec, outputFile, null, manifestFirstRowId); + boolean threw = true; + try (ManifestReader reader = read(manifest, io, specsById)) { + for (ManifestEntry entry : reader.liveEntries()) { + DataFile file = entry.file(); + ValidationException.check( + file.firstRowId() == null, "Data file already has a first row ID: %s", file.location()); + Long assignedFirstRowId = firstRowIdForFile.apply(file); + ValidationException.check( + assignedFirstRowId != null, "No first row ID for data file: %s", file.location()); + + long lastRowIdExclusive; + try { + lastRowIdExclusive = Math.addExact(assignedFirstRowId, file.recordCount()); + } catch (ArithmeticException e) { + throw new ValidationException( + e, "Row ID range overflows for data file: %s", file.location()); + } + ValidationException.check( + assignedFirstRowId >= rangeStart && lastRowIdExclusive <= rangeEnd, + "Row ID range [%s, %s) for data file %s is outside reserved range [%s, %s)", + assignedFirstRowId, + lastRowIdExclusive, + file.location(), + rangeStart, + rangeEnd); + + DataFile withFirstRowId = + DataFiles.builder(spec).copy(file).withFirstRowId(assignedFirstRowId).build(); + writer.existing( + withFirstRowId, + Preconditions.checkNotNull( + entry.snapshotId(), "Missing snapshot ID for data file: %s", file.location()), + Preconditions.checkNotNull( + entry.dataSequenceNumber(), + "Missing data sequence number for data file: %s", + file.location()), + entry.fileSequenceNumber()); + } + + threw = false; + } catch (IOException e) { + throw new RuntimeIOException(e, "Failed to close manifest: %s", manifest.path()); + } finally { + try { + writer.close(); + } catch (IOException e) { + if (!threw) { + throw new RuntimeIOException(e, "Failed to close manifest: %s", outputFile); + } + } + } + + return writer.toManifestFile(); + } + @VisibleForTesting static ManifestWriter newWriter( int formatVersion, diff --git a/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java b/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java index 4efa93079b9b..f5f78d481c2f 100644 --- a/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java +++ b/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java @@ -30,6 +30,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.stream.Collectors; import org.apache.iceberg.data.Record; import org.apache.iceberg.exceptions.ValidationException; @@ -537,7 +538,7 @@ public void testTableUpgrade(@TempDir File altLocation) { } @Test - public void testAssignExplicitFirstRowIdsAfterUpgrade(@TempDir File altLocation) { + public void testCommitDistributedRowIdManifestRewrite(@TempDir File altLocation) { BaseTable upgradeTable = TestTables.create( altLocation, @@ -562,12 +563,26 @@ public void testAssignExplicitFirstRowIdsAfterUpgrade(@TempDir File altLocation) Map firstRowIds = Map.of(FILE_A.location(), 0L, FILE_B.location(), 1_000L); long nextRowIdExclusive = 2_000L; + List oldDataManifests = + upgradeTable.currentSnapshot().dataManifests(upgradeTable.io()); + List replacementManifests = + oldDataManifests.stream() + .map( + manifest -> + rewriteManifestWithFirstRowIds( + upgradeTable, manifest, firstRowIds, 0L, nextRowIdExclusive)) + .collect(Collectors.toList()); + Transaction upgradeTransaction = upgradeTable.newTransaction(); upgradeTransaction.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit(); - upgradeTransaction - .rewriteManifests() - .assignFirstRowIds(file -> firstRowIds.get(file.location()), 0L, nextRowIdExclusive) - .commit(); + TableMetadata stagedMetadata = ((BaseTransaction) upgradeTransaction).currentMetadata(); + assertThat(stagedMetadata.formatVersion()).isEqualTo(3); + assertThat(stagedMetadata.nextRowId()).isEqualTo(0L); + RewriteManifests rewrite = + upgradeTransaction.rewriteManifests().setAssignedRowIdRange(0L, nextRowIdExclusive); + oldDataManifests.forEach(rewrite::deleteManifest); + replacementManifests.forEach(rewrite::addManifest); + rewrite.commit(); upgradeTransaction.commitTransaction(); upgradeTable.refresh(); @@ -605,89 +620,37 @@ public void testAssignExplicitFirstRowIdsAfterUpgrade(@TempDir File altLocation) } @Test - public void testRejectInvalidExplicitFirstRowIds(@TempDir File altLocation) { + public void testRejectInvalidDistributedRowIdManifestRewrite(@TempDir File altLocation) { BaseTable upgradeTable = upgradedTableWithTwoFiles(altLocation); - long snapshotIdBefore = upgradeTable.currentSnapshot().snapshotId(); - - assertThatThrownBy( - () -> - upgradeTable - .rewriteManifests() - .assignFirstRowIds( - file -> file.location().equals(FILE_A.location()) ? 0L : 100L, 0L, 225L) - .commit()) - .isInstanceOf(ValidationException.class) - .hasMessageContaining("Overlapping row ID ranges"); - - assertThatThrownBy( - () -> - upgradeTable - .rewriteManifests() - .assignFirstRowIds( - file -> file.location().equals(FILE_A.location()) ? 0L : 1_000L, 0L, 1_050L) - .commit()) - .isInstanceOf(ValidationException.class) - .hasMessageContaining("outside reserved range"); - - assertThatThrownBy( - () -> - upgradeTable - .rewriteManifests() - .assignFirstRowIds( - file -> file.location().equals(FILE_A.location()) ? 0L : null, 0L, 225L) - .commit()) - .isInstanceOf(ValidationException.class) - .hasMessageContaining("No first row ID"); assertThatThrownBy( () -> - upgradeTable - .rewriteManifests() - .assignFirstRowIds( - file -> file.location().equals(FILE_A.location()) ? -1L : 125L, 0L, 225L) - .commit()) + ManifestFiles.rewriteDataManifestWithFirstRowIds( + upgradeTable.currentSnapshot().dataManifests(upgradeTable.io()).get(0), + upgradeTable.io(), + upgradeTable.specs(), + upgradeTable + .io() + .newOutputFile(new File(altLocation, "overlapping-range.avro").toString()), + 0L, + file -> file.location().equals(FILE_A.location()) ? 0L : 100L, + 125L, + 225L)) .isInstanceOf(ValidationException.class) .hasMessageContaining("outside reserved range"); - assertThatThrownBy( - () -> - upgradeTable - .rewriteManifests() - .assignFirstRowIds( - file -> - file.location().equals(FILE_A.location()) ? Long.MAX_VALUE - 100L : 0L, - 0L, - Long.MAX_VALUE) - .commit()) - .isInstanceOf(ValidationException.class) - .hasMessageContaining("Row ID range overflows"); - - assertThat(upgradeTable.currentSnapshot().snapshotId()).isEqualTo(snapshotIdBefore); - assertThat(upgradeTable.operations().current().nextRowId()).isEqualTo(0L); - } - - @Test - public void testRejectCombiningExplicitAssignmentWithOtherRewriteModes( - @TempDir File altLocation) { - BaseTable upgradeTable = upgradedTableWithTwoFiles(altLocation); + assertThatThrownBy(() -> upgradeTable.rewriteManifests().setAssignedRowIdRange(10L, 9L)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must not be less than"); RewriteManifests selectedRewrite = upgradeTable.rewriteManifests().rewriteIf(ignored -> true); - assertThatThrownBy( - () -> selectedRewrite.assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount())) + assertThatThrownBy(() -> selectedRewrite.setAssignedRowIdRange(0L, 225L)) .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("cannot be combined"); - - RewriteManifests rowIdAssignment = - upgradeTable - .rewriteManifests() - .assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount() + FILE_B.recordCount()); - assertThatThrownBy(() -> rowIdAssignment.rewriteIf(ignored -> true)) - .isInstanceOf(IllegalStateException.class) - .hasMessageContaining("cannot be combined"); + .hasMessageContaining("direct manifest replacement"); } @Test - public void testRejectConcurrentSnapshotDuringExplicitAssignment(@TempDir File altLocation) { + public void testRejectConcurrentSnapshotDuringDistributedRowIdRewrite(@TempDir File altLocation) { BaseTable upgradeTable = TestTables.create( altLocation, @@ -700,8 +663,15 @@ public void testRejectConcurrentSnapshotDuringExplicitAssignment(@TempDir File a TestTables.upgrade(altLocation, "test_upgrade", 3); upgradeTable.refresh(); - RewriteManifests assignment = - upgradeTable.rewriteManifests().assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount()); + ManifestFile oldManifest = + Iterables.getOnlyElement(upgradeTable.currentSnapshot().dataManifests(upgradeTable.io())); + ManifestFile replacement = + rewriteManifestWithFirstRowIds( + upgradeTable, oldManifest, Map.of(FILE_A.location(), 0L), 0L, FILE_A.recordCount()); + RewriteManifests assignment = upgradeTable.rewriteManifests(); + assignment.deleteManifest(oldManifest); + assignment.addManifest(replacement); + assignment.setAssignedRowIdRange(0L, FILE_A.recordCount()); upgradeTable.newAppend().appendFile(FILE_B).commit(); assertThatThrownBy(assignment::commit) @@ -709,19 +679,6 @@ public void testRejectConcurrentSnapshotDuringExplicitAssignment(@TempDir File a .hasMessageContaining("table snapshot changed"); } - @Test - public void testRejectExplicitAssignmentBeforeFormatV3(@TempDir File altLocation) { - BaseTable v2Table = - TestTables.create(altLocation, "test_v2", SCHEMA, PartitionSpec.unpartitioned(), 2); - v2Table.newAppend().appendFile(FILE_A).commit(); - - assertThatThrownBy( - () -> - v2Table.rewriteManifests().assignFirstRowIds(file -> 0L, 0L, FILE_A.recordCount())) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("format version 2"); - } - private BaseTable upgradedTableWithTwoFiles(File altLocation) { BaseTable upgradeTable = TestTables.create( @@ -737,6 +694,24 @@ private BaseTable upgradedTableWithTwoFiles(File altLocation) { return upgradeTable; } + private static ManifestFile rewriteManifestWithFirstRowIds( + Table table, + ManifestFile manifest, + Map firstRowIds, + long rangeStart, + long rangeEnd) { + String outputPath = table.location() + "/metadata/row-id-" + UUID.randomUUID() + ".avro"; + return ManifestFiles.rewriteDataManifestWithFirstRowIds( + manifest, + table.io(), + table.specs(), + table.io().newOutputFile(outputPath), + rangeStart, + file -> firstRowIds.get(file.location()), + rangeStart, + rangeEnd); + } + private static Map liveDataEntriesByPath(Table table) { Map entriesByPath = new HashMap<>(); for (ManifestFile manifest : table.currentSnapshot().dataManifests(table.io())) { diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java index 714129ca86ad..54f5e18ea4c8 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java @@ -29,6 +29,7 @@ import java.util.Comparator; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; @@ -41,10 +42,13 @@ import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileScanTask; import org.apache.iceberg.Files; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; import org.apache.iceberg.Parameter; import org.apache.iceberg.ParameterizedTestExtension; import org.apache.iceberg.Parameters; import org.apache.iceberg.RewriteFiles; +import org.apache.iceberg.RewriteManifests; import org.apache.iceberg.Schema; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotChanges; @@ -67,6 +71,7 @@ import org.apache.iceberg.spark.SparkCatalogConfig; import org.apache.iceberg.spark.SparkReadConf; import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.util.SnapshotUtil; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.VoidFunction2; import org.apache.spark.sql.Dataset; @@ -767,10 +772,12 @@ public void testResumingStreamReadFromCheckpoint() throws Exception { public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { File writerCheckpoint = temp.resolve("upgrade-checkpoint").toFile(); File output = temp.resolve("upgrade-output").toFile(); - List beforeUpgrade = Lists.newArrayList(new SimpleRecord(1, "before")); - List afterUpgrade = Lists.newArrayList(new SimpleRecord(2, "after")); + List checkpointed = Lists.newArrayList(new SimpleRecord(1, "checkpointed")); + List pendingBeforeUpgrade = + Lists.newArrayList(new SimpleRecord(2, "pending-before-upgrade")); + List afterUpgrade = Lists.newArrayList(new SimpleRecord(3, "after-upgrade")); - appendData(beforeUpgrade); + appendData(checkpointed); table.refresh(); long checkpointSnapshotId = table.currentSnapshot().snapshotId(); @@ -789,6 +796,10 @@ public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { initialQuery.processAllAvailable(); initialQuery.stop(); + // This append is deliberately newer than the checkpoint. A resumed stream must still process + // it after walking through the v3 manifest-rewrite snapshot. + appendData(pendingBeforeUpgrade); + List activeFiles = Lists.newArrayList(); try (CloseableIterable tasks = table.newScan().planFiles()) { tasks.forEach(task -> activeFiles.add(task.file())); @@ -803,17 +814,26 @@ public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { } long nextRowIdExclusive = nextRowId; + List oldDataManifests = table.currentSnapshot().dataManifests(table.io()); + List replacementManifests = + oldDataManifests.stream() + .map( + manifest -> + rewriteManifestWithFirstRowIds(manifest, firstRowIds, 0L, nextRowIdExclusive)) + .collect(java.util.stream.Collectors.toList()); + Transaction upgrade = table.newTransaction(); upgrade.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit(); - upgrade - .rewriteManifests() - .assignFirstRowIds(file -> firstRowIds.get(file.location()), 0L, nextRowIdExclusive) - .commit(); + RewriteManifests rewrite = + upgrade.rewriteManifests().setAssignedRowIdRange(0L, nextRowIdExclusive); + oldDataManifests.forEach(rewrite::deleteManifest); + replacementManifests.forEach(rewrite::addManifest); + rewrite.commit(); upgrade.commitTransaction(); table.refresh(); assertThat(table.snapshot(checkpointSnapshotId)).isNotNull(); - assertThat(table.currentSnapshot().parentId()).isEqualTo(checkpointSnapshotId); + assertThat(SnapshotUtil.isAncestorOf(table, checkpointSnapshotId)).isTrue(); assertThat(table.currentSnapshot().operation()).isEqualTo(DataOperations.REPLACE); appendData(afterUpgrade); @@ -829,7 +849,22 @@ public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { List actual = spark.read().load(output.getPath()).as(Encoders.bean(SimpleRecord.class)).collectAsList(); assertThat(actual) - .containsExactlyInAnyOrderElementsOf(Iterables.concat(beforeUpgrade, afterUpgrade)); + .containsExactlyInAnyOrderElementsOf( + Iterables.concat(checkpointed, pendingBeforeUpgrade, afterUpgrade)); + } + + private ManifestFile rewriteManifestWithFirstRowIds( + ManifestFile manifest, Map firstRowIds, long rangeStart, long rangeEnd) { + String outputPath = table.location() + "/metadata/row-id-" + UUID.randomUUID() + ".avro"; + return ManifestFiles.rewriteDataManifestWithFirstRowIds( + manifest, + table.io(), + table.specs(), + table.io().newOutputFile(outputPath), + rangeStart, + file -> firstRowIds.get(file.location()), + rangeStart, + rangeEnd); } @TestTemplate From 725f41a8337a8e679be135d70c61e64800932965 Mon Sep 17 00:00:00 2001 From: Timothy Wang Date: Fri, 31 Jul 2026 10:18:32 +0000 Subject: [PATCH 3/3] Core: Harden distributed row ID manifest replacement Validate the reserved row ID range, reject partial legacy manifest replacement and concurrent row lineage changes, and cover checkpoint continuation across a v2-to-v3 upgrade. Generated-by: Codex GPT-5 --- .../org/apache/iceberg/RewriteManifests.java | 20 +-- .../apache/iceberg/BaseRewriteManifests.java | 43 ++++-- .../org/apache/iceberg/ManifestFiles.java | 59 ++++++-- .../org/apache/iceberg/SnapshotProducer.java | 4 +- .../iceberg/TestRowLineageAssignment.java | 140 ++++++++++++++++-- .../source/TestStructuredStreamingRead3.java | 7 +- 6 files changed, 217 insertions(+), 56 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/RewriteManifests.java b/api/src/main/java/org/apache/iceberg/RewriteManifests.java index 12daa21eca4a..5f8e0fdbcb4f 100644 --- a/api/src/main/java/org/apache/iceberg/RewriteManifests.java +++ b/api/src/main/java/org/apache/iceberg/RewriteManifests.java @@ -49,22 +49,24 @@ public interface RewriteManifests extends SnapshotUpdate { RewriteManifests clusterBy(Function func); /** - * Declares the row ID range assigned by replacement manifests. + * Reserves the row ID range assigned by replacement manifests. * - *

This is intended for distributed manifest rewrites during an upgrade to format version 3. - * The manifests added through {@link #addManifest(ManifestFile)} must contain the same live data - * files as the deleted manifests, with valid first row IDs in the declared range. Existing - * snapshot IDs, data sequence numbers, and file sequence numbers must be preserved. + *

This is a low-level operation for distributed manifest rewrites during an upgrade to format + * version 3. All current data manifests without assigned row IDs must be replaced. Added + * manifests must contain the same live data files as the deleted manifests and preserve each + * manifest entry's snapshot ID, data sequence number, and file sequence number. * - *

The declared range starts at the table's current {@code next-row-id} and ends at {@code - * nextRowIdExclusive}. This update reserves the range without loading all manifest entries into - * the committing process. + *

As with direct manifest replacement through {@link #addManifest(ManifestFile)}, the caller + * is responsible for the contents of the supplied manifests. In particular, the committing + * process does not load all manifest entries and does not validate that row ID ranges are + * globally disjoint. Every data file's row ID range must be contained in {@code + * [expectedFirstRowId, nextRowIdExclusive)} and must not overlap another data file's range. * * @param expectedFirstRowId expected table {@code next-row-id} before this operation * @param nextRowIdExclusive first row ID available after this operation * @return this for method chaining */ - default RewriteManifests setAssignedRowIdRange(long expectedFirstRowId, long nextRowIdExclusive) { + default RewriteManifests reserveRowIdRange(long expectedFirstRowId, long nextRowIdExclusive) { throw new UnsupportedOperationException("Explicit row ID ranges are not supported"); } diff --git a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java index 8523b39468a4..2aed7df43b60 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java +++ b/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java @@ -66,8 +66,8 @@ public class BaseRewriteManifests extends SnapshotProducer private Function clusterByFunc; private Predicate predicate; - private Long expectedFirstRowId; - private Long nextRowIdExclusive; + private Long reservedRangeStart; + private Long reservedRangeEndExclusive; private Long expectedSnapshotId; private final SnapshotSummary.Builder summaryBuilder = SnapshotSummary.builder(); @@ -125,7 +125,7 @@ public RewriteManifests clusterBy(Function func) { } @Override - public RewriteManifests setAssignedRowIdRange(long expectedFirstRowId, long nextRowIdExclusive) { + public RewriteManifests reserveRowIdRange(long expectedFirstRowId, long nextRowIdExclusive) { Preconditions.checkArgument( expectedFirstRowId >= 0, "Expected first row ID must be non-negative: %s", @@ -141,8 +141,8 @@ public RewriteManifests setAssignedRowIdRange(long expectedFirstRowId, long next predicate == null && clusterByFunc == null, "Assigned row ID ranges can only be used with direct manifest replacement"); - this.expectedFirstRowId = expectedFirstRowId; - this.nextRowIdExclusive = nextRowIdExclusive; + this.reservedRangeStart = expectedFirstRowId; + this.reservedRangeEndExclusive = nextRowIdExclusive; Snapshot currentSnapshot = ops().current().currentSnapshot(); this.expectedSnapshotId = currentSnapshot != null ? currentSnapshot.snapshotId() : null; @@ -400,7 +400,22 @@ private void validateAssignedRowIdRange() { manifest.firstRowId() != null, "Replacement data manifest must have first row ID: %s", manifest.path()); + ValidationException.check( + manifest.firstRowId() >= reservedRangeStart + && manifest.firstRowId() < reservedRangeEndExclusive, + "Replacement data manifest first row ID %s is outside reserved range [%s, %s]: %s", + manifest.firstRowId(), + reservedRangeStart, + reservedRangeEndExclusive, + manifest.path()); }); + + keptManifests.forEach( + manifest -> + ValidationException.check( + manifest.content() != ManifestContent.DATA || manifest.firstRowId() != null, + "All data manifests without assigned row IDs must be replaced: %s", + manifest.path())); } @Override @@ -410,8 +425,8 @@ protected void validate(TableMetadata currentMetadata, Snapshot snapshot) { } ValidationException.check( - currentMetadata.formatVersion() >= 3, - "Cannot assign row IDs to a table with format version %s", + currentMetadata.formatVersion() == 3, + "Can only assign row IDs to a table with format version 3, found %s", currentMetadata.formatVersion()); Long currentSnapshotId = snapshot != null ? snapshot.snapshotId() : null; ValidationException.check( @@ -420,9 +435,9 @@ protected void validate(TableMetadata currentMetadata, Snapshot snapshot) { expectedSnapshotId, currentSnapshotId); ValidationException.check( - currentMetadata.nextRowId() == expectedFirstRowId, + currentMetadata.nextRowId() == reservedRangeStart, "Cannot assign first row IDs after table next-row-id changed: expected %s, found %s", - expectedFirstRowId, + reservedRangeStart, currentMetadata.nextRowId()); } @@ -432,11 +447,11 @@ protected long assignedRows(TableMetadata base, long manifestListNextRowId) { return super.assignedRows(base, manifestListNextRowId); } - return Math.subtractExact(nextRowIdExclusive, expectedFirstRowId); + return Math.subtractExact(reservedRangeEndExclusive, reservedRangeStart); } private boolean hasAssignedRowIdRange() { - return expectedFirstRowId != null; + return reservedRangeStart != null; } private WriterWrapper getWriter(Object key, int partitionSpecId) { @@ -465,17 +480,13 @@ class WriterWrapper { } synchronized void addEntry(ManifestEntry entry) { - prepareWriter(); - writer.existing(entry); - } - - private void prepareWriter() { if (writer == null) { writer = newManifestWriter(spec); } else if (writer.length() >= getManifestTargetSizeBytes()) { close(); writer = newManifestWriter(spec); } + writer.existing(entry); } synchronized void close() { diff --git a/core/src/main/java/org/apache/iceberg/ManifestFiles.java b/core/src/main/java/org/apache/iceberg/ManifestFiles.java index ba69394cb957..e76f870e84da 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestFiles.java +++ b/core/src/main/java/org/apache/iceberg/ManifestFiles.java @@ -24,6 +24,7 @@ import java.math.RoundingMode; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; @@ -304,11 +305,17 @@ public static ManifestWriter write( * can therefore build replacement manifests in parallel and commit them with {@link * RewriteManifests#addManifest(ManifestFile)}. * + *

This method validates that assigned row ID ranges do not overlap within the replacement + * manifest. The caller must validate that ranges are globally disjoint across all replacement + * manifests. + * + *

The caller owns the output file until it is successfully committed. The output file may + * exist when this method throws and must be cleaned up by the caller. + * * @param manifest data manifest to rewrite * @param io file IO used to read the manifest * @param specsById table partition specs by ID * @param outputFile output file for the replacement manifest - * @param manifestFirstRowId non-null manifest-level first row ID * @param firstRowIdForFile function that returns the first row ID for each live data file * @param rangeStart inclusive start of the reserved row ID range * @param rangeEnd exclusive end of the reserved row ID range @@ -319,7 +326,6 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( FileIO io, Map specsById, OutputFile outputFile, - long manifestFirstRowId, Function firstRowIdForFile, long rangeStart, long rangeEnd) { @@ -328,7 +334,6 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( io, specsById, EncryptedFiles.plainAsEncryptedOutput(outputFile), - manifestFirstRowId, firstRowIdForFile, rangeStart, rangeEnd); @@ -337,15 +342,14 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( /** * Rewrites the live entries in a data manifest with explicit first row IDs. * - * @see #rewriteDataManifestWithFirstRowIds(ManifestFile, FileIO, Map, OutputFile, long, Function, - * long, long) + * @see #rewriteDataManifestWithFirstRowIds(ManifestFile, FileIO, Map, OutputFile, Function, long, + * long) */ public static ManifestFile rewriteDataManifestWithFirstRowIds( ManifestFile manifest, FileIO io, Map specsById, EncryptedOutputFile outputFile, - long manifestFirstRowId, Function firstRowIdForFile, long rangeStart, long rangeEnd) { @@ -354,10 +358,6 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( "Cannot assign row IDs in a delete manifest: %s", manifest.path()); Preconditions.checkArgument(firstRowIdForFile != null, "First row ID function cannot be null"); - Preconditions.checkArgument( - manifestFirstRowId >= 0, - "Manifest first row ID must be non-negative: %s", - manifestFirstRowId); Preconditions.checkArgument( rangeStart >= 0, "Row ID range start must be non-negative: %s", rangeStart); Preconditions.checkArgument( @@ -371,7 +371,8 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( specsById.get(manifest.partitionSpecId()), "Cannot find partition spec: %s", manifest.partitionSpecId()); - ManifestWriter writer = newWriter(3, spec, outputFile, null, manifestFirstRowId); + ManifestWriter writer = newWriter(3, spec, outputFile, null, rangeStart); + List assignedRanges = Lists.newArrayList(); boolean threw = true; try (ManifestReader reader = read(manifest, io, specsById)) { for (ManifestEntry entry : reader.liveEntries()) { @@ -397,6 +398,7 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( file.location(), rangeStart, rangeEnd); + assignedRanges.add(new RowIdRange(assignedFirstRowId, lastRowIdExclusive, file.location())); DataFile withFirstRowId = DataFiles.builder(spec).copy(file).withFirstRowId(assignedFirstRowId).build(); @@ -411,6 +413,7 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( entry.fileSequenceNumber()); } + validateNonOverlappingRowIdRanges(assignedRanges); threw = false; } catch (IOException e) { throw new RuntimeIOException(e, "Failed to close manifest: %s", manifest.path()); @@ -427,6 +430,40 @@ public static ManifestFile rewriteDataManifestWithFirstRowIds( return writer.toManifestFile(); } + private static void validateNonOverlappingRowIdRanges(List ranges) { + ranges.sort(Comparator.comparingLong(range -> range.firstRowId)); + + RowIdRange previous = null; + for (RowIdRange current : ranges) { + if (previous != null && current.firstRowId < previous.lastRowIdExclusive) { + throw new ValidationException( + "Overlapping row ID ranges for data files %s [%s, %s) and %s [%s, %s)", + previous.filePath, + previous.firstRowId, + previous.lastRowIdExclusive, + current.filePath, + current.firstRowId, + current.lastRowIdExclusive); + } + + if (previous == null || current.lastRowIdExclusive > previous.lastRowIdExclusive) { + previous = current; + } + } + } + + private static class RowIdRange { + private final long firstRowId; + private final long lastRowIdExclusive; + private final String filePath; + + private RowIdRange(long firstRowId, long lastRowIdExclusive, String filePath) { + this.firstRowId = firstRowId; + this.lastRowIdExclusive = lastRowIdExclusive; + this.filePath = filePath; + } + } + @VisibleForTesting static ManifestWriter newWriter( int formatVersion, diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index ca805f6732b6..20ebfc7a31f4 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -374,8 +374,8 @@ public Snapshot apply() { *

Snapshot producers that assign explicit row IDs may override this method when the reserved * range cannot be inferred from manifest-level row ID inheritance. */ - protected long assignedRows(TableMetadata base, long manifestListNextRowId) { - return Math.subtractExact(manifestListNextRowId, base.nextRowId()); + protected long assignedRows(TableMetadata tableMetadata, long manifestListNextRowId) { + return Math.subtractExact(manifestListNextRowId, tableMetadata.nextRowId()); } private void runValidations(Snapshot parentSnapshot) { diff --git a/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java b/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java index f5f78d481c2f..123ce4d5e031 100644 --- a/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java +++ b/core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java @@ -25,8 +25,6 @@ import java.io.File; import java.io.IOException; import java.io.UncheckedIOException; -import java.util.HashMap; -import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -37,6 +35,8 @@ import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Types; import org.apache.iceberg.types.Types.NestedField; import org.apache.iceberg.util.PartitionUtil; @@ -538,7 +538,7 @@ public void testTableUpgrade(@TempDir File altLocation) { } @Test - public void testCommitDistributedRowIdManifestRewrite(@TempDir File altLocation) { + public void commitDistributedRowIdManifestRewrite(@TempDir File altLocation) { BaseTable upgradeTable = TestTables.create( altLocation, @@ -558,7 +558,7 @@ public void testCommitDistributedRowIdManifestRewrite(@TempDir File altLocation) .map(ManifestFile::path) .collect(Collectors.toList()); long currentSnapshotIdBefore = upgradeTable.currentSnapshot().snapshotId(); - Set snapshotIdsBefore = new HashSet<>(); + Set snapshotIdsBefore = Sets.newHashSet(); upgradeTable.snapshots().forEach(snapshot -> snapshotIdsBefore.add(snapshot.snapshotId())); Map firstRowIds = Map.of(FILE_A.location(), 0L, FILE_B.location(), 1_000L); @@ -579,7 +579,7 @@ public void testCommitDistributedRowIdManifestRewrite(@TempDir File altLocation) assertThat(stagedMetadata.formatVersion()).isEqualTo(3); assertThat(stagedMetadata.nextRowId()).isEqualTo(0L); RewriteManifests rewrite = - upgradeTransaction.rewriteManifests().setAssignedRowIdRange(0L, nextRowIdExclusive); + upgradeTransaction.rewriteManifests().reserveRowIdRange(0L, nextRowIdExclusive); oldDataManifests.forEach(rewrite::deleteManifest); replacementManifests.forEach(rewrite::addManifest); rewrite.commit(); @@ -620,37 +620,90 @@ public void testCommitDistributedRowIdManifestRewrite(@TempDir File altLocation) } @Test - public void testRejectInvalidDistributedRowIdManifestRewrite(@TempDir File altLocation) { + public void rejectInvalidDistributedRowIdManifestRewrite(@TempDir File altLocation) { BaseTable upgradeTable = upgradedTableWithTwoFiles(altLocation); + ManifestFile oldManifest = + Iterables.getOnlyElement(upgradeTable.currentSnapshot().dataManifests(upgradeTable.io())); assertThatThrownBy( () -> ManifestFiles.rewriteDataManifestWithFirstRowIds( - upgradeTable.currentSnapshot().dataManifests(upgradeTable.io()).get(0), + oldManifest, upgradeTable.io(), upgradeTable.specs(), upgradeTable .io() .newOutputFile(new File(altLocation, "overlapping-range.avro").toString()), - 0L, file -> file.location().equals(FILE_A.location()) ? 0L : 100L, 125L, 225L)) .isInstanceOf(ValidationException.class) .hasMessageContaining("outside reserved range"); - assertThatThrownBy(() -> upgradeTable.rewriteManifests().setAssignedRowIdRange(10L, 9L)) + assertThatThrownBy( + () -> + ManifestFiles.rewriteDataManifestWithFirstRowIds( + oldManifest, + upgradeTable.io(), + upgradeTable.specs(), + upgradeTable + .io() + .newOutputFile(new File(altLocation, "overlapping-ids.avro").toString()), + ignored -> 0L, + 0L, + FILE_A.recordCount())) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Overlapping row ID ranges"); + + assertThatThrownBy(() -> upgradeTable.rewriteManifests().reserveRowIdRange(10L, 9L)) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("must not be less than"); RewriteManifests selectedRewrite = upgradeTable.rewriteManifests().rewriteIf(ignored -> true); - assertThatThrownBy(() -> selectedRewrite.setAssignedRowIdRange(0L, 225L)) + assertThatThrownBy(() -> selectedRewrite.reserveRowIdRange(0L, 225L)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("direct manifest replacement"); } @Test - public void testRejectConcurrentSnapshotDuringDistributedRowIdRewrite(@TempDir File altLocation) { + public void rejectPartialLegacyManifestReplacement(@TempDir File altLocation) { + BaseTable upgradeTable = + TestTables.create( + altLocation, + "test_upgrade", + SCHEMA, + PartitionSpec.unpartitioned(), + 2, + Map.of("random-snapshot-ids", "true")); + upgradeTable.newFastAppend().appendFile(FILE_A).commit(); + upgradeTable.newFastAppend().appendFile(FILE_B).commit(); + TestTables.upgrade(altLocation, "test_upgrade", 3); + upgradeTable.refresh(); + + List oldManifests = + upgradeTable.currentSnapshot().dataManifests(upgradeTable.io()); + assertThat(oldManifests).hasSize(2); + ManifestFile replacedManifest = oldManifests.get(0); + ManifestFile replacement = + rewriteManifestWithFirstRowIds( + upgradeTable, + replacedManifest, + Map.of(FILE_A.location(), 0L, FILE_B.location(), 0L), + 0L, + FILE_A.recordCount()); + + RewriteManifests rewrite = upgradeTable.rewriteManifests(); + rewrite.deleteManifest(replacedManifest); + rewrite.addManifest(replacement); + rewrite.reserveRowIdRange(0L, FILE_A.recordCount()); + + assertThatThrownBy(rewrite::commit) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("All data manifests without assigned row IDs must be replaced"); + } + + @Test + public void rejectConcurrentSnapshotDuringDistributedRowIdRewrite(@TempDir File altLocation) { BaseTable upgradeTable = TestTables.create( altLocation, @@ -671,7 +724,7 @@ public void testRejectConcurrentSnapshotDuringDistributedRowIdRewrite(@TempDir F RewriteManifests assignment = upgradeTable.rewriteManifests(); assignment.deleteManifest(oldManifest); assignment.addManifest(replacement); - assignment.setAssignedRowIdRange(0L, FILE_A.recordCount()); + assignment.reserveRowIdRange(0L, FILE_A.recordCount()); upgradeTable.newAppend().appendFile(FILE_B).commit(); assertThatThrownBy(assignment::commit) @@ -679,6 +732,66 @@ public void testRejectConcurrentSnapshotDuringDistributedRowIdRewrite(@TempDir F .hasMessageContaining("table snapshot changed"); } + @Test + public void rejectConcurrentNextRowIdChange(@TempDir File altLocation) { + BaseTable upgradeTable = + TestTables.create( + altLocation, + "test_upgrade", + SCHEMA, + PartitionSpec.unpartitioned(), + 2, + Map.of("random-snapshot-ids", "true")); + upgradeTable.newAppend().appendFile(FILE_A).commit(); + TestTables.upgrade(altLocation, "test_upgrade", 3); + upgradeTable.refresh(); + + long mainSnapshotId = upgradeTable.currentSnapshot().snapshotId(); + ManifestFile oldManifest = + Iterables.getOnlyElement(upgradeTable.currentSnapshot().dataManifests(upgradeTable.io())); + ManifestFile replacement = + rewriteManifestWithFirstRowIds( + upgradeTable, oldManifest, Map.of(FILE_A.location(), 0L), 0L, FILE_A.recordCount()); + RewriteManifests assignment = upgradeTable.rewriteManifests(); + assignment.deleteManifest(oldManifest); + assignment.addManifest(replacement); + assignment.reserveRowIdRange(0L, FILE_A.recordCount()); + + upgradeTable.manageSnapshots().createBranch("audit", mainSnapshotId).commit(); + upgradeTable.newAppend().appendFile(FILE_B).toBranch("audit").commit(); + assertThat(upgradeTable.currentSnapshot().snapshotId()).isEqualTo(mainSnapshotId); + + assertThatThrownBy(assignment::commit) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("table next-row-id changed"); + } + + @Test + public void rejectRowIdReservationForFormatV4(@TempDir File altLocation) { + BaseTable upgradeTable = upgradedTableWithTwoFiles(altLocation); + TestTables.upgrade(altLocation, "test_upgrade", 4); + upgradeTable.refresh(); + + ManifestFile oldManifest = + Iterables.getOnlyElement(upgradeTable.currentSnapshot().dataManifests(upgradeTable.io())); + long rangeEnd = Math.addExact(FILE_A.recordCount(), FILE_B.recordCount()); + ManifestFile replacement = + rewriteManifestWithFirstRowIds( + upgradeTable, + oldManifest, + Map.of(FILE_A.location(), 0L, FILE_B.location(), FILE_A.recordCount()), + 0L, + rangeEnd); + RewriteManifests assignment = upgradeTable.rewriteManifests(); + assignment.deleteManifest(oldManifest); + assignment.addManifest(replacement); + assignment.reserveRowIdRange(0L, rangeEnd); + + assertThatThrownBy(assignment::commit) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("format version 3, found 4"); + } + private BaseTable upgradedTableWithTwoFiles(File altLocation) { BaseTable upgradeTable = TestTables.create( @@ -706,14 +819,13 @@ private static ManifestFile rewriteManifestWithFirstRowIds( table.io(), table.specs(), table.io().newOutputFile(outputPath), - rangeStart, file -> firstRowIds.get(file.location()), rangeStart, rangeEnd); } private static Map liveDataEntriesByPath(Table table) { - Map entriesByPath = new HashMap<>(); + Map entriesByPath = Maps.newHashMap(); for (ManifestFile manifest : table.currentSnapshot().dataManifests(table.io())) { try (ManifestReader reader = ManifestFiles.read(manifest, table.io(), table.specs())) { diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java index 54f5e18ea4c8..de89aceb09e2 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java @@ -769,7 +769,7 @@ public void testResumingStreamReadFromCheckpoint() throws Exception { } @TestTemplate - public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { + public void resumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { File writerCheckpoint = temp.resolve("upgrade-checkpoint").toFile(); File output = temp.resolve("upgrade-output").toFile(); List checkpointed = Lists.newArrayList(new SimpleRecord(1, "checkpointed")); @@ -799,6 +799,7 @@ public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { // This append is deliberately newer than the checkpoint. A resumed stream must still process // it after walking through the v3 manifest-rewrite snapshot. appendData(pendingBeforeUpgrade); + table.refresh(); List activeFiles = Lists.newArrayList(); try (CloseableIterable tasks = table.newScan().planFiles()) { @@ -824,8 +825,7 @@ public void testResumeCheckpointAfterV2ToV3RowIdAssignment() throws Exception { Transaction upgrade = table.newTransaction(); upgrade.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit(); - RewriteManifests rewrite = - upgrade.rewriteManifests().setAssignedRowIdRange(0L, nextRowIdExclusive); + RewriteManifests rewrite = upgrade.rewriteManifests().reserveRowIdRange(0L, nextRowIdExclusive); oldDataManifests.forEach(rewrite::deleteManifest); replacementManifests.forEach(rewrite::addManifest); rewrite.commit(); @@ -861,7 +861,6 @@ private ManifestFile rewriteManifestWithFirstRowIds( table.io(), table.specs(), table.io().newOutputFile(outputPath), - rangeStart, file -> firstRowIds.get(file.location()), rangeStart, rangeEnd);