rdblue commented on code in PR #18853:
URL: https://github.com/apache/iceberg/pull/18853#discussion_r4518834288
##########
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());
Review Comment:
This shouldn't re-derive the partition type used in the file. Instead it
should be passed in as a separate argument. That will ensure that the actual
struct type used is what is produced here.
Also, I think we need to validate that the union type includes all of the
field IDs of each partition type so that we guarantee here that we are not
dropping fields.
--
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]
