feat(encryption) [15/N] write encrypted parquet data files

crates/iceberg/public-api.txt

@@ -3267,7 +3267,7 @@ pub async fn iceberg::writer::file_writer::ParquetWriter::close(self) -> iceberg
 pub async fn iceberg::writer::file_writer::ParquetWriter::write(&mut self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<()>
 pub struct iceberg::writer::file_writer::ParquetWriterBuilder
 impl iceberg::writer::file_writer::ParquetWriterBuilder
-pub fn iceberg::writer::file_writer::ParquetWriterBuilder::from_table_properties(table_props: &iceberg::spec::TableProperties, schema: iceberg::spec::SchemaRef) -> Self
+pub fn iceberg::writer::file_writer::ParquetWriterBuilder::from_table_properties(table_props: &iceberg::spec::TableProperties, schema: iceberg::spec::SchemaRef) -> iceberg::Result<Self>
 pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef) -> Self
 pub fn iceberg::writer::file_writer::ParquetWriterBuilder::new_with_match_mode(props: parquet::file::properties::WriterProperties, schema: iceberg::spec::SchemaRef, match_mode: iceberg::arrow::FieldMatchMode) -> Self
 pub fn iceberg::writer::file_writer::ParquetWriterBuilder::with_match_mode(self, match_mode: iceberg::arrow::FieldMatchMode) -> Self

crates/iceberg/src/arrow/test_utils.rs

@@ -19,7 +19,9 @@ use std::fs::File;
 
 use arrow_array::RecordBatch;
 use parquet::arrow::ArrowWriter;
+use parquet::arrow::arrow_reader::{ArrowReaderOptions, ParquetRecordBatchReaderBuilder};
 use parquet::basic::Compression;
+use parquet::encryption::decrypt::FileDecryptionProperties;
 use parquet::encryption::encrypt::FileEncryptionProperties;
 use parquet::file::properties::WriterProperties;
 
@@ -46,3 +48,26 @@ pub(crate) fn write_encrypted_parquet(
     writer.write(batch).expect("Writing batch");
     writer.close().unwrap();
 }
+
+/// Reads the Parquet file at `path` encrypted with `key` and `aad_prefix`, returning
+/// all record batches.
+pub(crate) fn read_encrypted_parquet(
+    path: &str,
+    key: &[u8],
+    aad_prefix: Option<&[u8]>,
+) -> Vec<RecordBatch> {
+    let mut builder = FileDecryptionProperties::builder(key.to_vec());
+    if let Some(aad) = aad_prefix {
+        builder = builder.with_aad_prefix(aad.to_vec());
+    }
+    let options =
+        ArrowReaderOptions::new().with_file_decryption_properties(builder.build().unwrap());
+
+    let file = File::open(path).unwrap();
+    ParquetRecordBatchReaderBuilder::try_new_with_options(file, options)
+        .unwrap()
+        .build()
+        .unwrap()
+        .map(|b| b.unwrap())
+        .collect()
+}

crates/iceberg/src/encryption/key_metadata.rs

@@ -20,7 +20,10 @@
 
 use std::fmt;
 
-use super::SecureKey;
+use aes_gcm::aead::OsRng;
+use aes_gcm::aead::rand_core::RngCore;
+
+use super::{AesKeySize, SecureKey};
 use crate::{Error, ErrorKind, Result};
 
 /// Standard key metadata for Iceberg table encryption.
@@ -108,6 +111,22 @@ impl From<SecureKey> for StandardKeyMetadata {
     }
 }
 
+/// AAD prefix length in bytes.
+const AAD_PREFIX_LENGTH: usize = 16;
+
+/// Generate a [`StandardKeyMetadata`] with a fresh random DEK and AAD prefix,
+/// sized to `key_size`.
+pub(crate) fn generate_standard_key_metadata(key_size: AesKeySize) -> StandardKeyMetadata {
+    let dek = SecureKey::generate(key_size);
+    StandardKeyMetadata::from(dek).with_aad_prefix(&generate_aad_prefix())
+}
+
+fn generate_aad_prefix() -> Box<[u8]> {
+    let mut prefix = vec![0u8; AAD_PREFIX_LENGTH];
+    OsRng.fill_bytes(&mut prefix);
+    prefix.into_boxed_slice()
+}
+
 mod _serde {
     use std::io::Cursor;
     use std::sync::{Arc, LazyLock};

crates/iceberg/src/encryption/manager.rs

@@ -28,8 +28,6 @@ use std::fmt;
 use std::sync::{Arc, RwLock};
 use std::time::Duration;
 
-use aes_gcm::aead::OsRng;
-use aes_gcm::aead::rand_core::RngCore;
 use chrono::Utc;
 use moka::future::Cache;
 use uuid::Uuid;
@@ -38,7 +36,7 @@ const MILLIS_IN_DAY: i64 = 24 * 60 * 60 * 1000;
 
 use super::crypto::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes};
 use super::io::EncryptedOutputFile;
-use super::key_metadata::StandardKeyMetadata;
+use super::key_metadata::{StandardKeyMetadata, generate_standard_key_metadata};
 use super::kms::KeyManagementClient;
 use crate::io::OutputFile;
 use crate::spec::{EncryptedKey, FormatVersion, TableMetadataRef};
@@ -54,10 +52,6 @@ const DEFAULT_KEK_LIFESPAN_DAYS: i64 = 730;
 /// Default cache TTL for unwrapped KEKs.
 const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(3600);
 
-/// Default AAD prefix length in bytes.
-/// Matches Java's `TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT`.
-const AAD_PREFIX_LENGTH: usize = 16;
-
 /// File-level encryption manager using two-layer envelope encryption.
 ///
 /// Uses an async cache for unwrapped KEK bytes to avoid repeated KMS calls.
@@ -151,10 +145,7 @@ impl EncryptionManager {
     /// Returns an [`EncryptedOutputFile`] that transparently encrypts on
     /// write, along with key metadata for later decryption.
     pub fn encrypt(&self, raw_output: OutputFile) -> EncryptedOutputFile {
-        let dek = SecureKey::generate(self.key_size);
-        let aad_prefix = Self::generate_aad_prefix();
-        let metadata = StandardKeyMetadata::from(dek).with_aad_prefix(&aad_prefix);
-        EncryptedOutputFile::new(raw_output, metadata)
+        EncryptedOutputFile::new(raw_output, generate_standard_key_metadata(self.key_size))
     }
 
     /// Wrap a manifest list key metadata with a KEK for storage in table metadata.
@@ -397,13 +388,6 @@ impl EncryptionManager {
             })
     }
 
-    /// Generate a random AAD prefix for file encryption.
-    fn generate_aad_prefix() -> Box<[u8]> {
-        let mut prefix = vec![0u8; AAD_PREFIX_LENGTH];
-        OsRng.fill_bytes(&mut prefix);
-        prefix.into_boxed_slice()
-    }
-
     /// Wrap a DEK with a KEK using local AES-GCM.
     fn wrap_dek_with_kek(
         &self,