Skip to content

Core: Fix row lineage last updated sequence inheritance #17039

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
amogh-jahagirdar merged 5 commits into from
Jul 9, 2026
Merged

Core: Fix row lineage last updated sequence inheritance #17039

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
78e103a
Spec: Fix row lineage sequence number inheritance
wgtmac Jul 2, 2026
6a6f812
Core: Fix row lineage last updated sequence inheritance
wgtmac Jul 2, 2026
a489551
Spark: Test row lineage sequence after upgrade
wgtmac Jul 3, 2026
8fe315a
Spark: Simplify row lineage upgrade test
wgtmac Jul 4, 2026
da082da
Spark: Clarify row lineage upgrade test
wgtmac Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/src/main/java/org/apache/iceberg/util/PartitionUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ private PartitionUtil() {}

idToConstant.put(
MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId(),
convertConstant.apply(Types.LongType.get(), task.file().fileSequenceNumber()));
convertConstant.apply(Types.LongType.get(), task.file().dataSequenceNumber()));

// add _file
idToConstant.put(
Expand Down
41 changes: 41 additions & 0 deletions core/src/test/java/org/apache/iceberg/TestRowLineageAssignment.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,14 @@
import java.io.UncheckedIOException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.iceberg.data.Record;
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.types.Types;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.util.PartitionUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -722,6 +724,45 @@ public void testRowDeltaAssignmentAfterUpgrade(@TempDir File altLocation) {
assertThat(manifests.get(1).path()).isEqualTo(existingManifests.get(1).path());
}

@Test
public void lastUpdatedAfterUpgrade(@TempDir File altLocation) throws IOException {

@amogh-jahagirdar amogh-jahagirdar Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

testLastUpdatedDataSequencePreservedAfterUpgrade? I think we should still prefix with test in this class to be consistent.

@RussellSpitzer RussellSpitzer Jul 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I thought we agreed we would no longer be restrained by the sins of our past :)

More seriously I did think we decided we were doing updates for new code. Just trying not to blanket modify everything when it doesn't matter

@amogh-jahagirdar amogh-jahagirdar Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

got it, in that case i'm good with how it is!

BaseTable upgradeTable =
TestTables.create(altLocation, "test_upgrade", SCHEMA, PartitionSpec.unpartitioned(), 2);

upgradeTable.newAppend().appendFile(FILE_A).commit();
Snapshot originalSnapshot = upgradeTable.currentSnapshot();
long originalSequenceNumber = originalSnapshot.sequenceNumber();

upgradeTable
.newRewrite()
.validateFromSnapshot(originalSnapshot.snapshotId())
.rewriteFiles(Set.of(FILE_A), Set.of(FILE_B), originalSequenceNumber)
.commit();

TestTables.upgrade(altLocation, "test_upgrade", 3);
upgradeTable.refresh();

// Assign row IDs to the upgraded metadata tree without rewriting the data file.
upgradeTable.newFastAppend().commit();

try (CloseableIterable<FileScanTask> tasks = upgradeTable.newScan().planFiles()) {
FileScanTask task = Iterables.getOnlyElement(tasks);

assertThat(task.file().location()).isEqualTo(FILE_B.location());
assertThat(task.file().dataSequenceNumber()).isEqualTo(originalSequenceNumber);
assertThat(task.file().fileSequenceNumber()).isGreaterThan(originalSequenceNumber);
assertThat(task.file().firstRowId()).isNotNull();

// Scan planning projects row lineage metadata columns through constantsMap.
// The upgraded data file is not rewritten, so validate the value readers see.
assertThat(
PartitionUtil.constantsMap(task)

@amogh-jahagirdar amogh-jahagirdar Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

It is a bit strange in these tests to have inspect the constantsMap but I get it because the sequence number is inherited metadata and not present on any of the in memory representations available. Maybe just an inline comment?

.get(MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId()))
.as("Last updated should preserve the original data sequence after upgrade")
.isEqualTo(originalSequenceNumber);
}
}

@Test
public void testUpgradeAssignmentWithManifestCompaction(@TempDir File altLocation) {
// create a non-empty upgrade table with FILE_A
Expand Down
43 changes: 43 additions & 0 deletions ...v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2067,6 +2067,49 @@ public void testUnpartitionedRewriteDataFilesPreservesLineage() throws NoSuchTab
assertEquals("Rows must match", expectedRecordsWithLineage, actualRecordsWithLineage);
}

@TestTemplate
public void testUpgradePreservesDataSequence() throws NoSuchTableException {
assumeThat(formatVersion).isEqualTo(2);

Table table = createTable();
writeRecords(2, 4);
table.refresh();
shouldHaveFiles(table, 2);
long committedDataSequence = table.currentSnapshot().sequenceNumber();

Result result = basicRewrite(table).execute();
assertThat(result.rewrittenDataFilesCount()).isEqualTo(2);
assertThat(result.addedDataFilesCount()).isOne();
table.refresh();
shouldHaveFiles(table, 1);

DataFile compactedFile = Iterables.getOnlyElement(currentDataFiles(table));
long dataSequenceNumber = compactedFile.dataSequenceNumber();
assertThat(dataSequenceNumber)
.as("Compaction must preserve the original data sequence number")
.isEqualTo(committedDataSequence);
assertThat(compactedFile.fileSequenceNumber())
.as("Compaction must bump the file sequence above the preserved data sequence")
.isGreaterThan(dataSequenceNumber);

table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
table.rewriteManifests().rewriteIf(manifest -> true).commit();
table.refresh();

List<Object[]> expectedLineage =
Lists.newArrayList(
row(0L, committedDataSequence, ANY, ANY, ANY),
row(1L, committedDataSequence, ANY, ANY, ANY),
row(2L, committedDataSequence, ANY, ANY, ANY),
row(3L, committedDataSequence, ANY, ANY, ANY));

assertEquals(
"First snapshot after upgrade to v3 assigns row IDs and inherits the committed data"
+ " sequence as _last_updated_sequence_number",
expectedLineage,
currentDataWithLineage());
}

@TestTemplate
public void testRewriteDataFilesPreservesLineage() throws NoSuchTableException {
assumeThat(formatVersion).isGreaterThan(2);
Expand Down
43 changes: 43 additions & 0 deletions ...v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2069,6 +2069,49 @@ public void testUnpartitionedRewriteDataFilesPreservesLineage() throws NoSuchTab
assertEquals("Rows must match", expectedRecordsWithLineage, actualRecordsWithLineage);
}

@TestTemplate

@amogh-jahagirdar amogh-jahagirdar Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I think this looks right but I'd cleanup the empty append and do a more realistic write, I'd also just use an explicit assertion on the records (just a bit more readable/less opaque)

+  @TestTemplate
+  public void testLastUpdatedSequenceInheritedFromDataSequenceAfterUpgrade() {
+    // Reproduces the inheritance bug for a v2-origin file carried into v3.
+    assumeThat(formatVersion).isEqualTo(2);
+
+    Table table = createTable();
+    writeRecords(2 /* files */, 4 /* records */);
+    table.refresh();
+    shouldHaveFiles(table, 2);
+    long originalSequenceNumber = table.currentSnapshot().sequenceNumber();
+    // Perform a compaction before upgrade which preserves the original sequence number
+    basicRewrite(table).execute();
+    table.refresh();
+    shouldHaveFiles(table, 1);
+
+    DataFile compacted = Iterables.getOnlyElement(currentDataFiles(table));
+    assertThat(compacted.dataSequenceNumber()).isEqualTo(originalSequenceNumber);
+    assertThat(compacted.fileSequenceNumber())
+        .as("Compaction must bump the file sequence number above the preserved data sequence")
+        .isGreaterThan(originalSequenceNumber);
+
+    // Upgrade to v3 so the row lineage metadata columns are exposed on read.
+    table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
+    table.refresh();
+
+    writeRecords(1 /* files */, 4 /* records */);
+    table.refresh();
+    long appendSequenceNumber = table.currentSnapshot().sequenceNumber();
+
+    // currentDataWithLineage() projects (_row_id, _last_updated_sequence_number, *) ordered by
+    // _row_id; the trailing data columns are irrelevant here, so match them with ANY.
+    List<Object[]> actualLineage = currentDataWithLineage();
+
+    List<Object[]> expectedLineage =
+        Lists.newArrayList(
+            // The post-upgrade append writes a new file, which is assigned the first row IDs (0-3)
+            // and carries the append commit's sequence number as its last-updated sequence.
+            row(0L, appendSequenceNumber, ANY, ANY, ANY),
+            row(1L, appendSequenceNumber, ANY, ANY, ANY),
+            row(2L, appendSequenceNumber, ANY, ANY, ANY),
+            row(3L, appendSequenceNumber, ANY, ANY, ANY),
+            row(4L, originalSequenceNumber, ANY, ANY, ANY),
+            row(5L, originalSequenceNumber, ANY, ANY, ANY),
+            row(6L, originalSequenceNumber, ANY, ANY, ANY),
+            row(7L, originalSequenceNumber, ANY, ANY, ANY));
+
+    assertEquals(
+        "Carried-over rows must inherit the data sequence",
+        expectedLineage,
+        actualLineage);
+  }
+

@amogh-jahagirdar amogh-jahagirdar Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Feel free to keep your test name as is, just what i had locally

@amogh-jahagirdar amogh-jahagirdar Jul 3, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

actually my test is arguably a bit brittle to how we organize the manifests in the manifest list, today it works but if there's a reordering the test would fail needlessly since the first row ID assignment would be different but still valid. Maybe we should key off a logical row to make it more resilient (but it becomes a bit more complex)

@amogh-jahagirdar amogh-jahagirdar Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Ok here's a more minimal test I came up with so we don't even have to append any records. We can just do a compact + upgrade + rewrite manifests to move the new file into a single v3 manifest. since there's a single data file/single manifest we can then reliably assert the row IDs regardless of any change in the implementation.

+  @TestTemplate
+  public void testLastUpdatedSequenceInheritedFromDataSequenceAfterUpgrade() {
+    assumeThat(formatVersion).isEqualTo(2);
+
+    Table table = createTable();
+
+    // Two data files in a single v2 commit; both share data sequence number 1.
+    writeRecords(2 /* files */, 4 /* records */);
+    table.refresh();
+    shouldHaveFiles(table, 2);
+    long committedDataSequence = table.currentSnapshot().sequenceNumber();
+
+    basicRewrite(table).execute();
+    table.refresh();
+    shouldHaveFiles(table, 1);
+
+    DataFile compacted = Iterables.getOnlyElement(currentDataFiles(table));
+    long dataSequence = compacted.dataSequenceNumber();
+    long fileSequence = compacted.fileSequenceNumber();
+    assertThat(dataSequence)
+        .as("Compaction must preserve the original data sequence number")
+        .isEqualTo(committedDataSequence);
+    assertThat(fileSequence)
+        .as("Compaction must bump the file sequence above the preserved data sequence")

+    shouldHaveFiles(table, 1);
+
+    DataFile compacted = Iterables.getOnlyElement(currentDataFiles(table));
+    long dataSequence = compacted.dataSequenceNumber();
+    long fileSequence = compacted.fileSequenceNumber();
+    assertThat(dataSequence)
+        .as("Compaction must preserve the original data sequence number")
+        .isEqualTo(committedDataSequence);
+    assertThat(fileSequence)
+        .as("Compaction must bump the file sequence above the preserved data sequence")
+        .isGreaterThan(dataSequence);
+
+    // Upgrade to v3, then move the carried-over file into a new v3 manifest 
+    table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
+    table.rewriteManifests().rewriteIf(manifest -> true).commit();
+    table.refresh();
+
+    List<Object[]> expected =
+        Lists.newArrayList(
+            row(0L, dataSequence, ANY, ANY, ANY),
+            row(1L, dataSequence, ANY, ANY, ANY),
+            row(2L, dataSequence, ANY, ANY, ANY),
+            row(3L, dataSequence, ANY, ANY, ANY));
+
+    assertEquals(
+        "Row IDs must be 0..3 and last-updated must inherit the data sequence",
+        expected,
+        currentDataWithLineage());
+  }

@wgtmac wgtmac Jul 6, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Changed according to your last comment.

public void testUpgradePreservesDataSequence() throws NoSuchTableException {
assumeThat(formatVersion).isEqualTo(2);

Table table = createTable();
writeRecords(2, 4);
table.refresh();
shouldHaveFiles(table, 2);
long committedDataSequence = table.currentSnapshot().sequenceNumber();

Result result = basicRewrite(table).execute();
assertThat(result.rewrittenDataFilesCount()).isEqualTo(2);
assertThat(result.addedDataFilesCount()).isOne();
table.refresh();
shouldHaveFiles(table, 1);

DataFile compactedFile = Iterables.getOnlyElement(currentDataFiles(table));
long dataSequenceNumber = compactedFile.dataSequenceNumber();
assertThat(dataSequenceNumber)
.as("Compaction must preserve the original data sequence number")
.isEqualTo(committedDataSequence);
assertThat(compactedFile.fileSequenceNumber())
.as("Compaction must bump the file sequence above the preserved data sequence")
.isGreaterThan(dataSequenceNumber);

table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
table.rewriteManifests().rewriteIf(manifest -> true).commit();
table.refresh();

List<Object[]> expectedLineage =
Lists.newArrayList(
row(0L, committedDataSequence, ANY, ANY, ANY),
row(1L, committedDataSequence, ANY, ANY, ANY),
row(2L, committedDataSequence, ANY, ANY, ANY),
row(3L, committedDataSequence, ANY, ANY, ANY));

assertEquals(
"First snapshot after upgrade to v3 assigns row IDs and inherits the committed data"
+ " sequence as _last_updated_sequence_number",
expectedLineage,
currentDataWithLineage());
}

@TestTemplate
public void testRewriteDataFilesPreservesLineage() throws NoSuchTableException {
assumeThat(formatVersion).isGreaterThan(2);
Expand Down
43 changes: 43 additions & 0 deletions ...v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2069,6 +2069,49 @@ public void testUnpartitionedRewriteDataFilesPreservesLineage() throws NoSuchTab
assertEquals("Rows must match", expectedRecordsWithLineage, actualRecordsWithLineage);
}

@TestTemplate
public void testUpgradePreservesDataSequence() throws NoSuchTableException {
assumeThat(formatVersion).isEqualTo(2);

Table table = createTable();
writeRecords(2, 4);
table.refresh();
shouldHaveFiles(table, 2);
long committedDataSequence = table.currentSnapshot().sequenceNumber();

Result result = basicRewrite(table).execute();
assertThat(result.rewrittenDataFilesCount()).isEqualTo(2);
assertThat(result.addedDataFilesCount()).isOne();
table.refresh();
shouldHaveFiles(table, 1);

DataFile compactedFile = Iterables.getOnlyElement(currentDataFiles(table));
long dataSequenceNumber = compactedFile.dataSequenceNumber();
assertThat(dataSequenceNumber)
.as("Compaction must preserve the original data sequence number")
.isEqualTo(committedDataSequence);
assertThat(compactedFile.fileSequenceNumber())
.as("Compaction must bump the file sequence above the preserved data sequence")
.isGreaterThan(dataSequenceNumber);

table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
table.rewriteManifests().rewriteIf(manifest -> true).commit();
table.refresh();

List<Object[]> expectedLineage =
Lists.newArrayList(
row(0L, committedDataSequence, ANY, ANY, ANY),
row(1L, committedDataSequence, ANY, ANY, ANY),
row(2L, committedDataSequence, ANY, ANY, ANY),
row(3L, committedDataSequence, ANY, ANY, ANY));

assertEquals(
"First snapshot after upgrade to v3 assigns row IDs and inherits the committed data"
+ " sequence as _last_updated_sequence_number",
expectedLineage,
currentDataWithLineage());
}

@TestTemplate
public void testRewriteDataFilesPreservesLineage() throws NoSuchTableException {
assumeThat(formatVersion).isGreaterThan(2);
Expand Down
Loading