diff --git a/api/src/main/java/org/apache/iceberg/ColumnFile.java b/api/src/main/java/org/apache/iceberg/ColumnFile.java new file mode 100644 index 000000000000..3a056ca1201b --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/ColumnFile.java @@ -0,0 +1,54 @@ +/* + * 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 java.util.List; +import org.apache.iceberg.types.Types; + +/** Information about a column file. */ +public interface ColumnFile { + Types.NestedField FIELD_IDS = + Types.NestedField.required( + 161, + "field_ids", + Types.ListType.ofRequired(162, Types.IntegerType.get()), + "Field IDs this column file contains"); + Types.NestedField LOCATION = + Types.NestedField.required( + 163, "location", Types.StringType.get(), "Location of the column file"); + Types.NestedField FILE_SIZE_IN_BYTES = + Types.NestedField.required( + 164, "file_size_in_bytes", Types.LongType.get(), "Total column file size in bytes"); + + static Types.StructType schema() { + return Types.StructType.of(FIELD_IDS, LOCATION, FILE_SIZE_IN_BYTES); + } + + /** Returns the field IDs contained in this column file. */ + List fieldIds(); + + /** Returns the location of the column file. */ + String location(); + + /** Returns the total size of the column file in bytes. */ + long fileSizeInBytes(); + + /** Copies this column file. */ + ColumnFile copy(); +} diff --git a/api/src/main/java/org/apache/iceberg/DataFile.java b/api/src/main/java/org/apache/iceberg/DataFile.java index 64338228349a..67e825d16b0c 100644 --- a/api/src/main/java/org/apache/iceberg/DataFile.java +++ b/api/src/main/java/org/apache/iceberg/DataFile.java @@ -164,4 +164,8 @@ default FileContent content() { default List equalityFieldIds() { return null; } + + default List columnFiles() { + return null; + } } diff --git a/core/src/main/java/org/apache/iceberg/ColumnFileStruct.java b/core/src/main/java/org/apache/iceberg/ColumnFileStruct.java new file mode 100644 index 000000000000..378d8ebd197f --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/ColumnFileStruct.java @@ -0,0 +1,176 @@ +/* + * 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 java.io.Serializable; +import java.util.Arrays; +import java.util.List; +import org.apache.iceberg.avro.SupportsIndexProjection; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ArrayUtil; + +/** Mutable {@link StructLike} implementation of {@link ColumnFile}. */ +class ColumnFileStruct extends SupportsIndexProjection implements ColumnFile, Serializable { + private static final Types.StructType BASE_TYPE = + Types.StructType.of(ColumnFile.FIELD_IDS, ColumnFile.LOCATION, ColumnFile.FILE_SIZE_IN_BYTES); + + private int[] fieldIds = null; + private String location = null; + private long fileSizeInBytes = -1L; + + /** Used by internal readers to instantiate this class with a projection schema. */ + ColumnFileStruct(Types.StructType projection) { + super(BASE_TYPE, projection); + } + + ColumnFileStruct(List fieldIds, String location, long fileSizeInBytes) { + super(BASE_TYPE.fields().size()); + this.fieldIds = ArrayUtil.toIntArray(fieldIds); + this.location = location; + this.fileSizeInBytes = fileSizeInBytes; + } + + /** Copy constructor. */ + private ColumnFileStruct(ColumnFileStruct toCopy) { + super(toCopy); + this.fieldIds = + toCopy.fieldIds != null ? Arrays.copyOf(toCopy.fieldIds, toCopy.fieldIds.length) : null; + this.location = toCopy.location; + this.fileSizeInBytes = toCopy.fileSizeInBytes; + } + + /** Constructor for Java serialization. */ + ColumnFileStruct() { + super(BASE_TYPE.fields().size()); + } + + @Override + public List fieldIds() { + return fieldIds != null ? ArrayUtil.toUnmodifiableIntList(fieldIds) : null; + } + + @Override + public String location() { + return location; + } + + @Override + public long fileSizeInBytes() { + return fileSizeInBytes; + } + + @Override + public ColumnFile copy() { + return new ColumnFileStruct(this); + } + + @Override + protected T internalGet(int pos, Class javaClass) { + return javaClass.cast(getByPos(pos)); + } + + private Object getByPos(int pos) { + switch (pos) { + case 0: + return fieldIds(); + case 1: + return location; + case 2: + return fileSizeInBytes; + default: + throw new UnsupportedOperationException("Unknown field ordinal: " + pos); + } + } + + @Override + @SuppressWarnings("unchecked") + protected void internalSet(int pos, T value) { + switch (pos) { + case 0: + this.fieldIds = ArrayUtil.toIntArray((List) value); + break; + case 1: + // always coerce to String for Serializable + this.location = value.toString(); + break; + case 2: + this.fileSizeInBytes = (long) value; + break; + default: + // ignore the object, it must be from a newer version of the format + } + } + + static Builder builder() { + return new Builder(); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("field_ids", fieldIds) + .add("location", location) + .add("file_size_in_bytes", fileSizeInBytes) + .toString(); + } + + static class Builder { + private List fieldIds = null; + private String location = null; + private Long fileSizeInBytes = null; + + Builder fieldIds(List newFieldIds) { + Preconditions.checkArgument(newFieldIds != null, "Invalid field IDs: null"); + Preconditions.checkArgument(!newFieldIds.isEmpty(), "Invalid field IDs: empty"); + Preconditions.checkArgument( + Sets.newHashSet(newFieldIds).size() == newFieldIds.size(), + "Invalid field IDs: duplicateD IDs found: %s", + newFieldIds); + this.fieldIds = newFieldIds; + return this; + } + + Builder location(String newLocation) { + Preconditions.checkArgument(newLocation != null, "Invalid location: null"); + Preconditions.checkArgument(!newLocation.isEmpty(), "Invalid location: empty"); + this.location = newLocation; + return this; + } + + Builder fileSizeInBytes(long newFileSizeInBytes) { + Preconditions.checkArgument( + newFileSizeInBytes >= 0, + "Invalid file size in bytes: %s (must be >= 0)", + newFileSizeInBytes); + this.fileSizeInBytes = newFileSizeInBytes; + return this; + } + + ColumnFile build() { + Preconditions.checkArgument(fieldIds != null, "Missing required value: fieldIds"); + Preconditions.checkArgument(location != null, "Missing required value: location"); + Preconditions.checkArgument( + fileSizeInBytes != null, "Missing required value: fileSizeInBytes"); + return new ColumnFileStruct(fieldIds, location, fileSizeInBytes); + } + } +} diff --git a/core/src/main/java/org/apache/iceberg/TrackedFile.java b/core/src/main/java/org/apache/iceberg/TrackedFile.java index 5b9cdd9ab46c..d954a737542d 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFile.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFile.java @@ -94,6 +94,9 @@ interface TrackedFile { "equality_ids", Types.ListType.ofRequired(136, Types.IntegerType.get()), "Field ids used to determine row equality in equality delete files"); + Types.NestedField COLUMN_FILES = + Types.NestedField.optional( + 158, "column_files", Types.ListType.ofRequired(159, ColumnFile.schema()), "Column files"); static Types.StructType schemaWithContentStats( Types.StructType partitionType, Types.StructType contentStatsType) { @@ -114,7 +117,8 @@ static Types.StructType schemaWithContentStats( MANIFEST_INFO, KEY_METADATA, SPLIT_OFFSETS, - EQUALITY_IDS); + EQUALITY_IDS, + COLUMN_FILES); } /** Returns the tracking information for this entry. */ @@ -165,6 +169,9 @@ static Types.StructType schemaWithContentStats( /** Returns the set of field IDs used for equality comparison in equality delete files. */ List equalityIds(); + /** Returns the column files for this file. */ + List columnFiles(); + /** Copies this tracked file. */ TrackedFile copy(); diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java index e0feafeda246..7de87c3808a3 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java @@ -233,6 +233,11 @@ public DataFile copyWithoutStats() { public DataFile copyWithStats(Set requestedColumnIds) { return new TrackedDataFile(file().copyWithStats(requestedColumnIds), spec()); } + + @Override + public List columnFiles() { + return file().columnFiles(); + } } /** Adapts a TrackedFile EQUALITY_DELETES entry to the {@link DeleteFile} interface. */ diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileBuilder.java b/core/src/main/java/org/apache/iceberg/TrackedFileBuilder.java index d7aa28c3290f..30a307c0a228 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileBuilder.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileBuilder.java @@ -142,7 +142,8 @@ private static TrackedFile terminal(TrackedFile source, Tracking tracking) { source.manifestInfo(), source.keyMetadata(), source.splitOffsets(), - source.equalityIds()); + source.equalityIds(), + null); } private TrackedFileBuilder(FileContent contentType, long snapshotId) { @@ -356,6 +357,7 @@ TrackedFile build() { manifestInfo, keyMetadata, splitOffsets, - equalityIds); + equalityIds, + null); } } diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java index 958ddfbbc4cf..546a3bcae42d 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java @@ -21,10 +21,14 @@ import java.io.Serializable; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.Collections; import java.util.List; +import java.util.Objects; import java.util.Set; +import java.util.stream.Collectors; import org.apache.iceberg.avro.SupportsIndexProjection; import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ArrayUtil; @@ -66,7 +70,8 @@ public PartitionData copy() { TrackedFile.MANIFEST_INFO, TrackedFile.KEY_METADATA, TrackedFile.SPLIT_OFFSETS, - TrackedFile.EQUALITY_IDS); + TrackedFile.EQUALITY_IDS, + TrackedFile.COLUMN_FILES); private FileContent contentType = null; private int formatVersion = -1; @@ -86,6 +91,7 @@ public PartitionData copy() { private byte[] keyMetadata = null; private long[] splitOffsets = null; private int[] equalityIds = null; + private List columnFiles = null; /** Used by internal readers to instantiate this class with a projection schema. */ TrackedFileStruct(Types.StructType projection) { @@ -118,7 +124,8 @@ public PartitionData copy() { ManifestInfo manifestInfo, ByteBuffer keyMetadata, List splitOffsets, - List equalityIds) { + List equalityIds, + List columnFiles) { super(BASE_TYPE.fields().size()); this.tracking = tracking; this.contentType = contentType; @@ -139,6 +146,7 @@ public PartitionData copy() { this.keyMetadata = ByteBuffers.toByteArray(keyMetadata); this.splitOffsets = ArrayUtil.toLongArray(splitOffsets); this.equalityIds = ArrayUtil.toIntArray(equalityIds); + this.columnFiles = columnFiles != null ? Lists.newArrayList(columnFiles) : null; } /** Copy constructor. */ @@ -176,6 +184,13 @@ private TrackedFileStruct(TrackedFileStruct toCopy, boolean withStats, Set equalityIds() { return equalityIds != null ? ArrayUtil.toUnmodifiableIntList(equalityIds) : null; } + @Override + public List columnFiles() { + return columnFiles != null ? Collections.unmodifiableList(columnFiles) : null; + } + @Override public TrackedFile copy() { return new TrackedFileStruct(this, true, null); @@ -291,6 +311,7 @@ private Object getByPos(int pos) { case 13 -> keyMetadata(); case 14 -> splitOffsets(); case 15 -> equalityIds(); + case 16 -> columnFiles(); default -> throw new UnsupportedOperationException("Unknown field ordinal: " + pos); }; } @@ -316,6 +337,7 @@ protected void internalSet(int pos, T value) { case 13 -> this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer) value); case 14 -> this.splitOffsets = ArrayUtil.toLongArray((List) value); case 15 -> this.equalityIds = ArrayUtil.toIntArray((List) value); + case 16 -> this.columnFiles = (List) value; default -> { // ignore the object, it must be from a newer version of the format } @@ -341,6 +363,7 @@ public String toString() { .add("key_metadata", keyMetadata == null ? "null" : "(redacted)") .add("split_offsets", splitOffsets == null ? "null" : splitOffsets()) .add("equality_ids", equalityIds == null ? "null" : equalityIds()) + .add("column_files", columnFiles) .toString(); } } diff --git a/core/src/main/java/org/apache/iceberg/Tracking.java b/core/src/main/java/org/apache/iceberg/Tracking.java index fcdc4e50b236..2d586e4bb66e 100644 --- a/core/src/main/java/org/apache/iceberg/Tracking.java +++ b/core/src/main/java/org/apache/iceberg/Tracking.java @@ -65,6 +65,12 @@ interface Tracking { "replaced_positions", Types.BinaryType.get(), "Bitmap of positions replaced in this snapshot"); + Types.NestedField LATEST_COLUMN_FILE_SNAPSHOT_ID = + Types.NestedField.optional( + 160, + "latest_column_file_snapshot_id", + Types.LongType.get(), + "Snapshot ID where the latest column file was added; null if there are no column files"); static Types.StructType schema() { return Types.StructType.of( @@ -75,7 +81,8 @@ static Types.StructType schema() { DV_SNAPSHOT_ID, FIRST_ROW_ID, DELETED_POSITIONS, - REPLACED_POSITIONS); + REPLACED_POSITIONS, + LATEST_COLUMN_FILE_SNAPSHOT_ID); } /** Returns the status of the entry. */ @@ -107,6 +114,12 @@ default boolean isLive() { /** Returns the bitmap of positions replaced in this snapshot. */ ByteBuffer replacedPositions(); + /** + * Returns the snapshot ID where the latest column file was added; null if there are no column + * files. + */ + Long latestColumnFileSnapshotId(); + /** Returns the manifest location this entry was read from, or null. */ String manifestLocation(); diff --git a/core/src/main/java/org/apache/iceberg/TrackingBuilder.java b/core/src/main/java/org/apache/iceberg/TrackingBuilder.java index c5a11bc53cee..1a830e12b244 100644 --- a/core/src/main/java/org/apache/iceberg/TrackingBuilder.java +++ b/core/src/main/java/org/apache/iceberg/TrackingBuilder.java @@ -25,13 +25,14 @@ class TrackingBuilder { private final long newSnapshotId; private final Long snapshotId; - private final Long dataSequenceNumber; private final Long fileSequenceNumber; private final Long firstRowId; private EntryStatus status; + private Long dataSequenceNumber; private Long dvSnapshotId; private byte[] deletedPositions; private byte[] replacedPositions; + private Long latestColumnFileSnapshotId; /** * Creates a builder for a newly added file. @@ -82,6 +83,7 @@ private TrackingBuilder(long newSnapshotId) { this.dvSnapshotId = null; this.deletedPositions = null; this.replacedPositions = null; + this.latestColumnFileSnapshotId = null; } private TrackingBuilder(Tracking source, long newSnapshotId) { @@ -96,6 +98,7 @@ private TrackingBuilder(Tracking source, long newSnapshotId) { this.dvSnapshotId = source.dvSnapshotId(); this.deletedPositions = null; this.replacedPositions = null; + this.latestColumnFileSnapshotId = source.latestColumnFileSnapshotId(); } /** Indicates that the DV has been updated for the new Tracking. */ @@ -111,6 +114,17 @@ TrackingBuilder dvUpdated() { return this; } + /** Indicates that the column files list has been updated for the new Tracking. */ + TrackingBuilder columnFilesUpdated() { + this.latestColumnFileSnapshotId = newSnapshotId; + if (status == EntryStatus.EXISTING) { + this.status = EntryStatus.MODIFIED; + } + // Bumping 'dataSequenceNumber' to avoid having both equality deletes and column files. + this.dataSequenceNumber = null; + return this; + } + /** Sets the positions deleted by this commit for a manifest entry. */ TrackingBuilder deletedPositions(ByteBuffer positions) { Preconditions.checkState( @@ -140,7 +154,8 @@ Tracking build() { dvSnapshotId, firstRowId, deletedPositions, - replacedPositions); + replacedPositions, + latestColumnFileSnapshotId); } private static Tracking terminal(EntryStatus to, Tracking source, long newSnapshotId) { @@ -154,7 +169,8 @@ private static Tracking terminal(EntryStatus to, Tracking source, long newSnapsh source.dvSnapshotId(), source.firstRowId(), null, - null); + null, + source.latestColumnFileSnapshotId()); } private static void validateSource(Tracking source) { diff --git a/core/src/main/java/org/apache/iceberg/TrackingStruct.java b/core/src/main/java/org/apache/iceberg/TrackingStruct.java index 8ae4b7e4ce88..975def87deab 100644 --- a/core/src/main/java/org/apache/iceberg/TrackingStruct.java +++ b/core/src/main/java/org/apache/iceberg/TrackingStruct.java @@ -40,6 +40,7 @@ class TrackingStruct extends SupportsIndexProjection implements Tracking, Serial Tracking.FIRST_ROW_ID, Tracking.DELETED_POSITIONS, Tracking.REPLACED_POSITIONS, + Tracking.LATEST_COLUMN_FILE_SNAPSHOT_ID, MetadataColumns.ROW_POSITION); private EntryStatus status = null; @@ -50,6 +51,7 @@ class TrackingStruct extends SupportsIndexProjection implements Tracking, Serial private Long firstRowId = null; private byte[] deletedPositions = null; private byte[] replacedPositions = null; + private Long latestColumnFileSnapshotId = null; // set by manifest readers, not written to manifests private String manifestLocation = null; @@ -80,6 +82,7 @@ private TrackingStruct(TrackingStruct toCopy) { toCopy.replacedPositions != null ? Arrays.copyOf(toCopy.replacedPositions, toCopy.replacedPositions.length) : null; + this.latestColumnFileSnapshotId = toCopy.latestColumnFileSnapshotId; this.manifestLocation = toCopy.manifestLocation; this.manifestPos = toCopy.manifestPos; } @@ -92,7 +95,8 @@ private TrackingStruct(TrackingStruct toCopy) { Long dvSnapshotId, Long firstRowId, byte[] deletedPositions, - byte[] replacedPositions) { + byte[] replacedPositions, + Long latestColumnFileSnapshotId) { super(BASE_TYPE.fields().size()); this.status = status; this.snapshotId = snapshotId; @@ -102,6 +106,7 @@ private TrackingStruct(TrackingStruct toCopy) { this.firstRowId = firstRowId; this.deletedPositions = deletedPositions; this.replacedPositions = replacedPositions; + this.latestColumnFileSnapshotId = latestColumnFileSnapshotId; } void inheritFrom(Tracking manifestTracking) { @@ -118,11 +123,13 @@ void inheritFrom(Tracking manifestTracking) { manifestTracking.dataSequenceNumber(), manifestTracking.fileSequenceNumber()); - if (status == EntryStatus.ADDED) { + if (status == EntryStatus.ADDED || status == EntryStatus.MODIFIED) { if (dataSequenceNumber == null) { this.dataSequenceNumber = manifestTracking.fileSequenceNumber(); } + } + if (status == EntryStatus.ADDED) { if (fileSequenceNumber == null) { this.fileSequenceNumber = manifestTracking.fileSequenceNumber(); } @@ -174,6 +181,11 @@ public ByteBuffer replacedPositions() { return replacedPositions != null ? ByteBuffer.wrap(replacedPositions) : null; } + @Override + public Long latestColumnFileSnapshotId() { + return latestColumnFileSnapshotId; + } + @Override public String manifestLocation() { return manifestLocation; @@ -195,62 +207,37 @@ protected T internalGet(int pos, Class javaClass) { } private Object getByPos(int pos) { - switch (pos) { - case 0: - return status != null ? status.id() : null; - case 1: - return snapshotId(); - case 2: - return dataSequenceNumber(); - case 3: - return fileSequenceNumber(); - case 4: - return dvSnapshotId; - case 5: - return firstRowId; - case 6: - return deletedPositions(); - case 7: - return replacedPositions(); - case 8: - return manifestPos; - default: - throw new UnsupportedOperationException("Unknown field ordinal: " + pos); - } + return switch (pos) { + case 0 -> status != null ? status.id() : null; + case 1 -> snapshotId(); + case 2 -> dataSequenceNumber(); + case 3 -> fileSequenceNumber(); + case 4 -> dvSnapshotId; + case 5 -> firstRowId; + case 6 -> deletedPositions(); + case 7 -> replacedPositions(); + case 8 -> latestColumnFileSnapshotId; + case 9 -> manifestPos; + default -> throw new UnsupportedOperationException("Unknown field ordinal: " + pos); + }; } @Override protected void internalSet(int pos, T value) { switch (pos) { - case 0: - this.status = EntryStatus.fromId((Integer) value); - break; - case 1: - this.snapshotId = (Long) value; - break; - case 2: - this.dataSequenceNumber = (Long) value; - break; - case 3: - this.fileSequenceNumber = (Long) value; - break; - case 4: - this.dvSnapshotId = (Long) value; - break; - case 5: - this.firstRowId = (Long) value; - break; - case 6: - this.deletedPositions = ByteBuffers.toByteArray((ByteBuffer) value); - break; - case 7: - this.replacedPositions = ByteBuffers.toByteArray((ByteBuffer) value); - break; - case 8: - this.manifestPos = (long) value; - break; - default: + case 0 -> this.status = EntryStatus.fromId((Integer) value); + case 1 -> this.snapshotId = (Long) value; + case 2 -> this.dataSequenceNumber = (Long) value; + case 3 -> this.fileSequenceNumber = (Long) value; + case 4 -> this.dvSnapshotId = (Long) value; + case 5 -> this.firstRowId = (Long) value; + case 6 -> this.deletedPositions = ByteBuffers.toByteArray((ByteBuffer) value); + case 7 -> this.replacedPositions = ByteBuffers.toByteArray((ByteBuffer) value); + case 8 -> this.latestColumnFileSnapshotId = (Long) value; + case 9 -> this.manifestPos = (long) value; + default -> { // ignore the object, it must be from a newer version of the format + } } } @@ -265,6 +252,7 @@ public String toString() { .add("first_row_id", firstRowId) .add("deleted_positions", deletedPositions == null ? "null" : "(binary)") .add("replaced_positions", replacedPositions == null ? "null" : "(binary)") + .add("latest_column_file_snapshot_id", latestColumnFileSnapshotId) .toString(); } } diff --git a/core/src/test/java/org/apache/iceberg/TestColumnFileStruct.java b/core/src/test/java/org/apache/iceberg/TestColumnFileStruct.java new file mode 100644 index 000000000000..e3ccc52f2f94 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestColumnFileStruct.java @@ -0,0 +1,197 @@ +/* + * 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.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.util.List; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +class TestColumnFileStruct { + + private static final List FIELD_IDS = Lists.newArrayList(1, 2, 3); + private static final String LOCATION = "s3://bucket/data/column.parquet"; + + @Test + void testFieldAccess() { + ColumnFile columnFile = new ColumnFileStruct(FIELD_IDS, LOCATION, 1024L); + + assertThat(columnFile.fieldIds()).containsExactlyElementsOf(FIELD_IDS); + assertThat(columnFile.location()).isEqualTo(LOCATION); + assertThat(columnFile.fileSizeInBytes()).isEqualTo(1024L); + } + + @Test + void testCopy() { + ColumnFile columnFile = new ColumnFileStruct(FIELD_IDS, LOCATION, 2048L); + + ColumnFile copy = columnFile.copy(); + + assertThat(copy.fieldIds()).containsExactlyElementsOf(FIELD_IDS); + assertThat(copy.location()).isEqualTo(LOCATION); + assertThat(copy.fileSizeInBytes()).isEqualTo(2048L); + } + + @Test + void testStructLikeSize() { + ColumnFileStruct columnFile = new ColumnFileStruct(); + assertThat(columnFile.size()).isEqualTo(3); + } + + @Test + void testSetFieldsByOrdinals() { + ColumnFileStruct columnFile = new ColumnFileStruct(); + + columnFile.set(0, FIELD_IDS); + columnFile.set(1, LOCATION); + columnFile.set(2, 128L); + + assertThat(columnFile.fieldIds()).containsExactlyElementsOf(FIELD_IDS); + assertThat(columnFile.location()).isEqualTo(LOCATION); + assertThat(columnFile.fileSizeInBytes()).isEqualTo(128L); + } + + @Test + void testGetFieldsByOrdinals() { + ColumnFileStruct columnFile = new ColumnFileStruct(FIELD_IDS, LOCATION, 128L); + + assertThat(columnFile.get(0, List.class)).containsExactlyElementsOf(FIELD_IDS); + assertThat(columnFile.get(1, String.class)).isEqualTo(LOCATION); + assertThat(columnFile.get(2, Long.class)).isEqualTo(128L); + } + + @Test + void testProjectedStructLike() { + Types.StructType projection = + Types.StructType.of(ColumnFile.LOCATION, ColumnFile.FILE_SIZE_IN_BYTES); + + ColumnFileStruct columnFile = new ColumnFileStruct(projection); + assertThat(columnFile.size()).isEqualTo(2); + + // projected position 0 maps to internal position 1 (location) + // projected position 1 maps to internal position 2 (file_size_in_bytes) + columnFile.set(0, LOCATION); + columnFile.set(1, 1024L); + + assertThat(columnFile.location()).isEqualTo(LOCATION); + assertThat(columnFile.fileSizeInBytes()).isEqualTo(1024L); + assertThat(columnFile.get(0, String.class)).isEqualTo(LOCATION); + assertThat(columnFile.get(1, Long.class)).isEqualTo(1024L); + } + + @ParameterizedTest + @MethodSource("org.apache.iceberg.TestHelpers#serializers") + void testSerializationRoundTrip(TestHelpers.RoundTripSerializer roundTripSerializer) + throws IOException, ClassNotFoundException { + ColumnFile columnFile = new ColumnFileStruct(FIELD_IDS, LOCATION, 1024L); + + ColumnFile deserialized = roundTripSerializer.apply(columnFile); + + assertThat(deserialized.fieldIds()).containsExactlyElementsOf(FIELD_IDS); + assertThat(deserialized.location()).isEqualTo(LOCATION); + assertThat(deserialized.fileSizeInBytes()).isEqualTo(1024L); + } + + @Test + void testInvalidBuilderValues() { + assertThatThrownBy( + () -> + ColumnFileStruct.builder() + .fieldIds(null) + .location(LOCATION) + .fileSizeInBytes(1024L) + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid field IDs: null"); + + assertThatThrownBy( + () -> + ColumnFileStruct.builder() + .fieldIds(Lists.newArrayList()) + .location(LOCATION) + .fileSizeInBytes(1024L) + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid field IDs: empty"); + + assertThatThrownBy( + () -> + ColumnFileStruct.builder() + .fieldIds(Lists.newArrayList(1, 2, 1)) + .location(LOCATION) + .fileSizeInBytes(1024L) + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid field IDs: duplicateD IDs found: [1, 2, 1]"); + + assertThatThrownBy( + () -> + ColumnFileStruct.builder() + .fieldIds(FIELD_IDS) + .location(null) + .fileSizeInBytes(1024L) + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid location: null"); + + assertThatThrownBy( + () -> + ColumnFileStruct.builder() + .fieldIds(FIELD_IDS) + .location("") + .fileSizeInBytes(1024L) + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid location: empty"); + + assertThatThrownBy( + () -> + ColumnFileStruct.builder() + .fieldIds(FIELD_IDS) + .location(LOCATION) + .fileSizeInBytes(-1) + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid file size in bytes: -1 (must be >= 0)"); + } + + @Test + void testMissingBuilderValues() { + assertThatThrownBy( + () -> ColumnFileStruct.builder().location(LOCATION).fileSizeInBytes(1024L).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Missing required value: fieldIds"); + + assertThatThrownBy( + () -> ColumnFileStruct.builder().fieldIds(FIELD_IDS).fileSizeInBytes(1024L).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Missing required value: location"); + + assertThatThrownBy( + () -> ColumnFileStruct.builder().fieldIds(FIELD_IDS).location(LOCATION).build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Missing required value: fileSizeInBytes"); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java index 170c01ef7dc4..e0372b3a6da5 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java @@ -60,7 +60,8 @@ public void schemaWithContentStatsFieldOrder() { "manifest_info", "key_metadata", "split_offsets", - "equality_ids"); + "equality_ids", + "column_files"); } @Test @@ -71,7 +72,7 @@ public void schemaWithContentStatsFieldIds() { assertThat(fields) .extracting(Types.NestedField::fieldId) .containsExactly( - 147, 134, 157, 100, 101, 103, 104, 141, 102, 146, 140, 148, 150, 131, 132, 135); + 147, 134, 157, 100, 101, 103, 104, 141, 102, 146, 140, 148, 150, 131, 132, 135, 158); } @Test diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java index f4d5675ee94a..cdb573794081 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java @@ -58,6 +58,19 @@ class TestTrackedFileAdapters { .withSpecId(PARTITIONED_SPEC_ID) .build(); private static final PartitionData PARTITION = partition("books"); + private static final ColumnFile COLUMN_FILE_1 = + ColumnFileStruct.builder() + .fieldIds(List.of(1, 2)) + .location("column_file_1.parquet") + .fileSizeInBytes(128L) + .build(); + private static final ColumnFile COLUMN_FILE_2 = + ColumnFileStruct.builder() + .fieldIds(List.of(3)) + .location("column_file_2.parquet") + .fileSizeInBytes(256L) + .build(); + private static final List COLUMN_FILES = List.of(COLUMN_FILE_1, COLUMN_FILE_2); // Tracking field ordinals, looked up from the schema so the tests do not hard-code offsets. private static final int DATA_SEQUENCE_NUMBER_ORDINAL = @@ -79,19 +92,24 @@ class TestTrackedFileAdapters { @Test void testDataFileAdapterDelegation() { TrackedFile file = - TrackedFileBuilder.data(42L) - .formatVersion(FORMAT_VERSION_V4) - .location(DATA_FILE_LOCATION) - .fileFormat(FileFormat.PARQUET) - .partition(PARTITION) - .recordCount(100L) - .fileSizeInBytes(1024L) - .specId(PARTITIONED_SPEC_ID) - .contentStats(createContentStats()) - .sortOrderId(3) - .keyMetadata(ByteBuffer.wrap(new byte[] {1, 2, 3})) - .splitOffsets(ImmutableList.of(50L, 100L)) - .build(); + new TrackedFileStruct( + TrackingBuilder.added(42L).build(), + FileContent.DATA, + FORMAT_VERSION_V4, + DATA_FILE_LOCATION, + FileFormat.PARQUET, + PARTITION, + 100L, + 1024L, + PARTITIONED_SPEC_ID, + createContentStats(), + 3, + null, + null, + ByteBuffer.wrap(new byte[] {1, 2, 3}), + ImmutableList.of(50L, 100L), + null, + COLUMN_FILES); populateTrackingFields(file); DataFile dataFile = TrackedFileAdapters.asDataFile(file, specsById(PARTITIONED_SPEC)); @@ -124,6 +142,7 @@ void testDataFileAdapterDelegation() { .containsOnly( Map.entry(1, Conversions.toByteBuffer(Types.IntegerType.get(), 1000)), Map.entry(2, Conversions.toByteBuffer(Types.FloatType.get(), 100.0f))); + assertThat(dataFile.columnFiles()).containsExactly(COLUMN_FILE_1, COLUMN_FILE_2); } @ParameterizedTest diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileBuilder.java b/core/src/test/java/org/apache/iceberg/TestTrackedFileBuilder.java index d6b0701fdfd7..518630dcedb7 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFileBuilder.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileBuilder.java @@ -824,7 +824,15 @@ private static TrackedFile sourceDeleteManifest(long snapshotId) { private static TrackedFile entryWithInheritedSeqNums(TrackedFile entry, long sequenceNumber) { Tracking manifestTrackingToInheritFrom = new TrackingStruct( - EntryStatus.EXISTING, 123L, sequenceNumber, sequenceNumber, null, null, null, null); + EntryStatus.EXISTING, + 123L, + sequenceNumber, + sequenceNumber, + null, + null, + null, + null, + null); ((TrackingStruct) entry.tracking()).inheritFrom(manifestTrackingToInheritFrom); return entry; diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java index 8e0d5c06824b..e65eccf18a32 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.List; import java.util.Set; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; @@ -35,6 +36,18 @@ class TestTrackedFileStruct { Types.StructType.of( Types.NestedField.optional(1000, "id_bucket", Types.IntegerType.get()), Types.NestedField.optional(1001, "category", Types.StringType.get())); + private static final ColumnFile COLUMN_FILE_1 = + ColumnFileStruct.builder() + .fieldIds(ImmutableList.of(1, 2)) + .location("s3://bucket/data/col-1.parquet") + .fileSizeInBytes(256L) + .build(); + private static final ColumnFile COLUMN_FILE_2 = + ColumnFileStruct.builder() + .fieldIds(ImmutableList.of(3)) + .location("s3://bucket/data/col-2.parquet") + .fileSizeInBytes(128L) + .build(); // Ordinals looked up from the TrackedFile schema so tests don't hard-code positions. private static final List SCHEMA_FIELDS = @@ -60,6 +73,7 @@ class TestTrackedFileStruct { private static final int KEY_METADATA_ORDINAL = SCHEMA_FIELDS.indexOf(TrackedFile.KEY_METADATA); private static final int SPLIT_OFFSETS_ORDINAL = SCHEMA_FIELDS.indexOf(TrackedFile.SPLIT_OFFSETS); private static final int EQUALITY_IDS_ORDINAL = SCHEMA_FIELDS.indexOf(TrackedFile.EQUALITY_IDS); + private static final int COLUMN_FILES_ORDINAL = SCHEMA_FIELDS.indexOf(TrackedFile.COLUMN_FILES); // Ordinal of MetadataColumns.ROW_POSITION within TrackingStruct's BASE_TYPE, // which appends ROW_POSITION after the Tracking schema fields. @@ -112,6 +126,7 @@ void testFieldAccess() { file.set(KEY_METADATA_ORDINAL, ByteBuffer.wrap(new byte[] {1, 2, 3})); file.set(SPLIT_OFFSETS_ORDINAL, ImmutableList.of(100L, 200L)); file.set(EQUALITY_IDS_ORDINAL, ImmutableList.of(1, 2, 3)); + file.set(COLUMN_FILES_ORDINAL, ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2)); assertThat(file.tracking()).isNotNull(); assertThat(file.tracking().status()).isEqualTo(EntryStatus.ADDED); @@ -129,6 +144,10 @@ void testFieldAccess() { assertThat(file.keyMetadata()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2, 3})); assertThat(file.splitOffsets()).containsExactly(100L, 200L); assertThat(file.equalityIds()).containsExactly(1, 2, 3); + assertThat(file.columnFiles()).hasSize(2); + verifyColumnFile(COLUMN_FILE_1, file.columnFiles().get(0)); + verifyColumnFile(COLUMN_FILE_2, file.columnFiles().get(1)); + // should return EMPTY_PARTITION_DATA assertThat(file.partition()).isNotNull(); assertThat(file.partition().size()).isEqualTo(0); @@ -215,7 +234,24 @@ void testCopy() { assertThat(copy.equalityIds()).isNull(); assertThat(copy.tracking().manifestLocation()).isEqualTo("s3://bucket/manifest.avro"); assertThat(copy.tracking().manifestPos()).isEqualTo(3L); + assertThat(copy.tracking().latestColumnFileSnapshotId()).isEqualTo(42L); assertThat(copy.partition()).isEqualTo(newPartition(7, "music")); + assertThat(copy.columnFiles()).hasSize(2); + assertThat(copy.columnFiles()).isNotSameAs(file.columnFiles()); + verifyColumnFile(COLUMN_FILE_1, copy.columnFiles().get(0)); + verifyColumnFile(COLUMN_FILE_2, copy.columnFiles().get(1)); + } + + @Test + void testCopyWithNullColumnFile() { + TrackedFileStruct file = createFullTrackedFile(); + file.set(COLUMN_FILES_ORDINAL, Arrays.asList(COLUMN_FILE_1, null, COLUMN_FILE_2)); + + TrackedFile copy = file.copy(); + + assertThat(copy.columnFiles()).hasSize(2); + verifyColumnFile(COLUMN_FILE_1, copy.columnFiles().get(0)); + verifyColumnFile(COLUMN_FILE_2, copy.columnFiles().get(1)); } @Test @@ -249,14 +285,13 @@ void testCopyIsDeep() { TrackedFile copy = file.copy(); - // keyMetadata should be a deep copy assertThat(copy.keyMetadata()).isNotSameAs(file.keyMetadata()); } @Test void testStructLikeSize() { TrackedFileStruct file = new TrackedFileStruct(); - assertThat(file.size()).isEqualTo(16); + assertThat(file.size()).isEqualTo(17); } @Test @@ -274,6 +309,13 @@ void testStructLikeGetSet() { file.set(SORT_ORDER_ID_ORDINAL, 3); assertThat(file.get(SORT_ORDER_ID_ORDINAL, Integer.class)).isEqualTo(3); + + file.set(COLUMN_FILES_ORDINAL, ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2)); + @SuppressWarnings("unchecked") + List roundTrippedColumnFiles = file.get(COLUMN_FILES_ORDINAL, List.class); + assertThat(roundTrippedColumnFiles).hasSize(2); + verifyColumnFile(COLUMN_FILE_1, roundTrippedColumnFiles.get(0)); + verifyColumnFile(COLUMN_FILE_2, roundTrippedColumnFiles.get(1)); } @Test @@ -346,7 +388,11 @@ void testJavaSerializationRoundTrip() throws IOException, ClassNotFoundException assertThat(deserialized.splitOffsets()).containsExactly(50L); assertThat(deserialized.tracking().manifestPos()).isEqualTo(3L); assertThat(deserialized.tracking().manifestLocation()).isEqualTo("s3://bucket/manifest.avro"); + assertThat(deserialized.tracking().latestColumnFileSnapshotId()).isEqualTo(42L); assertThat(deserialized.partition()).isEqualTo(newPartition(7, "music")); + assertThat(deserialized.columnFiles()).hasSize(2); + verifyColumnFile(COLUMN_FILE_1, deserialized.columnFiles().get(0)); + verifyColumnFile(COLUMN_FILE_2, deserialized.columnFiles().get(1)); } @Test @@ -370,7 +416,11 @@ void testKryoSerializationRoundTrip() throws IOException { assertThat(deserialized.splitOffsets()).containsExactly(50L); assertThat(deserialized.tracking().manifestPos()).isEqualTo(3L); assertThat(deserialized.tracking().manifestLocation()).isEqualTo("s3://bucket/manifest.avro"); + assertThat(deserialized.tracking().latestColumnFileSnapshotId()).isEqualTo(42L); assertThat(deserialized.partition()).isEqualTo(newPartition(7, "music")); + assertThat(deserialized.columnFiles()).hasSize(2); + verifyColumnFile(COLUMN_FILE_1, deserialized.columnFiles().get(0)); + verifyColumnFile(COLUMN_FILE_2, deserialized.columnFiles().get(1)); } static TrackedFileStruct createFullTrackedFile() { @@ -382,25 +432,31 @@ static TrackedFileStruct createFullTrackedFile() { .cardinality(5L) .build(); + Tracking tracking = TrackingBuilder.added(42L).columnFilesUpdated().build(); + TrackedFileStruct file = - (TrackedFileStruct) - TrackedFileBuilder.data(42L) - .formatVersion(FORMAT_VERSION_V4) - .location("s3://bucket/data/file.parquet") - .fileFormat(FileFormat.PARQUET) - .partition(newPartition(7, "music")) - .recordCount(100L) - .fileSizeInBytes(1024L) - .specId(0) - .sortOrderId(1) - .deletionVector(dv) - .keyMetadata(ByteBuffer.wrap(new byte[] {1, 2, 3})) - .splitOffsets(ImmutableList.of(50L)) - .build(); - - TrackingStruct tracking = (TrackingStruct) file.tracking(); - tracking.setManifestLocation("s3://bucket/manifest.avro"); - tracking.set(MANIFEST_POS_ORDINAL, 3L); + new TrackedFileStruct( + tracking, + FileContent.DATA, + FORMAT_VERSION_V4, + "s3://bucket/data/file.parquet", + FileFormat.PARQUET, + newPartition(7, "music"), + 100L, + 1024L, + 0, + null, + 1, + dv, + null, + ByteBuffer.wrap(new byte[] {1, 2, 3}), + ImmutableList.of(50L), + null, + ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2)); + + TrackingStruct trackingStruct = (TrackingStruct) file.tracking(); + trackingStruct.setManifestLocation("s3://bucket/manifest.avro"); + trackingStruct.set(MANIFEST_POS_ORDINAL, 3L); return file; } @@ -460,16 +516,31 @@ static TrackedFileStruct createTrackedFileWithStats() { .withFieldStats(fieldStatsList) .build(); - return (TrackedFileStruct) - TrackedFileBuilder.data(0L) - .formatVersion(FORMAT_VERSION_V4) - .location("s3://bucket/data/file.parquet") - .fileFormat(FileFormat.PARQUET) - .partition(new PartitionData(Types.StructType.of())) - .recordCount(100L) - .fileSizeInBytes(1024L) - .specId(0) - .contentStats(stats) - .build(); + Tracking tracking = TrackingBuilder.added(0L).columnFilesUpdated().build(); + + return new TrackedFileStruct( + tracking, + FileContent.DATA, + FORMAT_VERSION_V4, + "s3://bucket/data/file.parquet", + FileFormat.PARQUET, + new PartitionData(Types.StructType.of()), + 100L, + 1024L, + 0, + stats, + null, + null, + null, + null, + null, + null, + ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2)); + } + + private static void verifyColumnFile(ColumnFile expected, ColumnFile actual) { + assertThat(actual.fieldIds()).containsExactlyElementsOf(expected.fieldIds()); + assertThat(actual.location()).isEqualTo(expected.location()); + assertThat(actual.fileSizeInBytes()).isEqualTo(expected.fileSizeInBytes()); } } diff --git a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java index 0d1803022a23..cc6798f34d77 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackingStruct.java @@ -50,6 +50,8 @@ class TestTrackingStruct { TRACKING_FIELDS.indexOf(Tracking.DELETED_POSITIONS); private static final int REPLACED_POSITIONS_ORDINAL = TRACKING_FIELDS.indexOf(Tracking.REPLACED_POSITIONS); + private static final int LATEST_COLUMN_FILE_SNAPSHOT_ID_ORDINAL = + TRACKING_FIELDS.indexOf(Tracking.LATEST_COLUMN_FILE_SNAPSHOT_ID); @Test void testFieldAccess() { @@ -61,6 +63,7 @@ void testFieldAccess() { tracking.set(FILE_SEQUENCE_NUMBER_ORDINAL, 11L); tracking.set(DV_SNAPSHOT_ID_ORDINAL, 43L); tracking.set(FIRST_ROW_ID_ORDINAL, 1000L); + tracking.set(LATEST_COLUMN_FILE_SNAPSHOT_ID_ORDINAL, 15L); assertThat(tracking.status()).isEqualTo(EntryStatus.ADDED); assertThat(tracking.snapshotId()).isEqualTo(42L); @@ -70,6 +73,7 @@ void testFieldAccess() { assertThat(tracking.firstRowId()).isEqualTo(1000L); assertThat(tracking.deletedPositions()).isNull(); assertThat(tracking.replacedPositions()).isNull(); + assertThat(tracking.latestColumnFileSnapshotId()).isEqualTo(15L); } @Test @@ -78,6 +82,7 @@ void testCopy() { TrackingBuilder.from(manifestSourceTracking(), 1L) .deletedPositions(ByteBuffer.wrap(new byte[] {1, 2})) .replacedPositions(ByteBuffer.wrap(new byte[] {3, 4})) + .columnFilesUpdated() .build(); Tracking copy = tracking.copy(); @@ -90,6 +95,7 @@ void testCopy() { assertThat(copy.firstRowId()).isEqualTo(tracking.firstRowId()); assertThat(copy.deletedPositions()).isEqualTo(tracking.deletedPositions()); assertThat(copy.replacedPositions()).isEqualTo(tracking.replacedPositions()); + assertThat(copy.latestColumnFileSnapshotId()).isEqualTo(1L); // verify deep copy of ByteBuffer backing arrays assertThat(copy.deletedPositions().array()).isNotSameAs(tracking.deletedPositions().array()); @@ -142,7 +148,7 @@ void testDoNotInheritSequenceNumberForExistingEntries() { } @Test - void testDoNotInheritSequenceNumberForModifiedEntries() { + void testDoNotInheritSequenceNumberForModifiedIfAlreadySet() { TrackingStruct tracking = new TrackingStruct(Tracking.schema()); tracking.set(STATUS_ORDINAL, EntryStatus.MODIFIED.id()); tracking.set(DATA_SEQUENCE_NUMBER_ORDINAL, 5L); @@ -155,6 +161,24 @@ void testDoNotInheritSequenceNumberForModifiedEntries() { assertThat(tracking.fileSequenceNumber()).isEqualTo(6L); } + @Test + void testInheritDataSequenceNumberAfterColumnFilesChange() { + TrackingStruct source = new TrackingStruct(Tracking.schema()); + source.set(STATUS_ORDINAL, EntryStatus.MODIFIED.id()); + source.set(DATA_SEQUENCE_NUMBER_ORDINAL, 5L); + source.set(FILE_SEQUENCE_NUMBER_ORDINAL, 6L); + + Tracking tracking = TrackingBuilder.from(source, 10L).columnFilesUpdated().build(); + + assertThat(tracking.dataSequenceNumber()).isNull(); + assertThat(tracking.fileSequenceNumber()).isEqualTo(6); + + ((TrackingStruct) tracking).inheritFrom(createManifestTracking(100L, 60L)); + + assertThat(tracking.dataSequenceNumber()).isEqualTo(60L); + assertThat(tracking.fileSequenceNumber()).isEqualTo(6); + } + @Test void testExplicitValuesOverrideInheritance() { TrackingStruct tracking = new TrackingStruct(Tracking.schema()); @@ -222,13 +246,14 @@ private static Tracking createManifestTracking(long snapshotId, long sequenceNum @Test void testAddedWithSameCommitDvStaysAdded() { - Tracking tracking = TrackingBuilder.added(42L).dvUpdated().build(); + Tracking tracking = TrackingBuilder.added(42L).dvUpdated().columnFilesUpdated().build(); assertThat(tracking.status()).isEqualTo(EntryStatus.ADDED); assertThat(tracking.snapshotId()).isEqualTo(42L); assertThat(tracking.dvSnapshotId()).isEqualTo(42L); assertThat(tracking.deletedPositions()).isNull(); assertThat(tracking.replacedPositions()).isNull(); + assertThat(tracking.latestColumnFileSnapshotId()).isEqualTo(42L); // sequence numbers and firstRowId remain null; populated by inheritance assertThat(tracking.dataSequenceNumber()).isNull(); assertThat(tracking.fileSequenceNumber()).isNull(); @@ -247,6 +272,8 @@ void testExistingBuilderPreservesSourceFields() { assertThat(existing.fileSequenceNumber()).isEqualTo(source.fileSequenceNumber()); assertThat(existing.dvSnapshotId()).isEqualTo(source.dvSnapshotId()); assertThat(existing.firstRowId()).isEqualTo(source.firstRowId()); + assertThat(existing.latestColumnFileSnapshotId()) + .isEqualTo(source.latestColumnFileSnapshotId()); } @Test @@ -261,6 +288,7 @@ void testDeleteUpdatesSnapshotIdAndPreservesRest() { assertThat(deleted.fileSequenceNumber()).isEqualTo(source.fileSequenceNumber()); assertThat(deleted.dvSnapshotId()).isEqualTo(source.dvSnapshotId()); assertThat(deleted.firstRowId()).isEqualTo(source.firstRowId()); + assertThat(deleted.latestColumnFileSnapshotId()).isEqualTo(source.latestColumnFileSnapshotId()); } @Test @@ -275,6 +303,8 @@ void testReplaceUpdatesSnapshotIdAndPreservesRest() { assertThat(replaced.fileSequenceNumber()).isEqualTo(source.fileSequenceNumber()); assertThat(replaced.dvSnapshotId()).isEqualTo(source.dvSnapshotId()); assertThat(replaced.firstRowId()).isEqualTo(source.firstRowId()); + assertThat(replaced.latestColumnFileSnapshotId()) + .isEqualTo(source.latestColumnFileSnapshotId()); } @Test @@ -308,6 +338,12 @@ void testDvUpdatedProducesModifiedAndAdvancesDvSnapshotId() { assertThat(modified.dvSnapshotId()).isEqualTo(999L); } + @Test + void testExistingBuilderAllowsColumnFileMutation() { + Tracking existing = TrackingBuilder.from(sourceTracking(), 900L).columnFilesUpdated().build(); + assertThat(existing.latestColumnFileSnapshotId()).isEqualTo(900L); + } + @Test void testManifestDVMutatorsRejectedOnAdded() { assertThatThrownBy( @@ -463,6 +499,60 @@ void testManifestDVPositionsProduceModified() { assertThat(modified.deletedPositions()).isEqualTo(deletedBytes); } + @Test + void testManifestPositionsWithColumnFilesUpdated() { + ByteBuffer deletedBytes = ByteBuffer.wrap(new byte[] {1}); + Tracking withDeletedPositions = + TrackingBuilder.from(manifestSourceTracking(), 999L) + .columnFilesUpdated() + .deletedPositions(deletedBytes) + .build(); + + assertThat(withDeletedPositions.status()).isEqualTo(EntryStatus.MODIFIED); + assertThat(withDeletedPositions.latestColumnFileSnapshotId()).isEqualTo(999L); + assertThat(withDeletedPositions.dvSnapshotId()).isEqualTo(999L); + assertThat(withDeletedPositions.deletedPositions()).isEqualTo(deletedBytes); + + ByteBuffer replacedBytes = ByteBuffer.wrap(new byte[] {2}); + Tracking withReplacedPositions = + TrackingBuilder.from(manifestSourceTracking(), 999L) + .columnFilesUpdated() + .replacedPositions(replacedBytes) + .build(); + + assertThat(withReplacedPositions.status()).isEqualTo(EntryStatus.MODIFIED); + assertThat(withReplacedPositions.latestColumnFileSnapshotId()).isEqualTo(999L); + assertThat(withReplacedPositions.dvSnapshotId()).isEqualTo(999L); + assertThat(withReplacedPositions.replacedPositions()).isEqualTo(replacedBytes); + } + + @Test + void testColumnFilesUpdatedWithManifestPositions() { + ByteBuffer deletedBytes = ByteBuffer.wrap(new byte[] {1}); + Tracking withDeletedPositions = + TrackingBuilder.from(manifestSourceTracking(), 999L) + .deletedPositions(deletedBytes) + .columnFilesUpdated() + .build(); + + assertThat(withDeletedPositions.status()).isEqualTo(EntryStatus.MODIFIED); + assertThat(withDeletedPositions.latestColumnFileSnapshotId()).isEqualTo(999L); + assertThat(withDeletedPositions.dvSnapshotId()).isEqualTo(999L); + assertThat(withDeletedPositions.deletedPositions()).isEqualTo(deletedBytes); + + ByteBuffer replacedBytes = ByteBuffer.wrap(new byte[] {2}); + Tracking withReplacedPositions = + TrackingBuilder.from(manifestSourceTracking(), 999L) + .replacedPositions(replacedBytes) + .columnFilesUpdated() + .build(); + + assertThat(withReplacedPositions.status()).isEqualTo(EntryStatus.MODIFIED); + assertThat(withReplacedPositions.latestColumnFileSnapshotId()).isEqualTo(999L); + assertThat(withReplacedPositions.dvSnapshotId()).isEqualTo(999L); + assertThat(withReplacedPositions.replacedPositions()).isEqualTo(replacedBytes); + } + @Test void testIsLiveDelegatesToStatus() { assertThat(sourceTrackingWithStatus(EntryStatus.ADDED).isLive()).isTrue(); @@ -480,6 +570,7 @@ void testInternalSetIgnoresUnknownOrdinal() { tracking.set(FIRST_ROW_ID_ORDINAL, 1000L); tracking.set(DELETED_POSITIONS_ORDINAL, ByteBuffer.wrap(new byte[] {1, 2})); tracking.set(REPLACED_POSITIONS_ORDINAL, ByteBuffer.wrap(new byte[] {3, 4})); + tracking.set(LATEST_COLUMN_FILE_SNAPSHOT_ID_ORDINAL, 49L); // unknown ordinals from a newer format version are silently ignored tracking.internalSet(99, "value from a newer format"); @@ -493,6 +584,7 @@ void testInternalSetIgnoresUnknownOrdinal() { assertThat(tracking.firstRowId()).isEqualTo(1000L); assertThat(tracking.deletedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2})); assertThat(tracking.replacedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {3, 4})); + assertThat(tracking.latestColumnFileSnapshotId()).isEqualTo(49L); } @Test @@ -516,7 +608,7 @@ void testProjectedStructLike() { @Test void testJavaSerializationRoundTripForDataFile() throws IOException, ClassNotFoundException { - Tracking tracking = TrackingBuilder.added(42L).dvUpdated().build(); + Tracking tracking = TrackingBuilder.added(42L).dvUpdated().columnFilesUpdated().build(); Tracking deserialized = TestHelpers.roundTripSerialize(tracking); @@ -525,6 +617,7 @@ void testJavaSerializationRoundTripForDataFile() throws IOException, ClassNotFou assertThat(deserialized.dvSnapshotId()).isEqualTo(42L); assertThat(deserialized.deletedPositions()).isNull(); assertThat(deserialized.replacedPositions()).isNull(); + assertThat(deserialized.latestColumnFileSnapshotId()).isEqualTo(42L); } @Test @@ -533,6 +626,7 @@ void testJavaSerializationRoundTripForManifest() throws IOException, ClassNotFou TrackingBuilder.from(manifestSourceTracking(), 1L) .deletedPositions(ByteBuffer.wrap(new byte[] {1, 2})) .replacedPositions(ByteBuffer.wrap(new byte[] {3, 4})) + .columnFilesUpdated() .build(); Tracking deserialized = TestHelpers.roundTripSerialize(tracking); @@ -541,11 +635,12 @@ void testJavaSerializationRoundTripForManifest() throws IOException, ClassNotFou assertThat(deserialized.dvSnapshotId()).isEqualTo(1L); assertThat(deserialized.deletedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2})); assertThat(deserialized.replacedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {3, 4})); + assertThat(deserialized.latestColumnFileSnapshotId()).isEqualTo(1L); } @Test void testKryoSerializationRoundTripForDataFile() throws IOException { - Tracking tracking = TrackingBuilder.added(42L).dvUpdated().build(); + Tracking tracking = TrackingBuilder.added(42L).dvUpdated().columnFilesUpdated().build(); Tracking deserialized = TestHelpers.KryoHelpers.roundTripSerialize(tracking); @@ -554,6 +649,7 @@ void testKryoSerializationRoundTripForDataFile() throws IOException { assertThat(deserialized.dvSnapshotId()).isEqualTo(42L); assertThat(deserialized.deletedPositions()).isNull(); assertThat(deserialized.replacedPositions()).isNull(); + assertThat(deserialized.latestColumnFileSnapshotId()).isEqualTo(42L); } @Test @@ -562,6 +658,7 @@ void testKryoSerializationRoundTripForManifest() throws IOException { TrackingBuilder.from(manifestSourceTracking(), 1L) .deletedPositions(ByteBuffer.wrap(new byte[] {1, 2})) .replacedPositions(ByteBuffer.wrap(new byte[] {3, 4})) + .columnFilesUpdated() .build(); Tracking deserialized = TestHelpers.KryoHelpers.roundTripSerialize(tracking); @@ -570,6 +667,7 @@ void testKryoSerializationRoundTripForManifest() throws IOException { assertThat(deserialized.dvSnapshotId()).isEqualTo(1L); assertThat(deserialized.deletedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {1, 2})); assertThat(deserialized.replacedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {3, 4})); + assertThat(deserialized.latestColumnFileSnapshotId()).isEqualTo(1L); } private static TrackingStruct sourceTracking() {