diff --git a/api/src/main/java/org/apache/iceberg/FileWithKeyId.java b/api/src/main/java/org/apache/iceberg/FileWithKeyId.java new file mode 100644 index 000000000000..cb826c14465d --- /dev/null +++ b/api/src/main/java/org/apache/iceberg/FileWithKeyId.java @@ -0,0 +1,32 @@ +/* + * 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; + +/** + * A file that may be encrypted. If it is encrypted, its encrypted key metadata is tracked in the + * table metadata encryption keys and is referenced by a key ID. + */ +public interface FileWithKeyId { + + /** Location of the file. */ + String location(); + + /** Returns the encryption key ID for this file, or null if the file is not encrypted. */ + String keyId(); +} diff --git a/api/src/main/java/org/apache/iceberg/ManifestListFile.java b/api/src/main/java/org/apache/iceberg/ManifestListFile.java index e727a35a4e09..242ed554678f 100644 --- a/api/src/main/java/org/apache/iceberg/ManifestListFile.java +++ b/api/src/main/java/org/apache/iceberg/ManifestListFile.java @@ -21,14 +21,19 @@ import java.nio.ByteBuffer; import org.apache.iceberg.encryption.EncryptionManager; -public interface ManifestListFile { - - /** Location of manifest list file. */ - String location(); - +/** + * @deprecated since 1.12.0. Will be removed in 2.0.0; use {@link FileWithKeyId} instead. + */ +@Deprecated +public interface ManifestListFile extends FileWithKeyId { /** The manifest list key metadata can be encrypted. Returns ID of encryption key */ String encryptionKeyID(); + @Override + default String keyId() { + return encryptionKeyID(); + } + /** Decrypt and return the manifest list key metadata */ ByteBuffer decryptKeyMetadata(EncryptionManager em); } diff --git a/api/src/main/java/org/apache/iceberg/Snapshot.java b/api/src/main/java/org/apache/iceberg/Snapshot.java index 8a74dca6d053..5f58e77f131a 100644 --- a/api/src/main/java/org/apache/iceberg/Snapshot.java +++ b/api/src/main/java/org/apache/iceberg/Snapshot.java @@ -167,9 +167,32 @@ default Iterable removedDeleteFiles(FileIO io) { * Return the location of this snapshot's manifest list, or null if it is not separate. * * @return the location of the manifest list for this Snapshot + * @deprecated since 1.12.0. Will be removed in 2.0.0; use {@link #rootLocation()}, which returns + * the manifest list for v3 and earlier and the root manifest for v4+. */ + @Deprecated String manifestListLocation(); + /** + * Returns the location of this snapshot's root metadata file — a manifest list for v3 and + * earlier, or a root manifest for v4+. + * + * @return the location of the root file for this Snapshot + */ + default String rootLocation() { + return manifestListLocation(); + } + + /** + * Returns the format version this snapshot was written under, or 0 if the snapshot does not + * report one. + * + * @return the snapshot's format version + */ + default int formatVersion() { + return 0; + } + /** * Return the id of the schema used when this snapshot was created, or null if this information is * not available. diff --git a/api/src/main/java/org/apache/iceberg/encryption/EncryptingFileIO.java b/api/src/main/java/org/apache/iceberg/encryption/EncryptingFileIO.java index 6a14db3dd439..e9da3f6508eb 100644 --- a/api/src/main/java/org/apache/iceberg/encryption/EncryptingFileIO.java +++ b/api/src/main/java/org/apache/iceberg/encryption/EncryptingFileIO.java @@ -28,6 +28,7 @@ import org.apache.iceberg.ContentFile; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileWithKeyId; import org.apache.iceberg.ManifestFile; import org.apache.iceberg.ManifestListFile; import org.apache.iceberg.io.BulkDeletionFailureException; @@ -130,6 +131,11 @@ public InputFile newInputFile(ManifestFile manifest) { } } + /** + * @deprecated since 1.12.0. Will be removed in 2.0.0; use {@link #newInputFile(FileWithKeyId)} + * instead. + */ + @Deprecated @Override public InputFile newInputFile(ManifestListFile manifestList) { if (manifestList.encryptionKeyID() != null) { @@ -140,6 +146,16 @@ public InputFile newInputFile(ManifestListFile manifestList) { } } + @Override + public InputFile newInputFile(FileWithKeyId file) { + if (file.keyId() != null) { + ByteBuffer keyMetadata = em.decryptKeyMetadata(file.keyId()); + return newDecryptingInputFile(file.location(), keyMetadata); + } else { + return newInputFile(file.location()); + } + } + public InputFile newDecryptingInputFile(String path, ByteBuffer buffer) { return em.decrypt(wrap(io.newInputFile(path), buffer)); } diff --git a/api/src/main/java/org/apache/iceberg/encryption/EncryptionManager.java b/api/src/main/java/org/apache/iceberg/encryption/EncryptionManager.java index 22d2858599a8..3e09ac14915d 100644 --- a/api/src/main/java/org/apache/iceberg/encryption/EncryptionManager.java +++ b/api/src/main/java/org/apache/iceberg/encryption/EncryptionManager.java @@ -19,6 +19,7 @@ package org.apache.iceberg.encryption; import java.io.Serializable; +import java.nio.ByteBuffer; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; @@ -51,6 +52,17 @@ default Iterable decrypt(Iterable encrypted) { return Iterables.transform(encrypted, this::decrypt); } + /** + * Decrypt an encrypted key metadata referred by a key id. + * + * @param keyId the encryption key ID + * @return the decrypted key metadata buffer + */ + default ByteBuffer decryptKeyMetadata(String keyId) { + throw new UnsupportedOperationException( + this.getClass().getName() + " does not support key metadata decryption"); + } + /** * Given a handle on an {@link OutputFile} that writes raw bytes to the underlying file system, * return a bundle of an {@link EncryptedOutputFile#encryptingOutputFile()} that writes encrypted diff --git a/api/src/main/java/org/apache/iceberg/io/FileIO.java b/api/src/main/java/org/apache/iceberg/io/FileIO.java index 72d2d9b0fca2..119d6d4e44da 100644 --- a/api/src/main/java/org/apache/iceberg/io/FileIO.java +++ b/api/src/main/java/org/apache/iceberg/io/FileIO.java @@ -23,6 +23,7 @@ import java.util.Map; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileWithKeyId; import org.apache.iceberg.ManifestFile; import org.apache.iceberg.ManifestListFile; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -71,6 +72,11 @@ default InputFile newInputFile(ManifestFile manifest) { return newInputFile(manifest.path(), manifest.length()); } + /** + * @deprecated since 1.12.0. Will be removed in 2.0.0; use {@link #newInputFile(FileWithKeyId)} + * instead. + */ + @Deprecated default InputFile newInputFile(ManifestListFile manifestList) { Preconditions.checkArgument( manifestList.encryptionKeyID() == null, @@ -80,6 +86,13 @@ default InputFile newInputFile(ManifestListFile manifestList) { return newInputFile(manifestList.location()); } + default InputFile newInputFile(FileWithKeyId file) { + Preconditions.checkArgument( + file.keyId() == null, "Cannot decrypt file: %s (use EncryptingFileIO)", file.location()); + // cannot pass length because it is not tracked outside of key metadata + return newInputFile(file.location()); + } + /** Get a {@link OutputFile} instance to write bytes to the file at the given path. */ OutputFile newOutputFile(String path); diff --git a/core/src/main/java/org/apache/iceberg/AllManifestsTable.java b/core/src/main/java/org/apache/iceberg/AllManifestsTable.java index 8ceffc29c9e4..e01144d3fd2b 100644 --- a/core/src/main/java/org/apache/iceberg/AllManifestsTable.java +++ b/core/src/main/java/org/apache/iceberg/AllManifestsTable.java @@ -136,13 +136,13 @@ protected CloseableIterable doPlanFiles() { Iterables.transform( filteredSnapshots, snap -> { - if (snap.manifestListLocation() != null) { + if (snap.rootLocation() != null) { return new ManifestListReadTask( dataTableSchema, io, schema(), specs, - new BaseManifestListFile(snap.manifestListLocation(), snap.keyId()), + new BaseFileWithKeyId(snap.rootLocation(), snap.keyId()), residual, snap.snapshotId()); } else { @@ -165,7 +165,7 @@ static class ManifestListReadTask implements DataTask { private final FileIO io; private final Schema schema; private final Map specs; - private final ManifestListFile manifestList; + private final FileWithKeyId manifestList; private final Expression residual; private final long referenceSnapshotId; private DataFile lazyDataFile = null; @@ -175,7 +175,7 @@ static class ManifestListReadTask implements DataTask { FileIO io, Schema schema, Map specs, - ManifestListFile manifestList, + FileWithKeyId manifestList, Expression residual, long referenceSnapshotId) { this.dataTableSchema = dataTableSchema; @@ -276,7 +276,7 @@ Map specsById() { return specs; } - ManifestListFile manifestList() { + FileWithKeyId manifestList() { return manifestList; } diff --git a/core/src/main/java/org/apache/iceberg/AllManifestsTableTaskParser.java b/core/src/main/java/org/apache/iceberg/AllManifestsTableTaskParser.java index e6539f2d714f..8131305ad6e7 100644 --- a/core/src/main/java/org/apache/iceberg/AllManifestsTableTaskParser.java +++ b/core/src/main/java/org/apache/iceberg/AllManifestsTableTaskParser.java @@ -65,8 +65,8 @@ static void toJson(AllManifestsTable.ManifestListReadTask task, JsonGenerator ge generator.writeEndArray(); generator.writeStringField(MANIFEST_LIST_LOCATION, task.manifestList().location()); - if (task.manifestList().encryptionKeyID() != null) { - generator.writeStringField(MANIFEST_LIST_KEY_ID, task.manifestList().encryptionKeyID()); + if (task.manifestList().keyId() != null) { + generator.writeStringField(MANIFEST_LIST_KEY_ID, task.manifestList().keyId()); } generator.writeFieldName(RESIDUAL); diff --git a/core/src/main/java/org/apache/iceberg/BaseFileWithKeyId.java b/core/src/main/java/org/apache/iceberg/BaseFileWithKeyId.java new file mode 100644 index 000000000000..ec018947dc6f --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/BaseFileWithKeyId.java @@ -0,0 +1,41 @@ +/* + * 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; + +class BaseFileWithKeyId implements FileWithKeyId, Serializable { + private final String location; + private final String keyId; + + BaseFileWithKeyId(String location, String encryptionKeyID) { + this.location = location; + this.keyId = encryptionKeyID; + } + + @Override + public String location() { + return location; + } + + @Override + public String keyId() { + return keyId; + } +} diff --git a/core/src/main/java/org/apache/iceberg/BaseManifestListFile.java b/core/src/main/java/org/apache/iceberg/BaseManifestListFile.java index e0ecfd50c863..3e6af82ee3e1 100644 --- a/core/src/main/java/org/apache/iceberg/BaseManifestListFile.java +++ b/core/src/main/java/org/apache/iceberg/BaseManifestListFile.java @@ -18,28 +18,23 @@ */ package org.apache.iceberg; -import java.io.Serializable; import java.nio.ByteBuffer; import org.apache.iceberg.encryption.EncryptionManager; import org.apache.iceberg.encryption.EncryptionUtil; -class BaseManifestListFile implements ManifestListFile, Serializable { - private final String location; - private final String encryptionKeyID; +/** + * @deprecated since 1.12.0. Will be removed in 2.0.0; use {@link BaseFileWithKeyId} instead. + */ +@Deprecated +class BaseManifestListFile extends BaseFileWithKeyId implements ManifestListFile { BaseManifestListFile(String location, String encryptionKeyID) { - this.location = location; - this.encryptionKeyID = encryptionKeyID; - } - - @Override - public String location() { - return location; + super(location, encryptionKeyID); } @Override public String encryptionKeyID() { - return encryptionKeyID; + return keyId(); } @Override diff --git a/core/src/main/java/org/apache/iceberg/BaseSnapshot.java b/core/src/main/java/org/apache/iceberg/BaseSnapshot.java index 826b9624c0e6..1e9bd94ec02f 100644 --- a/core/src/main/java/org/apache/iceberg/BaseSnapshot.java +++ b/core/src/main/java/org/apache/iceberg/BaseSnapshot.java @@ -38,7 +38,7 @@ class BaseSnapshot implements Snapshot { private final Long parentId; private final long sequenceNumber; private final long timestampMillis; - private final String manifestListLocation; + private final String rootLocation; private final String operation; private final Map summary; private final Integer schemaId; @@ -64,7 +64,7 @@ class BaseSnapshot implements Snapshot { String operation, Map summary, Integer schemaId, - String manifestList, + String rootLocation, Long firstRowId, Long addedRows, String keyId) { @@ -86,7 +86,7 @@ class BaseSnapshot implements Snapshot { this.operation = operation; this.summary = summary; this.schemaId = schemaId; - this.manifestListLocation = manifestList; + this.rootLocation = rootLocation; this.v1ManifestLocations = null; this.firstRowId = firstRowId; this.addedRows = firstRowId != null ? addedRows : null; @@ -109,7 +109,7 @@ class BaseSnapshot implements Snapshot { this.operation = operation; this.summary = summary; this.schemaId = schemaId; - this.manifestListLocation = null; + this.rootLocation = null; this.v1ManifestLocations = v1ManifestLocations; this.firstRowId = null; this.addedRows = null; @@ -181,11 +181,10 @@ private void cacheManifests(FileIO fileIO) { } if (allManifests == null) { - // if manifests isn't set, then the snapshotFile is set and should be read to get the list + // if manifests isn't set, then the root location is set and should be read to get the list this.allManifests = ManifestLists.read( - ManifestLists.newInputFile( - fileIO, new BaseManifestListFile(manifestListLocation, keyId))); + ManifestLists.newInputFile(fileIO, new BaseFileWithKeyId(rootLocation, keyId))); } if (dataManifests == null || deleteManifests == null) { @@ -256,9 +255,14 @@ public Iterable removedDeleteFiles(FileIO fileIO) { return removedDeleteFiles; } + @Override + public String rootLocation() { + return rootLocation; + } + @Override public String manifestListLocation() { - return manifestListLocation; + return rootLocation; } private void cacheDeleteFileChanges(FileIO fileIO) { @@ -369,7 +373,7 @@ public String toString() { .add("timestamp_ms", timestampMillis) .add("operation", operation) .add("summary", summary) - .add("manifest-list", manifestListLocation) + .add("root-location", rootLocation) .add("schema-id", schemaId) .add("first-row-id", firstRowId) .add("added-rows", addedRows) diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java b/core/src/main/java/org/apache/iceberg/BaseTransaction.java index 46eda9e0c92e..56615c115697 100644 --- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java +++ b/core/src/main/java/org/apache/iceberg/BaseTransaction.java @@ -458,7 +458,7 @@ private static Set committedFiles(FileIO io, Set snapshots) { Set committedFiles = Sets.newHashSet(); for (Snapshot snap : snapshots) { - committedFiles.add(snap.manifestListLocation()); + committedFiles.add(snap.rootLocation()); snap.allManifests(io).forEach(manifest -> committedFiles.add(manifest.path())); } diff --git a/core/src/main/java/org/apache/iceberg/CatalogUtil.java b/core/src/main/java/org/apache/iceberg/CatalogUtil.java index 4fa9fc30f1d0..57b8debe1917 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogUtil.java +++ b/core/src/main/java/org/apache/iceberg/CatalogUtil.java @@ -102,14 +102,14 @@ public static void dropTableData(FileIO io, TableMetadata metadata) { // Reads and deletes are done using Tasks.foreach(...).suppressFailureWhenFinished to complete // as much of the delete work as possible and avoid orphaned data or manifest files. - Set manifestListsToDelete = Sets.newHashSet(); + Set snapshotFilesToDelete = Sets.newHashSet(); Set manifestsToDelete = Sets.newHashSet(); for (Snapshot snapshot : metadata.snapshots()) { // add all manifests to the delete set because both data and delete files should be removed Iterables.addAll(manifestsToDelete, snapshot.allManifests(io)); - // add the manifest list to the delete set, if present - if (snapshot.manifestListLocation() != null) { - manifestListsToDelete.add(snapshot.manifestListLocation()); + // add the top-level snapshot file (manifest list for v3, root manifest for v4+) if present + if (snapshot.rootLocation() != null) { + snapshotFilesToDelete.add(snapshot.rootLocation()); } } @@ -131,7 +131,7 @@ public static void dropTableData(FileIO io, TableMetadata metadata) { } deleteFiles(io, Iterables.transform(manifestsToDelete, ManifestFile::path), "manifest"); - deleteFiles(io, manifestListsToDelete, "manifest list"); + deleteFiles(io, snapshotFilesToDelete, "snapshot file"); deleteFiles( io, Iterables.transform(metadata.previousFiles(), TableMetadata.MetadataLogEntry::file), diff --git a/core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java b/core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java index 53652426c30b..7f67df6ec018 100644 --- a/core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java +++ b/core/src/main/java/org/apache/iceberg/FileCleanupStrategy.java @@ -92,9 +92,8 @@ public abstract void cleanFiles( ManifestFile.DELETED_FILES_COUNT.fieldId())); protected CloseableIterable readManifests(Snapshot snapshot) { - if (snapshot.manifestListLocation() != null) { - return InternalData.read( - FileFormat.AVRO, fileIO.newInputFile(snapshot.manifestListLocation())) + if (snapshot.rootLocation() != null) { + return InternalData.read(FileFormat.AVRO, fileIO.newInputFile(snapshot.rootLocation())) .setRootType(GenericManifestFile.class) .project(MANIFEST_PROJECTION) .reuseContainers() diff --git a/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java b/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java index 911a94c13f28..94b79acda271 100644 --- a/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java +++ b/core/src/main/java/org/apache/iceberg/IncrementalFileCleanup.java @@ -124,7 +124,7 @@ public void cleanFiles( LOG.warn( "Failed on snapshot {} while reading manifest list: {}", snapshot.snapshotId(), - snapshot.manifestListLocation(), + snapshot.rootLocation(), exc)) .run( snapshot -> { @@ -155,12 +155,12 @@ public void cleanFiles( } catch (IOException e) { throw new RuntimeIOException( - e, "Failed to close manifest list: %s", snapshot.manifestListLocation()); + e, "Failed to close manifest list: %s", snapshot.rootLocation()); } }); // find manifests to clean up that were only referenced by snapshots that have expired - Set manifestListsToDelete = ConcurrentHashMap.newKeySet(); + Set snapshotFilesToDelete = ConcurrentHashMap.newKeySet(); Set manifestsToDelete = ConcurrentHashMap.newKeySet(); Set manifestsToRevert = ConcurrentHashMap.newKeySet(); Tasks.foreach(beforeExpiration.snapshots()) @@ -172,7 +172,7 @@ public void cleanFiles( LOG.warn( "Failed on snapshot {} while reading manifest list: {}", snapshot.snapshotId(), - snapshot.manifestListLocation(), + snapshot.rootLocation(), exc)) .run( snapshot -> { @@ -248,12 +248,12 @@ public void cleanFiles( } } catch (IOException e) { throw new RuntimeIOException( - e, "Failed to close manifest list: %s", snapshot.manifestListLocation()); + e, "Failed to close manifest list: %s", snapshot.rootLocation()); } // add the manifest list to the delete set, if present - if (snapshot.manifestListLocation() != null) { - manifestListsToDelete.add(snapshot.manifestListLocation()); + if (snapshot.rootLocation() != null) { + snapshotFilesToDelete.add(snapshot.rootLocation()); } } }); @@ -268,8 +268,8 @@ public void cleanFiles( LOG.debug("Deleting {} manifest files", manifestsToDelete.size()); deleteFiles(manifestsToDelete, "manifest"); - LOG.debug("Deleting {} manifest-list files", manifestListsToDelete.size()); - deleteFiles(manifestListsToDelete, "manifest list"); + LOG.debug("Deleting {} manifest-list files", snapshotFilesToDelete.size()); + deleteFiles(snapshotFilesToDelete, "manifest list"); if (hasAnyStatisticsFiles(beforeExpiration)) { Set expiredStatisticsFilesLocations = diff --git a/core/src/main/java/org/apache/iceberg/ManifestListWriter.java b/core/src/main/java/org/apache/iceberg/ManifestListWriter.java index 378bb9dffbc2..fe3817ea2c5e 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestListWriter.java +++ b/core/src/main/java/org/apache/iceberg/ManifestListWriter.java @@ -93,7 +93,7 @@ public Long nextRowId() { return null; } - public ManifestListFile toManifestListFile() { + public FileWithKeyId toManifestListFile() { if (manifestListKeyMetadata != null && manifestListKeyMetadata.encryptionKey() != null) { String manifestListKeyID = standardEncryptionManager.addManifestListKeyMetadata( diff --git a/core/src/main/java/org/apache/iceberg/ManifestLists.java b/core/src/main/java/org/apache/iceberg/ManifestLists.java index dbe080584b13..b66743bd6a55 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestLists.java +++ b/core/src/main/java/org/apache/iceberg/ManifestLists.java @@ -32,7 +32,7 @@ class ManifestLists { private ManifestLists() {} - static InputFile newInputFile(FileIO io, ManifestListFile manifestList) { + static InputFile newInputFile(FileIO io, FileWithKeyId manifestList) { InputFile input = io.newInputFile(manifestList); if (ManifestFiles.cachingEnabled(io)) { return ManifestFiles.contentCache(io).tryCache(input); diff --git a/core/src/main/java/org/apache/iceberg/ManifestsTable.java b/core/src/main/java/org/apache/iceberg/ManifestsTable.java index 252f074b1b5d..dc6c23a64e83 100644 --- a/core/src/main/java/org/apache/iceberg/ManifestsTable.java +++ b/core/src/main/java/org/apache/iceberg/ManifestsTable.java @@ -77,7 +77,7 @@ MetadataTableType metadataTableType() { protected DataTask task(TableScan scan) { FileIO io = table().io(); - String location = scan.snapshot().manifestListLocation(); + String location = scan.snapshot().rootLocation(); Map specs = Maps.newHashMap(table().specs()); return StaticDataTask.of( diff --git a/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java b/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java index b5a565b81e7e..f8545996811f 100644 --- a/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java +++ b/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java @@ -59,7 +59,7 @@ public void cleanFiles( return; } - Set manifestListsToDelete = Sets.newHashSet(); + Set snapshotFilesToDelete = Sets.newHashSet(); Set snapshotsBeforeExpiration = Sets.newHashSet(beforeExpiration.snapshots()); Set snapshotsAfterExpiration = Sets.newHashSet(afterExpiration.snapshots()); @@ -67,8 +67,8 @@ public void cleanFiles( for (Snapshot snapshot : snapshotsBeforeExpiration) { if (!snapshotsAfterExpiration.contains(snapshot)) { expiredSnapshots.add(snapshot); - if (snapshot.manifestListLocation() != null) { - manifestListsToDelete.add(snapshot.manifestListLocation()); + if (snapshot.rootLocation() != null) { + snapshotFilesToDelete.add(snapshot.rootLocation()); } } } @@ -95,8 +95,8 @@ public void cleanFiles( } } - LOG.debug("Deleting {} manifest-list files", manifestListsToDelete.size()); - deleteFiles(manifestListsToDelete, "manifest list"); + LOG.debug("Deleting {} manifest-list files", snapshotFilesToDelete.size()); + deleteFiles(snapshotFilesToDelete, "manifest list"); if (hasAnyStatisticsFiles(beforeExpiration)) { Set expiredStatisticsFilesLocations = @@ -134,7 +134,7 @@ private Set pruneReferencedManifests( } } catch (IOException e) { throw new RuntimeIOException( - e, "Failed to close manifest list: %s", snapshot.manifestListLocation()); + e, "Failed to close manifest list: %s", snapshot.rootLocation()); } }); @@ -160,7 +160,7 @@ private Set readManifests(Set snapshots) { } } catch (IOException e) { throw new RuntimeIOException( - e, "Failed to close manifest list: %s", snapshot.manifestListLocation()); + e, "Failed to close manifest list: %s", snapshot.rootLocation()); } }); diff --git a/core/src/main/java/org/apache/iceberg/ReachableFileUtil.java b/core/src/main/java/org/apache/iceberg/ReachableFileUtil.java index fc53d6b79542..0f2206c1348b 100644 --- a/core/src/main/java/org/apache/iceberg/ReachableFileUtil.java +++ b/core/src/main/java/org/apache/iceberg/ReachableFileUtil.java @@ -99,36 +99,57 @@ private static TableMetadata findFirstExistentPreviousMetadata( } /** - * Returns locations of manifest lists in a table. - * - * @param table table for which manifestList needs to be fetched - * @return the location of manifest lists + * @deprecated since 1.12.0, will be removed in 1.13.0; use {@link #rootLocations(Table)}. The + * method returns v4+ root manifest locations too, so the name no longer matches its behavior. */ + @Deprecated public static List manifestListLocations(Table table) { - return manifestListLocations(table, null); + return rootLocations(table, null); } /** - * Returns locations of manifest lists in a table. - * - * @param table table for which manifestList needs to be fetched - * @param snapshotIds ids of snapshots for which manifest lists will be returned - * @return the location of manifest lists + * @deprecated since 1.12.0, will be removed in 1.13.0; use {@link #rootLocations(Table, Set)}. + * The method returns v4+ root manifest locations too, so the name no longer matches its + * behavior. */ + @Deprecated public static List manifestListLocations(Table table, Set snapshotIds) { + return rootLocations(table, snapshotIds); + } + + /** + * Returns the root location for every snapshot in the table — a manifest list for v3 and earlier, + * or a root manifest for v4+. + * + * @param table table whose root locations should be fetched + * @return the root location per snapshot + */ + public static List rootLocations(Table table) { + return rootLocations(table, null); + } + + /** + * Returns the root location for each snapshot filtered by id — a manifest list for v3 and + * earlier, or a root manifest for v4+. + * + * @param table table whose root locations should be fetched + * @param snapshotIds ids of snapshots to include, or null for every snapshot + * @return the root location per matching snapshot + */ + public static List rootLocations(Table table, Set snapshotIds) { Iterable snapshots = table.snapshots(); if (snapshotIds != null) { snapshots = Iterables.filter(snapshots, s -> snapshotIds.contains(s.snapshotId())); } - List manifestListLocations = Lists.newArrayList(); + List rootLocations = Lists.newArrayList(); for (Snapshot snapshot : snapshots) { - String manifestListLocation = snapshot.manifestListLocation(); - if (manifestListLocation != null) { - manifestListLocations.add(manifestListLocation); + String location = snapshot.rootLocation(); + if (location != null) { + rootLocations.add(location); } } - return manifestListLocations; + return rootLocations; } /** diff --git a/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java b/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java index de9217259cd4..0710614ee5b8 100644 --- a/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java +++ b/core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java @@ -242,8 +242,7 @@ private static List updatePathInSnapshots( TableMetadata metadata, String sourcePrefix, String targetPrefix) { List newSnapshots = Lists.newArrayListWithCapacity(metadata.snapshots().size()); for (Snapshot snapshot : metadata.snapshots()) { - String newManifestListLocation = - newPath(snapshot.manifestListLocation(), sourcePrefix, targetPrefix); + String newManifestListLocation = newPath(snapshot.rootLocation(), sourcePrefix, targetPrefix); Snapshot newSnapshot = new BaseSnapshot( snapshot.sequenceNumber(), @@ -308,7 +307,7 @@ public static RewriteResult rewriteManifestList( "{} of {} manifests in {} were not rewritten in this run and keep their source length", carriedOver, manifestFiles.size(), - snapshot.manifestListLocation()); + snapshot.rootLocation()); } EncryptionManager encryptionManager = @@ -343,7 +342,7 @@ public static RewriteResult rewriteManifestList( return result; } catch (IOException e) { throw new UncheckedIOException( - "Failed to rewrite the manifest list file " + snapshot.manifestListLocation(), e); + "Failed to rewrite the manifest list file " + snapshot.rootLocation(), e); } } @@ -351,7 +350,7 @@ private static List manifestFilesInSnapshot(FileIO io, Snapshot sn try { return snapshot.allManifests(io); } catch (RuntimeIOException e) { - LOG.warn("Failed to read manifest list {}", snapshot.manifestListLocation(), e); + LOG.warn("Failed to read manifest list {}", snapshot.rootLocation(), e); return ImmutableList.of(); } } diff --git a/core/src/main/java/org/apache/iceberg/SnapshotParser.java b/core/src/main/java/org/apache/iceberg/SnapshotParser.java index 53cec16dcd87..2011cae7d048 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotParser.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotParser.java @@ -82,10 +82,10 @@ static void toJson(Snapshot snapshot, JsonGenerator generator) throws IOExceptio generator.writeEndObject(); } - String manifestList = snapshot.manifestListLocation(); - if (manifestList != null) { + String snapshotFile = snapshot.rootLocation(); + if (snapshotFile != null) { // write just the location. manifests should not be embedded in JSON along with a list - generator.writeStringField(MANIFEST_LIST, manifestList); + generator.writeStringField(MANIFEST_LIST, snapshotFile); } else { // embed the manifest list in the JSON, v1 only JsonUtil.writeStringArray( diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index 06adc0bb9d79..c1a633aa3da1 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -106,7 +106,7 @@ public void accept(String file) { private final String commitUUID = UUID.randomUUID().toString(); private final AtomicInteger manifestCount = new AtomicInteger(0); private final AtomicInteger attempt = new AtomicInteger(0); - private final List manifestLists = Lists.newArrayList(); + private final List snapshotFiles = Lists.newArrayList(); private final long targetManifestSizeBytes; private final FileFormat manifestFormat; private final Map manifestWriterProps; @@ -313,8 +313,8 @@ public Snapshot apply() { base.nextRowId()); try (writer) { - // keep track of the manifest lists created - manifestLists.add(manifestList.location()); + // keep track of the snapshot files created + snapshotFiles.add(manifestList.location()); ManifestFile[] manifestFiles = new ManifestFile[manifests.size()]; @@ -364,7 +364,7 @@ public Snapshot apply() { manifestList.location(), nextRowId, assignedRows, - writer.toManifestListFile().encryptionKeyID()); + writer.toManifestListFile().keyId()); } private void runValidations(Snapshot parentSnapshot) { @@ -544,10 +544,12 @@ public void commit() { cleanUncommitted(Sets.newHashSet(saved.allManifests(ops.io()))); } - // also clean up unused manifest lists created by multiple attempts - for (String manifestList : manifestLists) { - if (!saved.manifestListLocation().equals(manifestList)) { - deleteFile(manifestList); + // also clean up unused snapshot files (manifest lists for v3, root manifests for v4) + // created by multiple attempts. + String committedLocation = saved.rootLocation(); + for (String snapshotFile : snapshotFiles) { + if (!snapshotFile.equals(committedLocation)) { + deleteFile(snapshotFile); } } } else { @@ -596,10 +598,10 @@ private void notifyListeners() { } protected void cleanAll() { - for (String manifestList : manifestLists) { - deleteFile(manifestList); + for (String snapshotFile : snapshotFiles) { + deleteFile(snapshotFile); } - manifestLists.clear(); + snapshotFiles.clear(); cleanUncommitted(EMPTY_SET); } diff --git a/core/src/main/java/org/apache/iceberg/SnapshotsTable.java b/core/src/main/java/org/apache/iceberg/SnapshotsTable.java index f948c5578345..2066b9f4e0c0 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotsTable.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotsTable.java @@ -37,7 +37,8 @@ public class SnapshotsTable extends BaseMetadataTable { Types.NestedField.optional( 6, "summary", - Types.MapType.ofRequired(7, 8, Types.StringType.get(), Types.StringType.get()))); + Types.MapType.ofRequired(7, 8, Types.StringType.get(), Types.StringType.get())), + Types.NestedField.optional(9, "root_location", Types.StringType.get())); SnapshotsTable(Table table) { this(table, table.name() + ".snapshots"); @@ -94,13 +95,16 @@ public CloseableIterable planFiles() { } } - private static StaticDataTask.Row snapshotToRow(Snapshot snap) { + static StaticDataTask.Row snapshotToRow(Snapshot snap) { + boolean adaptive = snap.formatVersion() != TableMetadata.UNREPORTED_FORMAT_VERSION; + String location = snap.rootLocation(); return StaticDataTask.Row.of( snap.timestampMillis() * 1000, snap.snapshotId(), snap.parentId(), snap.operation(), - snap.manifestListLocation(), - snap.summary()); + adaptive ? null : location, + snap.summary(), + location); } } diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java index a43d4145907a..36197da68b54 100644 --- a/core/src/main/java/org/apache/iceberg/TableMetadata.java +++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java @@ -56,6 +56,7 @@ public class TableMetadata implements Serializable { static final long INVALID_SEQUENCE_NUMBER = -1; static final int DEFAULT_TABLE_FORMAT_VERSION = 2; static final int SUPPORTED_TABLE_FORMAT_VERSION = 4; + static final int UNREPORTED_FORMAT_VERSION = 0; static final int MIN_FORMAT_VERSION_ROW_LINEAGE = 3; static final int MIN_FORMAT_VERSION_PARQUET_MANIFESTS = 4; static final int MIN_FORMAT_VERSION_OPTIONAL_LOCATION = 4; diff --git a/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java b/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java index 382d244883d6..77d9f746076a 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java +++ b/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java @@ -25,6 +25,7 @@ import java.util.Map; import java.util.Set; import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.FileWithKeyId; import org.apache.iceberg.ManifestListFile; import org.apache.iceberg.TableProperties; import org.apache.iceberg.common.DynConstructors; @@ -140,19 +141,36 @@ public static ByteBuffer setFileLength(ByteBuffer keyMetadata, long fileLength) * @param manifestList a ManifestListFile * @param em the table's EncryptionManager * @return a decrypted key metadata buffer + * @deprecated since 1.12.0. Will be removed in 2.0.0; use {@link + * #decryptKeyMetadata(FileWithKeyId, EncryptionManager)} instead. */ + @Deprecated public static ByteBuffer decryptManifestListKeyMetadata( ManifestListFile manifestList, EncryptionManager em) { + return decryptKeyMetadata(manifestList.encryptionKeyID(), em); + } + + /** + * Decrypt the key metadata of an encryptable file. + * + * @param file a FileWithKeyId + * @param em the table's EncryptionManager + * @return a decrypted key metadata buffer + */ + public static ByteBuffer decryptKeyMetadata(FileWithKeyId file, EncryptionManager em) { + return decryptKeyMetadata(file.keyId(), em); + } + + static ByteBuffer decryptKeyMetadata(String encryptionKeyId, EncryptionManager em) { Preconditions.checkState( em instanceof StandardEncryptionManager, "Snapshot key metadata encryption requires a StandardEncryptionManager"); StandardEncryptionManager sem = (StandardEncryptionManager) em; - String manifestListKeyId = manifestList.encryptionKeyID(); Map encryptionKeys = sem.encryptionKeys(); - EncryptedKey manifestListKey = encryptionKeys.get(manifestListKeyId); - ByteBuffer encryptedKeyMetadata = manifestListKey.encryptedKeyMetadata(); - String keyEncryptionKeyID = manifestListKey.encryptedById(); - ByteBuffer keyEncryptionKey = sem.encryptedByKey(manifestListKeyId); + EncryptedKey encryptionKey = encryptionKeys.get(encryptionKeyId); + ByteBuffer encryptedKeyMetadata = encryptionKey.encryptedKeyMetadata(); + String keyEncryptionKeyID = encryptionKey.encryptedById(); + ByteBuffer keyEncryptionKey = sem.encryptedByKey(encryptionKeyId); String keyEncryptionKeyTimestamp = encryptionKeys .get(keyEncryptionKeyID) diff --git a/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java b/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java index f1b19fa30489..ddf18e7e9312 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java +++ b/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java @@ -106,6 +106,11 @@ public Iterable decrypt(Iterable encrypted) { return Iterables.transform(encrypted, this::decrypt); } + @Override + public ByteBuffer decryptKeyMetadata(String keyId) { + return EncryptionUtil.decryptKeyMetadata(keyId, this); + } + private LoadingCache unwrappedKeyCache() { if (this.unwrappedKeyCache == null) { this.unwrappedKeyCache = diff --git a/core/src/test/java/org/apache/iceberg/TestAllManifestsTableTaskParser.java b/core/src/test/java/org/apache/iceberg/TestAllManifestsTableTaskParser.java index 05656926a881..3295b26b8c24 100644 --- a/core/src/test/java/org/apache/iceberg/TestAllManifestsTableTaskParser.java +++ b/core/src/test/java/org/apache/iceberg/TestAllManifestsTableTaskParser.java @@ -148,8 +148,7 @@ private void assertTaskEquals( assertThat(actual.specsById()).isEqualTo(expected.specsById()); assertThat(actual.manifestList().location()).isEqualTo(expected.manifestList().location()); - assertThat(actual.manifestList().encryptionKeyID()) - .isEqualTo(expected.manifestList().encryptionKeyID()); + assertThat(actual.manifestList().keyId()).isEqualTo(expected.manifestList().keyId()); assertThat(actual.residual().toString()).isEqualTo(expected.residual().toString()); assertThat(actual.referenceSnapshotId()).isEqualTo(expected.referenceSnapshotId()); } diff --git a/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java b/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java index b85d14525b68..65cace148d17 100644 --- a/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java +++ b/core/src/test/java/org/apache/iceberg/TestDataTaskParser.java @@ -46,17 +46,21 @@ public class TestDataTaskParser { Types.NestedField.optional( 6, "summary", - Types.MapType.ofRequired(7, 8, Types.StringType.get(), Types.StringType.get()))); + Types.MapType.ofRequired(7, 8, Types.StringType.get(), Types.StringType.get())), + Types.NestedField.optional(9, "root_location", Types.StringType.get())); - // copied from SnapshotsTable to avoid making it package public + // mirrors SnapshotsTable.snapshotToRow, which is package-private private static StaticDataTask.Row snapshotToRow(Snapshot snap) { + boolean adaptive = snap.formatVersion() != TableMetadata.UNREPORTED_FORMAT_VERSION; + String location = snap.rootLocation(); return StaticDataTask.Row.of( snap.timestampMillis() * 1000, snap.snapshotId(), snap.parentId(), snap.operation(), - snap.manifestListLocation(), - snap.summary()); + adaptive ? null : location, + snap.summary(), + location); } @Test @@ -134,7 +138,8 @@ public void missingFields() throws Exception { + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"}," + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\"," + "\"key-id\":7,\"key\":\"string\",\"value-id\":8," - + "\"value\":\"string\",\"value-required\":true}}]}," + + "\"value\":\"string\",\"value-required\":true}}," + + "{\"id\":9,\"name\":\"root_location\",\"required\":false,\"type\":\"string\"}]}," + "\"projection\":{\"type\":\"struct\",\"schema-id\":0," + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}," + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"}," @@ -143,7 +148,8 @@ public void missingFields() throws Exception { + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"}," + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\"," + "\"key-id\":7,\"key\":\"string\",\"value-id\":8," - + "\"value\":\"string\",\"value-required\":true}}]}," + + "\"value\":\"string\",\"value-required\":true}}," + + "{\"id\":9,\"name\":\"root_location\",\"required\":false,\"type\":\"string\"}]}," + "\"metadata-file\":{\"spec-id\":0,\"content\":\"data\"," + "\"file-path\":\"/tmp/metadata2.json\"," + "\"file-format\":\"metadata\",\"partition\":[]," @@ -253,7 +259,8 @@ private String snapshotsDataTaskJson() { + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"}," + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\"," + "\"key-id\":7,\"key\":\"string\",\"value-id\":8," - + "\"value\":\"string\",\"value-required\":true}}]}," + + "\"value\":\"string\",\"value-required\":true}}," + + "{\"id\":9,\"name\":\"root_location\",\"required\":false,\"type\":\"string\"}]}," + "\"projection\":{\"type\":\"struct\",\"schema-id\":0," + "\"fields\":[{\"id\":1,\"name\":\"committed_at\",\"required\":true,\"type\":\"timestamptz\"}," + "{\"id\":2,\"name\":\"snapshot_id\",\"required\":true,\"type\":\"long\"}," @@ -262,7 +269,8 @@ private String snapshotsDataTaskJson() { + "{\"id\":5,\"name\":\"manifest_list\",\"required\":false,\"type\":\"string\"}," + "{\"id\":6,\"name\":\"summary\",\"required\":false,\"type\":{\"type\":\"map\"," + "\"key-id\":7,\"key\":\"string\",\"value-id\":8," - + "\"value\":\"string\",\"value-required\":true}}]}," + + "\"value\":\"string\",\"value-required\":true}}," + + "{\"id\":9,\"name\":\"root_location\",\"required\":false,\"type\":\"string\"}]}," + "\"metadata-file\":{\"spec-id\":0,\"content\":\"data\"," + "\"file-path\":\"/tmp/metadata2.json\"," + "\"file-format\":\"metadata\",\"partition\":[]," @@ -272,13 +280,15 @@ private String snapshotsDataTaskJson() { + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\"," + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\"," + "\"total-position-deletes\",\"total-equality-deletes\"]," - + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}}," + + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"1\",\"10\",\"1\",\"0\",\"0\",\"0\"]}," + + "\"9\":\"file:/tmp/manifest1.avro\"}," + "{\"1\":\"2282-12-22T20:13:30+00:00\",\"2\":2,\"3\":1,\"4\":\"append\"," + "\"5\":\"file:/tmp/manifest2.avro\"," + "\"6\":{\"keys\":[\"added-data-files\",\"added-records\",\"added-files-size\",\"changed-partition-count\"," + "\"total-records\",\"total-files-size\",\"total-data-files\",\"total-delete-files\"," + "\"total-position-deletes\",\"total-equality-deletes\"]," - + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}}]}"; + + "\"values\":[\"1\",\"1\",\"10\",\"1\",\"2\",\"20\",\"2\",\"0\",\"0\",\"0\"]}," + + "\"9\":\"file:/tmp/manifest2.avro\"}]}"; } private void assertDataTaskEquals(StaticDataTask expected, StaticDataTask actual) { diff --git a/core/src/test/java/org/apache/iceberg/TestManifestListEncryption.java b/core/src/test/java/org/apache/iceberg/TestManifestListEncryption.java index 0acda5e21e23..593c5781a17c 100644 --- a/core/src/test/java/org/apache/iceberg/TestManifestListEncryption.java +++ b/core/src/test/java/org/apache/iceberg/TestManifestListEncryption.java @@ -220,7 +220,7 @@ private ManifestFile writeAndReadEncryptedManifestList(EncryptionManager em) thr SNAPSHOT_FIRST_ROW_ID); writer.add(TEST_MANIFEST); writer.close(); - ManifestListFile manifestListFile = writer.toManifestListFile(); + FileWithKeyId manifestListFile = writer.toManifestListFile(); // First try to read without decryption assertThatThrownBy(() -> ManifestLists.read(outputFile.toInputFile())) diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotsTable.java b/core/src/test/java/org/apache/iceberg/TestSnapshotsTable.java new file mode 100644 index 000000000000..3fb74bb55894 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestSnapshotsTable.java @@ -0,0 +1,68 @@ +/* + * 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.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.junit.jupiter.api.Test; + +public class TestSnapshotsTable { + + private static final String LOCATION = "file:/tmp/snap-1.avro"; + + @Test + public void legacySnapshotPopulatesManifestListAndRootLocation() { + Snapshot snap = stubSnapshot(TableMetadata.UNREPORTED_FORMAT_VERSION); + StaticDataTask.Row row = SnapshotsTable.snapshotToRow(snap); + + assertThat(row.get(4, String.class)) + .as("manifest_list should be populated for legacy snapshots") + .isEqualTo(LOCATION); + assertThat(row.get(6, String.class)) + .as("root_location should always be populated") + .isEqualTo(LOCATION); + } + + @Test + public void adaptiveSnapshotNullsManifestListAndPopulatesRootLocation() { + Snapshot snap = stubSnapshot(4); + StaticDataTask.Row row = SnapshotsTable.snapshotToRow(snap); + + assertThat(row.get(4, String.class)) + .as("manifest_list should be null for adaptive snapshots") + .isNull(); + assertThat(row.get(6, String.class)) + .as("root_location should be populated") + .isEqualTo(LOCATION); + } + + private static Snapshot stubSnapshot(int formatVersion) { + Snapshot snap = mock(Snapshot.class); + when(snap.formatVersion()).thenReturn(formatVersion); + when(snap.rootLocation()).thenReturn(LOCATION); + when(snap.timestampMillis()).thenReturn(1L); + when(snap.snapshotId()).thenReturn(1L); + when(snap.parentId()).thenReturn(null); + when(snap.operation()).thenReturn("append"); + when(snap.summary()).thenReturn(null); + return snap; + } +} diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestCatalogUtilDropTable.java b/core/src/test/java/org/apache/iceberg/hadoop/TestCatalogUtilDropTable.java index 7c1e284b27e9..8c6aa3bdd250 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/TestCatalogUtilDropTable.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/TestCatalogUtilDropTable.java @@ -31,11 +31,11 @@ import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileWithKeyId; import org.apache.iceberg.GenericBlobMetadata; import org.apache.iceberg.GenericStatisticsFile; import org.apache.iceberg.ImmutableGenericPartitionStatisticsFile; import org.apache.iceberg.ManifestFile; -import org.apache.iceberg.ManifestListFile; import org.apache.iceberg.PartitionStatisticsFile; import org.apache.iceberg.Snapshot; import org.apache.iceberg.SnapshotChanges; @@ -199,9 +199,8 @@ private static FileIO createMockFileIO(FileIO wrapped) { .thenAnswer( invocation -> wrapped.newInputFile(invocation.getArgument(0), invocation.getArgument(1))); - Mockito.when(mockIO.newInputFile(Mockito.any(ManifestListFile.class))) - .thenAnswer( - invocation -> wrapped.newInputFile((ManifestListFile) invocation.getArgument(0))); + Mockito.when(mockIO.newInputFile(Mockito.any(FileWithKeyId.class))) + .thenAnswer(invocation -> wrapped.newInputFile((FileWithKeyId) invocation.getArgument(0))); Mockito.when(mockIO.newInputFile(Mockito.any(ManifestFile.class))) .thenAnswer(invocation -> wrapped.newInputFile((ManifestFile) invocation.getArgument(0))); Mockito.when(mockIO.newInputFile(Mockito.any(DataFile.class))) diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java index f9000511c11f..e4bf9b1e714d 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java @@ -72,7 +72,7 @@ public void processElement(Trigger trigger, Context ctx, Collector colle .forEach( snapshot -> { // Manifest lists - collector.collect(snapshot.manifestListLocation()); + collector.collect(snapshot.rootLocation()); // Snapshot JSONs ReachableFileUtil.metadataFileLocations(table, false).forEach(collector::collect); // Statistics files diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java index f9000511c11f..e4bf9b1e714d 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java @@ -72,7 +72,7 @@ public void processElement(Trigger trigger, Context ctx, Collector colle .forEach( snapshot -> { // Manifest lists - collector.collect(snapshot.manifestListLocation()); + collector.collect(snapshot.rootLocation()); // Snapshot JSONs ReachableFileUtil.metadataFileLocations(table, false).forEach(collector::collect); // Statistics files diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java index f9000511c11f..e4bf9b1e714d 100644 --- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java +++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java @@ -72,7 +72,7 @@ public void processElement(Trigger trigger, Context ctx, Collector colle .forEach( snapshot -> { // Manifest lists - collector.collect(snapshot.manifestListLocation()); + collector.collect(snapshot.rootLocation()); // Snapshot JSONs ReachableFileUtil.metadataFileLocations(table, false).forEach(collector::collect); // Statistics files diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java index f9000511c11f..e4bf9b1e714d 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ListMetadataFiles.java @@ -72,7 +72,7 @@ public void processElement(Trigger trigger, Context ctx, Collector colle .forEach( snapshot -> { // Manifest lists - collector.collect(snapshot.manifestListLocation()); + collector.collect(snapshot.rootLocation()); // Snapshot JSONs ReachableFileUtil.metadataFileLocations(table, false).forEach(collector::collect); // Statistics files diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java index b178b713e45e..ee73e65917a9 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java @@ -507,7 +507,7 @@ private RewriteResult rewriteManifestList( Snapshot snapshot, TableMetadata tableMetadata, Map rewrittenManifestLengths) { RewriteResult result = new RewriteResult<>(); - String path = snapshot.manifestListLocation(); + String path = snapshot.rootLocation(); String outputPath = RewriteTablePathUtil.stagingPath(path, sourcePrefix, stagingDir); RewriteResult rewriteResult = RewriteTablePathUtil.rewriteManifestList( diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index f390acf1ff0d..9c12fcef0b10 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -939,6 +939,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-data-files", "1", "total-records", "1")) + .set("root_location", firstManifestList) .build(), builder .set("committed_at", secondSnapshotTimestamp * 1000) @@ -954,6 +955,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-records", "0", "total-data-files", "0")) + .set("root_location", secondManifestList) .build()); assertThat(actual).as("Snapshots table should have a row for each snapshot").hasSize(2); diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java index c637937d49b2..0c10f87c203c 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java @@ -216,7 +216,7 @@ public void testExpressionPushdown() { public void testMetadataTables() { assertEquals( "Snapshot metadata table", - ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY)), + ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY, ANY)), sql("SELECT * FROM %s.snapshots", tableName)); } diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java index b178b713e45e..ee73e65917a9 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java @@ -507,7 +507,7 @@ private RewriteResult rewriteManifestList( Snapshot snapshot, TableMetadata tableMetadata, Map rewrittenManifestLengths) { RewriteResult result = new RewriteResult<>(); - String path = snapshot.manifestListLocation(); + String path = snapshot.rootLocation(); String outputPath = RewriteTablePathUtil.stagingPath(path, sourcePrefix, stagingDir); RewriteResult rewriteResult = RewriteTablePathUtil.rewriteManifestList( diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index f390acf1ff0d..9c12fcef0b10 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -939,6 +939,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-data-files", "1", "total-records", "1")) + .set("root_location", firstManifestList) .build(), builder .set("committed_at", secondSnapshotTimestamp * 1000) @@ -954,6 +955,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-records", "0", "total-data-files", "0")) + .set("root_location", secondManifestList) .build()); assertThat(actual).as("Snapshots table should have a row for each snapshot").hasSize(2); diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java index 57457c0c3f31..ea3f5ac69e53 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java @@ -219,7 +219,7 @@ public void testExpressionPushdown() { public void testMetadataTables() { assertEquals( "Snapshot metadata table", - ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY)), + ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY, ANY)), sql("SELECT * FROM %s.snapshots", tableName)); } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java index b178b713e45e..ee73e65917a9 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java @@ -507,7 +507,7 @@ private RewriteResult rewriteManifestList( Snapshot snapshot, TableMetadata tableMetadata, Map rewrittenManifestLengths) { RewriteResult result = new RewriteResult<>(); - String path = snapshot.manifestListLocation(); + String path = snapshot.rootLocation(); String outputPath = RewriteTablePathUtil.stagingPath(path, sourcePrefix, stagingDir); RewriteResult rewriteResult = RewriteTablePathUtil.rewriteManifestList( diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 16e1754c1b48..c729a59b37ec 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -944,6 +944,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-data-files", "1", "total-records", "1")) + .set("root_location", firstManifestList) .build(), builder .set("committed_at", secondSnapshotTimestamp * 1000) @@ -959,6 +960,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-records", "0", "total-data-files", "0")) + .set("root_location", secondManifestList) .build()); assertThat(actual).as("Snapshots table should have a row for each snapshot").hasSize(2); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java index 2364cea11ddc..6ffc0538349c 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java @@ -218,7 +218,7 @@ public void testExpressionPushdown() { public void testMetadataTables() { assertEquals( "Snapshot metadata table", - ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY)), + ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY, ANY)), sql("SELECT * FROM %s.snapshots", tableName)); } diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java index b178b713e45e..ee73e65917a9 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java @@ -507,7 +507,7 @@ private RewriteResult rewriteManifestList( Snapshot snapshot, TableMetadata tableMetadata, Map rewrittenManifestLengths) { RewriteResult result = new RewriteResult<>(); - String path = snapshot.manifestListLocation(); + String path = snapshot.rootLocation(); String outputPath = RewriteTablePathUtil.stagingPath(path, sourcePrefix, stagingDir); RewriteResult rewriteResult = RewriteTablePathUtil.rewriteManifestList( diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java index 16e1754c1b48..c729a59b37ec 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java @@ -944,6 +944,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-data-files", "1", "total-records", "1")) + .set("root_location", firstManifestList) .build(), builder .set("committed_at", secondSnapshotTimestamp * 1000) @@ -959,6 +960,7 @@ public void testSnapshotsTable() { "changed-partition-count", "1", "total-records", "0", "total-data-files", "0")) + .set("root_location", secondManifestList) .build()); assertThat(actual).as("Snapshots table should have a row for each snapshot").hasSize(2); diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java index 2364cea11ddc..6ffc0538349c 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java @@ -218,7 +218,7 @@ public void testExpressionPushdown() { public void testMetadataTables() { assertEquals( "Snapshot metadata table", - ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY)), + ImmutableList.of(row(ANY, ANY, null, "append", ANY, ANY, ANY)), sql("SELECT * FROM %s.snapshots", tableName)); }