stevenzwu commented on code in PR #18853:
URL: https://github.com/apache/iceberg/pull/18853#discussion_r4528954237
##########
core/src/test/java/org/apache/iceberg/V4TestHelpers.java:
##########
@@ -204,14 +237,74 @@ static OutputFile writeTrackedFiles(
Types.StructType statsType,
Iterable<TrackedFile> files)
throws IOException {
- Schema writeSchema = TrackedFile.schema(partitionType, statsType);
- try (FileAppender<StructLike> appender =
- InternalData.write(format,
out).schema(writeSchema).named("tracked_file").build()) {
+ try (FileAppender<StructLike> appender = newAppender(out, format,
partitionType, statsType)) {
for (TrackedFile file : files) {
appender.add((StructLike) file);
}
}
return out;
}
+
+ static OutputFile writeRows(
Review Comment:
`writeRows` is added because the wrappers now only implement `StructLike`
(per the review on `TrackedFileWrapper`), so they can no longer be passed to
`writeTrackedFiles`, which takes `TrackedFile`. It writes `StructLike` rows the
same way a v4 manifest writer will consume the wrappers, and shares the
appender setup with `writeTrackedFiles`.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +545,576 @@ 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.sortOrderId();
+ case 10 ->
+ wrapped.deletionVector() != null ?
deletionVector.wrap(wrapped.deletionVector()) : null;
+ case 11 ->
+ wrapped.manifestInfo() != null ?
manifestInfo.wrap(wrapped.manifestInfo()) : null;
+ case 12 -> wrapped.keyMetadata();
+ case 13 -> wrapped.splitOffsets();
+ case 14 -> 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) {
+ if (commitSnapshotId != null) {
+ 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());
+ }
+ }
+
+ this.file = trackedFile;
+ this.wrapped = tracking;
+ this.assignedFirstRowId = firstRowId;
+ 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);
+ case 4 -> wrapped.modifiedSnapshotId();
+ case 5 -> firstRowId();
+ case 6 -> wrapped.deletedPositions();
+ case 7 -> wrapped.replacedPositions();
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private static Long assignSequenceNumber(Long sequenceNumber, Long
commitSequenceNumber) {
+ return commitSequenceNumber != null && isUnassigned(sequenceNumber)
+ ? commitSequenceNumber
+ : sequenceNumber;
+ }
+
+ private Long firstRowId() {
+ if (file.contentType() == FileContent.DELETE_MANIFEST) {
+ return null;
+ }
+
+ return assignedFirstRowId != null ? assignedFirstRowId :
wrapped.firstRowId();
+ }
+
+ private static boolean isUnassigned(Long sequenceNumber) {
+ return sequenceNumber == null || sequenceNumber ==
ManifestWriter.UNASSIGNED_SEQ;
+ }
+ }
+
+ /** Wrapper used to write {@link ManifestInfo} using the v4 manifest info
schema. */
+ private static class ManifestInfoWrapper implements StructLike {
+ private static final int SIZE = ManifestInfo.schema().fields().size();
+
+ private final long commitSequenceNumber;
+ private ManifestInfo wrapped = null;
+
+ private ManifestInfoWrapper(long commitSequenceNumber) {
+ this.commitSequenceNumber = commitSequenceNumber;
+ }
+
+ private ManifestInfoWrapper wrap(ManifestInfo info) {
+ this.wrapped = info;
+ return this;
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
ManifestInfoWrapper 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.addedFilesCount();
+ case 1 -> wrapped.existingFilesCount();
+ case 2 -> wrapped.deletedFilesCount();
+ case 3 -> wrapped.replacedFilesCount();
+ case 4 -> wrapped.modifiedFilesCount();
+ case 5 -> wrapped.addedRowsCount();
+ case 6 -> wrapped.existingRowsCount();
+ case 7 -> wrapped.deletedRowsCount();
+ case 8 -> wrapped.replacedRowsCount();
+ case 9 -> wrapped.modifiedRowsCount();
+ case 10 -> minSequenceNumber();
+ case 11 -> mdvBuffer();
+ case 12 -> wrapped.formatVersion();
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ private long minSequenceNumber() {
+ if (wrapped.minSequenceNumber() == ManifestWriter.UNASSIGNED_SEQ) {
+ return commitSequenceNumber;
+ }
+
+ return wrapped.minSequenceNumber();
+ }
+
+ private ByteBuffer mdvBuffer() {
+ ManifestBitmap mdv = wrapped.manifestDeletionVector();
+ return mdv != null ? mdv.buffer() : null;
+ }
+ }
+
+ /** Wrapper used to write {@link ContentStats} using a manifest's content
stats type. */
+ private static class ContentStatsWrapper implements StructLike {
+ private final int[] posToFieldId;
+ private final FieldStatsWrapper[] fieldStats;
+ private ContentStats wrapped = null;
+
+ private ContentStatsWrapper(Types.StructType type) {
+ List<Types.NestedField> fields = type.fields();
+ this.posToFieldId = ContentStatsStruct.posToId(type);
+ this.fieldStats = new FieldStatsWrapper[fields.size()];
+ for (int pos = 0; pos < fields.size(); pos += 1) {
+ fieldStats[pos] = new
FieldStatsWrapper(fields.get(pos).type().asStructType());
+ }
+ }
+
+ private ContentStatsWrapper wrap(ContentStats stats) {
+ this.wrapped = stats;
+ return this;
+ }
+
+ @Override
+ public int size() {
+ return posToFieldId.length;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify
ContentStatsWrapper wrapper via set");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ FieldStats<?> stats = wrapped.statsFor(posToFieldId[pos]);
+ return stats != null ? javaClass.cast(fieldStats[pos].wrap(stats)) :
null;
+ }
+ }
+
+ /** Wrapper used to write {@link DeletionVector} using the v4 deletion
vector schema. */
+ private static class DeletionVectorWrapper implements StructLike {
+ private static final int SIZE = DeletionVector.schema().fields().size();
+
+ private DeletionVector wrapped = null;
+
+ private DeletionVectorWrapper wrap(DeletionVector dv) {
+ this.wrapped = dv;
+ return this;
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException(
+ "Cannot modify DeletionVectorWrapper 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.location();
+ case 1 -> wrapped.offset();
+ case 2 -> wrapped.sizeInBytes();
+ case 3 -> wrapped.cardinality();
+ case 4 -> wrapped.keyMetadata();
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+ }
+
+ /** Wrapper used to write {@link ColumnFile} using the v4 column file
schema. */
+ private static class ColumnFileWrapper implements StructLike {
+ private static final int SIZE = ColumnFile.schema().fields().size();
+
+ private final ColumnFile wrapped;
+
+ private ColumnFileWrapper(ColumnFile columnFile) {
+ this.wrapped = columnFile;
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify ColumnFileWrapper
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.location();
+ case 1 -> ImmutableList.copyOf(wrapped.fieldIds());
Review Comment:
`ColumnFile.fieldIds()` returns a `Set`, but `field_ids` is a list in the
schema, so this wrapper copies it into a `List` for every column file it
writes. This is the mismatch @gaborkaszab pointed out in
[#16285](https://github.com/apache/iceberg/pull/16285#discussion_r4294649072)
when we discussed `Set` vs `List`
([thread](https://github.com/apache/iceberg/pull/16285#discussion_r4255093865)).
We went with `Set` because that's how callers consume the field IDs, and the
copy here is small and only paid by files with column files. Now that the write
path makes the conversion concrete, should we revisit that decision, or keep
`Set` and accept the copy? @gaborkaszab @rdblue @amogh-jahagirdar
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +548,730 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private static void checkNullSequenceNumbers(TrackedFile file, Tracking
tracking) {
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = tracking.status();
+ Preconditions.checkState(
+ tracking.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(
+ tracking.fileSequenceNumber() != null || status == EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be ADDED):
%s",
+ status,
+ file.location());
+ }
+
+ private abstract static class TrackedFileWrapper implements TrackedFile,
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 TrackedFile wrapped = null;
+ private Integer specId = null;
+ private TrackingWrapper tracking = null;
+ private ManifestInfoWrapper manifestInfo = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById, Types.StructType
contentStatsType) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ }
+
+ protected void wrapInternal(
+ TrackedFile file, TrackingWrapper newTracking, ManifestInfoWrapper
newManifestInfo) {
Review Comment:
Done. The file wrapper also checks that manifests have manifest info and
data files don't, then wraps `file.manifestInfo()` when it is present.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +548,730 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private static void checkNullSequenceNumbers(TrackedFile file, Tracking
tracking) {
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = tracking.status();
+ Preconditions.checkState(
+ tracking.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(
+ tracking.fileSequenceNumber() != null || status == EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be ADDED):
%s",
+ status,
+ file.location());
+ }
+
+ private abstract static class TrackedFileWrapper implements TrackedFile,
StructLike {
Review Comment:
Done. Since the wrappers no longer implement `TrackedFile`, there's nothing
left for a `Delegates` class to handle.
##########
core/src/test/java/org/apache/iceberg/TestV4TrackedFileWrappers.java:
##########
@@ -0,0 +1,810 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.apache.iceberg.V4TestHelpers.ADDED_TRACKING;
+import static org.apache.iceberg.V4TestHelpers.columnFile;
+import static org.apache.iceberg.V4TestHelpers.copyWithTracking;
+import static org.apache.iceberg.V4TestHelpers.dataFile;
+import static org.apache.iceberg.V4TestHelpers.dataFileWithColumnFiles;
+import static org.apache.iceberg.V4TestHelpers.dataFileWithStats;
+import static org.apache.iceberg.V4TestHelpers.deletionVector;
+import static org.apache.iceberg.V4TestHelpers.manifestRef;
+import static org.apache.iceberg.V4TestHelpers.manifestRefWithStats;
+import static org.apache.iceberg.V4TestHelpers.partition;
+import static org.apache.iceberg.V4TestHelpers.readTrackedFiles;
+import static org.apache.iceberg.V4TestHelpers.writeTrackedFiles;
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import org.apache.iceberg.V4Metadata.LeafTrackedFileWrapper;
+import org.apache.iceberg.V4Metadata.RootTrackedFileWrapper;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Conversions;
+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;
+
+class TestV4TrackedFileWrappers {
Review Comment:
Done. Added the test plan to the description. Besides the null-argument
tests, I also dropped tests that repeated a shared check through the other
wrapper.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +548,730 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private static void checkNullSequenceNumbers(TrackedFile file, Tracking
tracking) {
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = tracking.status();
+ Preconditions.checkState(
+ tracking.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(
+ tracking.fileSequenceNumber() != null || status == EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be ADDED):
%s",
+ status,
+ file.location());
+ }
+
+ private abstract static class TrackedFileWrapper implements TrackedFile,
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 TrackedFile wrapped = null;
+ private Integer specId = null;
+ private TrackingWrapper tracking = null;
+ private ManifestInfoWrapper manifestInfo = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById, Types.StructType
contentStatsType) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ }
+
+ protected void wrapInternal(
+ TrackedFile file, TrackingWrapper newTracking, ManifestInfoWrapper
newManifestInfo) {
+ 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.fields().isEmpty() ? fileSpecId :
null;
+ this.tracking = newTracking;
+ this.manifestInfo = newManifestInfo;
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify a read-only
tracked file");
+ }
+
+ @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.sortOrderId();
+ case 10 -> wrapped.deletionVector();
+ case 11 -> manifestInfo;
+ case 12 -> wrapped.keyMetadata();
+ case 13 -> wrapped.splitOffsets();
+ case 14 -> wrapped.columnFiles();
+ default -> throw new UnsupportedOperationException("Unknown field
ordinal: " + pos);
+ };
+ }
+
+ @Override
+ public Tracking tracking() {
+ return tracking;
+ }
+
+ @Override
+ public FileContent contentType() {
+ return wrapped.contentType();
+ }
+
+ @Override
+ public String location() {
+ return wrapped.location();
+ }
+
+ @Override
+ public FileFormat fileFormat() {
+ return wrapped.fileFormat();
+ }
+
+ @Override
+ public long recordCount() {
+ return wrapped.recordCount();
+ }
+
+ @Override
+ public long fileSizeInBytes() {
+ return wrapped.fileSizeInBytes();
+ }
+
+ @Override
+ public Integer specId() {
+ return specId;
+ }
+
+ @Override
+ public StructLike partition() {
+ return specId != null ? wrapped.partition() : null;
+ }
+
+ @Override
+ public ContentStats contentStats() {
+ return wrapped.contentStats();
+ }
+
+ @Override
+ public Integer sortOrderId() {
+ return wrapped.sortOrderId();
+ }
+
+ @Override
+ public DeletionVector deletionVector() {
+ return wrapped.deletionVector();
+ }
+
+ @Override
+ public ManifestInfo manifestInfo() {
+ return manifestInfo;
+ }
+
+ @Override
+ public ByteBuffer keyMetadata() {
+ return wrapped.keyMetadata();
+ }
+
+ @Override
+ public List<Long> splitOffsets() {
+ return wrapped.splitOffsets();
+ }
+
+ @Override
+ public List<ColumnFile> columnFiles() {
+ return wrapped.columnFiles();
+ }
+
+ @Override
+ public TrackedFile copy() {
+ return copy(null /* all stats */);
+ }
+
+ @Override
+ public TrackedFile copyWithStats(Set<Integer> requestedColumnIds) {
+ return copy(requestedColumnIds);
+ }
+
+ private TrackedFile copy(Set<Integer> statsIds) {
+ ContentStats stats = wrapped.contentStats();
+ ContentStats statsCopy;
+ if (stats == null || (statsIds != null && statsIds.isEmpty())) {
+ statsCopy = null;
+ } else {
+ statsCopy = statsIds != null ? stats.copy(statsIds) : stats.copy();
+ }
+
+ DeletionVector dv = wrapped.deletionVector();
+ return new TrackedFileStruct(
+ tracking.copy(),
+ wrapped.contentType(),
+ wrapped.location(),
+ wrapped.fileFormat(),
+ wrapped.recordCount(),
+ wrapped.fileSizeInBytes(),
+ specId,
+ StructLikeUtil.copy(partition()),
+ statsCopy,
+ wrapped.sortOrderId(),
+ dv != null ? dv.copy() : null,
+ manifestInfo != null ? manifestInfo.copy() : null,
+ wrapped.keyMetadata(),
+ wrapped.splitOffsets(),
+ copyColumnFiles(wrapped.columnFiles()));
+ }
+
+ private static List<ColumnFile> copyColumnFiles(List<ColumnFile>
columnFiles) {
+ if (columnFiles == null) {
+ return null;
+ }
+
+ List<ColumnFile> copies =
Lists.newArrayListWithCapacity(columnFiles.size());
+ for (ColumnFile columnFile : columnFiles) {
+ copies.add(columnFile.copy());
+ }
+
+ return copies;
+ }
+
+ private StructLike unionPartition() {
+ StructLike partition = partition();
+ if (partition == null) {
+ return null;
+ }
+
+ StructProjection projection = partitionProjections.get(specId);
+ return projection != null ? projection.wrap(partition) : partition;
+ }
+
+ private static Map<Integer, StructProjection> partitionProjections(
+ Map<Integer, PartitionSpec> specsById) {
+ Types.StructType unionType =
Partitioning.unionPartitionTypes(specsById.values());
+ Map<Integer, StructProjection> projections = Maps.newHashMap();
+ for (PartitionSpec spec : specsById.values()) {
+ Types.StructType partitionType = spec.partitionType();
+ 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 {
+ private final TrackingWrapper trackingWrapper = new TrackingWrapper();
+
+ LeafTrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById, Types.StructType
contentStatsType) {
+ super(specsById, contentStatsType);
+ }
+
+ 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());
+
+ checkNullSequenceNumbers(file, newTracking);
+
+ trackingWrapper.wrap(
+ newTracking,
+ newTracking.snapshotId(),
+ newTracking.dataSequenceNumber(),
+ newTracking.fileSequenceNumber(),
+ newTracking.firstRowId());
+ wrapInternal(file, trackingWrapper, null /* manifestInfo */);
+ return this;
+ }
+ }
+
+ /** Wrapper used to write a data file or leaf manifest to a v4 root
manifest. */
+ static class RootTrackedFileWrapper extends TrackedFileWrapper {
+ private final long commitSnapshotId;
+ private final long commitSequenceNumber;
+ private final TrackingWrapper trackingWrapper = new TrackingWrapper();
+ private final ManifestInfoWrapper manifestInfoWrapper = new
ManifestInfoWrapper();
+
+ RootTrackedFileWrapper(
+ long commitSnapshotId,
+ long commitSequenceNumber,
+ Map<Integer, PartitionSpec> specsById,
+ Types.StructType contentStatsType) {
+ super(specsById, contentStatsType);
+ this.commitSnapshotId = commitSnapshotId;
+ this.commitSequenceNumber = 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");
+
+ boolean isManifest = file.contentType() != FileContent.DATA;
+ Preconditions.checkArgument(
+ !isManifest || file.manifestInfo() != null,
+ "Missing manifest info for manifest: %s",
+ file.location());
+ Preconditions.checkArgument(
+ newTracking.snapshotId() != null, "Missing snapshot ID for file:
%s", file.location());
+ checkNullSequenceNumbers(file, newTracking);
+
+ long snapshotId = newTracking.snapshotId();
+ // a modified entry inherits the data sequence number of the commit that
modified it
+ Long dataSequenceSnapshotId =
+ newTracking.status() == EntryStatus.MODIFIED
+ ? newTracking.modifiedSnapshotId()
+ : newTracking.snapshotId();
+ trackingWrapper.wrap(
+ newTracking,
+ snapshotId,
+ resolveSequenceNumber(file, newTracking.dataSequenceNumber(),
dataSequenceSnapshotId),
+ resolveSequenceNumber(file, newTracking.fileSequenceNumber(),
snapshotId),
+ resolveFirstRowId(file, newTracking, assignedFirstRowId));
+
+ if (isManifest) {
+ ManifestInfo info = file.manifestInfo();
+ manifestInfoWrapper.wrap(
+ info, resolveSequenceNumber(file, info.minSequenceNumber(),
snapshotId));
+ wrapInternal(file, trackingWrapper, manifestInfoWrapper);
+ } else {
+ wrapInternal(file, trackingWrapper, null /* manifestInfo */);
+ }
+
+ return this;
+ }
+
+ private long resolveSequenceNumber(
+ TrackedFile file, Long existingSequenceNumber, Long entrySnapshotId) {
+ if (existingSequenceNumber == null
+ || existingSequenceNumber == ManifestWriter.UNASSIGNED_SEQ) {
+ // the sequence number is assigned here, so the file must be changed
by the current commit
+ Preconditions.checkState(
+ entrySnapshotId != null && entrySnapshotId == commitSnapshotId,
+ "Cannot assign sequence number to file from snapshot %s
(committing snapshot %s): %s",
+ entrySnapshotId,
+ commitSnapshotId,
+ file.location());
+ return commitSequenceNumber;
+ }
+
+ return existingSequenceNumber;
+ }
+
+ private static Long resolveFirstRowId(
Review Comment:
Update: validation ended up in `wrap` rather than `get`. `wrap` checks the
tracking and the assigned first row ID against the commit, and `get` only picks
the stored or commit value for each field, so validation stays separate from
value selection. The leaf uses the same tracking wrapper without commit values,
so it writes the tracking unchanged.
--
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]
