Updates from review

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

@@ -39,7 +39,7 @@ public interface EncryptedOutputFile {
   EncryptionKeyMetadata keyMetadata();
 
   /** Underlying output file for native encryption. */
-  default OutputFile rawOutputFile() {
+  default OutputFile plainOutputFile() {
     throw new UnsupportedOperationException("Not implemented");
-  };
+  }
 }

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

@@ -156,7 +156,7 @@ private CatalogProperties() {}
 
   public static final String AUTH_SESSION_TIMEOUT_MS = "auth.session-timeout-ms";
   public static final long AUTH_SESSION_TIMEOUT_MS_DEFAULT = TimeUnit.HOURS.toMillis(1);
+
   public static final String ENCRYPTION_KMS_TYPE = "encryption.kms-type";
-  public static final String ENCRYPTION_KMS_CUSTOM_TYPE = "custom";
-  public static final String ENCRYPTION_KMS_CLIENT_IMPL = "encryption.kms.client-impl";
+  public static final String ENCRYPTION_KMS_IMPL = "encryption.kms-impl";
 }

core/src/main/java/org/apache/iceberg/encryption/EncryptedFiles.java

@@ -57,9 +57,5 @@ public static EncryptedOutputFile encryptedOutput(
         encryptedOutputFile, BaseEncryptionKeyMetadata.fromByteArray(keyMetadata));
   }
 
-  public static EncryptedOutputFile plainAsEncryptedOutput(OutputFile encryptingOutputFile) {
-    return new BaseEncryptedOutputFile(encryptingOutputFile, EncryptionKeyMetadata.empty());
-  }
-
   private EncryptedFiles() {}
 }

core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java

@@ -18,72 +18,40 @@
  */
 package org.apache.iceberg.encryption;
 
-import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
-import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
-import static org.apache.iceberg.TableProperties.ENCRYPTION_DEK_LENGTH;
-import static org.apache.iceberg.TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT;
-import static org.apache.iceberg.TableProperties.ENCRYPTION_TABLE_KEY;
-
-import java.nio.ByteBuffer;
 import java.util.Map;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.PropertyUtil;
 
 public class EncryptionUtil {
 
   private EncryptionUtil() {}
 
-  public static KeyMetadata parseKeyMetadata(ByteBuffer metadataBuffer) {
-    return KeyMetadata.parse(metadataBuffer);
-  }
-
-  public static EncryptionKeyMetadata createKeyMetadata(ByteBuffer key, ByteBuffer aadPrefix) {
-    return new KeyMetadata(key, aadPrefix);
-  }
-
-  public static long gcmEncryptionLength(long plainFileLength) {
-    int numberOfFullBlocks = Math.toIntExact(plainFileLength / Ciphers.PLAIN_BLOCK_SIZE);
-    int plainBytesInLastBlock =
-        Math.toIntExact(plainFileLength - numberOfFullBlocks * Ciphers.PLAIN_BLOCK_SIZE);
-    boolean fullBlocksOnly = (0 == plainBytesInLastBlock);
-    int cipherBytesInLastBlock =
-        fullBlocksOnly ? 0 : plainBytesInLastBlock + Ciphers.NONCE_LENGTH + Ciphers.GCM_TAG_LENGTH;
-    int cipherBlockSize = Ciphers.PLAIN_BLOCK_SIZE + Ciphers.NONCE_LENGTH + Ciphers.GCM_TAG_LENGTH;
-    return (long) Ciphers.GCM_STREAM_HEADER_LENGTH
-        + numberOfFullBlocks * cipherBlockSize
-        + cipherBytesInLastBlock;
-  }
-
   public static KeyManagementClient createKmsClient(Map<String, String> catalogProperties) {
     String kmsType = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_TYPE);
+    String kmsImpl = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_IMPL);
 
-    if (kmsType == null) {
-      throw new IllegalStateException(
-          "Cannot create StandardEncryptionManagerFactory without KMS type");
-    }
+    Preconditions.checkArgument(
+        kmsType == null || kmsImpl == null,
+        "Cannot set both KMS type (%s) and KMS impl (%s)",
+        kmsType,
+        kmsImpl);
 
-    if (!kmsType.equals(CatalogProperties.ENCRYPTION_KMS_CUSTOM_TYPE)) {
-      // Currently support only custom types
-      throw new UnsupportedOperationException("Undefined KMS type " + kmsType);
-    }
-
-    String kmsClientImpl = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_CLIENT_IMPL);
-
-    if (kmsClientImpl == null) {
-      throw new IllegalStateException("Custom KMS client class is not defined");
-    }
+    // TODO: Add KMS implementations
+    Preconditions.checkArgument(kmsType == null, "Unsupported KMS type: %s", kmsType);
 
     KeyManagementClient kmsClient;
     DynConstructors.Ctor<KeyManagementClient> ctor;
     try {
-      ctor = DynConstructors.builder(KeyManagementClient.class).impl(kmsClientImpl).buildChecked();
+      ctor = DynConstructors.builder(KeyManagementClient.class).impl(kmsImpl).buildChecked();
     } catch (NoSuchMethodException e) {
       throw new IllegalArgumentException(
           String.format(
               "Cannot initialize KeyManagementClient, missing no-arg constructor for class %s",
-              kmsClientImpl),
+              kmsImpl),
           e);
     }
 
@@ -93,7 +61,7 @@ public static KeyManagementClient createKmsClient(Map<String, String> catalogPro
       throw new IllegalArgumentException(
           String.format(
               "Cannot initialize kms client, %s does not implement KeyManagementClient interface",
-              kmsClientImpl),
+              kmsImpl),
           e);
     }
 
@@ -104,20 +72,19 @@ public static KeyManagementClient createKmsClient(Map<String, String> catalogPro
 
   public static EncryptionManager createEncryptionManager(
       Map<String, String> tableProperties, KeyManagementClient kmsClient) {
-    String tableKeyId = tableProperties.get(ENCRYPTION_TABLE_KEY);
+    Preconditions.checkArgument(kmsClient != null, "Invalid KMS client: null");
+    String tableKeyId = tableProperties.get(TableProperties.ENCRYPTION_TABLE_KEY);
 
     if (null == tableKeyId) {
       // Unencrypted table
       return PlaintextEncryptionManager.instance();
     }
 
-    if (kmsClient == null) {
-      throw new IllegalStateException("Encrypted table. No KMS client is configured in catalog");
-    }
-
     String fileFormat =
         PropertyUtil.propertyAsString(
-            tableProperties, DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT);
+            tableProperties,
+            TableProperties.DEFAULT_FILE_FORMAT,
+            TableProperties.DEFAULT_FILE_FORMAT_DEFAULT);
 
     if (FileFormat.fromString(fileFormat) != FileFormat.PARQUET) {
       throw new UnsupportedOperationException(
@@ -126,12 +93,15 @@ public static EncryptionManager createEncryptionManager(
 
     int dataKeyLength =
         PropertyUtil.propertyAsInt(
-            tableProperties, ENCRYPTION_DEK_LENGTH, ENCRYPTION_DEK_LENGTH_DEFAULT);
+            tableProperties,
+            TableProperties.ENCRYPTION_DEK_LENGTH,
+            TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT);
 
-    return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient);
-  }
+    Preconditions.checkState(
+        dataKeyLength == 16 || dataKeyLength == 24 || dataKeyLength == 32,
+        "Invalid data key length: %s (must be 16, 24, or 32)",
+        dataKeyLength);
 
-  public static boolean useNativeEncryption(EncryptionKeyMetadata keyMetadata) {
-    return keyMetadata != null && keyMetadata instanceof KeyMetadata;
+    return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient);
   }
 }

core/src/main/java/org/apache/iceberg/encryption/KeyMetadataDecoder.java

@@ -28,9 +28,9 @@
 import org.apache.iceberg.data.avro.RawDecoder;
 import org.apache.iceberg.relocated.com.google.common.collect.MapMaker;
 
-class KeyMetadataDecoder extends MessageDecoder.BaseDecoder<KeyMetadata> {
+class KeyMetadataDecoder extends MessageDecoder.BaseDecoder<StandardKeyMetadata> {
   private final org.apache.iceberg.Schema readSchema;
-  private final Map<Byte, RawDecoder<KeyMetadata>> decoders = new MapMaker().makeMap();
+  private final Map<Byte, RawDecoder<StandardKeyMetadata>> decoders = new MapMaker().makeMap();
 
   /**
    * Creates a new decoder that constructs key metadata instances described by schema version.
@@ -39,11 +39,11 @@ class KeyMetadataDecoder extends MessageDecoder.BaseDecoder<KeyMetadata> {
    * instances created by this class will are described by the expected schema.
    */
   KeyMetadataDecoder(byte readSchemaVersion) {
-    this.readSchema = KeyMetadata.supportedSchemaVersions().get(readSchemaVersion);
+    this.readSchema = StandardKeyMetadata.supportedSchemaVersions().get(readSchemaVersion);
   }
 
   @Override
-  public KeyMetadata decode(InputStream stream, KeyMetadata reuse) {
+  public StandardKeyMetadata decode(InputStream stream, StandardKeyMetadata reuse) {
     byte writeSchemaVersion;
 
     try {
@@ -56,14 +56,14 @@ public KeyMetadata decode(InputStream stream, KeyMetadata reuse) {
       throw new RuntimeException("Version byte - end of stream reached");
     }
 
-    Schema writeSchema = KeyMetadata.supportedAvroSchemaVersions().get(writeSchemaVersion);
+    Schema writeSchema = StandardKeyMetadata.supportedAvroSchemaVersions().get(writeSchemaVersion);
 
     if (writeSchema == null) {
       throw new UnsupportedOperationException(
           "Cannot resolve schema for version: " + writeSchemaVersion);
     }
 
-    RawDecoder<KeyMetadata> decoder = decoders.get(writeSchemaVersion);
+    RawDecoder<StandardKeyMetadata> decoder = decoders.get(writeSchemaVersion);
 
     if (decoder == null) {
       decoder = new RawDecoder<>(readSchema, GenericAvroReader::create, writeSchema);

core/src/main/java/org/apache/iceberg/encryption/KeyMetadataEncoder.java

@@ -29,18 +29,18 @@
 import org.apache.avro.message.MessageEncoder;
 import org.apache.iceberg.avro.GenericAvroWriter;
 
-class KeyMetadataEncoder implements MessageEncoder<KeyMetadata> {
+class KeyMetadataEncoder implements MessageEncoder<StandardKeyMetadata> {
   private static final ThreadLocal<BufferOutputStream> TEMP =
       ThreadLocal.withInitial(BufferOutputStream::new);
   private static final ThreadLocal<BinaryEncoder> ENCODER = new ThreadLocal<>();
 
   private final byte schemaVersion;
   private final boolean copyOutputBytes;
-  private final DatumWriter<KeyMetadata> writer;
+  private final DatumWriter<StandardKeyMetadata> writer;
 
   /**
-   * Creates a new {@link MessageEncoder} that will deconstruct {@link KeyMetadata} instances
-   * described by the schema version.
+   * Creates a new {@link MessageEncoder} that will deconstruct {@link StandardKeyMetadata}
+   * instances described by the schema version.
    *
    * <p>Buffers returned by {@code encode} are copied and will not be modified by future calls to
    * {@code encode}.
@@ -50,8 +50,8 @@ class KeyMetadataEncoder implements MessageEncoder<KeyMetadata> {
   }
 
   /**
-   * Creates a new {@link MessageEncoder} that will deconstruct {@link KeyMetadata} instances
-   * described by the schema version.
+   * Creates a new {@link MessageEncoder} that will deconstruct {@link StandardKeyMetadata}
+   * instances described by the schema version.
    *
    * <p>If {@code shouldCopy} is true, then buffers returned by {@code encode} are copied and will
    * not be modified by future calls to {@code encode}.
@@ -62,7 +62,7 @@ class KeyMetadataEncoder implements MessageEncoder<KeyMetadata> {
    * next call to {@code encode}.
    */
   KeyMetadataEncoder(byte schemaVersion, boolean shouldCopy) {
-    Schema writeSchema = KeyMetadata.supportedAvroSchemaVersions().get(schemaVersion);
+    Schema writeSchema = StandardKeyMetadata.supportedAvroSchemaVersions().get(schemaVersion);
 
     if (writeSchema == null) {
       throw new UnsupportedOperationException(
@@ -75,7 +75,7 @@ class KeyMetadataEncoder implements MessageEncoder<KeyMetadata> {
   }
 
   @Override
-  public ByteBuffer encode(KeyMetadata datum) throws IOException {
+  public ByteBuffer encode(StandardKeyMetadata datum) throws IOException {
     BufferOutputStream temp = TEMP.get();
     temp.reset();
     temp.write(schemaVersion);
@@ -89,7 +89,7 @@ public ByteBuffer encode(KeyMetadata datum) throws IOException {
   }
 
   @Override
-  public void encode(KeyMetadata datum, OutputStream stream) throws IOException {
+  public void encode(StandardKeyMetadata datum, OutputStream stream) throws IOException {
     BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(stream, ENCODER.get());
     ENCODER.set(encoder);
     writer.write(datum, encoder);