Add native subclasses for InputFile and OutputFile to simplify.

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

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

core/src/main/java/org/apache/iceberg/avro/Avro.java

@@ -95,7 +95,7 @@ public static WriteBuilder write(OutputFile file) {
   public static WriteBuilder write(EncryptedOutputFile file) {
     Preconditions.checkState(
         file.keyMetadata() == null || file.keyMetadata() == EncryptionKeyMetadata.EMPTY,
-        "Currenty, encryption of data files in Avro format is not supported");
+        "Avro encryption is not supported");
     return new WriteBuilder(file.encryptingOutputFile());
   }
 
@@ -283,7 +283,7 @@ public static DataWriteBuilder writeData(OutputFile file) {
   public static DataWriteBuilder writeData(EncryptedOutputFile file) {
     Preconditions.checkState(
         file.keyMetadata() == null || file.keyMetadata() == EncryptionKeyMetadata.EMPTY,
-        "Currenty, encryption of data files in Avro format is not supported");
+        "Avro encryption is not supported");
     return new DataWriteBuilder(file.encryptingOutputFile());
   }
 
@@ -386,7 +386,7 @@ public static DeleteWriteBuilder writeDeletes(OutputFile file) {
   public static DeleteWriteBuilder writeDeletes(EncryptedOutputFile file) {
     Preconditions.checkState(
         file.keyMetadata() == null || file.keyMetadata() == EncryptionKeyMetadata.EMPTY,
-        "Currenty, encryption of delete files in Avro format is not supported");
+        "Avro encryption is not supported");
     return new DeleteWriteBuilder(file.encryptingOutputFile());
   }
 

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

@@ -20,7 +20,6 @@
 
 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.io.OutputFile;
@@ -81,14 +80,6 @@ public static EncryptionManager createEncryptionManager(
       return PlaintextEncryptionManager.instance();
     }
 
-    String fileFormat =
-        PropertyUtil.propertyAsString(
-            tableProperties,
-            TableProperties.DEFAULT_FILE_FORMAT,
-            TableProperties.DEFAULT_FILE_FORMAT_DEFAULT);
-
-    boolean nativeDataEncryption = (FileFormat.fromString(fileFormat) == FileFormat.PARQUET);
-
     int dataKeyLength =
         PropertyUtil.propertyAsInt(
             tableProperties,
@@ -100,8 +91,7 @@ public static EncryptionManager createEncryptionManager(
         "Invalid data key length: %s (must be 16, 24, or 32)",
         dataKeyLength);
 
-    return new StandardEncryptionManager(
-        tableKeyId, dataKeyLength, kmsClient, nativeDataEncryption);
+    return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient);
   }
 
   public static EncryptedOutputFile plainAsEncryptedOutput(OutputFile encryptingOutputFile) {

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

@@ -0,0 +1,27 @@
+/*
+ * 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.encryption;
+
+import org.apache.iceberg.io.InputFile;
+
+/** An {@link EncryptedInputFile} that can be used for format-native encryption. */
+public interface NativeEncryptionInputFile extends EncryptedInputFile, InputFile {
+  @Override
+  NativeEncryptionKeyMetadata keyMetadata();
+}

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

@@ -0,0 +1,30 @@
+/*
+ * 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.encryption;
+
+import java.nio.ByteBuffer;
+
+/** {@link EncryptionKeyMetadata} for use with format-native encryption. */
+public interface NativeEncryptionKeyMetadata extends EncryptionKeyMetadata {
+  /** Encryption key as a {@link ByteBuffer} */
+  ByteBuffer encryptionKey();
+
+  /** Additional authentication data as a {@link ByteBuffer} */
+  ByteBuffer aadPrefix();
+}

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

@@ -0,0 +1,30 @@
+/*
+ * 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.encryption;
+
+import org.apache.iceberg.io.OutputFile;
+
+/** An {@link EncryptedOutputFile} that can be used for format-native encryption. */
+public interface NativeEncryptionOutputFile extends EncryptedOutputFile {
+  @Override
+  NativeEncryptionKeyMetadata keyMetadata();
+
+  /** An {@link OutputFile} instance for the underlying (plaintext) output stream. */
+  OutputFile plainOutputFile();
+}

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

@@ -32,7 +32,6 @@ public class StandardEncryptionManager implements EncryptionManager {
   private final transient KeyManagementClient kmsClient;
   private final String tableKeyId;
   private final int dataKeyLength;
-  private final boolean nativeDataEncryption;
 
   private transient volatile SecureRandom lazyRNG = null;
 
@@ -42,10 +41,7 @@ public class StandardEncryptionManager implements EncryptionManager {
    * @param kmsClient Client of KMS used to wrap/unwrap keys in envelope encryption
    */
   public StandardEncryptionManager(
-      String tableKeyId,
-      int dataKeyLength,
-      KeyManagementClient kmsClient,
-      boolean nativeDataEncryption) {
+      String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient) {
     Preconditions.checkNotNull(tableKeyId, "Invalid encryption key ID: null");
     Preconditions.checkArgument(
         dataKeyLength == 16 || dataKeyLength == 24 || dataKeyLength == 32,
@@ -55,32 +51,22 @@ public StandardEncryptionManager(
     this.tableKeyId = tableKeyId;
     this.kmsClient = kmsClient;
     this.dataKeyLength = dataKeyLength;
-    this.nativeDataEncryption = nativeDataEncryption;
   }
 
   @Override
-  public EncryptedOutputFile encrypt(OutputFile plainOutput) {
+  public NativeEncryptionOutputFile encrypt(OutputFile plainOutput) {
     return new StandardEncryptedOutputFile(plainOutput, dataKeyLength);
   }
 
   @Override
-  public InputFile decrypt(EncryptedInputFile encrypted) {
+  public NativeEncryptionInputFile decrypt(EncryptedInputFile encrypted) {
     // this input file will lazily parse key metadata in case the file is not an AES GCM stream.
     return new StandardDecryptedInputFile(encrypted);
   }
 
   @Override
   public Iterable<InputFile> decrypt(Iterable<EncryptedInputFile> encrypted) {
-    // Bulk decrypt is only applied to data files. Returning source input files for parquet.
-    if (nativeDataEncryption) {
-      return Iterables.transform(encrypted, this::getSourceFile);
-    } else {
-      return Iterables.transform(encrypted, this::decrypt);
-    }
-  }
-
-  private InputFile getSourceFile(EncryptedInputFile encryptedFile) {
-    return encryptedFile.encryptedInputFile();
+    return Iterables.transform(encrypted, this::decrypt);
   }
 
   private SecureRandom workerRNG() {
@@ -109,7 +95,7 @@ public ByteBuffer unwrapKey(ByteBuffer wrappedSecretKey) {
     return kmsClient.unwrapKey(wrappedSecretKey, tableKeyId);
   }
 
-  private class StandardEncryptedOutputFile implements EncryptedOutputFile {
+  private class StandardEncryptedOutputFile implements NativeEncryptionOutputFile {
     private final OutputFile plainOutputFile;
     private final int dataKeyLength;
     private StandardKeyMetadata lazyKeyMetadata = null;
@@ -154,7 +140,7 @@ public OutputFile plainOutputFile() {
     }
   }
 
-  private static class StandardDecryptedInputFile implements InputFile {
+  private static class StandardDecryptedInputFile implements NativeEncryptionInputFile {
     private final EncryptedInputFile encryptedInputFile;
     private StandardKeyMetadata lazyKeyMetadata = null;
     private AesGcmInputFile lazyDecryptedInputFile = null;
@@ -163,7 +149,13 @@ private StandardDecryptedInputFile(EncryptedInputFile encryptedInputFile) {
       this.encryptedInputFile = encryptedInputFile;
     }
 
-    private StandardKeyMetadata keyMetadata() {
+    @Override
+    public InputFile encryptedInputFile() {
+      return encryptedInputFile.encryptedInputFile();
+    }
+
+    @Override
+    public StandardKeyMetadata keyMetadata() {
       if (null == lazyKeyMetadata) {
         this.lazyKeyMetadata = StandardKeyMetadata.castOrParse(encryptedInputFile.keyMetadata());
       }

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

@@ -31,7 +31,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.types.Types;
 
-public class StandardKeyMetadata implements EncryptionKeyMetadata, IndexedRecord {
+class StandardKeyMetadata implements NativeEncryptionKeyMetadata, IndexedRecord {
   private static final byte V1 = 1;
   private static final Schema SCHEMA_V1 =
       new Schema(
@@ -73,10 +73,12 @@ static Map<Byte, org.apache.avro.Schema> supportedAvroSchemaVersions() {
     return avroSchemaVersions;
   }
 
+  @Override
   public ByteBuffer encryptionKey() {
     return encryptionKey;
   }
 
+  @Override
   public ByteBuffer aadPrefix() {
     return aadPrefix;
   }
@@ -95,7 +97,7 @@ static StandardKeyMetadata castOrParse(EncryptionKeyMetadata keyMetadata) {
     return parse(kmBuffer);
   }
 
-  public static StandardKeyMetadata parse(ByteBuffer buffer) {
+  static StandardKeyMetadata parse(ByteBuffer buffer) {
     try {
       return KEY_METADATA_DECODER.decode(buffer);
     } catch (IOException e) {

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

@@ -36,7 +36,6 @@
 import org.apache.iceberg.deletes.DeleteCounter;
 import org.apache.iceberg.deletes.Deletes;
 import org.apache.iceberg.deletes.PositionDeleteIndex;
-import org.apache.iceberg.encryption.StandardKeyMetadata;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.InputFile;
@@ -282,12 +281,6 @@ private CloseableIterable<Record> openDeletes(DeleteFile deleteFile, Schema dele
           builder.filter(Expressions.equal(MetadataColumns.DELETE_FILE_PATH.name(), filePath));
         }
 
-        if (deleteFile.keyMetadata() != null) {
-          StandardKeyMetadata keyMetadata = StandardKeyMetadata.parse(deleteFile.keyMetadata());
-          builder.withFileEncryptionKey(keyMetadata.encryptionKey());
-          builder.withAADPrefix(keyMetadata.aadPrefix());
-        }
-
         return builder.build();
 
       case ORC:

data/src/main/java/org/apache/iceberg/data/GenericReader.java

@@ -29,7 +29,6 @@
 import org.apache.iceberg.data.avro.DataReader;
 import org.apache.iceberg.data.orc.GenericOrcReader;
 import org.apache.iceberg.data.parquet.GenericParquetReaders;
-import org.apache.iceberg.encryption.StandardKeyMetadata;
 import org.apache.iceberg.expressions.Evaluator;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
@@ -127,12 +126,6 @@ private CloseableIterable<Record> openFile(FileScanTask task, Schema fileProject
           parquet.reuseContainers();
         }
 
-        if (task.file().keyMetadata() != null) {
-          StandardKeyMetadata keyMetadata = StandardKeyMetadata.parse(task.file().keyMetadata());
-          parquet.withFileEncryptionKey(keyMetadata.encryptionKey());
-          parquet.withAADPrefix(keyMetadata.aadPrefix());
-        }
-
         return parquet.build();
 
       case ORC: