rdblue commented on code in PR #18853:
URL: https://github.com/apache/iceberg/pull/18853#discussion_r4530953382
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
Review Comment:
I don't think this comment is accurate. These checks validate that sequence
numbers are provided when they are required. When they are not required and
they are not supplied, then the value will be inherited for leaf manifests but
is known and must be written for root manifests.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
Review Comment:
I think these checks should be in the `Tracking` wrap method, not here.
These only use `file` for the error message, and `file` is available within
`TrackingWrapper.wrap`.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
Review Comment:
I think it makes sense to check the content type and that manifest info
matches here. I'd prefer to have the specific types covered and to validate
that the other types are not present:
```java
if (file.contentType() == FileContent.DATA) {
Preconditions.checkArgument(
file.manifestInfo() == null,
"Invalid data file, manifest info must be null: %s",
file.location());
} else if (file.contentType() == FileContent.DATA_MANIFEST ||
file.contentType() == FileContent.DELETE_MANIFEST) {
Preconditions.checkArgument(
file.manifestInfo() != null,
"Invalid manifest, missing manifest info: %s",
file.location());
} else {
throw new IllegalArgumentException(String.format("Cannot write %s to
a v4 manifest: %s", file.contentType(), file));
}
```
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.tightStats();
+ case 10 -> wrapped.sortOrderId();
+ case 11 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 12 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 13 -> wrapped.keyMetadata();
+ case 14 -> wrapped.splitOffsets();
+ case 15 -> wrapColumnFiles(wrapped.columnFiles());
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static List<StructLike> wrapColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<StructLike> wrappers =
Lists.newArrayListWithCapacity(columnFiles.size());
+ for (ColumnFile columnFile : columnFiles) {
+ wrappers.add(new ColumnFileWrapper(columnFile));
+ }
+
+ return wrappers;
+ }
+
+ private StructLike unionPartition() {
+ if (specId == null) {
+ return null;
+ }
+
+ StructProjection projection = partitionProjections.get(specId);
+ return projection != null ? projection.wrap(wrapped.partition()) :
wrapped.partition();
+ }
+
+ private static Map<Integer, StructProjection> partitionProjections(
+ Iterable<PartitionSpec> specs, Types.StructType unionType) {
+ Map<Integer, StructProjection> projections = Maps.newHashMap();
+ for (PartitionSpec spec : specs) {
+ Types.StructType partitionType = spec.partitionType();
+ for (Types.NestedField field : partitionType.fields()) {
Review Comment:
I'd prefer to refactor this into its own validation method.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
Review Comment:
In addition to checking `manifestInfo`, this can also validate that
`deletionVector` is only present for `DATA`. It is not allowed for manifests.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.tightStats();
+ case 10 -> wrapped.sortOrderId();
+ case 11 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 12 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 13 -> wrapped.keyMetadata();
+ case 14 -> wrapped.splitOffsets();
+ case 15 -> wrapColumnFiles(wrapped.columnFiles());
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static List<StructLike> wrapColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<StructLike> wrappers =
Lists.newArrayListWithCapacity(columnFiles.size());
+ for (ColumnFile columnFile : columnFiles) {
+ wrappers.add(new ColumnFileWrapper(columnFile));
+ }
+
+ return wrappers;
+ }
+
+ private StructLike unionPartition() {
+ if (specId == null) {
+ return null;
+ }
+
+ StructProjection projection = partitionProjections.get(specId);
+ return projection != null ? projection.wrap(wrapped.partition()) :
wrapped.partition();
+ }
+
+ private static Map<Integer, StructProjection> partitionProjections(
+ Iterable<PartitionSpec> specs, Types.StructType unionType) {
+ Map<Integer, StructProjection> projections = Maps.newHashMap();
+ for (PartitionSpec spec : specs) {
+ Types.StructType partitionType = spec.partitionType();
+ for (Types.NestedField field : partitionType.fields()) {
+ Preconditions.checkArgument(
+ unionType.field(field.fieldId()) != null,
+ "Invalid union partition type, missing field %s from spec %s:
%s",
+ field.fieldId(),
+ spec.specId(),
+ unionType);
+ }
+
+ StructProjection projection =
+ partitionType.equals(unionType)
+ ? null
+ : StructProjection.createAllowMissing(partitionType,
unionType);
+ projections.put(spec.specId(), projection);
+ }
+
+ return projections;
+ }
+ }
+
+ /** Wrapper used to write a data file to a v4 leaf manifest. */
+ static class LeafTrackedFileWrapper extends TrackedFileWrapper {
+ LeafTrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(),
+ null /* leaf manifests only contain data files */);
+ }
+
+ LeafTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ Preconditions.checkArgument(
+ file.contentType() == FileContent.DATA,
+ "Invalid content type for leaf manifest: %s",
+ file.contentType());
+
+ wrapInternal(file, newTracking, null /* assignedFirstRowId */);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write a data file or leaf manifest to a v4 root
manifest. */
+ static class RootTrackedFileWrapper extends TrackedFileWrapper {
+ RootTrackedFileWrapper(
+ long commitSnapshotId,
+ long commitSequenceNumber,
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(commitSnapshotId, commitSequenceNumber),
+ new ManifestInfoWrapper(commitSequenceNumber));
+ }
+
+ /**
+ * Re-points this wrapper at a file and the tracking to write for it.
+ *
+ * @param file a data file or manifest
+ * @param newTracking the tracking to write for the file
+ * @param assignedFirstRowId a first row ID to assign to the file, or null
if it is already
+ * assigned
+ * @return this for method chaining
+ */
+ RootTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+
+ FileContent content = file.contentType();
+ Preconditions.checkArgument(
+ content == FileContent.DATA
+ || content == FileContent.DATA_MANIFEST
+ || content == FileContent.DELETE_MANIFEST,
+ "Invalid content type for root manifest: %s",
+ content);
+ Preconditions.checkArgument(
+ newTracking.snapshotId() != null, "Missing snapshot ID for file:
%s", file.location());
+
+ wrapInternal(file, newTracking, assignedFirstRowId);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write {@link Tracking} using the v4 tracking schema. */
+ private static class TrackingWrapper implements StructLike {
+ private static final int SIZE = Tracking.schema().fields().size();
+
+ private final Long commitSnapshotId;
+ private final Long commitSequenceNumber;
+
+ private TrackedFile file = null;
+ private Tracking wrapped = null;
+ private Long assignedFirstRowId = null;
+
+ private TrackingWrapper() {
+ this.commitSnapshotId = null;
+ this.commitSequenceNumber = null;
+ }
+
+ private TrackingWrapper(long commitSnapshotId, long commitSequenceNumber) {
+ this.commitSnapshotId = commitSnapshotId;
+ this.commitSequenceNumber = commitSequenceNumber;
+ }
+
+ private TrackingWrapper wrap(TrackedFile trackedFile, Tracking tracking,
Long firstRowId) {
+ this.file = trackedFile;
+ this.wrapped = tracking;
+ this.assignedFirstRowId = firstRowId;
+ if (commitSnapshotId == null) {
+ return this;
+ }
+
+ if (isUnassigned(tracking.dataSequenceNumber())) {
+ // a modified entry inherits the data sequence number of the commit
that modified it
+ Long dataSnapshotId =
+ tracking.status() == EntryStatus.MODIFIED
+ ? tracking.modifiedSnapshotId()
+ : tracking.snapshotId();
+ Preconditions.checkState(
+ commitSnapshotId.equals(dataSnapshotId),
+ "Cannot assign sequence number to file from snapshot %s
(committing snapshot %s): %s",
+ dataSnapshotId,
+ commitSnapshotId,
+ trackedFile.location());
+ }
+
+ if (isUnassigned(tracking.fileSequenceNumber())) {
+ Preconditions.checkState(
+ commitSnapshotId.equals(tracking.snapshotId()),
+ "Cannot assign sequence number to file from snapshot %s
(committing snapshot %s): %s",
+ tracking.snapshotId(),
+ commitSnapshotId,
+ trackedFile.location());
+ }
+
+ if (trackedFile.contentType() == FileContent.DELETE_MANIFEST) {
Review Comment:
This depends on the `FileContent` checks that were done in the
`*TrackedFile.wrap` methods, which are not at all obvious to the
reader/reviewer. I think that splitting the logic in several places makes this
code brittle. I'd prefer to have checks that are co-located and for the type of
manifest file to be known.
I think it would make more sense to also have `LeafTrackingWrapper` and
`RootTrackingWrapper` to handle this, or to have separate `wrap` methods called
by the `TrackedFile` wrap methods. This would clean up the call chain, too,
which is fairly complicated.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.tightStats();
+ case 10 -> wrapped.sortOrderId();
+ case 11 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 12 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 13 -> wrapped.keyMetadata();
+ case 14 -> wrapped.splitOffsets();
+ case 15 -> wrapColumnFiles(wrapped.columnFiles());
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static List<StructLike> wrapColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<StructLike> wrappers =
Lists.newArrayListWithCapacity(columnFiles.size());
Review Comment:
This should reuse the array and `ColumnFileWrapper` instances. If the array
is too short, then new instances should be created and added to the array.
The return value should be `wrappers.subList(0, columnFiles.size())`.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
Review Comment:
`wrapColumnFiles` is null tolerant. Why not make `contentStats.wrap` and
others handle null as well? That would make this more readable.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.tightStats();
+ case 10 -> wrapped.sortOrderId();
+ case 11 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 12 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 13 -> wrapped.keyMetadata();
+ case 14 -> wrapped.splitOffsets();
+ case 15 -> wrapColumnFiles(wrapped.columnFiles());
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static List<StructLike> wrapColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<StructLike> wrappers =
Lists.newArrayListWithCapacity(columnFiles.size());
+ for (ColumnFile columnFile : columnFiles) {
+ wrappers.add(new ColumnFileWrapper(columnFile));
+ }
+
+ return wrappers;
+ }
+
+ private StructLike unionPartition() {
+ if (specId == null) {
+ return null;
+ }
+
+ StructProjection projection = partitionProjections.get(specId);
+ return projection != null ? projection.wrap(wrapped.partition()) :
wrapped.partition();
+ }
+
+ private static Map<Integer, StructProjection> partitionProjections(
+ Iterable<PartitionSpec> specs, Types.StructType unionType) {
+ Map<Integer, StructProjection> projections = Maps.newHashMap();
+ for (PartitionSpec spec : specs) {
+ Types.StructType partitionType = spec.partitionType();
+ for (Types.NestedField field : partitionType.fields()) {
+ Preconditions.checkArgument(
+ unionType.field(field.fieldId()) != null,
+ "Invalid union partition type, missing field %s from spec %s:
%s",
+ field.fieldId(),
+ spec.specId(),
+ unionType);
+ }
+
+ StructProjection projection =
+ partitionType.equals(unionType)
+ ? null
+ : StructProjection.createAllowMissing(partitionType,
unionType);
+ projections.put(spec.specId(), projection);
+ }
+
+ return projections;
+ }
+ }
+
+ /** Wrapper used to write a data file to a v4 leaf manifest. */
+ static class LeafTrackedFileWrapper extends TrackedFileWrapper {
+ LeafTrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(),
+ null /* leaf manifests only contain data files */);
+ }
+
+ LeafTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
Review Comment:
These are duplicated. Why not have a `validate` method that is implemented
by children instead of having children call `wrapInternal`?
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
Review Comment:
This doesn't account for `UNASSIGNED_SEQ`.
I think this is fine and we should not check for `UNASSIGNED_SEQ` elsewhere
either. The reason for using `UNASSIGNED_SEQ` was that the `ManifestFile`
interface defines `sequenceNumber` as a primitive (`long`) and we had to handle
cases where it was not assigned using -1. The new interface, `Tracking` uses
`Long` so we don't need to use `UNASSIGNED_SEQ`.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.tightStats();
+ case 10 -> wrapped.sortOrderId();
+ case 11 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 12 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 13 -> wrapped.keyMetadata();
+ case 14 -> wrapped.splitOffsets();
+ case 15 -> wrapColumnFiles(wrapped.columnFiles());
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static List<StructLike> wrapColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<StructLike> wrappers =
Lists.newArrayListWithCapacity(columnFiles.size());
+ for (ColumnFile columnFile : columnFiles) {
+ wrappers.add(new ColumnFileWrapper(columnFile));
+ }
+
+ return wrappers;
+ }
+
+ private StructLike unionPartition() {
+ if (specId == null) {
+ return null;
+ }
+
+ StructProjection projection = partitionProjections.get(specId);
+ return projection != null ? projection.wrap(wrapped.partition()) :
wrapped.partition();
+ }
+
+ private static Map<Integer, StructProjection> partitionProjections(
+ Iterable<PartitionSpec> specs, Types.StructType unionType) {
+ Map<Integer, StructProjection> projections = Maps.newHashMap();
+ for (PartitionSpec spec : specs) {
+ Types.StructType partitionType = spec.partitionType();
+ for (Types.NestedField field : partitionType.fields()) {
+ Preconditions.checkArgument(
+ unionType.field(field.fieldId()) != null,
+ "Invalid union partition type, missing field %s from spec %s:
%s",
+ field.fieldId(),
+ spec.specId(),
+ unionType);
+ }
+
+ StructProjection projection =
+ partitionType.equals(unionType)
+ ? null
+ : StructProjection.createAllowMissing(partitionType,
unionType);
+ projections.put(spec.specId(), projection);
+ }
+
+ return projections;
+ }
+ }
+
+ /** Wrapper used to write a data file to a v4 leaf manifest. */
+ static class LeafTrackedFileWrapper extends TrackedFileWrapper {
+ LeafTrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(),
+ null /* leaf manifests only contain data files */);
+ }
+
+ LeafTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ Preconditions.checkArgument(
+ file.contentType() == FileContent.DATA,
+ "Invalid content type for leaf manifest: %s",
+ file.contentType());
+
+ wrapInternal(file, newTracking, null /* assignedFirstRowId */);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write a data file or leaf manifest to a v4 root
manifest. */
+ static class RootTrackedFileWrapper extends TrackedFileWrapper {
+ RootTrackedFileWrapper(
+ long commitSnapshotId,
+ long commitSequenceNumber,
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(commitSnapshotId, commitSequenceNumber),
+ new ManifestInfoWrapper(commitSequenceNumber));
+ }
+
+ /**
+ * Re-points this wrapper at a file and the tracking to write for it.
+ *
+ * @param file a data file or manifest
+ * @param newTracking the tracking to write for the file
+ * @param assignedFirstRowId a first row ID to assign to the file, or null
if it is already
+ * assigned
+ * @return this for method chaining
+ */
+ RootTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+
+ FileContent content = file.contentType();
+ Preconditions.checkArgument(
+ content == FileContent.DATA
+ || content == FileContent.DATA_MANIFEST
+ || content == FileContent.DELETE_MANIFEST,
+ "Invalid content type for root manifest: %s",
+ content);
+ Preconditions.checkArgument(
+ newTracking.snapshotId() != null, "Missing snapshot ID for file:
%s", file.location());
+
+ wrapInternal(file, newTracking, assignedFirstRowId);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write {@link Tracking} using the v4 tracking schema. */
+ private static class TrackingWrapper implements StructLike {
+ private static final int SIZE = Tracking.schema().fields().size();
+
+ private final Long commitSnapshotId;
+ private final Long commitSequenceNumber;
+
+ private TrackedFile file = null;
+ private Tracking wrapped = null;
+ private Long assignedFirstRowId = null;
+
+ private TrackingWrapper() {
+ this.commitSnapshotId = null;
+ this.commitSequenceNumber = null;
+ }
+
+ private TrackingWrapper(long commitSnapshotId, long commitSequenceNumber) {
+ this.commitSnapshotId = commitSnapshotId;
+ this.commitSequenceNumber = commitSequenceNumber;
+ }
+
+ private TrackingWrapper wrap(TrackedFile trackedFile, Tracking tracking,
Long firstRowId) {
+ this.file = trackedFile;
+ this.wrapped = tracking;
+ this.assignedFirstRowId = firstRowId;
+ if (commitSnapshotId == null) {
+ return this;
+ }
+
+ if (isUnassigned(tracking.dataSequenceNumber())) {
+ // a modified entry inherits the data sequence number of the commit
that modified it
+ Long dataSnapshotId =
+ tracking.status() == EntryStatus.MODIFIED
+ ? tracking.modifiedSnapshotId()
+ : tracking.snapshotId();
+ Preconditions.checkState(
+ commitSnapshotId.equals(dataSnapshotId),
+ "Cannot assign sequence number to file from snapshot %s
(committing snapshot %s): %s",
+ dataSnapshotId,
+ commitSnapshotId,
+ trackedFile.location());
+ }
+
+ if (isUnassigned(tracking.fileSequenceNumber())) {
+ Preconditions.checkState(
+ commitSnapshotId.equals(tracking.snapshotId()),
+ "Cannot assign sequence number to file from snapshot %s
(committing snapshot %s): %s",
+ tracking.snapshotId(),
+ commitSnapshotId,
+ trackedFile.location());
+ }
+
+ if (trackedFile.contentType() == FileContent.DELETE_MANIFEST) {
+ Preconditions.checkState(
+ firstRowId == null,
+ "Cannot assign first-row-id to delete manifest: %s",
+ trackedFile.location());
+ } else if (firstRowId != null) {
+ Preconditions.checkState(
+ tracking.status().isLive(),
+ "Cannot assign first-row-id to file with status %s: %s",
+ tracking.status(),
+ trackedFile.location());
+ Preconditions.checkState(
+ tracking.firstRowId() == null,
+ "Cannot assign first-row-id to file with existing first-row-id %s:
%s",
+ tracking.firstRowId(),
+ trackedFile.location());
+ } else {
+ Preconditions.checkState(
+ tracking.firstRowId() != null,
+ "Missing first-row-id for file: %s",
+ trackedFile.location());
+ }
+
+ return this;
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify TrackingWrapper
wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> wrapped.status().id();
+ case 1 -> wrapped.snapshotId();
+ case 2 -> assignSequenceNumber(wrapped.dataSequenceNumber(),
commitSequenceNumber);
+ case 3 -> assignSequenceNumber(wrapped.fileSequenceNumber(),
commitSequenceNumber);
Review Comment:
I think this should just be called `sequenceNumber` and not
`assignSequenceNumber`. It is selecting the one that should be used.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.tightStats();
+ case 10 -> wrapped.sortOrderId();
+ case 11 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 12 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 13 -> wrapped.keyMetadata();
+ case 14 -> wrapped.splitOffsets();
+ case 15 -> wrapColumnFiles(wrapped.columnFiles());
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static List<StructLike> wrapColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<StructLike> wrappers =
Lists.newArrayListWithCapacity(columnFiles.size());
+ for (ColumnFile columnFile : columnFiles) {
+ wrappers.add(new ColumnFileWrapper(columnFile));
+ }
+
+ return wrappers;
+ }
+
+ private StructLike unionPartition() {
+ if (specId == null) {
+ return null;
+ }
+
+ StructProjection projection = partitionProjections.get(specId);
+ return projection != null ? projection.wrap(wrapped.partition()) :
wrapped.partition();
+ }
+
+ private static Map<Integer, StructProjection> partitionProjections(
+ Iterable<PartitionSpec> specs, Types.StructType unionType) {
+ Map<Integer, StructProjection> projections = Maps.newHashMap();
+ for (PartitionSpec spec : specs) {
+ Types.StructType partitionType = spec.partitionType();
+ for (Types.NestedField field : partitionType.fields()) {
+ Preconditions.checkArgument(
+ unionType.field(field.fieldId()) != null,
+ "Invalid union partition type, missing field %s from spec %s:
%s",
+ field.fieldId(),
+ spec.specId(),
+ unionType);
+ }
+
+ StructProjection projection =
+ partitionType.equals(unionType)
+ ? null
+ : StructProjection.createAllowMissing(partitionType,
unionType);
+ projections.put(spec.specId(), projection);
+ }
+
+ return projections;
+ }
+ }
+
+ /** Wrapper used to write a data file to a v4 leaf manifest. */
+ static class LeafTrackedFileWrapper extends TrackedFileWrapper {
+ LeafTrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(),
+ null /* leaf manifests only contain data files */);
+ }
+
+ LeafTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ Preconditions.checkArgument(
+ file.contentType() == FileContent.DATA,
+ "Invalid content type for leaf manifest: %s",
+ file.contentType());
+
+ wrapInternal(file, newTracking, null /* assignedFirstRowId */);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write a data file or leaf manifest to a v4 root
manifest. */
+ static class RootTrackedFileWrapper extends TrackedFileWrapper {
+ RootTrackedFileWrapper(
+ long commitSnapshotId,
+ long commitSequenceNumber,
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(commitSnapshotId, commitSequenceNumber),
+ new ManifestInfoWrapper(commitSequenceNumber));
+ }
+
+ /**
+ * Re-points this wrapper at a file and the tracking to write for it.
+ *
+ * @param file a data file or manifest
+ * @param newTracking the tracking to write for the file
+ * @param assignedFirstRowId a first row ID to assign to the file, or null
if it is already
+ * assigned
+ * @return this for method chaining
Review Comment:
This is not used for method chaining. Do we actually need Javadoc? I think
it's pretty clear what is happening here.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,578 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private abstract static class TrackedFileWrapper implements StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+ private final DeletionVectorWrapper deletionVector = new
DeletionVectorWrapper();
+ private final TrackingWrapper tracking;
+ private final ManifestInfoWrapper manifestInfo;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType,
+ TrackingWrapper tracking,
+ ManifestInfoWrapper manifestInfo) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById.values(),
partitionType);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ this.tracking = tracking;
+ this.manifestInfo = manifestInfo;
+ }
+
+ protected void wrapInternal(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ if (file.contentType() == FileContent.DATA) {
+ Preconditions.checkArgument(
+ file.manifestInfo() == null,
+ "Invalid data file, manifest info must be null: %s",
+ file.location());
+ } else {
+ Preconditions.checkArgument(
+ file.manifestInfo() != null,
+ "Invalid manifest, missing manifest info: %s",
+ file.location());
+ }
+
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = newTracking.status();
+ Preconditions.checkState(
+ newTracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ newTracking.fileSequenceNumber() != null || status ==
EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ status,
+ file.location());
+
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.isUnpartitioned() ? fileSpecId :
null;
+ tracking.wrap(file, newTracking, assignedFirstRowId);
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
TrackedFileWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.tightStats();
+ case 10 -> wrapped.sortOrderId();
+ case 11 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 12 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 13 -> wrapped.keyMetadata();
+ case 14 -> wrapped.splitOffsets();
+ case 15 -> wrapColumnFiles(wrapped.columnFiles());
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static List<StructLike> wrapColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<StructLike> wrappers =
Lists.newArrayListWithCapacity(columnFiles.size());
+ for (ColumnFile columnFile : columnFiles) {
+ wrappers.add(new ColumnFileWrapper(columnFile));
+ }
+
+ return wrappers;
+ }
+
+ private StructLike unionPartition() {
+ if (specId == null) {
+ return null;
+ }
+
+ StructProjection projection = partitionProjections.get(specId);
+ return projection != null ? projection.wrap(wrapped.partition()) :
wrapped.partition();
+ }
+
+ private static Map<Integer, StructProjection> partitionProjections(
+ Iterable<PartitionSpec> specs, Types.StructType unionType) {
+ Map<Integer, StructProjection> projections = Maps.newHashMap();
+ for (PartitionSpec spec : specs) {
+ Types.StructType partitionType = spec.partitionType();
+ for (Types.NestedField field : partitionType.fields()) {
+ Preconditions.checkArgument(
+ unionType.field(field.fieldId()) != null,
+ "Invalid union partition type, missing field %s from spec %s:
%s",
+ field.fieldId(),
+ spec.specId(),
+ unionType);
+ }
+
+ StructProjection projection =
+ partitionType.equals(unionType)
+ ? null
+ : StructProjection.createAllowMissing(partitionType,
unionType);
+ projections.put(spec.specId(), projection);
+ }
+
+ return projections;
+ }
+ }
+
+ /** Wrapper used to write a data file to a v4 leaf manifest. */
+ static class LeafTrackedFileWrapper extends TrackedFileWrapper {
+ LeafTrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(),
+ null /* leaf manifests only contain data files */);
+ }
+
+ LeafTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+ Preconditions.checkArgument(
+ file.contentType() == FileContent.DATA,
+ "Invalid content type for leaf manifest: %s",
+ file.contentType());
+
+ wrapInternal(file, newTracking, null /* assignedFirstRowId */);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write a data file or leaf manifest to a v4 root
manifest. */
+ static class RootTrackedFileWrapper extends TrackedFileWrapper {
+ RootTrackedFileWrapper(
+ long commitSnapshotId,
+ long commitSequenceNumber,
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType partitionType,
+ Types.StructType contentStatsType) {
+ super(
+ specsById,
+ partitionType,
+ contentStatsType,
+ new TrackingWrapper(commitSnapshotId, commitSequenceNumber),
+ new ManifestInfoWrapper(commitSequenceNumber));
+ }
+
+ /**
+ * Re-points this wrapper at a file and the tracking to write for it.
+ *
+ * @param file a data file or manifest
+ * @param newTracking the tracking to write for the file
+ * @param assignedFirstRowId a first row ID to assign to the file, or null
if it is already
+ * assigned
+ * @return this for method chaining
+ */
+ RootTrackedFileWrapper wrap(TrackedFile file, Tracking newTracking, Long
assignedFirstRowId) {
+ Preconditions.checkArgument(file != null, "Invalid file: null");
+ Preconditions.checkArgument(newTracking != null, "Invalid tracking:
null");
+
+ FileContent content = file.contentType();
+ Preconditions.checkArgument(
+ content == FileContent.DATA
+ || content == FileContent.DATA_MANIFEST
+ || content == FileContent.DELETE_MANIFEST,
+ "Invalid content type for root manifest: %s",
+ content);
+ Preconditions.checkArgument(
+ newTracking.snapshotId() != null, "Missing snapshot ID for file:
%s", file.location());
+
+ wrapInternal(file, newTracking, assignedFirstRowId);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write {@link Tracking} using the v4 tracking schema. */
+ private static class TrackingWrapper implements StructLike {
+ private static final int SIZE = Tracking.schema().fields().size();
+
+ private final Long commitSnapshotId;
+ private final Long commitSequenceNumber;
+
+ private TrackedFile file = null;
+ private Tracking wrapped = null;
+ private Long assignedFirstRowId = null;
+
+ private TrackingWrapper() {
+ this.commitSnapshotId = null;
+ this.commitSequenceNumber = null;
+ }
+
+ private TrackingWrapper(long commitSnapshotId, long commitSequenceNumber) {
+ this.commitSnapshotId = commitSnapshotId;
+ this.commitSequenceNumber = commitSequenceNumber;
+ }
+
+ private TrackingWrapper wrap(TrackedFile trackedFile, Tracking tracking,
Long firstRowId) {
+ this.file = trackedFile;
+ this.wrapped = tracking;
+ this.assignedFirstRowId = firstRowId;
+ if (commitSnapshotId == null) {
Review Comment:
This is true when writing a leaf that is not yet associated with a commit.
For instance, writing a leaf that will be committed using
`AppendFiles.appendManifest`.
It doesn't mean that the remaining checks should not be performed. I don't
see a reason to return early here, since we know that the only way to reach
this with a null snapshot ID is through `appendManifest`. In that case, I think
there are still checks that should run, like the first row ID checks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
