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

crates/iceberg/public-api.txt

@@ -353,6 +353,7 @@ pub fn iceberg::encryption::StandardKeyMetadata::decode(bytes: &[u8]) -> iceberg
 pub fn iceberg::encryption::StandardKeyMetadata::encode(&self) -> iceberg::Result<alloc::boxed::Box<[u8]>>
 pub fn iceberg::encryption::StandardKeyMetadata::encryption_key(&self) -> &iceberg::encryption::SecureKey
 pub fn iceberg::encryption::StandardKeyMetadata::file_length(&self) -> core::option::Option<u64>
+pub fn iceberg::encryption::StandardKeyMetadata::generate(key_size: iceberg::encryption::AesKeySize) -> Self
 pub fn iceberg::encryption::StandardKeyMetadata::try_new(encryption_key: &[u8]) -> iceberg::Result<Self>
 pub fn iceberg::encryption::StandardKeyMetadata::with_aad_prefix(self, aad_prefix: &[u8]) -> Self
 pub fn iceberg::encryption::StandardKeyMetadata::with_file_length(self, length: u64) -> Self
@@ -3277,7 +3278,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.
@@ -59,6 +62,12 @@ impl StandardKeyMetadata {
         Ok(Self::from(SecureKey::new(encryption_key)?))
     }
 
+    /// Generates a `StandardKeyMetadata` carrying a fresh random DEK of
+    /// `key_size` together with a fresh random AAD prefix.
+    pub fn generate(key_size: AesKeySize) -> Self {
+        Self::from(SecureKey::generate(key_size)).with_aad_prefix(&generate_aad_prefix())
+    }
+
     /// Adds an AAD prefix.
     pub fn with_aad_prefix(mut self, aad_prefix: &[u8]) -> Self {
         self.aad_prefix = Some(aad_prefix.into());
@@ -108,6 +117,15 @@ impl From<SecureKey> for StandardKeyMetadata {
     }
 }
 
+/// AAD prefix length in bytes.
+const AAD_PREFIX_LENGTH: usize = 16;
+
+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;
@@ -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, StandardKeyMetadata::generate(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,

crates/iceberg/src/spec/mod.rs

@@ -50,8 +50,8 @@ pub use sort::*;
 pub use statistic_file::*;
 pub use table_metadata::*;
 pub(crate) use table_metadata_builder::FIRST_FIELD_ID;
-pub(crate) use table_properties::parse_metadata_file_compression;
 pub use table_properties::*;
+pub(crate) use table_properties::{data_encryption_key_size, parse_metadata_file_compression};
 pub use transform::*;
 pub(crate) use values::decimal_utils;
 pub use values::*;

crates/iceberg/src/spec/table_properties.rs

@@ -20,6 +20,7 @@ use std::fmt::Display;
 use std::str::FromStr;
 
 use crate::compression::CompressionCodec;
+use crate::encryption::AesKeySize;
 use crate::error::{Error, ErrorKind, Result};
 
 fn parse_property<T: FromStr>(
@@ -40,6 +41,19 @@ where
     })
 }
 
+/// The AES key size to use when generating data encryption keys, derived from
+/// `encryption.data-key-length`.
+///
+/// Returns `None` when the table is not configured for encryption.
+/// Returns an error when `encryption.data-key-length` is not a valid AES key length.
+pub(crate) fn data_encryption_key_size(props: &TableProperties) -> Result<Option<AesKeySize>> {
+    props
+        .encryption_key_id
+        .is_some()
+        .then(|| AesKeySize::from_key_length(props.encryption_data_key_length))
+        .transpose()
+}
+
 /// Strips trailing slashes from a location, preserving a bare URI scheme root
 fn strip_trailing_slash(path: &str) -> &str {
     let mut path = path;