diff --git a/api/src/main/java/org/apache/iceberg/ManifestBitmap.java b/api/src/main/java/org/apache/iceberg/ManifestBitmap.java new file mode 100644 index 000000000000..a6a41af41479 --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/ManifestBitmap.java @@ -0,0 +1,33 @@ +/* + * 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.nio.ByteBuffer; + +/** A bitmap that is serialized inline in a metadata file. */ +public interface ManifestBitmap { + /** Number of bits set in this bitmap. */ + int cardinality(); + + /** Return whether the bit at {@code position} is set. */ + boolean isSet(int position); + + /** Return the serialized bitmap as a {@link ByteBuffer}. */ + ByteBuffer buffer(); +} diff --git a/api/src/main/java/org/apache/iceberg/ManifestFile.java b/api/src/main/java/org/apache/iceberg/ManifestFile.java index 2f732aef427f..b831c2059190 100644 --- a/api/src/main/java/org/apache/iceberg/ManifestFile.java +++ b/api/src/main/java/org/apache/iceberg/ManifestFile.java @@ -126,7 +126,7 @@ static Schema schema() { /** Returns length of the manifest file. */ long length(); - /** Returns iD of the {@link PartitionSpec} used to write the manifest file. */ + /** Returns ID of the {@link PartitionSpec} used to write the manifest file. */ int partitionSpecId(); /** Returns the content stored in the manifest; either DATA or DELETES. */ @@ -138,7 +138,7 @@ static Schema schema() { /** Returns the lowest data sequence number of any live file in the manifest. */ long minSequenceNumber(); - /** Returns iD of the snapshot that added the manifest file to table metadata. */ + /** Returns ID of the snapshot that added the manifest file to table metadata. */ Long snapshotId(); /** @@ -210,6 +210,11 @@ default Long firstRowId() { return null; } + /** Returns the manifest deletion vector, or null if absent. */ + default ManifestBitmap manifestDeletionVector() { + return null; + } + /** * Copies this {@link ManifestFile manifest file}. Readers can reuse manifest file instances; use * this method to make defensive copies. diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java index 8d3e5e950d6a..98cf78088702 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java @@ -24,7 +24,7 @@ import java.util.Set; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; -/** Adapts {@link TrackedFile} entries to the {@link DataFile} and {@link DeleteFile} APIs. */ +/** Adapts {@link TrackedFile} entries to their read APIs, for example {@link DataFile}. */ class TrackedFileAdapters { private TrackedFileAdapters() {} @@ -53,7 +53,16 @@ static DeleteFile asEqualityDeleteFile(TrackedFile file, Map> implements ContentFile { private final TrackedFile file; @@ -415,6 +424,134 @@ public DeleteFile copyWithStats(Set requestedColumnIds) { } } + /** Adapts a TrackedFile to {@link ManifestFile}. */ + private static class TrackedManifestFile implements ManifestFile { + private final TrackedFile file; + + private TrackedManifestFile(TrackedFile file) { + this.file = file; + } + + @Override + public String path() { + return file.location(); + } + + @Override + public long length() { + return file.fileSizeInBytes(); + } + + @Override + public int partitionSpecId() { + throw new UnsupportedOperationException( + "v4 manifests are not bound to a single partition spec"); + } + + @Override + public ManifestContent content() { + switch (file.contentType()) { + case DATA_MANIFEST: + return ManifestContent.DATA; + case DELETE_MANIFEST: + return ManifestContent.DELETES; + default: + throw new UnsupportedOperationException( + "Unsupported content type for manifests: " + file.contentType()); + } + } + + @Override + public long sequenceNumber() { + return file.tracking().dataSequenceNumber(); + } + + @Override + public long minSequenceNumber() { + return file.manifestInfo().minSequenceNumber(); + } + + @Override + public Long snapshotId() { + return file.tracking().snapshotId(); + } + + @Override + public Integer addedFilesCount() { + return file.manifestInfo().addedFilesCount(); + } + + @Override + public Long addedRowsCount() { + return file.manifestInfo().addedRowsCount(); + } + + @Override + public Integer existingFilesCount() { + return file.manifestInfo().existingFilesCount(); + } + + @Override + public Long existingRowsCount() { + return file.manifestInfo().existingRowsCount(); + } + + @Override + public Integer deletedFilesCount() { + return file.manifestInfo().deletedFilesCount(); + } + + @Override + public Long deletedRowsCount() { + return file.manifestInfo().deletedRowsCount(); + } + + @Override + public List partitions() { + return null; + } + + @Override + public ByteBuffer keyMetadata() { + return file.keyMetadata(); + } + + @Override + public Long firstRowId() { + return file.tracking().firstRowId(); + } + + @Override + public ManifestBitmap manifestDeletionVector() { + ByteBuffer dv = file.manifestInfo().dv(); + if (dv == null) { + return null; + } + + return new ManifestBitmap() { + @Override + public int cardinality() { + throw new UnsupportedOperationException("Bitmap decoding has not been implemented"); + } + + @Override + public boolean isSet(int position) { + throw new UnsupportedOperationException("Bitmap decoding has not been implemented"); + } + + @Override + public ByteBuffer buffer() { + return dv; + } + }; + } + + @Override + public ManifestFile copy() { + return new TrackedManifestFile(file.copy()); + } + } + private static PartitionSpec resolveSpec( TrackedFile file, Map specsById) { Integer specId = file.specId(); diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java index 6fa71fc7999f..646407fae356 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java @@ -33,6 +33,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import org.mockito.Mockito; class TestTrackedFileAdapters { @@ -40,9 +41,11 @@ class TestTrackedFileAdapters { private static final String MANIFEST_LOCATION = "s3://bucket/table/manifest.parquet"; private static final String DATA_FILE_LOCATION = "s3://bucket/data/file.parquet"; private static final String DV_LOCATION = "s3://bucket/puffin/dv-file.bin"; + private static final long MANIFEST_FILE_SIZE = 2048L; // Tracking values that the delegation tests validate. private static final long MANIFEST_POS = 3L; + private static final long SNAPSHOT_ID = 42L; private static final long DATA_SEQUENCE_NUMBER = 10L; private static final long FILE_SEQUENCE_NUMBER = 11L; private static final long FIRST_ROW_ID = 1000L; @@ -91,6 +94,36 @@ class TestTrackedFileAdapters { CONTENT_STATS.setStats(3, GEOM_STATS); } + private static final Tracking MANIFEST_TRACKING = + new TrackingStruct( + EntryStatus.ADDED, + SNAPSHOT_ID, + DATA_SEQUENCE_NUMBER, + FILE_SEQUENCE_NUMBER, + null, // dvSnapshotId + FIRST_ROW_ID, + null, // deletedPositions + null); // replacedPositions + + private static final byte[] MANIFEST_DV = new byte[] {1, 2, 3}; + + private static final ByteBuffer MANIFEST_KEY_METADATA = ByteBuffer.wrap(new byte[] {7, 8, 9}); + + private static final ManifestInfo MANIFEST_INFO = + ManifestInfoStruct.builder() + .addedFilesCount(3) + .existingFilesCount(5) + .deletedFilesCount(2) + .replacedFilesCount(0) + .addedRowsCount(300L) + .existingRowsCount(500L) + .deletedRowsCount(200L) + .replacedRowsCount(0L) + .minSequenceNumber(7L) + .dv(ByteBuffer.wrap(MANIFEST_DV)) + .dvCardinality(4L) + .build(); + @Test void dataFileAdapterDelegation() { TrackingStruct tracking = @@ -350,6 +383,83 @@ void dvDeleteFileAdapterRejectsNullDeletionVector() { .hasMessage("Cannot create DV delete file: no deletion vector"); } + @ParameterizedTest + @EnumSource( + value = FileContent.class, + names = {"DATA_MANIFEST", "DELETE_MANIFEST"}) + void manifestFileAdapterDelegation(FileContent contentType) { + TrackedFile file = + new TrackedFileStruct( + MANIFEST_TRACKING, + contentType, + FORMAT_VERSION_V4, + MANIFEST_LOCATION, + FileFormat.PARQUET, + 10L, // recordCount + MANIFEST_FILE_SIZE, + null, // specId + null, // partition + null, // contentStats + null, // sortOrderId + null, // deletionVector + MANIFEST_INFO, + MANIFEST_KEY_METADATA, + null, // splitOffsets + null); // equalityIds + + ManifestFile manifest = TrackedFileAdapters.asManifestFile(file); + + ManifestContent expectedContent = + contentType == FileContent.DATA_MANIFEST ? ManifestContent.DATA : ManifestContent.DELETES; + assertThat(manifest.path()).isEqualTo(MANIFEST_LOCATION); + assertThat(manifest.length()).isEqualTo(MANIFEST_FILE_SIZE); + assertThat(manifest.content()).isEqualTo(expectedContent); + assertThat(manifest.sequenceNumber()).isEqualTo(DATA_SEQUENCE_NUMBER); + assertThat(manifest.minSequenceNumber()).isEqualTo(MANIFEST_INFO.minSequenceNumber()); + assertThat(manifest.snapshotId()).isEqualTo(SNAPSHOT_ID); + assertThat(manifest.addedFilesCount()).isEqualTo(MANIFEST_INFO.addedFilesCount()); + assertThat(manifest.addedRowsCount()).isEqualTo(MANIFEST_INFO.addedRowsCount()); + assertThat(manifest.existingFilesCount()).isEqualTo(MANIFEST_INFO.existingFilesCount()); + assertThat(manifest.existingRowsCount()).isEqualTo(MANIFEST_INFO.existingRowsCount()); + assertThat(manifest.deletedFilesCount()).isEqualTo(MANIFEST_INFO.deletedFilesCount()); + assertThat(manifest.deletedRowsCount()).isEqualTo(MANIFEST_INFO.deletedRowsCount()); + assertThat(manifest.firstRowId()).isEqualTo(FIRST_ROW_ID); + assertThat(manifest.keyMetadata()).isEqualTo(MANIFEST_KEY_METADATA); + assertThat(manifest.manifestDeletionVector().buffer()).isEqualTo(ByteBuffer.wrap(MANIFEST_DV)); + assertThat(manifest.partitions()).isNull(); + assertThatThrownBy(manifest::partitionSpecId) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("v4 manifests are not bound to a single partition spec"); + } + + @Test + void manifestFileAdapterCopy() { + TrackedFile file = Mockito.mock(TrackedFile.class); + TrackedFile fileCopy = Mockito.mock(TrackedFile.class); + Mockito.when(file.contentType()).thenReturn(FileContent.DATA_MANIFEST); + Mockito.when(file.copy()).thenReturn(fileCopy); + Mockito.when(fileCopy.location()).thenReturn(MANIFEST_LOCATION); + + ManifestFile copy = TrackedFileAdapters.asManifestFile(file).copy(); + + // copy() delegates to the tracked file's copy(), which deep-copies the nested structs. + Mockito.verify(file).copy(); + assertThat(copy.path()).isEqualTo(MANIFEST_LOCATION); + } + + @ParameterizedTest + @EnumSource( + value = FileContent.class, + mode = EnumSource.Mode.EXCLUDE, + names = {"DATA_MANIFEST", "DELETE_MANIFEST"}) + void manifestFileAdapterRejectsNonManifestContent(FileContent contentType) { + TrackedFileStruct file = dummyTrackedFile(contentType); + + assertThatThrownBy(() -> TrackedFileAdapters.asManifestFile(file)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid content type for ManifestFile: %s", contentType); + } + @Test void dataFileWithoutDeletionVectorReturnsNull() { TrackedFile fileWithoutDv = mock(TrackedFile.class);