diff --git a/core/src/main/java/org/apache/iceberg/ColumnFile.java b/core/src/main/java/org/apache/iceberg/ColumnFile.java new file mode 100644 index 000000000000..6f0dc9de9c7d --- /dev/null +++ b/core/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. */ +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 info. */ + ColumnFile copy(); +} 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..a8c7e21d22fa --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/ColumnFileStruct.java @@ -0,0 +1,171 @@ +/* + * 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.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); + } + + private ColumnFileStruct(int[] fieldIds, String location, long fileSizeInBytes) { + super(BASE_TYPE.fields().size()); + this.fieldIds = 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 == null ? "null" : fieldIds()) + .add("location", location) + .add("file_size_in_bytes", fileSizeInBytes) + .toString(); + } + + static class Builder { + private int[] 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"); + this.fieldIds = ArrayUtil.toIntArray(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 8a6335972888..b27ae5029b7e 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFile.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFile.java @@ -91,6 +91,12 @@ 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 update files"); static Types.StructType schemaWithContentStats( Types.StructType partitionType, Types.StructType contentStatsType) { @@ -110,7 +116,8 @@ static Types.StructType schemaWithContentStats( MANIFEST_INFO, KEY_METADATA, SPLIT_OFFSETS, - EQUALITY_IDS); + EQUALITY_IDS, + COLUMN_FILES); } /** Returns the tracking information for this entry. */ @@ -158,6 +165,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/TrackedFileStruct.java b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java index 4830f69d6bf1..44575b8422e5 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java @@ -21,8 +21,10 @@ import java.io.Serializable; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.Collections; import java.util.List; 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.types.Type; @@ -65,7 +67,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 String location = null; @@ -84,6 +87,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) { @@ -155,6 +159,10 @@ 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); @@ -279,6 +292,8 @@ private Object getByPos(int pos) { return splitOffsets(); case 14: return equalityIds(); + case 15: + return columnFiles(); default: throw new UnsupportedOperationException("Unknown field ordinal: " + pos); } @@ -333,6 +348,10 @@ protected void internalSet(int pos, T value) { case 14: this.equalityIds = ArrayUtil.toIntArray((List) value); break; + case 15: + this.columnFiles = + ((List) value).stream().map(ColumnFile::copy).collect(Collectors.toList()); + break; default: // ignore the object, it must be from a newer version of the format } @@ -356,6 +375,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 8003ed82ea9c..09b85617e04c 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 bf3cb3a0666c..82c9dd87401d 100644 --- a/core/src/main/java/org/apache/iceberg/TrackingBuilder.java +++ b/core/src/main/java/org/apache/iceberg/TrackingBuilder.java @@ -33,6 +33,7 @@ class TrackingBuilder { private Long dvSnapshotId; private byte[] deletedPositions; private byte[] replacedPositions; + private Long latestColumnFileSnapshotId; /** * Creates a builder for a newly added file. @@ -83,6 +84,7 @@ private TrackingBuilder(long newSnapshotId) { this.dvSnapshotId = null; this.deletedPositions = null; this.replacedPositions = null; + this.latestColumnFileSnapshotId = null; } private TrackingBuilder(Tracking source, long newSnapshotId) { @@ -97,6 +99,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. */ @@ -109,6 +112,11 @@ TrackingBuilder dvUpdated() { return this; } + TrackingBuilder columnFileUpdated() { + this.latestColumnFileSnapshotId = newSnapshotId; + return this; + } + TrackingBuilder deletedPositions(ByteBuffer positions) { Preconditions.checkState( status == EntryStatus.EXISTING, "Cannot set deleted positions on %s entry", status); @@ -140,7 +148,8 @@ Tracking build() { dvSnapshotId, firstRowId, deletedPositions, - replacedPositions); + replacedPositions, + latestColumnFileSnapshotId); } private static Tracking terminal(EntryStatus to, Tracking source, long newSnapshotId) { @@ -154,7 +163,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..eedb24df38eb 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) { @@ -174,6 +179,11 @@ public ByteBuffer replacedPositions() { return replacedPositions != null ? ByteBuffer.wrap(replacedPositions) : null; } + @Override + public Long latestColumnFileSnapshotId() { + return latestColumnFileSnapshotId; + } + @Override public String manifestLocation() { return manifestLocation; @@ -213,6 +223,8 @@ private Object getByPos(int pos) { case 7: return replacedPositions(); case 8: + return latestColumnFileSnapshotId; + case 9: return manifestPos; default: throw new UnsupportedOperationException("Unknown field ordinal: " + pos); @@ -247,6 +259,9 @@ protected void internalSet(int pos, T value) { this.replacedPositions = ByteBuffers.toByteArray((ByteBuffer) value); break; case 8: + this.latestColumnFileSnapshotId = (Long) value; + break; + case 9: this.manifestPos = (long) value; break; default: @@ -265,6 +280,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..7de57d4ee93a --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestColumnFileStruct.java @@ -0,0 +1,208 @@ +/* + * 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; + +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 = + ColumnFileStruct.builder() + .fieldIds(FIELD_IDS) + .location(LOCATION) + .fileSizeInBytes(1024L) + .build(); + + assertThat(columnFile.fieldIds()).containsExactly(1, 2, 3); + assertThat(columnFile.location()).isEqualTo(LOCATION); + assertThat(columnFile.fileSizeInBytes()).isEqualTo(1024L); + } + + @Test + void testCopy() { + ColumnFile columnFile = + ColumnFileStruct.builder() + .fieldIds(FIELD_IDS) + .location(LOCATION) + .fileSizeInBytes(2048L) + .build(); + + ColumnFile copy = columnFile.copy(); + + assertThat(copy.fieldIds()).containsExactly(1, 2, 3); + assertThat(copy.location()).isEqualTo(LOCATION); + assertThat(copy.fileSizeInBytes()).isEqualTo(2048L); + + // verify deep copy + assertThat(copy.fieldIds()).isNotSameAs(columnFile.fieldIds()); + } + + @Test + void testStructLikeSize() { + ColumnFileStruct columnFile = new ColumnFileStruct(ColumnFile.schema()); + assertThat(columnFile.size()).isEqualTo(3); + } + + @Test + void testStructLikeGetSet() { + ColumnFileStruct columnFile = new ColumnFileStruct(ColumnFile.schema()); + + columnFile.set(0, Lists.newArrayList(1, 2, 3, 4)); + columnFile.set(1, LOCATION); + columnFile.set(2, 128L); + + assertThat(columnFile.get(0, List.class)).containsExactly(1, 2, 3, 4); + 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); + } + + @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(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"); + } + + @Test + void testJavaSerializationRoundTrip() throws IOException, ClassNotFoundException { + ColumnFile columnFile = + ColumnFileStruct.builder() + .fieldIds(FIELD_IDS) + .location(LOCATION) + .fileSizeInBytes(1024L) + .build(); + + ColumnFile deserialized = TestHelpers.roundTripSerialize(columnFile); + + assertThat(deserialized.fieldIds()).containsExactly(1, 2, 3); + assertThat(deserialized.location()).isEqualTo(LOCATION); + assertThat(deserialized.fileSizeInBytes()).isEqualTo(1024L); + } + + @Test + void testKryoSerializationRoundTrip() throws IOException { + ColumnFile columnFile = + ColumnFileStruct.builder() + .fieldIds(FIELD_IDS) + .location(LOCATION) + .fileSizeInBytes(1024L) + .build(); + + ColumnFile deserialized = TestHelpers.KryoHelpers.roundTripSerialize(columnFile); + + assertThat(deserialized.fieldIds()).containsExactly(1, 2, 3); + assertThat(deserialized.location()).isEqualTo(LOCATION); + assertThat(deserialized.fileSizeInBytes()).isEqualTo(1024L); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java index 0d850ee4c886..955b50adf663 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java @@ -59,7 +59,8 @@ public void schemaWithContentStatsFieldOrder() { "manifest_info", "key_metadata", "split_offsets", - "equality_ids"); + "equality_ids", + "column_files"); } @Test @@ -69,7 +70,8 @@ public void schemaWithContentStatsFieldIds() { assertThat(fields) .extracting(Types.NestedField::fieldId) - .containsExactly(147, 134, 100, 101, 103, 104, 141, 102, 146, 140, 148, 150, 131, 132, 135); + .containsExactly( + 147, 134, 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/TestTrackedFileStruct.java b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java index 8891db408be5..fa93e1df24d8 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java @@ -34,6 +34,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(); // Ordinal of MetadataColumns.ROW_POSITION within TrackingStruct's BASE_TYPE, // which appends ROW_POSITION after the Tracking schema fields. @@ -76,6 +88,7 @@ void testFieldAccess() { file.set(12, ByteBuffer.wrap(new byte[] {1, 2, 3})); file.set(13, ImmutableList.of(100L, 200L)); file.set(14, ImmutableList.of(1, 2, 3)); + file.set(15, ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2)); assertThat(file.tracking()).isNotNull(); assertThat(file.tracking().status()).isEqualTo(EntryStatus.ADDED); @@ -92,6 +105,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); @@ -177,7 +194,11 @@ 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); + verifyColumnFile(COLUMN_FILE_1, copy.columnFiles().get(0)); + verifyColumnFile(COLUMN_FILE_2, copy.columnFiles().get(1)); } @Test @@ -211,14 +232,18 @@ void testCopyIsDeep() { TrackedFile copy = file.copy(); - // keyMetadata should be a deep copy assertThat(copy.keyMetadata()).isNotSameAs(file.keyMetadata()); + assertThat(copy.columnFiles()).isNotSameAs(file.columnFiles()); + assertThat(copy.columnFiles()).hasSize(file.columnFiles().size()); + for (int i = 0; i < file.columnFiles().size(); ++i) { + assertThat(copy.columnFiles().get(i)).isNotSameAs(file.columnFiles().get(i)); + } } @Test void testStructLikeSize() { TrackedFileStruct file = new TrackedFileStruct(); - assertThat(file.size()).isEqualTo(15); + assertThat(file.size()).isEqualTo(16); } @Test @@ -233,6 +258,13 @@ void testStructLikeGetSet() { file.set(4, 999L); assertThat(file.get(4, Long.class)).isEqualTo(999L); + + file.set(15, ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2)); + @SuppressWarnings("unchecked") + List roundTrippedColumnFiles = file.get(15, List.class); + assertThat(roundTrippedColumnFiles).hasSize(2); + verifyColumnFile(COLUMN_FILE_1, roundTrippedColumnFiles.get(0)); + verifyColumnFile(COLUMN_FILE_2, roundTrippedColumnFiles.get(1)); } @Test @@ -275,6 +307,12 @@ void testContentStatsNullWhenNotSet() { assertThat(file.contentStats()).isNull(); } + @Test + void testColumnFilesNullWhenNotSet() { + TrackedFileStruct file = new TrackedFileStruct(); + assertThat(file.columnFiles()).isNull(); + } + @Test void testAllFileContentTypesSupported() { for (FileContent content : FileContent.values()) { @@ -304,7 +342,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 @@ -327,11 +369,16 @@ 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() { - TrackingStruct tracking = (TrackingStruct) TrackingBuilder.added(42L).build(); + TrackingStruct tracking = + (TrackingStruct) TrackingBuilder.added(42L).columnFileUpdated().build(); tracking.setManifestLocation("s3://bucket/manifest.avro"); tracking.set(MANIFEST_POS_ORDINAL, 3L); @@ -357,6 +404,7 @@ static TrackedFileStruct createFullTrackedFile() { file.set(10, dv); file.set(12, ByteBuffer.wrap(new byte[] {1, 2, 3})); file.set(13, ImmutableList.of(50L)); + file.set(15, ImmutableList.of(COLUMN_FILE_1, COLUMN_FILE_2)); return file; } @@ -430,4 +478,10 @@ static TrackedFileStruct createTrackedFileWithStats() { return file; } + + 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 0bbd8b348006..c37b51030881 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})) + .columnFileUpdated() .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()); @@ -225,13 +231,14 @@ private static Tracking createManifestTracking(long snapshotId, long sequenceNum @Test void testAddedBuilder() { - Tracking tracking = TrackingBuilder.added(42L).dvUpdated().build(); + Tracking tracking = TrackingBuilder.added(42L).dvUpdated().columnFileUpdated().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(); @@ -250,6 +257,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 @@ -264,6 +273,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 @@ -278,6 +288,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 @@ -305,6 +317,12 @@ void testExistingBuilderAllowsDvMutation() { assertThat(existing.dvSnapshotId()).isEqualTo(999L); } + @Test + void testExistingBuilderAllowsColumnFileMutation() { + Tracking existing = TrackingBuilder.from(sourceTracking(), 900L).columnFileUpdated().build(); + assertThat(existing.latestColumnFileSnapshotId()).isEqualTo(900L); + } + @Test void testManifestDVMutatorsRejectedOnAdded() { assertThatThrownBy( @@ -444,6 +462,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"); @@ -457,6 +476,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 @@ -538,6 +558,33 @@ void testExistingWithManifestDVPositionsKryoSerializationRoundTrip() throws IOEx assertThat(deserialized.replacedPositions()).isEqualTo(ByteBuffer.wrap(new byte[] {3, 4})); } + @Test + void testAddedWithColumnFileSnapshotIdJavaSerializationRoundTrip() + throws IOException, ClassNotFoundException { + Tracking tracking = TrackingBuilder.added(42L).columnFileUpdated().build(); + + Tracking deserialized = TestHelpers.roundTripSerialize(tracking); + + assertThat(deserialized.status()).isEqualTo(EntryStatus.ADDED); + assertThat(deserialized.snapshotId()).isEqualTo(42L); + assertThat(deserialized.deletedPositions()).isNull(); + assertThat(deserialized.replacedPositions()).isNull(); + assertThat(deserialized.latestColumnFileSnapshotId()).isEqualTo(42L); + } + + @Test + void testAddedWithColumnFileSnapshotIdKryoSerializationRoundTrip() throws IOException { + Tracking tracking = TrackingBuilder.added(42L).columnFileUpdated().build(); + + Tracking deserialized = TestHelpers.KryoHelpers.roundTripSerialize(tracking); + + assertThat(deserialized.status()).isEqualTo(EntryStatus.ADDED); + assertThat(deserialized.snapshotId()).isEqualTo(42L); + assertThat(deserialized.deletedPositions()).isNull(); + assertThat(deserialized.replacedPositions()).isNull(); + assertThat(deserialized.latestColumnFileSnapshotId()).isEqualTo(42L); + } + private static TrackingStruct sourceTracking() { TrackingStruct tracking = new TrackingStruct(Tracking.schema()); tracking.set(STATUS_ORDINAL, EntryStatus.ADDED.id());