stevenzwu commented on code in PR #18853:
URL: https://github.com/apache/iceberg/pull/18853#discussion_r4517088329
##########
core/src/test/java/org/apache/iceberg/V4TestHelpers.java:
##########
@@ -54,6 +58,16 @@ static TrackedFile dataFile(
return dataFile(ADDED_TRACKING, location, specId, partition, stats, dv,
null);
}
+ static TrackedFile dataFile(
Review Comment:
a new overload for the column files
##########
core/src/main/java/org/apache/iceberg/TrackedFileStruct.java:
##########
@@ -102,7 +102,7 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
long recordCount,
long fileSizeInBytes,
Integer specId,
- PartitionData partition,
+ StructLike partition,
Review Comment:
Widening this from `PartitionData` to `StructLike` so the v4 write wrapper's
`copy()` can pass `StructLikeUtil.copy(wrapped.partition())`.
`TrackedFile.partition()` returns a `StructLike` and isn't always a
`PartitionData` (e.g. the `StructProjection` view returned when a file's spec
differs from the union partition type, or an engine `DataFile`'s partition), so
`PartitionData.copy()` doesn't apply.
This also matches the existing `TrackedFileStruct` copy constructor, which
already copies the partition with `StructLikeUtil.copy`, and the field itself
is already a `StructLike`.
Existing callers all pass `PartitionData` or null, so nothing else changes.
##########
core/src/test/java/org/apache/iceberg/TestV4TrackedFileWrappers.java:
##########
@@ -0,0 +1,810 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.apache.iceberg.V4TestHelpers.ADDED_TRACKING;
+import static org.apache.iceberg.V4TestHelpers.columnFile;
+import static org.apache.iceberg.V4TestHelpers.copyWithTracking;
+import static org.apache.iceberg.V4TestHelpers.dataFile;
+import static org.apache.iceberg.V4TestHelpers.dataFileWithColumnFiles;
+import static org.apache.iceberg.V4TestHelpers.dataFileWithStats;
+import static org.apache.iceberg.V4TestHelpers.deletionVector;
+import static org.apache.iceberg.V4TestHelpers.manifestRef;
+import static org.apache.iceberg.V4TestHelpers.manifestRefWithStats;
+import static org.apache.iceberg.V4TestHelpers.partition;
+import static org.apache.iceberg.V4TestHelpers.readTrackedFiles;
+import static org.apache.iceberg.V4TestHelpers.writeTrackedFiles;
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import org.apache.iceberg.V4Metadata.LeafTrackedFileWrapper;
+import org.apache.iceberg.V4Metadata.RootTrackedFileWrapper;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+class TestV4TrackedFileWrappers {
+ private static final long COMMIT_SNAPSHOT_ID = 42L;
+ private static final long COMMIT_SEQUENCE_NUMBER = 34L;
+ private static final long PREVIOUS_SNAPSHOT_ID = 7L;
+ private static final long PREVIOUS_SEQUENCE_NUMBER = 5L;
+ private static final long PREVIOUS_FIRST_ROW_ID = 500L;
+ private static final long FIRST_ROW_ID = 1_000L;
+
+ private static final Schema TABLE_SCHEMA =
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()), optional(2, "data",
Types.StringType.get()));
+ private static final PartitionSpec ID_SPEC =
+
PartitionSpec.builderFor(TABLE_SCHEMA).withSpecId(1).identity("id").build();
+ private static final PartitionSpec DATA_SPEC =
+ PartitionSpec.builderFor(TABLE_SCHEMA)
+ .withSpecId(2)
+ .add(2 /* sourceId */, 1001 /* fieldId */, "data",
Transforms.identity())
+ .build();
+ private static final Map<Integer, PartitionSpec> ID_SPECS =
+ ImmutableMap.of(ID_SPEC.specId(), ID_SPEC);
+ private static final Map<Integer, PartitionSpec> UNPARTITIONED_SPECS =
+ ImmutableMap.of(PartitionSpec.unpartitioned().specId(),
PartitionSpec.unpartitioned());
+
+ private static final Types.StructType STATS_TYPE =
+ StatsUtil.statsWriteSchema(
+ TABLE_SCHEMA, MetricsTestUtil.from(ImmutableMap.of(), TABLE_SCHEMA));
+ private static final FieldStats<Integer> ID_STATS =
+ new FieldStatsStruct<>(
+ STATS_TYPE.fieldType("id").asStructType(),
+ 1 /* lowerBound */,
+ 99 /* upperBound */,
+ true /* tightBounds */,
+ 100L /* valueCount */,
+ 0L /* nullValueCount */,
+ 0L /* nanValueCount */,
+ 0L /* totalBytes */);
+ private static final FieldStats<String> DATA_STATS =
+ new FieldStatsStruct<>(
+ STATS_TYPE.fieldType("data").asStructType(),
+ "a",
+ "z",
+ false /* tightBounds */,
+ 100L /* valueCount */,
+ 10L /* nullValueCount */,
+ 0L /* nanValueCount */,
+ 500L /* totalBytes */);
+
+ private static final Tracking EXISTING_TRACKING =
+ new TrackingStruct(
+ EntryStatus.EXISTING,
+ PREVIOUS_SNAPSHOT_ID,
+ PREVIOUS_SEQUENCE_NUMBER,
+ PREVIOUS_SEQUENCE_NUMBER,
+ null /* modifiedSnapshotId */,
+ PREVIOUS_FIRST_ROW_ID,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+ private static final ManifestInfo MANIFEST_INFO =
+ new ManifestInfoStruct(
+ 49 /* addedFilesCount */,
+ 51 /* existingFilesCount */,
+ 0 /* deletedFilesCount */,
+ 0 /* replacedFilesCount */,
+ 0 /* modifiedFilesCount */,
+ 4_900L /* addedRowsCount */,
+ 5_100L /* existingRowsCount */,
+ 0L /* deletedRowsCount */,
+ 0L /* replacedRowsCount */,
+ 0L /* modifiedRowsCount */,
+ PREVIOUS_SEQUENCE_NUMBER,
+ null /* dv */,
+ 4 /* formatVersion */);
+
+ private final InMemoryFileIO io = new InMemoryFileIO();
+
+ @Test
+ void leafWritesDataFile() throws IOException {
+ TrackedFile file =
+ dataFile(
+ ADDED_TRACKING,
+ "s3://bucket/table/id=1/file.parquet",
+ ID_SPEC.specId(),
+ partition(ID_SPEC, 1),
+ null /* stats */,
+ deletionVector("s3://bucket/table/dv.puffin"),
+ ImmutableList.of(
+ columnFile("s3://bucket/table/id=1/file-cols-1.parquet",
ImmutableSet.of(1)),
+ columnFile(
+ "s3://bucket/table/id=1/file-cols-2.parquet",
+ ImmutableSet.of(2),
+ ByteBuffer.wrap(new byte[] {4, 5}))));
+ TrackedFile expected = copyWithTracking(file, EXISTING_TRACKING);
+
+ LeafTrackedFileWrapper wrapper = new LeafTrackedFileWrapper(ID_SPECS,
Types.StructType.of());
+ TrackedFile actual =
+ writeAndRead(ID_SPEC.partitionType(), wrapper.wrap(file,
EXISTING_TRACKING));
+
+ assertThat(actual)
+
.usingComparator(V4TestComparators.trackedFile(ID_SPEC.partitionType()))
+ .isEqualTo(expected);
+ }
+
+ @Test
+ void leafWritesAdaptedDataFile() throws IOException {
+ DataFile dataFile =
+ DataFiles.builder(ID_SPEC)
+ .withPath("s3://bucket/table/id=1/file.parquet")
+ .withPartition(partition(ID_SPEC, 1))
+ .withFileSizeInBytes(1024L)
+ .withMetrics(
+ new Metrics(
+ 100L /* rowCount */,
+ null /* columnSizes */,
+ ImmutableMap.of(1, 100L, 2, 100L),
+ ImmutableMap.of(1, 0L, 2, 10L),
+ null /* nanValueCounts */,
+ ImmutableMap.of(
+ 1, Conversions.toByteBuffer(Types.IntegerType.get(),
1),
+ 2, Conversions.toByteBuffer(Types.StringType.get(),
"a")),
+ ImmutableMap.of(
+ 1, Conversions.toByteBuffer(Types.IntegerType.get(),
99),
+ 2, Conversions.toByteBuffer(Types.StringType.get(),
"z"))))
+
.withSortOrder(SortOrder.builderFor(TABLE_SCHEMA).withOrderId(1).asc("id").build())
+ .withEncryptionKeyMetadata(ByteBuffer.wrap(new byte[] {1, 2, 3}))
+ .withSplitOffsets(ImmutableList.of(4L, 512L))
+ .build();
+ TrackedFile file =
TrackedFileAdapters.forDataFile(TABLE_SCHEMA).wrap(dataFile);
+ TrackedFile expected = copyWithTracking(file, ADDED_TRACKING);
+
+ LeafTrackedFileWrapper wrapper = new LeafTrackedFileWrapper(ID_SPECS,
STATS_TYPE);
+ TrackedFile actual =
+ writeAndRead(ID_SPEC.partitionType(), STATS_TYPE, wrapper.wrap(file,
ADDED_TRACKING));
+
+ assertThat(actual)
+
.usingComparator(V4TestComparators.trackedFile(ID_SPEC.partitionType()))
+ .isEqualTo(expected);
+ }
+
+ @Test
+ void leafWritesNullSpecIdForUnpartitionedFile() throws IOException {
+ DataFile dataFile =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("s3://bucket/table/file.parquet")
+ .withFileSizeInBytes(1024L)
+ .withRecordCount(100L)
+ .build();
+ TrackedFile file =
TrackedFileAdapters.forDataFile(TABLE_SCHEMA).wrap(dataFile);
+
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+ TrackedFile actual = writeAndRead(Types.StructType.of(),
wrapper.wrap(file, ADDED_TRACKING));
+
+ assertThat(actual.specId()).isNull();
+ assertThat(actual.partition()).isNull();
+ }
+
+ @Test
+ void leafProjectsPartitionToUnionType() throws IOException {
+ Map<Integer, PartitionSpec> specs =
+ ImmutableMap.of(ID_SPEC.specId(), ID_SPEC, DATA_SPEC.specId(),
DATA_SPEC);
+ Types.StructType unionType =
Partitioning.unionPartitionTypes(specs.values());
+ TrackedFile file =
+ dataFile(
+ "s3://bucket/table/data=a/file.parquet", DATA_SPEC.specId(),
partition(DATA_SPEC, "a"));
+
+ LeafTrackedFileWrapper wrapper = new LeafTrackedFileWrapper(specs,
Types.StructType.of());
+ TrackedFile actual = writeAndRead(unionType, wrapper.wrap(file,
file.tracking()));
+
+ PartitionData expected = new PartitionData(unionType);
+ expected.set(unionType.fields().indexOf(unionType.field("data")), "a");
+ assertThat(actual.partition())
+ .usingComparator(Comparators.forType(unionType))
+ .isEqualTo(expected);
+ }
+
+ @Test
+ void leafWritesStatsUsingManifestStatsType() throws IOException {
+ Types.StructType reorderedType =
+ Types.StructType.of(STATS_TYPE.field("data"), STATS_TYPE.field("id"));
+ ContentStats stats = StatsTestUtil.contentStats(reorderedType, ID_STATS,
DATA_STATS);
+ TrackedFile file = dataFileWithStats("s3://bucket/table/file.parquet",
stats);
+
+ LeafTrackedFileWrapper wrapper = new
LeafTrackedFileWrapper(UNPARTITIONED_SPECS, STATS_TYPE);
+ TrackedFile actual =
+ writeAndRead(Types.StructType.of(), STATS_TYPE, wrapper.wrap(file,
file.tracking()));
+
+ assertThat(actual.contentStats())
+ .usingComparator(V4TestComparators.CONTENT_STATS)
+ .isEqualTo(stats);
+ }
+
+ @Test
+ void leafWritesNullStatsForColumnWithoutStats() throws IOException {
+ ContentStats stats = StatsTestUtil.contentStats(STATS_TYPE, ID_STATS);
+ TrackedFile file = dataFileWithStats("s3://bucket/table/file.parquet",
stats);
+
+ LeafTrackedFileWrapper wrapper = new
LeafTrackedFileWrapper(UNPARTITIONED_SPECS, STATS_TYPE);
+ TrackedFile actual =
+ writeAndRead(Types.StructType.of(), STATS_TYPE, wrapper.wrap(file,
file.tracking()));
+
+ assertThat(actual.contentStats().statsFor(DATA_STATS.fieldId())).isNull();
+ }
+
+ @Test
+ void leafReturnsPartitionWithSpecType() {
+ Map<Integer, PartitionSpec> specs =
+ ImmutableMap.of(ID_SPEC.specId(), ID_SPEC, DATA_SPEC.specId(),
DATA_SPEC);
+ PartitionData partition = partition(DATA_SPEC, "a");
+ TrackedFile file =
+ dataFile("s3://bucket/table/data=a/file.parquet", DATA_SPEC.specId(),
partition);
+
+ LeafTrackedFileWrapper wrapper = new LeafTrackedFileWrapper(specs,
Types.StructType.of());
+ TrackedFile wrapped = wrapper.wrap(file, file.tracking());
+
+ assertThat(wrapped.partition())
+ .usingComparator(Comparators.forType(DATA_SPEC.partitionType()))
+ .isEqualTo(partition);
+ }
+
+ @Test
+ void leafRejectsUnassignedDataSequenceNumberForExistingFile() {
+ Tracking tracking =
+ new TrackingStruct(
+ EntryStatus.EXISTING,
+ COMMIT_SNAPSHOT_ID,
+ null /* dataSequenceNumber */,
+ null /* fileSequenceNumber */,
+ null /* modifiedSnapshotId */,
+ null /* firstRowId */,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+ TrackedFile file = dataFile(tracking, "s3://bucket/table/file.parquet");
+
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, file.tracking()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cannot write null data sequence number for status %s (must be
ADDED or MODIFIED): %s",
+ EntryStatus.EXISTING, file.location());
+ }
+
+ @Test
+ void leafRejectsUnassignedFileSequenceNumberForModifiedFile() {
+ Tracking tracking =
+ new TrackingStruct(
+ EntryStatus.MODIFIED,
+ PREVIOUS_SNAPSHOT_ID,
+ null /* dataSequenceNumber */,
+ null /* fileSequenceNumber */,
+ COMMIT_SNAPSHOT_ID /* modifiedSnapshotId */,
+ PREVIOUS_FIRST_ROW_ID,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+ TrackedFile file = dataFile(tracking, "s3://bucket/table/file.parquet");
+
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, file.tracking()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ EntryStatus.MODIFIED, file.location());
+ }
+
+ @Test
+ void leafAllowsUnassignedDataSequenceNumberForModifiedFile() {
+ Tracking tracking =
+ TrackingBuilder.from(EXISTING_TRACKING,
COMMIT_SNAPSHOT_ID).columnFilesUpdated().build();
+ TrackedFile file =
+ dataFileWithColumnFiles(
+ tracking,
+ "s3://bucket/table/file.parquet",
+ ImmutableList.of(
+ columnFile("s3://bucket/table/file-cols.parquet",
ImmutableSet.of(2))));
+
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+ TrackedFile wrapped = wrapper.wrap(file, file.tracking());
+
+ assertThat(wrapped.tracking().dataSequenceNumber()).isNull();
+ }
+
+ @Test
+ void leafRejectsManifest() {
+ TrackedFile manifest =
+ manifestRef(FileContent.DATA_MANIFEST,
"s3://bucket/table/leaf.parquet", MANIFEST_INFO);
+
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(manifest, manifest.tracking()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid content type for leaf manifest: DATA_MANIFEST");
+ }
+
+ @Test
+ void leafRejectsUnknownSpecId() {
+ TrackedFile file =
+ dataFile("s3://bucket/table/id=1/file.parquet", ID_SPEC.specId(),
partition(ID_SPEC, 1));
+
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, ADDED_TRACKING))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot find partition spec for spec ID: %s",
ID_SPEC.specId());
+ }
+
+ @Test
+ void leafRejectsNullFile() {
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+
+ assertThatThrownBy(() -> wrapper.wrap(null, ADDED_TRACKING))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid file: null");
+ }
+
+ @Test
+ void leafRejectsNullTracking() {
+ TrackedFile file = dataFile(ADDED_TRACKING,
"s3://bucket/table/file.parquet");
+
+ LeafTrackedFileWrapper wrapper =
+ new LeafTrackedFileWrapper(UNPARTITIONED_SPECS, Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid tracking: null");
+ }
+
+ @Test
+ void rootWritesDataFile() throws IOException {
+ DataFile dataFile =
+ DataFiles.builder(ID_SPEC)
+ .withPath("s3://bucket/table/id=1/file.parquet")
+ .withPartition(partition(ID_SPEC, 1))
+ .withFileSizeInBytes(1024L)
+ .withRecordCount(100L)
+ .build();
+ TrackedFile file =
TrackedFileAdapters.forDataFile(TABLE_SCHEMA).wrap(dataFile);
+ TrackedFile expected =
+ copyWithTracking(
+ file,
+ new TrackingStruct(
+ EntryStatus.ADDED,
+ COMMIT_SNAPSHOT_ID,
+ COMMIT_SEQUENCE_NUMBER,
+ COMMIT_SEQUENCE_NUMBER,
+ null /* modifiedSnapshotId */,
+ FIRST_ROW_ID,
+ null /* deletedPositions */,
+ null /* replacedPositions */));
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, ID_SPECS,
Types.StructType.of());
+ TrackedFile actual =
+ writeAndRead(ID_SPEC.partitionType(), wrapper.wrap(file,
ADDED_TRACKING, FIRST_ROW_ID));
+
+ assertThat(actual)
+
.usingComparator(V4TestComparators.trackedFile(ID_SPEC.partitionType()))
+ .isEqualTo(expected);
+ }
+
+ @Test
+ void rootAssignsCommitDataSequenceNumberToModifiedDataFile() {
+ Tracking tracking =
+ TrackingBuilder.from(EXISTING_TRACKING,
COMMIT_SNAPSHOT_ID).columnFilesUpdated().build();
+ TrackedFile file =
+ dataFileWithColumnFiles(
+ tracking,
+ "s3://bucket/table/file.parquet",
+ ImmutableList.of(
+ columnFile("s3://bucket/table/file-cols.parquet",
ImmutableSet.of(2))));
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ TrackedFile wrapped = wrapper.wrap(file, file.tracking(), null /*
assignedFirstRowId */);
+
+
assertThat(wrapped.tracking().dataSequenceNumber()).isEqualTo(COMMIT_SEQUENCE_NUMBER);
+
assertThat(wrapped.tracking().fileSequenceNumber()).isEqualTo(PREVIOUS_SEQUENCE_NUMBER);
+ }
+
+ @Test
+ void rootRejectsUnassignedFileSequenceNumberForModifiedDataFile() {
+ Tracking tracking =
+ new TrackingStruct(
+ EntryStatus.MODIFIED,
+ PREVIOUS_SNAPSHOT_ID,
+ null /* dataSequenceNumber */,
+ null /* fileSequenceNumber */,
+ COMMIT_SNAPSHOT_ID /* modifiedSnapshotId */,
+ PREVIOUS_FIRST_ROW_ID,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+ TrackedFile file = dataFile(tracking, "s3://bucket/table/file.parquet");
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, file.tracking(), null /*
assignedFirstRowId */))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cannot write null file sequence number for status %s (must be
ADDED): %s",
+ EntryStatus.MODIFIED, file.location());
+ }
+
+ @Test
+ void rootRejectsNullSnapshotIdForDataFile() {
+ Tracking tracking =
+ new TrackingStruct(
+ EntryStatus.ADDED,
+ null /* snapshotId */,
+ null /* dataSequenceNumber */,
+ null /* fileSequenceNumber */,
+ null /* modifiedSnapshotId */,
+ null /* firstRowId */,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+ TrackedFile file = dataFile(tracking, "s3://bucket/table/file.parquet");
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, file.tracking(), FIRST_ROW_ID))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Missing snapshot ID for file: %s", file.location());
+ }
+
+ @Test
+ void rootRejectsUnassignedFirstRowIdForDataFile() {
+ TrackedFile file = dataFile(ADDED_TRACKING,
"s3://bucket/table/file.parquet");
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, file.tracking(), null /*
assignedFirstRowId */))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Missing first-row-id for file: %s", file.location());
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = EntryStatus.class,
+ names = {"DELETED", "REPLACED"})
+ void rootRejectsFirstRowIdAssignmentForNonLiveDataFile(EntryStatus status) {
+ Tracking tracking =
+ new TrackingStruct(
+ status,
+ COMMIT_SNAPSHOT_ID,
+ PREVIOUS_SEQUENCE_NUMBER,
+ PREVIOUS_SEQUENCE_NUMBER,
+ null /* modifiedSnapshotId */,
+ null /* firstRowId */,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+ TrackedFile file = dataFile(tracking, "s3://bucket/table/file.parquet");
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, file.tracking(), FIRST_ROW_ID))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cannot assign first-row-id to file with status %s: %s", status,
file.location());
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = EntryStatus.class,
+ names = {"EXISTING", "MODIFIED"})
+ void rootRejectsFirstRowIdAssignmentForAssignedDataFile(EntryStatus status) {
+ Tracking tracking =
+ new TrackingStruct(
+ status,
+ PREVIOUS_SNAPSHOT_ID,
+ PREVIOUS_SEQUENCE_NUMBER,
+ PREVIOUS_SEQUENCE_NUMBER,
+ null /* modifiedSnapshotId */,
+ PREVIOUS_FIRST_ROW_ID,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+ TrackedFile file = dataFile(tracking, "s3://bucket/table/file.parquet");
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ assertThatThrownBy(() -> wrapper.wrap(file, file.tracking(), FIRST_ROW_ID))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cannot assign first-row-id to file with existing first-row-id %s:
%s",
+ PREVIOUS_FIRST_ROW_ID, file.location());
+ }
+
+ @Test
+ void rootWritesManifest() throws IOException {
+ TrackedFile manifest =
+ manifestRef(FileContent.DATA_MANIFEST,
"s3://bucket/table/leaf.parquet", MANIFEST_INFO);
+ Tracking tracking =
+ new TrackingStruct(
+ EntryStatus.MODIFIED,
+ PREVIOUS_SNAPSHOT_ID,
+ PREVIOUS_SEQUENCE_NUMBER,
+ PREVIOUS_SEQUENCE_NUMBER,
+ COMMIT_SNAPSHOT_ID /* modifiedSnapshotId */,
+ PREVIOUS_FIRST_ROW_ID,
+ new byte[] {1} /* deletedPositions */,
+ new byte[] {2} /* replacedPositions */);
+ TrackedFile expected = copyWithTracking(manifest, tracking);
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ TrackedFile actual =
+ writeAndRead(
+ Types.StructType.of(), wrapper.wrap(manifest, tracking, null /*
assignedFirstRowId */));
+
+ assertThat(actual)
+ .usingComparator(V4TestComparators.trackedFile(Types.StructType.of()))
+ .isEqualTo(expected);
+ }
+
+ @Test
+ void rootWritesManifestStats() throws IOException {
+ ContentStats stats = StatsTestUtil.contentStats(STATS_TYPE, ID_STATS,
DATA_STATS);
+ TrackedFile manifest =
+ manifestRefWithStats(
+ FileContent.DATA_MANIFEST, "s3://bucket/table/leaf.parquet",
stats, MANIFEST_INFO);
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
STATS_TYPE);
+ TrackedFile actual =
+ writeAndRead(
+ Types.StructType.of(),
+ STATS_TYPE,
+ wrapper.wrap(manifest, EXISTING_TRACKING, null /*
assignedFirstRowId */));
+
+ assertThat(actual.contentStats())
+ .usingComparator(V4TestComparators.CONTENT_STATS)
+ .isEqualTo(stats);
+ }
+
+ @Test
+ void rootWritesNullSpecIdForUnpartitionedManifest() throws IOException {
+ ManifestFile manifest = newManifest(ManifestContent.DATA,
COMMIT_SNAPSHOT_ID);
+ TrackedFile file = TrackedFileAdapters.forManifestFile().wrap(manifest);
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ TrackedFile actual =
+ writeAndRead(Types.StructType.of(), wrapper.wrap(file,
file.tracking(), FIRST_ROW_ID));
+
+ assertThat(actual.specId()).isNull();
+ }
+
+ @Test
+ void rootAssignsCommitSequenceNumberToNewManifest() {
+ ManifestFile manifest = newManifest(ManifestContent.DATA,
COMMIT_SNAPSHOT_ID);
+ TrackedFile file = TrackedFileAdapters.forManifestFile().wrap(manifest);
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ TrackedFile wrapped = wrapper.wrap(file, file.tracking(), FIRST_ROW_ID);
+
+
assertThat(wrapped.tracking().dataSequenceNumber()).isEqualTo(COMMIT_SEQUENCE_NUMBER);
+
assertThat(wrapped.tracking().fileSequenceNumber()).isEqualTo(COMMIT_SEQUENCE_NUMBER);
+
assertThat(wrapped.manifestInfo().minSequenceNumber()).isEqualTo(COMMIT_SEQUENCE_NUMBER);
+ }
+
+ @Test
+ void rootReturnsNullFirstRowIdForDeleteManifest() {
+ TrackedFile manifest =
+ manifestRef(FileContent.DELETE_MANIFEST,
"s3://bucket/table/deletes.avro", MANIFEST_INFO);
+ Tracking tracking =
+ new TrackingStruct(
+ EntryStatus.EXISTING,
+ PREVIOUS_SNAPSHOT_ID,
+ PREVIOUS_SEQUENCE_NUMBER,
+ PREVIOUS_SEQUENCE_NUMBER,
+ null /* modifiedSnapshotId */,
+ PREVIOUS_FIRST_ROW_ID,
+ null /* deletedPositions */,
+ null /* replacedPositions */);
+
+ RootTrackedFileWrapper wrapper =
+ new RootTrackedFileWrapper(
+ COMMIT_SNAPSHOT_ID, COMMIT_SEQUENCE_NUMBER, UNPARTITIONED_SPECS,
Types.StructType.of());
+ TrackedFile wrapped = wrapper.wrap(manifest, tracking, null /*
assignedFirstRowId */);
+
+ assertThat(wrapped.tracking().firstRowId()).isNull();
+ }
+
+ @Test
+ void rootRejectsUnassignedSequenceNumberForManifestFromAnotherSnapshot() {
Review Comment:
This guard matches the v2/v3 `ManifestFileWrapper` check that a manifest
with an unassigned sequence number must come from the committing snapshot. It
isn't reachable through public APIs today: `appendManifest` requires an
unassigned snapshot ID and producer-written manifests use the producer's own
ID. It protects against a producer bug that would silently give another
snapshot's manifest this commit's sequence number. v2/v3 don't test their
check; this test covers the v4 equivalent and is the only coverage of that
branch.
##########
core/src/main/java/org/apache/iceberg/V4Metadata.java:
##########
@@ -541,4 +548,730 @@ public Long pos() {
return null;
}
}
+
+ /** Base wrapper used to write a {@link TrackedFile} using the v4 tracked
file schema. */
+ private static void checkNullSequenceNumbers(TrackedFile file, Tracking
tracking) {
+ // null sequence numbers are inherited from the commit: ADDED entries
inherit both and
+ // MODIFIED entries inherit only the data sequence number
+ EntryStatus status = tracking.status();
+ Preconditions.checkState(
+ tracking.dataSequenceNumber() != null
+ || status == EntryStatus.ADDED
+ || status == EntryStatus.MODIFIED,
+ "Cannot write null data sequence number for status %s (must be ADDED
or MODIFIED): %s",
+ status,
+ file.location());
+ Preconditions.checkState(
+ tracking.fileSequenceNumber() != null || status == EntryStatus.ADDED,
+ "Cannot write null file sequence number for status %s (must be ADDED):
%s",
+ status,
+ file.location());
+ }
+
+ private abstract static class TrackedFileWrapper implements TrackedFile,
StructLike {
+ private static final int SIZE =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).columns().size();
+
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Map<Integer, StructProjection> partitionProjections;
+ private final ContentStatsWrapper contentStats;
+
+ private TrackedFile wrapped = null;
+ private Integer specId = null;
+ private TrackingWrapper tracking = null;
+ private ManifestInfoWrapper manifestInfo = null;
+
+ private TrackedFileWrapper(
+ Map<Integer, PartitionSpec> specsById, Types.StructType
contentStatsType) {
+ this.specsById = specsById;
+ this.partitionProjections = partitionProjections(specsById);
+ this.contentStats = new ContentStatsWrapper(contentStatsType);
+ }
+
+ protected void wrapInternal(
+ TrackedFile file, TrackingWrapper newTracking, ManifestInfoWrapper
newManifestInfo) {
+ Integer fileSpecId = file.specId();
+ PartitionSpec spec = fileSpecId != null ? specsById.get(fileSpecId) :
null;
+ Preconditions.checkArgument(
+ fileSpecId == null || spec != null,
+ "Cannot find partition spec for spec ID: %s",
+ fileSpecId);
+
+ this.wrapped = file;
+ // v4 metadata stores a null spec ID and partition for unpartitioned
files
+ this.specId = spec != null && !spec.fields().isEmpty() ? fileSpecId :
null;
+ this.tracking = newTracking;
+ this.manifestInfo = newManifestInfo;
+ }
+
+ @Override
+ public int size() {
+ return SIZE;
+ }
+
+ @Override
+ public <T> void set(int pos, T value) {
+ throw new UnsupportedOperationException("Cannot modify a read-only
tracked file");
+ }
+
+ @Override
+ public <T> T get(int pos, Class<T> javaClass) {
+ return javaClass.cast(get(pos));
+ }
+
+ private Object get(int pos) {
+ return switch (pos) {
+ case 0 -> tracking;
+ case 1 -> wrapped.contentType().id();
+ case 2 -> wrapped.location();
+ case 3 -> wrapped.fileFormat().toString();
+ case 4 -> wrapped.recordCount();
+ case 5 -> wrapped.fileSizeInBytes();
+ case 6 -> specId;
+ case 7 -> unionPartition();
+ case 8 -> wrapped.contentStats() != null ?
contentStats.wrap(wrapped.contentStats()) : null;
+ case 9 -> wrapped.sortOrderId();
+ case 10 -> wrapped.deletionVector();
Review Comment:
No write-side wrapper for the DV or column files, unlike tracking, manifest
info, and stats. Those need one because the source may not be `StructLike` or
may have positions that differ from the write type. Neither applies here:
`DeletionVectorStruct` and `ColumnFileStruct` are the only implementations, and
their field orders match the fixed `DeletionVector.schema()` and
`ColumnFile.schema()`. `leafWritesDataFile` round-trips both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
