rdblue commented on code in PR #18853:
URL: https://github.com/apache/iceberg/pull/18853#discussion_r4530639461
##########
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(),
Review Comment:
This isn't correct. For a leaf, there is always a snapshot ID from the
caller. It may be null if the leaf is not yet associated with a commit, but it
must be passed in. I think that this wrapper should accept a `Long snapshotId`
and pass it here.
--
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]
