This is an automated email from the ASF dual-hosted git repository.
amogh-jahagirdar 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 94f318dc99 Core: Scope V4ManifestReader row ID assignment to each
iterator (#18795)
94f318dc99 is described below
commit 94f318dc9949473c7f39faff98667a35823dca24
Author: Anoop Johnson <[email protected]>
AuthorDate: Thu Nov 5 14:32:23 2026 -0800
Core: Scope V4ManifestReader row ID assignment to each iterator (#18795)
---
.../java/org/apache/iceberg/V4ManifestReader.java | 50 ++++++++++++++--------
.../org/apache/iceberg/TestV4ManifestReader.java | 27 ++++++++++++
2 files changed, 60 insertions(+), 17 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
index a91c32d4ff..42b426ae07 100644
--- a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
+++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
@@ -22,6 +22,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.function.Function;
import org.apache.iceberg.expressions.Binder;
import org.apache.iceberg.expressions.Evaluator;
import org.apache.iceberg.expressions.Expression;
@@ -80,8 +81,6 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
private final boolean isUncommitted;
private final ScanMetrics scanMetrics;
- private Long nextRowId = null;
-
private V4ManifestReader(
ManifestFile manifest,
FileIO io,
@@ -104,9 +103,6 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
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
@@ -118,8 +114,7 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
public CloseableIterator<TrackedFile> iterator() {
// first row ID assignment must happen first to maintain consistent
assignment when files are
// removed or filtered
- CloseableIterable<TrackedFile> files =
- CloseableIterable.transform(open(), this::applyInheritance);
+ CloseableIterable<TrackedFile> files = CloseableIterable.transform(open(),
applyInheritance());
files = CloseableIterable.transform(files, this::projectPartition);
@@ -151,20 +146,41 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
return CloseableIterable.transform(files, this::copyResolved).iterator();
}
- private TrackedFile applyInheritance(TrackedFile file) {
- // the reader uses TrackingStruct to read tracking so this cast is safe
- TrackingStruct tracking = (TrackingStruct) file.tracking();
+ private Function<TrackedFile, TrackedFile> applyInheritance() {
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 -> {
+ // the reader uses TrackingStruct to read tracking so this cast is safe
+ ((TrackingStruct) file.tracking()).inherit(manifest.snapshotId());
+ return file;
+ };
+ }
+
+ Long firstRowId = manifest.firstRowId();
+ if (firstRowId != null) {
+ return new Function<>() {
+ private long nextRowId = firstRowId;
+
+ @Override
+ public TrackedFile apply(TrackedFile file) {
+ TrackingStruct tracking = (TrackingStruct) file.tracking();
+ tracking.inherit(manifest.snapshotId(), manifest.sequenceNumber());
+ if (tracking.assignFirstRowId(nextRowId)) {
+ nextRowId += file.recordCount();
+ }
+
+ return file;
+ }
+ };
}
- return file;
+ // committed manifest with null manifest-level first row ID (pre-v3
upgrade path)
+ return file -> {
+ TrackingStruct tracking = (TrackingStruct) file.tracking();
+ tracking.inherit(manifest.snapshotId(), manifest.sequenceNumber());
+ tracking.assignFirstRowId(null);
+ return file;
+ };
}
private TrackedFile projectPartition(TrackedFile file) {
diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
index 864a0e1eec..1dff57e04c 100644
--- a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
+++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
@@ -54,6 +54,7 @@ import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.geospatial.GeospatialBound;
import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.io.CloseableIterator;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.metrics.DefaultMetricsContext;
import org.apache.iceberg.metrics.ScanMetrics;
@@ -720,6 +721,32 @@ class TestV4ManifestReader {
.isEqualTo(expected);
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ void inheritanceFirstRowIdIsIndependentBetweenIterators(FileFormat format)
throws IOException {
+ TrackedFile fileA = dataFile(ADDED_TRACKING,
"s3://bucket/table/file-a.parquet");
+ TrackedFile fileB = dataFile(ADDED_TRACKING,
"s3://bucket/table/file-b.parquet");
+
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE, ImmutableList.of(fileA,
fileB));
+ when(manifest.firstRowId()).thenReturn(10_000L);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+
+ try (V4ManifestReader reader = builder.build();
+ CloseableIterator<TrackedFile> first = reader.iterator();
+ CloseableIterator<TrackedFile> second = reader.iterator()) {
+ assertThat(first.next().tracking().firstRowId()).isEqualTo(10_000L);
+ assertThat(second.next().tracking().firstRowId())
+ .as("a second iterator assigns from the manifest's first row ID, not
the first's counter")
+ .isEqualTo(10_000L);
+ assertThat(first.next().tracking().firstRowId()).isEqualTo(10_000L +
RECORD_COUNT);
+ assertThat(second.next().tracking().firstRowId()).isEqualTo(10_000L +
RECORD_COUNT);
+ }
+ }
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void inheritanceUncommittedSkipsFirstRowId(FileFormat format) throws
IOException {
