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 3957d26daa Core: Implement v4 first row ID assignment (#18225)
3957d26daa is described below
commit 3957d26daa535eb5f03e101f1210ee234e978f9a
Author: Ryan Blue <[email protected]>
AuthorDate: Thu Sep 24 11:59:34 2026 -0700
Core: Implement v4 first row ID assignment (#18225)
---
.../java/org/apache/iceberg/TrackingStruct.java | 47 +++++++
.../java/org/apache/iceberg/V4ManifestReader.java | 36 ++++-
.../org/apache/iceberg/TestTrackingStruct.java | 65 +++++++++
.../org/apache/iceberg/TestV4ManifestReader.java | 153 +++++++++++++++++++++
4 files changed, 297 insertions(+), 4 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
index 860fdda881..e5b3517111 100644
--- a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
+++ b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
@@ -116,6 +116,25 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
this.replacedPositions = replacedPositions;
}
+ /**
+ * Inherit a commit's snapshot ID if it is unset.
+ *
+ * <p>Used when rewriting uncommitted manifests before the sequence number
is determined.
+ *
+ * @param manifestSnapshotId the manifest's snapshot ID
+ */
+ void inherit(long manifestSnapshotId) {
+ if (null == snapshotId) {
+ this.snapshotId = manifestSnapshotId;
+ }
+ }
+
+ /**
+ * Inherit a commit's snapshot ID and sequence number if either is unset.
+ *
+ * @param manifestSnapshotId the manifest's snapshot ID
+ * @param manifestSeqNumber the manifest's sequence number, determined by a
successful commit
+ */
void inherit(long manifestSnapshotId, long manifestSeqNumber) {
if (null == snapshotId) {
this.snapshotId = manifestSnapshotId;
@@ -132,6 +151,34 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
}
}
+ /**
+ * Assign the first row ID to the given next row ID if it is unassigned.
+ *
+ * <p>If the {@code nextRowId} is null, the first row ID will also be set to
null. This is used
+ * when reading snapshots from older format versions that do not have
assigned row IDs.
+ *
+ * @param nextRowId the next row ID to assign, or null when reading v2 or
earlier snapshots
+ * @return true if the first row ID is assigned a non-null value
+ */
+ boolean assignFirstRowId(Long nextRowId) {
+ if (null == nextRowId) {
+ // null manifest first row ID from pre-v3 upgrade path
+ // defensively set the first row ID for every entry to be null
+ this.firstRowId = null;
+ return false;
+ }
+
+ // ADDED will assign for new files
+ // EXISTING will assign to handle existing files upgraded from pre-v3
+ // MODIFIED must assign when a pre-v3 EXISTING file is modified in the
first v4 commit
+ if (status.isLive() && null == firstRowId) {
+ this.firstRowId = nextRowId;
+ return true;
+ }
+
+ return false;
+ }
+
@Override
public EntryStatus status() {
return status;
diff --git a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
index 6ee7486ff5..1122cb5750 100644
--- a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
+++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
@@ -59,7 +59,12 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
static Builder builder(
ManifestFile manifest, FileIO io, Schema tableSchema, Map<Integer,
PartitionSpec> specsById) {
- return new Builder(manifest, io, tableSchema, specsById);
+ return new Builder(manifest, io, tableSchema, specsById, false /*
committed */);
+ }
+
+ static Builder uncommitted(
+ ManifestFile manifest, FileIO io, Schema tableSchema, Map<Integer,
PartitionSpec> specsById) {
+ return new Builder(manifest, io, tableSchema, specsById, true /*
uncommitted */);
}
private final ManifestFile manifest;
@@ -70,8 +75,11 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
private final Map<Integer, Pair<Evaluator, StructProjection>>
partitionFilters; // by spec ID
private final boolean includeAll;
private final Set<Integer> requestedStatsFieldIds;
+ private final boolean isUncommitted;
private final ScanMetrics scanMetrics;
+ private Long nextRowId = null;
+
private V4ManifestReader(
ManifestFile manifest,
FileIO io,
@@ -81,6 +89,7 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
Map<Integer, Pair<Evaluator, StructProjection>> partitionFilters,
boolean includeAll,
Set<Integer> requestedStatsFieldIds,
+ boolean isUncommitted,
ScanMetrics scanMetrics) {
this.manifest = manifest;
this.io = io;
@@ -90,7 +99,11 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
this.partitionFilters = partitionFilters;
this.includeAll = includeAll;
this.requestedStatsFieldIds = requestedStatsFieldIds;
+ this.isUncommitted = isUncommitted;
this.scanMetrics = scanMetrics;
+
+ // initialize the next row ID to the manifest's first row ID
+ this.nextRowId = manifest.firstRowId();
}
@VisibleForTesting
@@ -130,7 +143,16 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
private TrackedFile applyInheritance(TrackedFile file) {
// the reader uses TrackingStruct to read tracking so this cast is safe
TrackingStruct tracking = (TrackingStruct) file.tracking();
- tracking.inherit(manifest.snapshotId(), manifest.sequenceNumber());
+ if (isUncommitted) {
+ // uncommitted files cannot have a sequence number or assign first row ID
+ tracking.inherit(manifest.snapshotId());
+ } else {
+ tracking.inherit(manifest.snapshotId(), manifest.sequenceNumber());
+ if (tracking.assignFirstRowId(nextRowId)) {
+ this.nextRowId += file.recordCount();
+ }
+ }
+
return file;
}
@@ -235,6 +257,7 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
private final Schema tableSchema;
private final Types.StructType unionPartitionType;
private final Map<Integer, PartitionSpec> specsById;
+ private final boolean isUncommitted;
private String tableLocation = null;
private Expression rowFilter = Expressions.alwaysTrue();
private boolean caseSensitive = true;
@@ -250,7 +273,8 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
ManifestFile manifest,
FileIO io,
Schema tableSchema,
- Map<Integer, PartitionSpec> specsById) {
+ Map<Integer, PartitionSpec> specsById,
+ boolean isUncommitted) {
Preconditions.checkArgument(tableSchema != null, "Invalid table schema:
null");
int formatVersion = manifest.formatVersion();
Preconditions.checkArgument(
@@ -275,8 +299,9 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
this.manifest = manifest;
this.io = io;
this.tableSchema = tableSchema;
- this.specsById = specsById;
this.unionPartitionType =
Partitioning.unionPartitionTypes(specsById.values());
+ this.specsById = specsById;
+ this.isUncommitted = isUncommitted;
}
/**
@@ -311,6 +336,8 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
/** Configures the reader to select the minimal fields needed for scan
planning. */
Builder forScanPlanning() {
+ Preconditions.checkState(
+ !isUncommitted, "Cannot plan scan using uncommitted manifest: %s",
manifest.path());
Preconditions.checkState(
requestedColumns == null && requestedProjection == null,
"Cannot use forScanPlanning() with select(Iterable<String>) or
project(Schema)");
@@ -395,6 +422,7 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
partitionFilters,
includeAll,
requestedStatsFieldIds,
+ isUncommitted,
scanMetrics);
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
index d5f0e8d425..253384a683 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
@@ -133,6 +133,17 @@ class TestTrackingStruct {
assertThat(copy.replacedPositions().array()).isNotSameAs(tracking.replacedPositions().array());
}
+ @ParameterizedTest
+ @EnumSource(EntryStatus.class)
+ void inheritSnapshotIdOnly(EntryStatus status) {
+ TrackingStruct tracking = new TrackingStruct(status, null, null, null,
null, null, null, null);
+
+ tracking.inherit(100L);
+
+ assertThat(tracking.snapshotId()).isEqualTo(100L);
+ assertThat(tracking.dataSequenceNumber()).isNull();
+ }
+
@ParameterizedTest
@EnumSource(EntryStatus.class)
void inheritSnapshotId(EntryStatus status) {
@@ -193,6 +204,60 @@ class TestTrackingStruct {
assertThat(tracking.fileSequenceNumber()).isNull();
}
+ private static final List<EntryStatus> ASSIGNING_STATUSES =
+ List.of(EntryStatus.ADDED, EntryStatus.EXISTING, EntryStatus.MODIFIED);
+ private static final List<EntryStatus> NON_ASSIGNING_STATUSES =
+ List.of(EntryStatus.DELETED, EntryStatus.REPLACED);
+
+ @ParameterizedTest
+ @FieldSource("ASSIGNING_STATUSES")
+ void assignmentFirstRowIdAssigned(EntryStatus status) {
+ TrackingStruct tracking = new TrackingStruct(status, 42L, null, null,
null, null, null, null);
+
+ assertThat(tracking.assignFirstRowId(10_000L)).as("Should return true when
assigned").isTrue();
+
+ assertThat(tracking.firstRowId()).isEqualTo(10_000L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("NON_ASSIGNING_STATUSES")
+ void assignmentFirstRowIdNotAssigned(EntryStatus status) {
+ TrackingStruct tracking = new TrackingStruct(status, 42L, null, null,
null, null, null, null);
+
+ assertThat(tracking.assignFirstRowId(10_000L))
+ .as("Should return false when not assigned")
+ .isFalse();
+
+ assertThat(tracking.firstRowId()).isNull();
+ }
+
+ @ParameterizedTest
+ @EnumSource(EntryStatus.class)
+ void assignmentFirstRowIdAlreadyAssigned(EntryStatus status) {
+ TrackingStruct tracking = new TrackingStruct(status, 42L, null, null,
null, 5_000L, null, null);
+
+ assertThat(tracking.assignFirstRowId(10_000L))
+ .as("Should return false when not assigned")
+ .isFalse();
+
+ assertThat(tracking.firstRowId())
+ .as("Should use existing value when already assigned")
+ .isEqualTo(5_000L);
+ }
+
+ @ParameterizedTest
+ @EnumSource(EntryStatus.class)
+ void assignmentNullFirstRowId(EntryStatus status) {
+ TrackingStruct tracking =
+ new TrackingStruct(status, 42L, null, null, null, 10_000L, null, null);
+
+ assertThat(tracking.assignFirstRowId(null))
+ .as("Should return false when not assigned")
+ .isFalse();
+
+ assertThat(tracking.firstRowId()).as("Should override a value with
null").isNull();
+ }
+
@ParameterizedTest
@EnumSource(EntryStatus.class)
void isLiveDelegatesToStatus(EntryStatus status) {
diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
index c7c6f76a4f..0a9c55ced6 100644
--- a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
+++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
@@ -345,6 +345,44 @@ class TestV4ManifestReader {
.containsExactlyElementsOf(files);
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceUncommittedOnlyInheritsSnapshotId(FileFormat format)
throws IOException {
+ Tracking trackingWithSnapshotId =
+ new TrackingStruct(EntryStatus.ADDED, 1234567L, null, null, null,
null, null, null);
+ TrackedFile withSnapshotId =
+ unpartitionedDataFile(trackingWithSnapshotId,
"s3://bucket/table/file-b.parquet");
+ Tracking trackingWithoutSnapshotId =
+ new TrackingStruct(EntryStatus.ADDED, null, null, null, null, null,
null, null);
+ TrackedFile withoutSnapshotId =
+ unpartitionedDataFile(trackingWithoutSnapshotId,
"s3://bucket/table/file-a.parquet");
+
+ ManifestFile manifest =
+ writeManifest(
+ format, UNPARTITIONED_TYPE, ImmutableList.of(withSnapshotId,
withoutSnapshotId));
+
+ assertThat(manifest.sequenceNumber())
+ .as("Manifest metadata has a sequence number that will not be
inherited")
+ .isNotNull();
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.uncommitted(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> actual = read(builder);
+
+ assertThat(actual)
+ .extracting(file -> file.tracking().snapshotId())
+ .containsExactly(1234567L, SNAPSHOT_ID);
+ assertThat(actual)
+ .as("Uncommitted files do not inherit data sequence number")
+ .extracting(file -> file.tracking().dataSequenceNumber())
+ .containsOnlyNulls();
+ assertThat(actual)
+ .as("Uncommitted files do not inherit file sequence number")
+ .extracting(file -> file.tracking().fileSequenceNumber())
+ .containsOnlyNulls();
+ }
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void inheritanceSnapshotId(FileFormat format) throws IOException {
@@ -447,6 +485,121 @@ class TestV4ManifestReader {
.containsExactly(MANIFEST_SEQ, 500L);
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceFirstRowId(FileFormat format) throws IOException {
+ Tracking trackingWithoutFirstRowId =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null, null,
null, null, null);
+ TrackedFile withoutFirstRowId =
+ unpartitionedDataFile(trackingWithoutFirstRowId,
"s3://bucket/table/file-a.parquet");
+ Tracking trackingWithFirstRowId =
+ new TrackingStruct(EntryStatus.EXISTING, SNAPSHOT_ID, null, null,
null, 5_000L, null, null);
+ TrackedFile withFirstRowId =
+ unpartitionedDataFile(trackingWithFirstRowId,
"s3://bucket/table/file-c.parquet");
+
+ ManifestFile manifest =
+ writeManifest(
+ format, UNPARTITIONED_TYPE, ImmutableList.of(withoutFirstRowId,
withFirstRowId));
+ when(manifest.firstRowId()).thenReturn(10_000L);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> actual = read(builder);
+
+ assertThat(actual)
+ .extracting(file -> file.tracking().firstRowId())
+ .containsExactly(10_000L, 5_000L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceFirstRowIdIncrementsByRecordCount(FileFormat format)
throws IOException {
+ Tracking trackingWithoutFirstRowId =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null, null,
null, null, null);
+ Tracking trackingWithFirstRowId =
+ new TrackingStruct(EntryStatus.EXISTING, SNAPSHOT_ID, null, null,
null, 5_000L, null, null);
+
+ TrackedFile fileA =
+ unpartitionedDataFile(trackingWithoutFirstRowId,
"s3://bucket/table/file-a.parquet");
+ TrackedFile fileB =
+ unpartitionedDataFile(trackingWithoutFirstRowId,
"s3://bucket/table/file-b.parquet");
+ TrackedFile fileC =
+ unpartitionedDataFile(trackingWithoutFirstRowId,
"s3://bucket/table/file-c.parquet");
+ TrackedFile fileD =
+ unpartitionedDataFile(trackingWithFirstRowId,
"s3://bucket/table/file-d.parquet");
+ TrackedFile fileE =
+ unpartitionedDataFile(trackingWithoutFirstRowId,
"s3://bucket/table/file-e.parquet");
+
+ ManifestFile manifest =
+ writeManifest(
+ format, UNPARTITIONED_TYPE, ImmutableList.of(fileA, fileB, fileC,
fileD, fileE));
+ when(manifest.firstRowId()).thenReturn(10_000L);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> actual = read(builder);
+
+ assertThat(actual)
+ .as("Each assignment should increment by rocord_count=100")
+ .extracting(file -> file.tracking().firstRowId())
+ .containsExactly(10_000L, 10_100L, 10_200L, 5_000L, 10_300L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceUncommittedSkipsFirstRowId(FileFormat format) throws
IOException {
+ Tracking trackingWithoutFirstRowId =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null, null,
null, null, null);
+ TrackedFile withoutFirstRowId =
+ unpartitionedDataFile(trackingWithoutFirstRowId,
"s3://bucket/table/file-a.parquet");
+ Tracking trackingWithFirstRowId =
+ new TrackingStruct(EntryStatus.EXISTING, SNAPSHOT_ID, null, null,
null, 5_000L, null, null);
+ TrackedFile withFirstRowId =
+ unpartitionedDataFile(trackingWithFirstRowId,
"s3://bucket/table/file-c.parquet");
+
+ ManifestFile manifest =
+ writeManifest(
+ format, UNPARTITIONED_TYPE, ImmutableList.of(withoutFirstRowId,
withFirstRowId));
+ when(manifest.firstRowId()).thenReturn(10_000L);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.uncommitted(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> actual = read(builder);
+
+ assertThat(actual)
+ .extracting(file -> file.tracking().firstRowId())
+ .containsExactly(null, 5_000L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceFirstRowIdNullFromUpgradedTable(FileFormat format)
throws IOException {
+ Tracking trackingWithoutFirstRowId =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null, null,
null, null, null);
+ TrackedFile withoutFirstRowId =
+ unpartitionedDataFile(trackingWithoutFirstRowId,
"s3://bucket/table/file-a.parquet");
+ Tracking trackingWithFirstRowId =
+ new TrackingStruct(EntryStatus.EXISTING, SNAPSHOT_ID, null, null,
null, 5_000L, null, null);
+ TrackedFile withFirstRowId =
+ unpartitionedDataFile(trackingWithFirstRowId,
"s3://bucket/table/file-c.parquet");
+
+ ManifestFile manifest =
+ writeManifest(
+ format, UNPARTITIONED_TYPE, ImmutableList.of(withoutFirstRowId,
withFirstRowId));
+ // first row ID is null in upgraded tables, which is propagated to all
files
+ when(manifest.firstRowId()).thenReturn(null);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> actual = read(builder);
+
+ assertThat(actual).extracting(file ->
file.tracking().firstRowId()).containsOnlyNulls();
+ }
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void inheritanceManifestLocationAndPosition(FileFormat format) throws
IOException {
