API, Core, Spark: add Snapshot.rootManifestLocation()

api/src/main/java/org/apache/iceberg/FileWithEncryptedKey.java

@@ -0,0 +1,43 @@
+/*
+ * 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 org.apache.iceberg.encryption.EncryptedKey;
+
+/**
+ * A file that may be encrypted using an {@link EncryptedKey}.
+ *
+ * <p>The encrypted key is either referenced by its ID from the table metadata encryption keys, or
+ * stored directly with the file's metadata that describe the file. If neither is present, the file
+ * is not encrypted.
+ */
+public interface FileWithEncryptedKey {
+
+  /** Returns the location of the file. */
+  String location();
+
+  /** Returns the length of the file in bytes, or null if unknown. */
+  Long length();
+
+  /** Returns the ID of the encrypted key in table metadata, or null if not set. */
+  String keyId();
+
+  /** Returns the encrypted key stored directly with the file, or null if not set. */
+  EncryptedKey encryptedKey();
+}

api/src/main/java/org/apache/iceberg/ManifestListFile.java

@@ -21,6 +21,10 @@
 import java.nio.ByteBuffer;
 import org.apache.iceberg.encryption.EncryptionManager;
 
+/**
+ * @deprecated since 1.13.0, will be removed in 2.0.0; use {@link FileWithEncryptedKey} instead.
+ */
+@Deprecated
 public interface ManifestListFile {
 
   /** Location of manifest list file. */

api/src/main/java/org/apache/iceberg/Snapshot.java

@@ -166,10 +166,26 @@ default Iterable<DeleteFile> 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
+   * <p>Only set for tables with format version 1 through 3. Snapshots in v4 tables use {@link
+   * #rootManifestLocation()} instead.
+   *
+   * @return the location of the manifest list for this snapshot
    */
   String manifestListLocation();
 
+  /**
+   * Return the location of this snapshot's root manifest, or null if this snapshot has no root
+   * manifest.
+   *
+   * <p>v4 snapshots store this in {@code root-manifest} instead of a manifest list. At most one of
+   * this and {@link #manifestListLocation()} is set.
+   *
+   * @return the location of the root manifest for this snapshot
+   */
+  default String rootManifestLocation() {
+    return null;
+  }
+
   /**
    * Return the id of the schema used when this snapshot was created, or null if this information is
    * not available.

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.FileWithEncryptedKey;
 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.13.0, will be removed in 2.0.0; use {@link
+   *     #newInputFile(FileWithEncryptedKey)} instead.
+   */
+  @Deprecated
   @Override
   public InputFile newInputFile(ManifestListFile manifestList) {
     if (manifestList.encryptionKeyID() != null) {
@@ -140,6 +146,20 @@ public InputFile newInputFile(ManifestListFile manifestList) {
     }
   }
 
+  @Override
+  public InputFile newInputFile(FileWithEncryptedKey file) {
+    ByteBuffer keyMetadata = em.decryptKeyMetadata(file);
+    if (keyMetadata != null) {
+      return file.length() != null
+          ? newDecryptingInputFile(file.location(), file.length(), keyMetadata)
+          : newDecryptingInputFile(file.location(), keyMetadata);
+    } else {
+      return file.length() != null
+          ? newInputFile(file.location(), file.length())
+          : newInputFile(file.location());
+    }
+  }
+
   public InputFile newDecryptingInputFile(String path, ByteBuffer buffer) {
     return em.decrypt(wrap(io.newInputFile(path), buffer));
   }

api/src/main/java/org/apache/iceberg/encryption/EncryptionManager.java

@@ -19,6 +19,8 @@
 package org.apache.iceberg.encryption;
 
 import java.io.Serializable;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.FileWithEncryptedKey;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
@@ -51,6 +53,17 @@ default Iterable<InputFile> decrypt(Iterable<EncryptedInputFile> encrypted) {
     return Iterables.transform(encrypted, this::decrypt);
   }
 
+  /**
+   * Decrypt the key metadata of a file that may be encrypted using an {@link EncryptedKey}.
+   *
+   * @param file the file to decrypt the key metadata for
+   * @return the decrypted key metadata buffer, or null if the file is not encrypted
+   */
+  default ByteBuffer decryptKeyMetadata(FileWithEncryptedKey file) {
+    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

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.FileWithEncryptedKey;
 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.13.0, will be removed in 2.0.0; use {@link
+   *     #newInputFile(FileWithEncryptedKey)} instead.
+   */
+  @Deprecated
   default InputFile newInputFile(ManifestListFile manifestList) {
     Preconditions.checkArgument(
         manifestList.encryptionKeyID() == null,
@@ -80,6 +86,18 @@ default InputFile newInputFile(ManifestListFile manifestList) {
     return newInputFile(manifestList.location());
   }
 
+  default InputFile newInputFile(FileWithEncryptedKey file) {
+    Preconditions.checkArgument(
+        file.keyId() == null && file.encryptedKey() == null,
+        "Cannot decrypt file: %s (use EncryptingFileIO)",
+        file.location());
+    if (file.length() != null) {
+      return newInputFile(file.location(), file.length());
+    }
+
+    return newInputFile(file.location());
+  }
+
   /** Get a {@link OutputFile} instance to write bytes to the file at the given path. */
   OutputFile newOutputFile(String path);
 

core/src/main/java/org/apache/iceberg/AllManifestsTable.java

@@ -142,7 +142,8 @@ protected CloseableIterable<FileScanTask> doPlanFiles() {
                       io,
                       schema(),
                       specs,
-                      new BaseManifestListFile(snap.manifestListLocation(), snap.keyId()),
+                      new BaseFileWithEncryptedKey(
+                          snap.manifestListLocation(), null, snap.keyId(), null),
                       residual,
                       snap.snapshotId());
                 } else {
@@ -165,7 +166,7 @@ static class ManifestListReadTask implements DataTask {
     private final FileIO io;
     private final Schema schema;
     private final Map<Integer, PartitionSpec> specs;
-    private final ManifestListFile manifestList;
+    private final FileWithEncryptedKey manifestList;
     private final Expression residual;
     private final long referenceSnapshotId;
     private DataFile lazyDataFile = null;
@@ -175,7 +176,7 @@ static class ManifestListReadTask implements DataTask {
         FileIO io,
         Schema schema,
         Map<Integer, PartitionSpec> specs,
-        ManifestListFile manifestList,
+        FileWithEncryptedKey manifestList,
         Expression residual,
         long referenceSnapshotId) {
       this.dataTableSchema = dataTableSchema;
@@ -276,7 +277,7 @@ Map<Integer, PartitionSpec> specsById() {
       return specs;
     }
 
-    ManifestListFile manifestList() {
+    FileWithEncryptedKey manifestList() {
       return manifestList;
     }
 

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);
@@ -105,7 +105,7 @@ static AllManifestsTable.ManifestListReadTask fromJson(JsonNode jsonNode) {
         fileIO,
         schema,
         specsById,
-        new BaseManifestListFile(manifestListLocation, manifestListKeyId),
+        new BaseFileWithEncryptedKey(manifestListLocation, null, manifestListKeyId, null),
         residualFilter,
         referenceSnapshotId);
   }

core/src/main/java/org/apache/iceberg/BaseManifestListFile.java → core/src/main/java/org/apache/iceberg/BaseFileWithEncryptedKey.java

@@ -19,17 +19,25 @@
 package org.apache.iceberg;
 
 import java.io.Serializable;
-import java.nio.ByteBuffer;
-import org.apache.iceberg.encryption.EncryptionManager;
-import org.apache.iceberg.encryption.EncryptionUtil;
+import org.apache.iceberg.encryption.EncryptedKey;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
-class BaseManifestListFile implements ManifestListFile, Serializable {
+class BaseFileWithEncryptedKey implements FileWithEncryptedKey, Serializable {
   private final String location;
-  private final String encryptionKeyID;
+  private final Long length;
+  private final String keyId;
+  private final EncryptedKey encryptedKey;
 
-  BaseManifestListFile(String location, String encryptionKeyID) {
+  BaseFileWithEncryptedKey(String location, Long length, String keyId, EncryptedKey encryptedKey) {
+    Preconditions.checkArgument(location != null, "Invalid file location: null");
+    Preconditions.checkArgument(
+        keyId == null || encryptedKey == null,
+        "Cannot set both key ID and encrypted key for file: %s",
+        location);
     this.location = location;
-    this.encryptionKeyID = encryptionKeyID;
+    this.length = length;
+    this.keyId = keyId;
+    this.encryptedKey = encryptedKey;
   }
 
   @Override
@@ -38,12 +46,17 @@ public String location() {
   }
 
   @Override
-  public String encryptionKeyID() {
-    return encryptionKeyID;
+  public Long length() {
+    return length;
   }
 
   @Override
-  public ByteBuffer decryptKeyMetadata(EncryptionManager em) {
-    return EncryptionUtil.decryptManifestListKeyMetadata(this, em);
+  public String keyId() {
+    return keyId;
+  }
+
+  @Override
+  public EncryptedKey encryptedKey() {
+    return encryptedKey;
   }
 }