You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2026/06/11 21:27:19 UTC
(iceberg) branch main updated: Core: Add EntryStatus.MODIFIED and TrackingBuilder status derivation (#16689)
This is an automated email from the ASF dual-hosted git repository.
rdblue pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new cc7fec9f28 Core: Add EntryStatus.MODIFIED and TrackingBuilder status derivation (#16689)
cc7fec9f28 is described below
commit cc7fec9f28c68aa182090ca71953f0674cc45a5d
Author: Anoop Johnson <an...@apache.org>
AuthorDate: Fri Jun 12 02:57:11 2026 +0530
Core: Add EntryStatus.MODIFIED and TrackingBuilder status derivation (#16689)
Co-authored-by: Steven Zhen Wu <st...@gmail.com>
---
.../main/java/org/apache/iceberg/EntryStatus.java | 22 +++-
.../src/main/java/org/apache/iceberg/Tracking.java | 8 +-
.../java/org/apache/iceberg/TrackingBuilder.java | 28 ++---
.../java/org/apache/iceberg/TestEntryStatus.java | 10 ++
.../org/apache/iceberg/TestTrackingStruct.java | 130 +++++++++++++--------
5 files changed, 126 insertions(+), 72 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/EntryStatus.java b/core/src/main/java/org/apache/iceberg/EntryStatus.java
index ceabeb5624..ccc0d23ef5 100644
--- a/core/src/main/java/org/apache/iceberg/EntryStatus.java
+++ b/core/src/main/java/org/apache/iceberg/EntryStatus.java
@@ -20,24 +20,34 @@ package org.apache.iceberg;
/** Status of an entry in a manifest file. */
enum EntryStatus {
- EXISTING(0),
- ADDED(1),
- DELETED(2),
- /** Indicates an entry that has been replaced by a column update or DV change. Added in v4. */
- REPLACED(3);
+ EXISTING(0, true),
+ ADDED(1, true),
+ DELETED(2, false),
+ /**
+ * The old (replaced) state of an entry that has been modified. Paired with MODIFIED. Added in v4.
+ */
+ REPLACED(3, false),
+ /** The new (live) state of an entry that has been modified. Added in v4. */
+ MODIFIED(4, true);
private static final EntryStatus[] VALUES = EntryStatus.values();
private final int id;
+ private final boolean live;
- EntryStatus(int id) {
+ EntryStatus(int id, boolean live) {
this.id = id;
+ this.live = live;
}
public int id() {
return id;
}
+ boolean isLive() {
+ return live;
+ }
+
static EntryStatus fromId(int id) {
return VALUES[id];
}
diff --git a/core/src/main/java/org/apache/iceberg/Tracking.java b/core/src/main/java/org/apache/iceberg/Tracking.java
index 8003ed82ea..fcdc4e50b2 100644
--- a/core/src/main/java/org/apache/iceberg/Tracking.java
+++ b/core/src/main/java/org/apache/iceberg/Tracking.java
@@ -28,13 +28,13 @@ interface Tracking {
0,
"status",
Types.IntegerType.get(),
- "Entry status: 0=existing, 1=added, 2=deleted, 3=replaced");
+ "Entry status: 0=existing, 1=added, 2=deleted, 3=replaced, 4=modified");
Types.NestedField SNAPSHOT_ID =
Types.NestedField.optional(
1,
"snapshot_id",
Types.LongType.get(),
- "Snapshot ID where the file was added or deleted");
+ "Snapshot ID where the file was added, deleted, or replaced");
Types.NestedField SEQUENCE_NUMBER =
Types.NestedField.optional(
3, "sequence_number", Types.LongType.get(), "Data sequence number of the file");
@@ -83,10 +83,10 @@ interface Tracking {
/** Returns whether this entry is live. */
default boolean isLive() {
- return status() == EntryStatus.ADDED || status() == EntryStatus.EXISTING;
+ return status().isLive();
}
- /** Returns the snapshot ID where the file was added or deleted. */
+ /** Returns the snapshot ID where the file was added, deleted, or replaced. */
Long snapshotId();
/** Returns the data sequence number of the file. */
diff --git a/core/src/main/java/org/apache/iceberg/TrackingBuilder.java b/core/src/main/java/org/apache/iceberg/TrackingBuilder.java
index bf3cb3a066..c5a11bc53c 100644
--- a/core/src/main/java/org/apache/iceberg/TrackingBuilder.java
+++ b/core/src/main/java/org/apache/iceberg/TrackingBuilder.java
@@ -23,13 +23,12 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.util.ByteBuffers;
class TrackingBuilder {
- private final EntryStatus status;
+ private final long newSnapshotId;
private final Long snapshotId;
private final Long dataSequenceNumber;
private final Long fileSequenceNumber;
private final Long firstRowId;
- // ID of the snapshot in which the new Tracking instance will be committed.
- private final long newSnapshotId;
+ private EntryStatus status;
private Long dvSnapshotId;
private byte[] deletedPositions;
private byte[] replacedPositions;
@@ -101,33 +100,34 @@ class TrackingBuilder {
/** Indicates that the DV has been updated for the new Tracking. */
TrackingBuilder dvUpdated() {
- // DV applies to data files; deleted/replaced positions apply to manifest files
Preconditions.checkState(
deletedPositions == null && replacedPositions == null,
"Cannot mark DV updated on a manifest entry (deleted/replaced positions are set)");
this.dvSnapshotId = newSnapshotId;
+ if (status == EntryStatus.EXISTING) {
+ this.status = EntryStatus.MODIFIED;
+ }
+
return this;
}
+ /** Sets the positions deleted by this commit for a manifest entry. */
TrackingBuilder deletedPositions(ByteBuffer positions) {
Preconditions.checkState(
- status == EntryStatus.EXISTING, "Cannot set deleted positions on %s entry", status);
- // DV applies to data files; deleted positions apply to manifest files
- Preconditions.checkState(
- dvSnapshotId == null,
- "Cannot set deleted positions on a data file entry (DV snapshot ID is set)");
+ status != EntryStatus.ADDED, "Cannot set deleted positions on ADDED entry");
this.deletedPositions = ByteBuffers.toByteArray(positions);
+ this.dvSnapshotId = newSnapshotId;
+ this.status = EntryStatus.MODIFIED;
return this;
}
+ /** Sets the positions replaced by this commit for a manifest entry. */
TrackingBuilder replacedPositions(ByteBuffer positions) {
Preconditions.checkState(
- status == EntryStatus.EXISTING, "Cannot set replaced positions on %s entry", status);
- // DV applies to data files; replaced positions apply to manifest files
- Preconditions.checkState(
- dvSnapshotId == null,
- "Cannot set replaced positions on a data file entry (DV snapshot ID is set)");
+ status != EntryStatus.ADDED, "Cannot set replaced positions on ADDED entry");
this.replacedPositions = ByteBuffers.toByteArray(positions);
+ this.dvSnapshotId = newSnapshotId;
+ this.status = EntryStatus.MODIFIED;
return this;
}
diff --git a/core/src/test/java/org/apache/iceberg/TestEntryStatus.java b/core/src/test/java/org/apache/iceberg/TestEntryStatus.java
index c395cdcece..a103bd98cf 100644
--- a/core/src/test/java/org/apache/iceberg/TestEntryStatus.java
+++ b/core/src/test/java/org/apache/iceberg/TestEntryStatus.java
@@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.util.stream.IntStream;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
@@ -45,4 +46,13 @@ class TestEntryStatus {
.isInstanceOf(ArrayIndexOutOfBoundsException.class)
.hasMessageContaining(String.valueOf(id));
}
+
+ @Test
+ void isLive() {
+ assertThat(EntryStatus.EXISTING.isLive()).isTrue();
+ assertThat(EntryStatus.ADDED.isLive()).isTrue();
+ assertThat(EntryStatus.MODIFIED.isLive()).isTrue();
+ assertThat(EntryStatus.DELETED.isLive()).isFalse();
+ assertThat(EntryStatus.REPLACED.isLive()).isFalse();
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
index 0bbd8b3480..0d1803022a 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
@@ -82,11 +82,11 @@ class TestTrackingStruct {
Tracking copy = tracking.copy();
- assertThat(copy.status()).isEqualTo(EntryStatus.EXISTING);
+ assertThat(copy.status()).isEqualTo(EntryStatus.MODIFIED);
assertThat(copy.snapshotId()).isEqualTo(tracking.snapshotId());
assertThat(copy.dataSequenceNumber()).isEqualTo(tracking.dataSequenceNumber());
assertThat(copy.fileSequenceNumber()).isEqualTo(tracking.fileSequenceNumber());
- assertThat(copy.dvSnapshotId()).isNull();
+ assertThat(copy.dvSnapshotId()).isEqualTo(tracking.dvSnapshotId());
assertThat(copy.firstRowId()).isEqualTo(tracking.firstRowId());
assertThat(copy.deletedPositions()).isEqualTo(tracking.deletedPositions());
assertThat(copy.replacedPositions()).isEqualTo(tracking.replacedPositions());
@@ -104,23 +104,6 @@ class TestTrackingStruct {
assertThat(tracking.status()).isEqualTo(status);
}
- @Test
- void testIsLive() {
- TrackingStruct tracking = new TrackingStruct(Tracking.schema());
-
- tracking.set(STATUS_ORDINAL, EntryStatus.ADDED.id());
- assertThat(tracking.isLive()).isTrue();
-
- tracking.set(STATUS_ORDINAL, EntryStatus.EXISTING.id());
- assertThat(tracking.isLive()).isTrue();
-
- tracking.set(STATUS_ORDINAL, EntryStatus.DELETED.id());
- assertThat(tracking.isLive()).isFalse();
-
- tracking.set(STATUS_ORDINAL, EntryStatus.REPLACED.id());
- assertThat(tracking.isLive()).isFalse();
- }
-
@Test
void testInheritSnapshotId() {
TrackingStruct tracking = new TrackingStruct(Tracking.schema());
@@ -158,6 +141,20 @@ class TestTrackingStruct {
assertThat(tracking.fileSequenceNumber()).isEqualTo(6L);
}
+ @Test
+ void testDoNotInheritSequenceNumberForModifiedEntries() {
+ TrackingStruct tracking = new TrackingStruct(Tracking.schema());
+ tracking.set(STATUS_ORDINAL, EntryStatus.MODIFIED.id());
+ tracking.set(DATA_SEQUENCE_NUMBER_ORDINAL, 5L);
+ tracking.set(FILE_SEQUENCE_NUMBER_ORDINAL, 6L);
+
+ tracking.inheritFrom(createManifestTracking(100L, 60L));
+
+ // sequence numbers are not inherited for MODIFIED entries
+ assertThat(tracking.dataSequenceNumber()).isEqualTo(5L);
+ assertThat(tracking.fileSequenceNumber()).isEqualTo(6L);
+ }
+
@Test
void testExplicitValuesOverrideInheritance() {
TrackingStruct tracking = new TrackingStruct(Tracking.schema());
@@ -224,7 +221,7 @@ class TestTrackingStruct {
}
@Test
- void testAddedBuilder() {
+ void testAddedWithSameCommitDvStaysAdded() {
Tracking tracking = TrackingBuilder.added(42L).dvUpdated().build();
assertThat(tracking.status()).isEqualTo(EntryStatus.ADDED);
@@ -300,9 +297,15 @@ class TestTrackingStruct {
}
@Test
- void testExistingBuilderAllowsDvMutation() {
- Tracking existing = TrackingBuilder.from(sourceTracking(), 999L).dvUpdated().build();
- assertThat(existing.dvSnapshotId()).isEqualTo(999L);
+ void testDvUpdatedProducesModifiedAndAdvancesDvSnapshotId() {
+ Tracking source = sourceTracking();
+ Tracking modified = TrackingBuilder.from(source, 999L).dvUpdated().build();
+
+ assertThat(modified.status()).isEqualTo(EntryStatus.MODIFIED);
+ // the entry snapshot id is preserved so we still know when the base file was added
+ assertThat(modified.snapshotId()).isEqualTo(source.snapshotId()).isNotEqualTo(999L);
+ // only the DV snapshot id advances to the commit snapshot
+ assertThat(modified.dvSnapshotId()).isEqualTo(999L);
}
@Test
@@ -319,27 +322,20 @@ class TestTrackingStruct {
}
@Test
- void testDvSnapshotIdAndManifestDVPositionsAreMutuallyExclusive() {
- // sourceTracking has dvSnapshotId=43, inherited by existing(source)
+ void testDvUpdatedRejectedWhenManifestPositionsSet() {
assertThatThrownBy(
() ->
- TrackingBuilder.from(sourceTracking(), 1L)
- .deletedPositions(ByteBuffer.wrap(new byte[] {1})))
- .isInstanceOf(IllegalStateException.class)
- .hasMessage("Cannot set deleted positions on a data file entry (DV snapshot ID is set)");
-
- assertThatThrownBy(
- () ->
- TrackingBuilder.from(sourceTracking(), 1L)
- .replacedPositions(ByteBuffer.wrap(new byte[] {1})))
+ TrackingBuilder.from(manifestSourceTracking(), 999L)
+ .deletedPositions(ByteBuffer.wrap(new byte[] {1}))
+ .dvUpdated())
.isInstanceOf(IllegalStateException.class)
- .hasMessage("Cannot set replaced positions on a data file entry (DV snapshot ID is set)");
+ .hasMessage(
+ "Cannot mark DV updated on a manifest entry (deleted/replaced positions are set)");
- // Setting MDV positions first then dvUpdated is also rejected
assertThatThrownBy(
() ->
- TrackingBuilder.from(manifestSourceTracking(), 1L)
- .deletedPositions(ByteBuffer.wrap(new byte[] {1}))
+ TrackingBuilder.from(manifestSourceTracking(), 999L)
+ .replacedPositions(ByteBuffer.wrap(new byte[] {1}))
.dvUpdated())
.isInstanceOf(IllegalStateException.class)
.hasMessage(
@@ -433,6 +429,46 @@ class TestTrackingStruct {
assertThat(replaced.snapshotId()).isEqualTo(999L);
}
+ @Test
+ void testExistingPreservesSourceSnapshotId() {
+ Tracking source = sourceTracking();
+ Tracking existing = TrackingBuilder.from(source, 999L).build();
+ assertThat(existing.status()).isEqualTo(EntryStatus.EXISTING);
+ assertThat(existing.snapshotId()).isEqualTo(source.snapshotId()).isNotEqualTo(999L);
+ }
+
+ @Test
+ void testCarryForwardFromModifiedSourceChangesToExisting() {
+ Tracking modifiedSource = sourceTrackingWithStatus(EntryStatus.MODIFIED);
+ Tracking carried = TrackingBuilder.from(modifiedSource, 999L).build();
+ assertThat(carried.status()).isEqualTo(EntryStatus.EXISTING);
+ assertThat(carried.snapshotId()).isEqualTo(modifiedSource.snapshotId()).isNotEqualTo(999L);
+ assertThat(carried.dvSnapshotId()).isEqualTo(modifiedSource.dvSnapshotId()).isNotEqualTo(999L);
+ assertThat(carried.dataSequenceNumber()).isEqualTo(modifiedSource.dataSequenceNumber());
+ assertThat(carried.fileSequenceNumber()).isEqualTo(modifiedSource.fileSequenceNumber());
+ assertThat(carried.firstRowId()).isEqualTo(modifiedSource.firstRowId());
+ }
+
+ @Test
+ void testManifestDVPositionsProduceModified() {
+ ByteBuffer deletedBytes = ByteBuffer.wrap(new byte[] {1, 2});
+
+ Tracking addedSource = manifestSourceTracking();
+ Tracking modified =
+ TrackingBuilder.from(addedSource, 999L).deletedPositions(deletedBytes).build();
+ assertThat(modified.status()).isEqualTo(EntryStatus.MODIFIED);
+ // the entry snapshot id is preserved; only the DV snapshot id advances to the commit snapshot
+ assertThat(modified.snapshotId()).isEqualTo(addedSource.snapshotId()).isNotEqualTo(999L);
+ assertThat(modified.dvSnapshotId()).isEqualTo(999L);
+ assertThat(modified.deletedPositions()).isEqualTo(deletedBytes);
+ }
+
+ @Test
+ void testIsLiveDelegatesToStatus() {
+ assertThat(sourceTrackingWithStatus(EntryStatus.ADDED).isLive()).isTrue();
+ assertThat(sourceTrackingWithStatus(EntryStatus.DELETED).isLive()).isFalse();
+ }
+
@Test
void testInternalSetIgnoresUnknownOrdinal() {
TrackingStruct tracking = new TrackingStruct(Tracking.schema());
@@ -479,8 +515,7 @@ class TestTrackingStruct {
}
@Test
- void testAddedWithDvSnapshotIdJavaSerializationRoundTrip()
- throws IOException, ClassNotFoundException {
+ void testJavaSerializationRoundTripForDataFile() throws IOException, ClassNotFoundException {
Tracking tracking = TrackingBuilder.added(42L).dvUpdated().build();
Tracking deserialized = TestHelpers.roundTripSerialize(tracking);
@@ -493,8 +528,7 @@ class TestTrackingStruct {
}
@Test
- void testExistingWithManifestDVPositionsJavaSerializationRoundTrip()
- throws IOException, ClassNotFoundException {
+ void testJavaSerializationRoundTripForManifest() throws IOException, ClassNotFoundException {
Tracking tracking =
TrackingBuilder.from(manifestSourceTracking(), 1L)
.deletedPositions(ByteBuffer.wrap(new byte[] {1, 2}))
@@ -503,14 +537,14 @@ class TestTrackingStruct {
Tracking deserialized = TestHelpers.roundTripSerialize(tracking);
- assertThat(deserialized.status()).isEqualTo(EntryStatus.EXISTING);
- assertThat(deserialized.dvSnapshotId()).isNull();
+ assertThat(deserialized.status()).isEqualTo(EntryStatus.MODIFIED);
+ assertThat(deserialized.dvSnapshotId()).isEqualTo(1L);
assertThat(deserialized.deletedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2}));
assertThat(deserialized.replacedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {3, 4}));
}
@Test
- void testAddedWithDvSnapshotIdKryoSerializationRoundTrip() throws IOException {
+ void testKryoSerializationRoundTripForDataFile() throws IOException {
Tracking tracking = TrackingBuilder.added(42L).dvUpdated().build();
Tracking deserialized = TestHelpers.KryoHelpers.roundTripSerialize(tracking);
@@ -523,7 +557,7 @@ class TestTrackingStruct {
}
@Test
- void testExistingWithManifestDVPositionsKryoSerializationRoundTrip() throws IOException {
+ void testKryoSerializationRoundTripForManifest() throws IOException {
Tracking tracking =
TrackingBuilder.from(manifestSourceTracking(), 1L)
.deletedPositions(ByteBuffer.wrap(new byte[] {1, 2}))
@@ -532,8 +566,8 @@ class TestTrackingStruct {
Tracking deserialized = TestHelpers.KryoHelpers.roundTripSerialize(tracking);
- assertThat(deserialized.status()).isEqualTo(EntryStatus.EXISTING);
- assertThat(deserialized.dvSnapshotId()).isNull();
+ assertThat(deserialized.status()).isEqualTo(EntryStatus.MODIFIED);
+ assertThat(deserialized.dvSnapshotId()).isEqualTo(1L);
assertThat(deserialized.deletedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2}));
assertThat(deserialized.replacedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {3, 4}));
}