API, Core, Spark, Flink: add version-agnostic Snapshot.rootLocation()

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();
+}

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);
 }

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

@@ -167,9 +167,32 @@ 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
+   * @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.

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));
   }

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<InputFile> decrypt(Iterable<EncryptedInputFile> 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

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);
 

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

@@ -136,13 +136,13 @@ protected CloseableIterable<FileScanTask> 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<Integer, PartitionSpec> 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<Integer, PartitionSpec> specs,
-        ManifestListFile manifestList,
+        FileWithKeyId manifestList,
         Expression residual,
         long referenceSnapshotId) {
       this.dataTableSchema = dataTableSchema;
@@ -276,7 +276,7 @@ Map<Integer, PartitionSpec> specsById() {
       return specs;
     }
 
-    ManifestListFile manifestList() {
+    FileWithKeyId 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);

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;
+  }
+}

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

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<String, String> summary;
   private final Integer schemaId;
@@ -64,7 +64,7 @@ class BaseSnapshot implements Snapshot {
       String operation,
       Map<String, String> 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<DeleteFile> 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)