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 17cf4b4a58 Core: Add inheritance to V4ManifestReader (#18171)
17cf4b4a58 is described below
commit 17cf4b4a585a7a8801d15b328399141f5289bd7d
Author: Ryan Blue <[email protected]>
AuthorDate: Tue Sep 22 14:24:59 2026 -0700
Core: Add inheritance to V4ManifestReader (#18171)
---
.../java/org/apache/iceberg/TrackingStruct.java | 43 +++----
.../java/org/apache/iceberg/V4ManifestReader.java | 23 ++--
.../main/java/org/apache/iceberg/avro/Avro.java | 6 +
.../org/apache/iceberg/avro/InternalReader.java | 11 +-
.../apache/iceberg/TestTrackedFileAdapters.java | 21 ++--
.../org/apache/iceberg/TestTrackingStruct.java | 125 ++++++++-------------
.../org/apache/iceberg/TestV4ManifestReader.java | 115 ++++++++++++++++++-
.../java/org/apache/iceberg/InternalParquet.java | 3 +-
.../iceberg/data/parquet/InternalReader.java | 13 ++-
9 files changed, 224 insertions(+), 136 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
index 283e0a51ff..860fdda881 100644
--- a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
+++ b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
@@ -21,10 +21,8 @@ package org.apache.iceberg;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.Arrays;
-import java.util.Objects;
import org.apache.iceberg.avro.SupportsIndexProjection;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.ByteBuffers;
@@ -41,6 +39,7 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
Tracking.FIRST_ROW_ID,
Tracking.DELETED_POSITIONS,
Tracking.REPLACED_POSITIONS,
+ MetadataColumns.FILE_PATH,
MetadataColumns.ROW_POSITION);
// tracking fields read on the scan path; row_position backs manifestPos.
@@ -52,6 +51,7 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
Tracking.SEQUENCE_NUMBER,
Tracking.FILE_SEQUENCE_NUMBER,
Tracking.FIRST_ROW_ID,
+ MetadataColumns.FILE_PATH,
MetadataColumns.ROW_POSITION);
private EntryStatus status = null;
@@ -116,34 +116,20 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
this.replacedPositions = replacedPositions;
}
- void inheritFrom(Tracking manifestTracking) {
- if (manifestTracking != null) {
- if (snapshotId == null) {
- this.snapshotId = manifestTracking.snapshotId();
- }
-
- // manifests do not distinguish between data and file sequence numbers
- Preconditions.checkArgument(
- Objects.equals(
- manifestTracking.dataSequenceNumber(),
manifestTracking.fileSequenceNumber()),
- "Manifest data and file sequence numbers must be equal, got %s and
%s",
- manifestTracking.dataSequenceNumber(),
- manifestTracking.fileSequenceNumber());
+ void inherit(long manifestSnapshotId, long manifestSeqNumber) {
+ if (null == snapshotId) {
+ this.snapshotId = manifestSnapshotId;
+ }
- if (status == EntryStatus.ADDED) {
- if (dataSequenceNumber == null) {
- this.dataSequenceNumber = manifestTracking.fileSequenceNumber();
- }
+ boolean isAdded = status == EntryStatus.ADDED;
- if (fileSequenceNumber == null) {
- this.fileSequenceNumber = manifestTracking.fileSequenceNumber();
- }
- }
+ if (null == dataSequenceNumber && (isAdded || manifestSeqNumber == 0)) {
+ this.dataSequenceNumber = manifestSeqNumber;
}
- }
- void setManifestLocation(String location) {
- this.manifestLocation = location;
+ if (null == fileSequenceNumber && (isAdded || manifestSeqNumber == 0)) {
+ this.fileSequenceNumber = manifestSeqNumber;
+ }
}
@Override
@@ -225,6 +211,8 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
case 7:
return replacedPositions();
case 8:
+ return manifestLocation;
+ case 9:
return manifestPos;
default:
throw new UnsupportedOperationException("Unknown field ordinal: " +
pos);
@@ -259,6 +247,9 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
this.replacedPositions = ByteBuffers.toByteArray((ByteBuffer) value);
break;
case 8:
+ this.manifestLocation = (String) value;
+ break;
+ case 9:
this.manifestPos = (long) value;
break;
default:
diff --git a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
index 48843a1d83..6ee7486ff5 100644
--- a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
+++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
@@ -52,7 +52,7 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
private static final int SUPPORTED_FORMAT_VERSION = 4;
private static final Set<Integer> REQUIRED_COLUMN_IDS =
ImmutableSet.of(
- Tracking.STATUS.fieldId(), // needed to filter live files
+ Tracking.STATUS.fieldId(), // needed to filter live files and for
inheritance
MetadataColumns.ROW_POSITION.fieldId(), // needed to apply metadata
DVs
TrackedFile.CONTENT_TYPE.fieldId(), // needed for content filtering
TrackedFile.RECORD_COUNT.fieldId()); // needed for first_row_id
assignment and filtering
@@ -98,10 +98,10 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
return readSchema;
}
- /** Returns copies of the tracked files that match this reader's configured
filters. */
@Override
public CloseableIterator<TrackedFile> iterator() {
- CloseableIterable<TrackedFile> files = CloseableIterable.transform(open(),
this::prepare);
+ CloseableIterable<TrackedFile> files =
+ CloseableIterable.transform(open(), this::applyInheritance);
if (!includeAll) {
files = CloseableIterable.filter(files, file ->
file.tracking().isLive());
@@ -127,6 +127,13 @@ 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();
+ tracking.inherit(manifest.snapshotId(), manifest.sequenceNumber());
+ return file;
+ }
+
private boolean matchesPartition(TrackedFile trackedFile) {
Integer specId = trackedFile.specId();
if (specId == null) {
@@ -196,16 +203,6 @@ class V4ManifestReader extends CloseableGroup implements
CloseableIterable<Track
return reader;
}
- private TrackedFile prepare(TrackedFile trackedFile) {
- Tracking tracking = trackedFile.tracking();
- // manifestLocation is not stored in the manifest; the reader fills it in
- if (tracking instanceof TrackingStruct) {
- ((TrackingStruct) tracking).setManifestLocation(manifest.path());
- }
-
- return trackedFile;
- }
-
// resolves stored locations against the table location
private TrackedFile copyResolved(TrackedFile trackedFile) {
TrackedFileStruct copy =
diff --git a/core/src/main/java/org/apache/iceberg/avro/Avro.java
b/core/src/main/java/org/apache/iceberg/avro/Avro.java
index 4eca2d7816..b5bc6a798b 100644
--- a/core/src/main/java/org/apache/iceberg/avro/Avro.java
+++ b/core/src/main/java/org/apache/iceberg/avro/Avro.java
@@ -47,6 +47,7 @@ import org.apache.avro.specific.SpecificData;
import org.apache.iceberg.FieldMetrics;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.InternalData;
+import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.MetricsConfig;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.SchemaParser;
@@ -716,6 +717,11 @@ public class Avro {
reader = (DatumReader<D>) defaultCreateReaderFunc.apply(schema);
}
+ if (reader instanceof InternalReader) {
+ ((InternalReader<D>) reader)
+ .setConstant(MetadataColumns.FILE_PATH_COLUMN_ID, file.location());
+ }
+
if (reader instanceof SupportsCustomRecords) {
((SupportsCustomRecords) reader).setClassLoader(loader);
((SupportsCustomRecords) reader).setRenames(renames);
diff --git a/core/src/main/java/org/apache/iceberg/avro/InternalReader.java
b/core/src/main/java/org/apache/iceberg/avro/InternalReader.java
index af3c4f1a82..e52b5d2439 100644
--- a/core/src/main/java/org/apache/iceberg/avro/InternalReader.java
+++ b/core/src/main/java/org/apache/iceberg/avro/InternalReader.java
@@ -28,7 +28,6 @@ import org.apache.avro.Schema;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
import org.apache.iceberg.StructLike;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
@@ -47,7 +46,7 @@ public class InternalReader<T> implements DatumReader<T>,
SupportsRowPosition, S
private final Types.StructType expectedType;
private final Map<Integer, Class<? extends StructLike>> typeMap =
Maps.newHashMap();
- private final Map<Integer, Object> idToConstant = ImmutableMap.of();
+ private final Map<Integer, Object> idToConstant = Maps.newHashMap();
private Schema fileSchema = null;
private ValueReader<T> reader = null;
@@ -70,6 +69,14 @@ public class InternalReader<T> implements DatumReader<T>,
SupportsRowPosition, S
AccessByID.instance());
}
+ public void setConstants(Map<Integer, Object> constantMap) {
+ this.idToConstant.putAll(constantMap);
+ }
+
+ public void setConstant(int fieldId, Object value) {
+ this.idToConstant.put(fieldId, value);
+ }
+
@Override
public void setSchema(Schema schema) {
this.fileSchema = schema;
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
index ad1a957215..a515f0fef0 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
@@ -67,8 +67,9 @@ class TestTrackedFileAdapters {
.build();
private static final PartitionData PARTITION = partition("books");
- // manifestPos is populated by readers using the setter with the position of
the field.
- private static final int MANIFEST_POS_ORDINAL =
Tracking.schema().fields().size();
+ // these are populated by readers using the setter with the position of the
field.
+ private static final int MANIFEST_LOCATION_ORDINAL =
Tracking.schema().fields().size();
+ private static final int MANIFEST_POSITION_ORDINAL =
Tracking.schema().fields().size() + 1;
private static final Schema TABLE_SCHEMA =
new Schema(
@@ -134,8 +135,8 @@ class TestTrackedFileAdapters {
FIRST_ROW_ID,
null,
null);
- tracking.setManifestLocation(MANIFEST_LOCATION);
- tracking.set(MANIFEST_POS_ORDINAL, MANIFEST_POS);
+ tracking.set(MANIFEST_LOCATION_ORDINAL, MANIFEST_LOCATION);
+ tracking.set(MANIFEST_POSITION_ORDINAL, MANIFEST_POS);
DeletionVector dv = mock(DeletionVector.class);
TrackedFile file =
@@ -215,8 +216,8 @@ class TestTrackedFileAdapters {
FIRST_ROW_ID,
null,
null);
- tracking.setManifestLocation(MANIFEST_LOCATION);
- tracking.set(MANIFEST_POS_ORDINAL, MANIFEST_POS);
+ tracking.set(MANIFEST_LOCATION_ORDINAL, MANIFEST_LOCATION);
+ tracking.set(MANIFEST_POSITION_ORDINAL, MANIFEST_POS);
TrackedFile file =
new TrackedFileStruct(
@@ -304,8 +305,8 @@ class TestTrackedFileAdapters {
FIRST_ROW_ID,
null,
null);
- tracking.setManifestLocation(MANIFEST_LOCATION);
- tracking.set(MANIFEST_POS_ORDINAL, MANIFEST_POS);
+ tracking.set(MANIFEST_LOCATION_ORDINAL, MANIFEST_LOCATION);
+ tracking.set(MANIFEST_POSITION_ORDINAL, MANIFEST_POS);
TrackedFile file =
new TrackedFileStruct(
@@ -428,8 +429,8 @@ class TestTrackedFileAdapters {
FIRST_ROW_ID,
null,
null);
- tracking.setManifestLocation(MANIFEST_LOCATION);
- tracking.set(MANIFEST_POS_ORDINAL, MANIFEST_POS);
+ tracking.set(MANIFEST_LOCATION_ORDINAL, MANIFEST_LOCATION);
+ tracking.set(MANIFEST_POSITION_ORDINAL, MANIFEST_POS);
DeletionVector dv =
DeletionVectorStruct.builder()
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
index 5eab8c215d..d5f0e8d425 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java
@@ -19,7 +19,6 @@
package org.apache.iceberg;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -28,20 +27,22 @@ import org.apache.iceberg.types.Types;
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.FieldSource;
import org.junit.jupiter.params.provider.MethodSource;
class TestTrackingStruct {
private static final byte[] DELETED_POSITIONS = new byte[] {1, 2};
private static final byte[] REPLACED_POSITIONS = new byte[] {3, 4};
- private static final int MANIFEST_POSITION_ORDINAL =
Tracking.schema().fields().size();
+ private static final int MANIFEST_LOCATION_ORDINAL =
Tracking.schema().fields().size();
+ private static final int MANIFEST_POSITION_ORDINAL =
Tracking.schema().fields().size() + 1;
@Test
void fieldAccess() {
TrackingStruct tracking =
new TrackingStruct(
EntryStatus.ADDED, 42L, 10L, 11L, 43L, 1000L, DELETED_POSITIONS,
REPLACED_POSITIONS);
- tracking.setManifestLocation("manifest-location");
+ tracking.set(MANIFEST_LOCATION_ORDINAL, "manifest-location");
tracking.set(MANIFEST_POSITION_ORDINAL, 7L);
assertThat(tracking.status()).isEqualTo(EntryStatus.ADDED);
@@ -68,6 +69,7 @@ class TestTrackingStruct {
tracking.set(pos("first_row_id"), 1000L);
tracking.set(pos("deleted_positions"), ByteBuffer.wrap(DELETED_POSITIONS));
tracking.set(pos("replaced_positions"),
ByteBuffer.wrap(REPLACED_POSITIONS));
+ tracking.set(MANIFEST_LOCATION_ORDINAL, "manifest-location");
tracking.set(MANIFEST_POSITION_ORDINAL, 7L);
assertThat(tracking.status()).isEqualTo(EntryStatus.ADDED);
@@ -78,6 +80,7 @@ class TestTrackingStruct {
assertThat(tracking.firstRowId()).isEqualTo(1000L);
assertThat(tracking.deletedPositions()).isEqualTo(ByteBuffer.wrap(DELETED_POSITIONS));
assertThat(tracking.replacedPositions()).isEqualTo(ByteBuffer.wrap(REPLACED_POSITIONS));
+ assertThat(tracking.manifestLocation()).isEqualTo("manifest-location");
assertThat(tracking.manifestPos()).isEqualTo(7L);
}
@@ -86,7 +89,7 @@ class TestTrackingStruct {
TrackingStruct tracking =
new TrackingStruct(
EntryStatus.ADDED, 42L, 10L, 11L, 43L, 1000L, DELETED_POSITIONS,
REPLACED_POSITIONS);
- tracking.setManifestLocation("manifest-location");
+ tracking.set(MANIFEST_LOCATION_ORDINAL, "manifest-location");
tracking.set(MANIFEST_POSITION_ORDINAL, 7L);
assertThat(tracking.get(pos("status"),
Integer.class)).isEqualTo(EntryStatus.ADDED.id());
@@ -99,6 +102,8 @@ class TestTrackingStruct {
.isEqualTo(ByteBuffer.wrap(DELETED_POSITIONS));
assertThat(tracking.get(pos("replaced_positions"), ByteBuffer.class))
.isEqualTo(ByteBuffer.wrap(REPLACED_POSITIONS));
+ assertThat(tracking.get(MANIFEST_LOCATION_ORDINAL, String.class))
+ .isEqualTo("manifest-location");
assertThat(tracking.get(MANIFEST_POSITION_ORDINAL,
Long.class)).isEqualTo(7L);
}
@@ -107,7 +112,7 @@ class TestTrackingStruct {
TrackingStruct tracking =
new TrackingStruct(
EntryStatus.MODIFIED, 42L, 10L, 11L, 43L, 1000L,
DELETED_POSITIONS, REPLACED_POSITIONS);
- tracking.setManifestLocation("manifest-location");
+ tracking.set(MANIFEST_LOCATION_ORDINAL, "manifest-location");
tracking.set(MANIFEST_POSITION_ORDINAL, 7L);
Tracking copy = tracking.copy();
@@ -128,108 +133,66 @@ class TestTrackingStruct {
assertThat(copy.replacedPositions().array()).isNotSameAs(tracking.replacedPositions().array());
}
- @Test
- void inheritSnapshotId() {
- TrackingStruct tracking =
- new TrackingStruct(EntryStatus.ADDED, null, null, null, null, null,
null, null);
+ @ParameterizedTest
+ @EnumSource(EntryStatus.class)
+ void inheritSnapshotId(EntryStatus status) {
+ TrackingStruct tracking = new TrackingStruct(status, null, null, null,
null, null, null, null);
- tracking.inheritFrom(createManifestTracking(100L, 60L));
+ tracking.inherit(100L, 60L);
- // snapshotId is null, should inherit from manifest
assertThat(tracking.snapshotId()).isEqualTo(100L);
}
- @Test
- void inheritSequenceNumberForAddedEntries() {
- TrackingStruct tracking =
- new TrackingStruct(EntryStatus.ADDED, 42L, null, null, null, null,
null, null);
-
- tracking.inheritFrom(createManifestTracking(100L, 60L));
-
- // sequence numbers are null and status is ADDED, should inherit
- assertThat(tracking.dataSequenceNumber()).isEqualTo(60L);
- assertThat(tracking.fileSequenceNumber()).isEqualTo(60L);
- }
-
- @Test
- void doNotInheritSequenceNumberForExistingEntries() {
- TrackingStruct tracking =
- new TrackingStruct(EntryStatus.EXISTING, 42L, 5L, 6L, null, null,
null, null);
-
- tracking.inheritFrom(createManifestTracking(100L, 60L));
-
- // sequence numbers are not inherited for EXISTING entries
- assertThat(tracking.dataSequenceNumber()).isEqualTo(5L);
- assertThat(tracking.fileSequenceNumber()).isEqualTo(6L);
- }
-
- @Test
- void doNotInheritSequenceNumberForModifiedEntries() {
- TrackingStruct tracking =
- new TrackingStruct(EntryStatus.MODIFIED, 42L, 5L, 6L, null, null,
null, null);
-
- 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 explicitValuesOverrideInheritance() {
- TrackingStruct tracking =
- new TrackingStruct(EntryStatus.ADDED, 200L, 75L, 76L, null, null,
null, null);
+ @ParameterizedTest
+ @EnumSource(EntryStatus.class)
+ void inheritancePreservesExplicitValues(EntryStatus status) {
+ TrackingStruct tracking = new TrackingStruct(status, 200L, 75L, 76L, null,
null, null, null);
- tracking.inheritFrom(createManifestTracking(100L, 60L));
+ tracking.inherit(100L, 60L);
- // explicit values should take precedence
assertThat(tracking.snapshotId()).isEqualTo(200L);
assertThat(tracking.dataSequenceNumber()).isEqualTo(75L);
assertThat(tracking.fileSequenceNumber()).isEqualTo(76L);
}
- @Test
- void inheritFromRejectsUnequalSequenceNumbers() {
- TrackingStruct tracking =
- new TrackingStruct(EntryStatus.ADDED, 42L, null, null, null, null,
null, null);
+ @ParameterizedTest
+ @EnumSource(EntryStatus.class)
+ void inheritanceAllStatusesInheritSeq0(EntryStatus status) {
+ // Sequence number 0 is inherited for all statuses for reading pre-seq v1
tables
+ TrackingStruct tracking = new TrackingStruct(status, 42L, null, null,
null, null, null, null);
- TrackingStruct manifestTracking =
- new TrackingStruct(EntryStatus.ADDED, 100L, 50L, 60L, null, null,
null, null);
+ tracking.inherit(100L, 0L);
- assertThatThrownBy(() -> tracking.inheritFrom(manifestTracking))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessage("Manifest data and file sequence numbers must be equal,
got 50 and 60");
+ assertThat(tracking.dataSequenceNumber()).isEqualTo(0L);
+ assertThat(tracking.fileSequenceNumber()).isEqualTo(0L);
}
@Test
- void noDefaultingWithoutInheritance() {
+ void inheritanceAddedEntriesInheritSequenceNumber() {
TrackingStruct tracking =
- new TrackingStruct(EntryStatus.ADDED, null, null, null, null, null,
null, null);
+ new TrackingStruct(EntryStatus.ADDED, 42L, null, null, null, null,
null, null);
- // no inheritance, nulls stay null
- assertThat(tracking.snapshotId()).isNull();
- assertThat(tracking.dataSequenceNumber()).isNull();
- assertThat(tracking.fileSequenceNumber()).isNull();
+ tracking.inherit(100L, 60L);
+
+ assertThat(tracking.dataSequenceNumber()).isEqualTo(60L);
+ assertThat(tracking.fileSequenceNumber()).isEqualTo(60L);
}
- @Test
- void inheritFromNullIsNoOp() {
- TrackingStruct tracking =
- new TrackingStruct(EntryStatus.ADDED, null, null, null, null, null,
null, null);
+ private static final List<EntryStatus> NON_INHERITING_STATUSES =
+ List.of(
+ EntryStatus.EXISTING, EntryStatus.MODIFIED, EntryStatus.DELETED,
EntryStatus.REPLACED);
+
+ @ParameterizedTest
+ @FieldSource("NON_INHERITING_STATUSES")
+ void inheritanceWithNonInheritingStatus(EntryStatus status) {
+ TrackingStruct tracking = new TrackingStruct(status, 42L, null, null,
null, null, null, null);
- tracking.inheritFrom(null);
+ tracking.inherit(100L, 60L);
- // null source is a no-op; all unset fields stay null
- assertThat(tracking.snapshotId()).isNull();
assertThat(tracking.dataSequenceNumber()).isNull();
assertThat(tracking.fileSequenceNumber()).isNull();
}
- private static Tracking createManifestTracking(long snapshotId, long
sequenceNumber) {
- return new TrackingStruct(
- EntryStatus.ADDED, snapshotId, sequenceNumber, sequenceNumber, null,
null, null, null);
- }
-
@ParameterizedTest
@EnumSource(EntryStatus.class)
void isLiveDelegatesToStatus(EntryStatus status) {
@@ -284,7 +247,7 @@ class TestTrackingStruct {
TrackingStruct tracking =
new TrackingStruct(
EntryStatus.MODIFIED, 42L, 10L, 11L, 43L, 1000L,
DELETED_POSITIONS, REPLACED_POSITIONS);
- tracking.setManifestLocation("manifest-location");
+ tracking.set(MANIFEST_LOCATION_ORDINAL, "manifest-location");
tracking.set(MANIFEST_POSITION_ORDINAL, 7L);
Tracking deserialized = roundTripSerializer.apply(tracking);
diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
index 1e6f72d864..c7c6f76a4f 100644
--- a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
+++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
@@ -65,6 +65,7 @@ class TestV4ManifestReader {
}
private static final long SNAPSHOT_ID = 42L;
+ private static final long MANIFEST_SEQ = 309L;
private static final int FORMAT_VERSION_V4 = 4;
private static final long RECORD_COUNT = 100L;
private static final long FILE_SIZE_IN_BYTES = 1024L;
@@ -344,6 +345,108 @@ class TestV4ManifestReader {
.containsExactlyElementsOf(files);
}
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceSnapshotId(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));
+
+ 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().snapshotId())
+ .containsExactly(1234567L, SNAPSHOT_ID);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceDVSnapshotIdNotInherited(FileFormat format) throws
IOException {
+ Tracking trackingWithDVSnapshotId =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null,
1234567L, null, null, null);
+ TrackedFile withDVSnapshotId =
+ unpartitionedDataFile(trackingWithDVSnapshotId,
"s3://bucket/table/file-b.parquet");
+ Tracking trackingWithoutDVSnapshotId =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null, null,
null, null, null);
+ TrackedFile withoutDVSnapshotId =
+ unpartitionedDataFile(trackingWithoutDVSnapshotId,
"s3://bucket/table/file-a.parquet");
+
+ ManifestFile manifest =
+ writeManifest(
+ format, UNPARTITIONED_TYPE, ImmutableList.of(withDVSnapshotId,
withoutDVSnapshotId));
+
+ 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().dvSnapshotId())
+ .containsExactly(1234567L, null);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceAddedDataSequenceNumber(FileFormat format) throws
IOException {
+ Tracking trackingWithoutSeq =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null, null,
null, null, null);
+ TrackedFile withoutSeq =
+ unpartitionedDataFile(trackingWithoutSeq,
"s3://bucket/table/file-a.parquet");
+ Tracking trackingWithDataSeq =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, 500L, null, null,
null, null, null);
+ TrackedFile withDataSeq =
+ unpartitionedDataFile(trackingWithDataSeq,
"s3://bucket/table/file-b.parquet");
+
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE, ImmutableList.of(withoutSeq,
withDataSeq));
+
+ 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().dataSequenceNumber())
+ .containsExactly(MANIFEST_SEQ, 500L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceAddedFileSequenceNumber(FileFormat format) throws
IOException {
+ Tracking trackingWithoutSeq =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null, null,
null, null, null);
+ TrackedFile withoutSeq =
+ unpartitionedDataFile(trackingWithoutSeq,
"s3://bucket/table/file-a.parquet");
+ Tracking trackingWithFileSeq =
+ new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, 500L, null,
null, null, null);
+ TrackedFile withFileSeq =
+ unpartitionedDataFile(trackingWithFileSeq,
"s3://bucket/table/file-c.parquet");
+
+ ManifestFile manifest =
+ writeManifest(format, UNPARTITIONED_TYPE, ImmutableList.of(withoutSeq,
withFileSeq));
+
+ 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().fileSequenceNumber())
+ .containsExactly(MANIFEST_SEQ, 500L);
+ }
+
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
public void inheritanceManifestLocationAndPosition(FileFormat format) throws
IOException {
@@ -351,6 +454,7 @@ class TestV4ManifestReader {
ImmutableList.of(FILE_A, FILE_B, DATA_MANIFEST_REF,
DELETE_MANIFEST_REF);
ManifestFile manifest = writeManifest(format, UNPARTITIONED_TYPE, files);
+ String location = manifest.path();
V4ManifestReader.Builder builder =
V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
@@ -358,8 +462,9 @@ class TestV4ManifestReader {
List<TrackedFile> read = read(builder);
assertThat(read)
- .allSatisfy(
- file ->
assertThat(file.tracking().manifestLocation()).isEqualTo(manifest.path()));
+ .hasSize(4)
+ .extracting(file -> file.tracking().manifestLocation())
+ .containsOnly(location);
assertThat(read)
.extracting(file -> file.tracking().manifestPos())
.containsExactly(0L, 1L, 2L, 3L);
@@ -1440,6 +1545,10 @@ class TestV4ManifestReader {
null); // equality_ids
}
+ private static TrackedFile unpartitionedDataFile(Tracking tracking, String
location) {
+ return unpartitionedDataFile(tracking, location, null /* no stats */, null
/* no DV */);
+ }
+
private static TrackedFile unpartitionedFileWithStatus(EntryStatus status,
String location) {
Tracking tracking =
new TrackingStruct(
@@ -1512,6 +1621,8 @@ class TestV4ManifestReader {
when(manifest.path()).thenReturn(path);
when(manifest.formatVersion()).thenReturn(FORMAT_VERSION_V4);
when(manifest.content()).thenReturn(ManifestContent.DATA);
+ when(manifest.snapshotId()).thenReturn(SNAPSHOT_ID);
+ when(manifest.sequenceNumber()).thenReturn(MANIFEST_SEQ);
return manifest;
}
diff --git a/parquet/src/main/java/org/apache/iceberg/InternalParquet.java
b/parquet/src/main/java/org/apache/iceberg/InternalParquet.java
index aa5b56ca53..f9800d7623 100644
--- a/parquet/src/main/java/org/apache/iceberg/InternalParquet.java
+++ b/parquet/src/main/java/org/apache/iceberg/InternalParquet.java
@@ -37,6 +37,7 @@ public class InternalParquet {
}
private static Parquet.ReadBuilder readInternal(InputFile inputFile) {
- return
Parquet.read(inputFile).createReaderFunc(InternalReader.readerFunction());
+ return Parquet.read(inputFile)
+ .createReaderFunc(InternalReader.readerFunction(inputFile.location()));
}
}
diff --git
a/parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java
b/parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java
index 44d35ef859..614bb89a8b 100644
--- a/parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java
+++ b/parquet/src/main/java/org/apache/iceberg/data/parquet/InternalReader.java
@@ -21,6 +21,7 @@ package org.apache.iceberg.data.parquet;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
+import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.data.Record;
@@ -53,6 +54,10 @@ public class InternalReader<T extends StructLike> extends
BaseParquetReaders<T>
}
public static Parquet.ReadBuilder.ReaderFunction readerFunction() {
+ return readerFunction(null);
+ }
+
+ public static Parquet.ReadBuilder.ReaderFunction readerFunction(String
location) {
InternalReader<?> reader = new InternalReader<>();
return new Parquet.ReadBuilder.ReaderFunction() {
@@ -60,7 +65,13 @@ public class InternalReader<T extends StructLike> extends
BaseParquetReaders<T>
@Override
public Function<MessageType, ParquetValueReader<?>> apply() {
- return messageType -> reader.createReader(schema, messageType);
+ return messageType ->
+ reader.createReader(
+ schema,
+ messageType,
+ location != null
+ ? Map.of(MetadataColumns.FILE_PATH_COLUMN_ID, location)
+ : Map.of());
}
@Override
