You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by li...@apache.org on 2026/07/16 01:56:52 UTC
(iceberg-rust) branch main updated: feat(encryption) [9/N] Read encrypted parquet data-files (#2584)
This is an automated email from the ASF dual-hosted git repository.
blackmwk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 64cb19a4f feat(encryption) [9/N] Read encrypted parquet data-files (#2584)
64cb19a4f is described below
commit 64cb19a4f798634c717773f81afeaead9ef7f757
Author: Xander <za...@googlemail.com>
AuthorDate: Thu Jul 16 02:56:47 2026 +0100
feat(encryption) [9/N] Read encrypted parquet data-files (#2584)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Working towards https://github.com/apache/iceberg-rust/issues/2034
We're currently not able to decrypt PME encrypted files, all that's
needed for this PR is to wire through the key_metadata from the manifest
entries into the scan tasks and then configure the parquet reader with
the correct encryption key and aad.
It's important to note that no encryption manager is needed here since
the key_metadata on the manifest files contain the plaintext keys.
## What changes are included in this PR?
<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->
## Are these changes tested?
Tests follow the pattern on manually writing an encrypted parquet file
using the raw arrow-rs writer and then using the FileScanTask and
iceberg reader to read that tmp file.
I looked at using the [parquet testing
repo](https://github.com/apache/parquet-testing) but the encrypted files
there all use column based encryption which isn't supported by the spec.
I'll separately track adding some files there that we can use here but
the testing strategy here is consistent with how Java does things so I
think this should be sufficient for us to continue.
<!--
Specify what test covers (unit test, integration test, etc.).
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
---
crates/iceberg/public-api.txt | 6 +-
.../src/arrow/caching_delete_file_loader.rs | 16 +-
crates/iceberg/src/arrow/delete_file_loader.rs | 160 +++++++++++++-
crates/iceberg/src/arrow/mod.rs | 3 +
crates/iceberg/src/arrow/reader/pipeline.rs | 234 ++++++++++++++++++++-
crates/iceberg/src/arrow/reader/row_filter.rs | 3 +
crates/iceberg/src/arrow/test_utils.rs | 48 +++++
crates/iceberg/src/encryption/key_metadata.rs | 40 +++-
crates/iceberg/src/scan/context.rs | 1 +
crates/iceberg/src/scan/task.rs | 31 +++
10 files changed, 530 insertions(+), 12 deletions(-)
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index c2013294e..2de70bf26 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -1244,6 +1244,7 @@ pub iceberg::scan::FileScanTask::data_file_format: iceberg::spec::DataFileFormat
pub iceberg::scan::FileScanTask::data_file_path: alloc::string::String
pub iceberg::scan::FileScanTask::deletes: alloc::vec::Vec<iceberg::scan::FileScanTaskDeleteFile>
pub iceberg::scan::FileScanTask::file_size_in_bytes: u64
+pub iceberg::scan::FileScanTask::key_metadata: core::option::Option<alloc::boxed::Box<[u8]>>
pub iceberg::scan::FileScanTask::length: u64
pub iceberg::scan::FileScanTask::name_mapping: core::option::Option<alloc::sync::Arc<iceberg::spec::NameMapping>>
pub iceberg::scan::FileScanTask::partition: core::option::Option<iceberg::spec::Struct>
@@ -1267,7 +1268,7 @@ impl core::fmt::Debug for iceberg::scan::FileScanTask
pub fn iceberg::scan::FileScanTask::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTask
impl iceberg::scan::FileScanTask
-pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), ())>
+pub fn iceberg::scan::FileScanTask::builder() -> FileScanTaskBuilder<((), (), (), (), (), (), (), (), (), (), (), (), (), (), ())>
impl serde_core::ser::Serialize for iceberg::scan::FileScanTask
pub fn iceberg::scan::FileScanTask::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTask
@@ -1277,6 +1278,7 @@ pub iceberg::scan::FileScanTaskDeleteFile::equality_ids: core::option::Option<al
pub iceberg::scan::FileScanTaskDeleteFile::file_path: alloc::string::String
pub iceberg::scan::FileScanTaskDeleteFile::file_size_in_bytes: u64
pub iceberg::scan::FileScanTaskDeleteFile::file_type: iceberg::spec::DataContentType
+pub iceberg::scan::FileScanTaskDeleteFile::key_metadata: core::option::Option<alloc::boxed::Box<[u8]>>
pub iceberg::scan::FileScanTaskDeleteFile::partition_spec_id: i32
impl core::clone::Clone for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::clone(&self) -> iceberg::scan::FileScanTaskDeleteFile
@@ -1286,7 +1288,7 @@ impl core::fmt::Debug for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
impl core::marker::StructuralPartialEq for iceberg::scan::FileScanTaskDeleteFile
impl iceberg::scan::FileScanTaskDeleteFile
-pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), ())>
+pub fn iceberg::scan::FileScanTaskDeleteFile::builder() -> FileScanTaskDeleteFileBuilder<((), (), (), (), (), ())>
impl serde_core::ser::Serialize for iceberg::scan::FileScanTaskDeleteFile
pub fn iceberg::scan::FileScanTaskDeleteFile::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
impl<'de> serde_core::de::Deserialize<'de> for iceberg::scan::FileScanTaskDeleteFile
diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
index de31d678a..25704e5bc 100644
--- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
@@ -251,7 +251,11 @@ impl CachingDeleteFileLoader {
PosDelLoadAction::Load => Ok(DeleteFileContext::PosDels {
file_path: task.file_path.clone(),
stream: basic_delete_file_loader
- .parquet_to_batch_stream(&task.file_path, task.file_size_in_bytes)
+ .parquet_to_batch_stream(
+ &task.file_path,
+ task.file_size_in_bytes,
+ task.key_metadata.as_deref(),
+ )
.await?,
}),
}
@@ -270,7 +274,11 @@ impl CachingDeleteFileLoader {
let equality_ids_vec = task.equality_ids.clone().unwrap();
let evolved_stream = BasicDeleteFileLoader::evolve_schema(
basic_delete_file_loader
- .parquet_to_batch_stream(&task.file_path, task.file_size_in_bytes)
+ .parquet_to_batch_stream(
+ &task.file_path,
+ task.file_size_in_bytes,
+ task.key_metadata.as_deref(),
+ )
.await?,
schema,
&equality_ids_vec,
@@ -647,6 +655,7 @@ mod tests {
.parquet_to_batch_stream(
&eq_delete_file_path,
std::fs::metadata(&eq_delete_file_path).unwrap().len(),
+ None,
)
.await
.expect("could not get batch stream");
@@ -977,6 +986,7 @@ mod tests {
.parquet_to_batch_stream(
&delete_file_path,
std::fs::metadata(&delete_file_path).unwrap().len(),
+ None,
)
.await
.unwrap();
@@ -1158,7 +1168,7 @@ mod tests {
let basic_delete_file_loader =
BasicDeleteFileLoader::new(file_io.clone(), ScanMetrics::new());
let record_batch_stream = basic_delete_file_loader
- .parquet_to_batch_stream(&path, std::fs::metadata(&path).unwrap().len())
+ .parquet_to_batch_stream(&path, std::fs::metadata(&path).unwrap().len(), None)
.await
.expect("could not get batch stream");
diff --git a/crates/iceberg/src/arrow/delete_file_loader.rs b/crates/iceberg/src/arrow/delete_file_loader.rs
index 134b02961..9bcc96519 100644
--- a/crates/iceberg/src/arrow/delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/delete_file_loader.rs
@@ -67,6 +67,7 @@ impl BasicDeleteFileLoader {
&self,
data_file_path: &str,
file_size_in_bytes: u64,
+ key_metadata: Option<&[u8]>,
) -> Result<ArrowRecordBatchStream> {
/*
Essentially a super-cut-down ArrowReader. We can't use ArrowReader directly
@@ -80,6 +81,7 @@ impl BasicDeleteFileLoader {
file_size_in_bytes,
parquet_read_options,
self.scan_metrics.bytes_read_counter(),
+ key_metadata,
)
.await?;
@@ -121,7 +123,11 @@ impl DeleteFileLoader for BasicDeleteFileLoader {
schema: SchemaRef,
) -> Result<ArrowRecordBatchStream> {
let raw_batch_stream = self
- .parquet_to_batch_stream(&task.file_path, task.file_size_in_bytes)
+ .parquet_to_batch_stream(
+ &task.file_path,
+ task.file_size_in_bytes,
+ task.key_metadata.as_deref(),
+ )
.await?;
// For equality deletes, only evolve the equality_ids columns.
@@ -141,6 +147,7 @@ mod tests {
use super::*;
use crate::arrow::delete_filter::tests::setup;
+ use crate::arrow::test_utils::write_encrypted_parquet;
#[tokio::test]
async fn test_basic_delete_file_loader_read_delete_file() {
@@ -165,4 +172,155 @@ mod tests {
assert_eq!(result.len(), 1);
}
+
+ #[tokio::test]
+ async fn test_read_encrypted_positional_delete_file() {
+ use std::sync::Arc;
+
+ use arrow_array::{Int64Array, RecordBatch, StringArray};
+
+ use crate::arrow::delete_filter::tests::create_pos_del_schema;
+ use crate::encryption::StandardKeyMetadata;
+ use crate::scan::FileScanTaskDeleteFile;
+ use crate::spec::DataContentType;
+
+ let encryption_key = b"0123456789abcdef";
+ let aad_prefix = b"aad_prefix";
+
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().to_str().unwrap();
+ let file_io = FileIO::new_with_fs();
+
+ let positional_delete_schema = create_pos_del_schema();
+ let file_path_col = Arc::new(StringArray::from_iter_values(vec!["data.parquet"; 4]));
+ let pos_col = Arc::new(Int64Array::from(vec![0i64, 1, 5, 10]));
+ let batch = RecordBatch::try_new(positional_delete_schema.clone(), vec![
+ file_path_col,
+ pos_col,
+ ])
+ .unwrap();
+
+ let del_path = format!("{table_location}/encrypted-pos-del.parquet");
+ write_encrypted_parquet(&del_path, &batch, encryption_key, Some(aad_prefix));
+
+ let key_metadata = StandardKeyMetadata::new(encryption_key)
+ .with_aad_prefix(aad_prefix)
+ .encode()
+ .unwrap();
+
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ crate::spec::NestedField::required(
+ 2147483546,
+ "file_path",
+ crate::spec::Type::Primitive(crate::spec::PrimitiveType::String),
+ )
+ .into(),
+ crate::spec::NestedField::required(
+ 2147483545,
+ "pos",
+ crate::spec::Type::Primitive(crate::spec::PrimitiveType::Long),
+ )
+ .into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let task = FileScanTaskDeleteFile {
+ file_path: del_path.clone(),
+ file_size_in_bytes: std::fs::metadata(&del_path).unwrap().len(),
+ file_type: DataContentType::PositionDeletes,
+ partition_spec_id: 0,
+ equality_ids: None,
+ key_metadata: Some(Box::from(key_metadata.as_ref())),
+ };
+
+ let scan_metrics = ScanMetrics::new();
+ let delete_file_loader = BasicDeleteFileLoader::new(file_io, scan_metrics);
+
+ let result = delete_file_loader
+ .read_delete_file(&task, schema)
+ .await
+ .unwrap();
+
+ let batches: Vec<_> = result.try_collect().await.unwrap();
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].num_rows(), 4);
+ }
+
+ #[tokio::test]
+ async fn test_read_encrypted_equality_delete_file() {
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ use arrow_array::{Int64Array, RecordBatch};
+ use parquet::arrow::PARQUET_FIELD_ID_META_KEY;
+
+ use crate::encryption::StandardKeyMetadata;
+ use crate::scan::FileScanTaskDeleteFile;
+ use crate::spec::DataContentType;
+
+ let encryption_key = b"0123456789abcdef";
+ let aad_prefix = b"my-table-uuid!!";
+
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().to_str().unwrap();
+ let file_io = FileIO::new_with_fs();
+
+ let arrow_schema = Arc::new(arrow_schema::Schema::new(vec![
+ arrow_schema::Field::new("id", arrow_schema::DataType::Int64, false).with_metadata(
+ HashMap::from([(PARQUET_FIELD_ID_META_KEY.to_string(), "1".to_string())]),
+ ),
+ ]));
+
+ let id_col = Arc::new(Int64Array::from(vec![100i64, 200, 300]));
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id_col]).unwrap();
+
+ let del_path = format!("{table_location}/encrypted-eq-del.parquet");
+ write_encrypted_parquet(&del_path, &batch, encryption_key, Some(aad_prefix));
+
+ let key_metadata = StandardKeyMetadata::new(encryption_key)
+ .with_aad_prefix(aad_prefix)
+ .encode()
+ .unwrap();
+
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ crate::spec::NestedField::required(
+ 1,
+ "id",
+ crate::spec::Type::Primitive(crate::spec::PrimitiveType::Long),
+ )
+ .into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let task = FileScanTaskDeleteFile {
+ file_path: del_path.clone(),
+ file_size_in_bytes: std::fs::metadata(&del_path).unwrap().len(),
+ file_type: DataContentType::EqualityDeletes,
+ partition_spec_id: 0,
+ equality_ids: Some(vec![1]),
+ key_metadata: Some(Box::from(key_metadata.as_ref())),
+ };
+
+ let scan_metrics = ScanMetrics::new();
+ let delete_file_loader = BasicDeleteFileLoader::new(file_io, scan_metrics);
+
+ let result = delete_file_loader
+ .read_delete_file(&task, schema)
+ .await
+ .unwrap();
+
+ let batches: Vec<_> = result.try_collect().await.unwrap();
+ assert_eq!(batches.len(), 1);
+ assert_eq!(batches[0].num_rows(), 3);
+ }
}
diff --git a/crates/iceberg/src/arrow/mod.rs b/crates/iceberg/src/arrow/mod.rs
index bf53633cf..4e0286cbd 100644
--- a/crates/iceberg/src/arrow/mod.rs
+++ b/crates/iceberg/src/arrow/mod.rs
@@ -44,3 +44,6 @@ pub use partition_value_calculator::*;
/// Record batch partition splitter for partitioned tables
pub mod record_batch_partition_splitter;
pub use record_batch_partition_splitter::*;
+
+#[cfg(test)]
+pub(crate) mod test_utils;
diff --git a/crates/iceberg/src/arrow/reader/pipeline.rs b/crates/iceberg/src/arrow/reader/pipeline.rs
index a4072854e..f05c3a8b4 100644
--- a/crates/iceberg/src/arrow/reader/pipeline.rs
+++ b/crates/iceberg/src/arrow/reader/pipeline.rs
@@ -26,6 +26,7 @@ use std::sync::atomic::AtomicU64;
use futures::{StreamExt, TryStreamExt};
use parquet::arrow::arrow_reader::{ArrowReaderMetadata, ArrowReaderOptions};
use parquet::arrow::{PARQUET_FIELD_ID_META_KEY, ParquetRecordBatchStreamBuilder};
+use parquet::encryption::decrypt::FileDecryptionProperties;
use super::{
ArrowFileReader, ArrowReader, ParquetReadOptions, add_fallback_field_ids_to_arrow_schema,
@@ -35,6 +36,7 @@ use crate::arrow::caching_delete_file_loader::CachingDeleteFileLoader;
use crate::arrow::int96::coerce_int96_timestamps;
use crate::arrow::record_batch_transformer::RecordBatchTransformerBuilder;
use crate::arrow::scan_metrics::{CountingFileRead, ScanMetrics, ScanResult};
+use crate::encryption::StandardKeyMetadata;
use crate::error::Result;
use crate::io::{FileIO, FileMetadata, FileRead};
use crate::metadata_columns::{
@@ -122,6 +124,7 @@ impl FileScanTaskReader {
task.file_size_in_bytes,
parquet_read_options,
self.scan_metrics.bytes_read_counter(),
+ task.key_metadata.as_deref(),
)
.await?;
@@ -431,6 +434,7 @@ impl ArrowReader {
file_size_in_bytes: u64,
parquet_read_options: ParquetReadOptions,
bytes_read: &Arc<AtomicU64>,
+ key_metadata: Option<&[u8]>,
) -> Result<(ArrowFileReader, ArrowReaderMetadata)> {
let parquet_file = file_io.new_input(data_file_path)?;
let counting_reader =
@@ -439,6 +443,7 @@ impl ArrowReader {
Box::new(counting_reader),
file_size_in_bytes,
parquet_read_options,
+ key_metadata,
)
.await
}
@@ -447,6 +452,7 @@ impl ArrowReader {
parquet_reader: Box<dyn FileRead>,
file_size_in_bytes: u64,
parquet_read_options: ParquetReadOptions,
+ key_metadata: Option<&[u8]>,
) -> Result<(ArrowFileReader, ArrowReaderMetadata)> {
let mut reader = ArrowFileReader::new(
FileMetadata {
@@ -456,7 +462,9 @@ impl ArrowReader {
)
.with_parquet_read_options(parquet_read_options);
- let arrow_metadata = ArrowReaderMetadata::load_async(&mut reader, Default::default())
+ let arrow_reader_options = Self::build_arrow_reader_options(key_metadata)?;
+
+ let arrow_metadata = ArrowReaderMetadata::load_async(&mut reader, arrow_reader_options)
.await
.map_err(|e| {
Error::new(ErrorKind::Unexpected, "Failed to load Parquet metadata").with_source(e)
@@ -464,6 +472,34 @@ impl ArrowReader {
Ok((reader, arrow_metadata))
}
+
+ /// Builds `ArrowReaderOptions`, adding `FileDecryptionProperties` when
+ /// key metadata is present for Parquet Modular Encryption.
+ fn build_arrow_reader_options(key_metadata: Option<&[u8]>) -> Result<ArrowReaderOptions> {
+ match key_metadata {
+ Some(km) => {
+ let standard_key_metadata = StandardKeyMetadata::decode(km)?;
+ let mut builder = FileDecryptionProperties::builder(
+ standard_key_metadata.encryption_key().as_bytes().to_vec(),
+ );
+ if let Some(aad) = standard_key_metadata.aad_prefix() {
+ builder = builder.with_aad_prefix(aad.to_vec());
+ }
+ let decryption_properties = builder.build().map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "Failed to build Parquet file decryption properties",
+ )
+ .with_source(e)
+ })?;
+ Ok(
+ ArrowReaderOptions::new()
+ .with_file_decryption_properties(decryption_properties),
+ )
+ }
+ None => Ok(ArrowReaderOptions::default()),
+ }
+ }
}
#[cfg(test)]
@@ -473,7 +509,7 @@ mod tests {
use std::sync::Arc;
use arrow_array::cast::AsArray;
- use arrow_array::{Array, ArrayRef, RecordBatch};
+ use arrow_array::{Array, ArrayRef, Int32Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use futures::TryStreamExt;
use parquet::arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY};
@@ -483,6 +519,7 @@ mod tests {
use crate::Runtime;
use crate::arrow::ArrowReaderBuilder;
+ use crate::arrow::test_utils::write_encrypted_parquet;
use crate::io::FileIO;
use crate::scan::{FileScanTask, FileScanTaskStream};
use crate::spec::{DataFileFormat, NestedField, PrimitiveType, Schema, SchemaRef, Type};
@@ -663,6 +700,199 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_read_encrypted_parquet() {
+ let encryption_key = b"0123456789abcdef";
+ let aad_prefix = b"aad_prefix";
+
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "1".to_string(),
+ )])),
+ ]));
+
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().to_str().unwrap().to_string();
+ let file_io = FileIO::new_with_fs();
+
+ let id_data = Arc::new(Int32Array::from(vec![10, 20, 30])) as ArrayRef;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id_data]).unwrap();
+
+ let file_path = format!("{table_location}/encrypted.parquet");
+ write_encrypted_parquet(&file_path, &batch, encryption_key, Some(aad_prefix));
+
+ let key_metadata = crate::encryption::StandardKeyMetadata::new(encryption_key)
+ .with_aad_prefix(aad_prefix)
+ .encode()
+ .unwrap();
+
+ let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build();
+
+ let task = FileScanTask::builder()
+ .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(file_path)
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema)
+ .with_project_field_ids(vec![1])
+ .with_case_sensitive(false)
+ .with_key_metadata(Some(key_metadata))
+ .build();
+
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+ let batches: Vec<RecordBatch> = reader
+ .read(tasks)
+ .unwrap()
+ .stream()
+ .try_collect()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 1);
+ let ids = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(ids.values(), &[10, 20, 30]);
+ }
+
+ #[tokio::test]
+ async fn test_read_encrypted_parquet_without_key_metadata_fails() {
+ let encryption_key = b"0123456789abcdef";
+
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "1".to_string(),
+ )])),
+ ]));
+
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().to_str().unwrap().to_string();
+ let file_io = FileIO::new_with_fs();
+
+ let id_data = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id_data]).unwrap();
+
+ let file_path = format!("{table_location}/encrypted_no_key.parquet");
+ write_encrypted_parquet(&file_path, &batch, encryption_key, None);
+
+ let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build();
+
+ let task = FileScanTask::builder()
+ .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(file_path)
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema)
+ .with_project_field_ids(vec![1])
+ .with_case_sensitive(false)
+ .build();
+
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+ let result: Result<Vec<RecordBatch>, _> =
+ reader.read(tasks).unwrap().stream().try_collect().await;
+
+ let err = result.unwrap_err();
+ assert_eq!(err.kind(), crate::ErrorKind::Unexpected);
+ let err_str = format!("{err}");
+ assert!(
+ err_str.contains("encrypted footer"),
+ "Expected error about encrypted footer, got: {err_str}"
+ );
+ assert!(
+ err_str.contains("decryption properties were not provided"),
+ "Expected error about missing decryption properties, got: {err_str}"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_read_encrypted_parquet_with_wrong_key_fails() {
+ let encryption_key = b"0123456789abcdef";
+ let wrong_key = b"fedcba9876543210";
+
+ let schema = Arc::new(
+ Schema::builder()
+ .with_schema_id(1)
+ .with_fields(vec![
+ NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(),
+ ])
+ .build()
+ .unwrap(),
+ );
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([(
+ PARQUET_FIELD_ID_META_KEY.to_string(),
+ "1".to_string(),
+ )])),
+ ]));
+
+ let tmp_dir = TempDir::new().unwrap();
+ let table_location = tmp_dir.path().to_str().unwrap().to_string();
+ let file_io = FileIO::new_with_fs();
+
+ let id_data = Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef;
+ let batch = RecordBatch::try_new(arrow_schema.clone(), vec![id_data]).unwrap();
+
+ let file_path = format!("{table_location}/encrypted_wrong_key.parquet");
+ write_encrypted_parquet(&file_path, &batch, encryption_key, None);
+
+ let wrong_key_metadata = crate::encryption::StandardKeyMetadata::new(wrong_key)
+ .encode()
+ .unwrap();
+
+ let reader = ArrowReaderBuilder::new(file_io, Runtime::current()).build();
+
+ let task = FileScanTask::builder()
+ .with_file_size_in_bytes(std::fs::metadata(&file_path).unwrap().len())
+ .with_start(0)
+ .with_length(0)
+ .with_data_file_path(file_path)
+ .with_data_file_format(DataFileFormat::Parquet)
+ .with_schema(schema)
+ .with_project_field_ids(vec![1])
+ .with_case_sensitive(false)
+ .with_key_metadata(Some(wrong_key_metadata))
+ .build();
+
+ let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+ let result: Result<Vec<RecordBatch>, _> =
+ reader.read(tasks).unwrap().stream().try_collect().await;
+
+ let err = result.unwrap_err();
+ assert_eq!(err.kind(), crate::ErrorKind::Unexpected);
+ let err_str = format!("{err}");
+ assert!(
+ err_str.contains("unable to decrypt parquet footer"),
+ "Expected error about decryption failure, got: {err_str}"
+ );
+ }
+
/// Test that concurrency=1 reads all files correctly and in deterministic order.
/// This verifies the fast-path optimization for single concurrency.
#[tokio::test]
diff --git a/crates/iceberg/src/arrow/reader/row_filter.rs b/crates/iceberg/src/arrow/reader/row_filter.rs
index a0ad5b627..623ef71f2 100644
--- a/crates/iceberg/src/arrow/reader/row_filter.rs
+++ b/crates/iceberg/src/arrow/reader/row_filter.rs
@@ -1148,6 +1148,7 @@ mod tests {
partition_spec: None,
name_mapping: None,
case_sensitive: false,
+ key_metadata: None,
};
let stream = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
@@ -1240,11 +1241,13 @@ mod tests {
partition_spec_id: 0,
equality_ids: None,
file_size_in_bytes: std::fs::metadata(&pos_del_path).unwrap().len(),
+ key_metadata: None,
}],
partition: None,
partition_spec: None,
name_mapping: None,
case_sensitive: false,
+ key_metadata: None,
};
let stream_sub2 =
diff --git a/crates/iceberg/src/arrow/test_utils.rs b/crates/iceberg/src/arrow/test_utils.rs
new file mode 100644
index 000000000..15ec0dc8d
--- /dev/null
+++ b/crates/iceberg/src/arrow/test_utils.rs
@@ -0,0 +1,48 @@
+// 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.
+
+use std::fs::File;
+
+use arrow_array::RecordBatch;
+use parquet::arrow::ArrowWriter;
+use parquet::basic::Compression;
+use parquet::encryption::encrypt::FileEncryptionProperties;
+use parquet::file::properties::WriterProperties;
+
+/// Writes `batch` to `path` as a Parquet file encrypted with `key` and `aad_prefix`.
+pub(crate) fn write_encrypted_parquet(
+ path: &str,
+ batch: &RecordBatch,
+ key: &[u8],
+ aad_prefix: Option<&[u8]>,
+) {
+ let mut builder = FileEncryptionProperties::builder(key.to_vec());
+ if let Some(aad) = aad_prefix {
+ builder = builder.with_aad_prefix(aad.to_vec());
+ }
+ let encryption_properties = builder.build().unwrap();
+
+ let props = WriterProperties::builder()
+ .set_compression(Compression::SNAPPY)
+ .with_file_encryption_properties(encryption_properties)
+ .build();
+
+ let file = File::create(path).unwrap();
+ let mut writer = ArrowWriter::try_new(file, batch.schema(), Some(props)).unwrap();
+ writer.write(batch).expect("Writing batch");
+ writer.close().unwrap();
+}
diff --git a/crates/iceberg/src/encryption/key_metadata.rs b/crates/iceberg/src/encryption/key_metadata.rs
index 4ef66ce39..9e536e4d8 100644
--- a/crates/iceberg/src/encryption/key_metadata.rs
+++ b/crates/iceberg/src/encryption/key_metadata.rs
@@ -20,7 +20,7 @@
use std::fmt;
-use super::SensitiveBytes;
+use super::{SecureKey, SensitiveBytes};
use crate::{Error, ErrorKind, Result};
/// Standard key metadata for Iceberg table encryption.
@@ -97,7 +97,17 @@ impl StandardKeyMetadata {
/// Decodes from Java-compatible format.
pub fn decode(bytes: &[u8]) -> Result<Self> {
- _serde::StandardKeyMetadataV1::decode(bytes).map(Self::from)
+ let metadata = _serde::StandardKeyMetadataV1::decode(bytes).map(Self::from)?;
+ // Validate the DEK is a usable AES key (16/24/32 bytes) up front, so a
+ // malformed key surfaces a clear error.
+ SecureKey::new(metadata.encryption_key.as_bytes()).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "Invalid encryption key in key metadata",
+ )
+ .with_source(e)
+ })?;
+ Ok(metadata)
}
}
@@ -276,11 +286,33 @@ mod tests {
#[test]
fn test_roundtrip_without_aad() {
- let metadata = StandardKeyMetadata::new(&[1, 2, 3, 4]);
+ let key = b"0123456789012345";
+ let metadata = StandardKeyMetadata::new(key);
let serialized = metadata.encode().unwrap();
let parsed = StandardKeyMetadata::decode(&serialized).unwrap();
- assert_eq!(parsed.encryption_key().as_bytes(), &[1, 2, 3, 4]);
+ assert_eq!(parsed.encryption_key().as_bytes(), key);
assert_eq!(parsed.aad_prefix(), None);
}
+
+ #[test]
+ fn test_decode_rejects_invalid_key_length() {
+ // 24-byte (AES-192) and 32-byte (AES-256) keys are accepted.
+ for len in [16usize, 24, 32] {
+ let metadata = StandardKeyMetadata::new(&vec![0u8; len]);
+ let serialized = metadata.encode().unwrap();
+ assert!(StandardKeyMetadata::decode(&serialized).is_ok());
+ }
+
+ for len in [0usize, 4, 15, 20, 33] {
+ let metadata = StandardKeyMetadata::new(&vec![0u8; len]);
+ let serialized = metadata.encode().unwrap();
+ let err = StandardKeyMetadata::decode(&serialized).unwrap_err();
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(
+ err.to_string()
+ .contains("Invalid encryption key in key metadata")
+ );
+ }
+ }
}
diff --git a/crates/iceberg/src/scan/context.rs b/crates/iceberg/src/scan/context.rs
index 6a6b2ebd6..8462eb89f 100644
--- a/crates/iceberg/src/scan/context.rs
+++ b/crates/iceberg/src/scan/context.rs
@@ -142,6 +142,7 @@ impl ManifestEntryContext {
.with_partition_spec(self.partition_spec.clone())
.with_name_mapping(self.name_mapping)
.with_case_sensitive(self.case_sensitive)
+ .with_key_metadata(self.manifest_entry.data_file.key_metadata().map(Box::from))
.build())
}
}
diff --git a/crates/iceberg/src/scan/task.rs b/crates/iceberg/src/scan/task.rs
index f3b556bcb..faeac51be 100644
--- a/crates/iceberg/src/scan/task.rs
+++ b/crates/iceberg/src/scan/task.rs
@@ -118,6 +118,19 @@ pub struct FileScanTask {
/// Whether this scan task should treat column names as case-sensitive when binding predicates.
pub case_sensitive: bool,
+
+ /// Key metadata for encrypted data files (Parquet Modular Encryption).
+ /// When present, the reader uses this to build `FileDecryptionProperties`.
+ ///
+ /// Note on the trust boundary: for the standard encryption scheme this
+ /// carries `StandardKeyMetadata`, whose payload is the *plaintext* DEK.
+ /// Because `FileScanTask` derives `Serialize`, that plaintext DEK is part
+ /// of the serialized scan plan should these tasks ever be serialized and sent
+ /// over the network.
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[builder(default)]
+ pub key_metadata: Option<Box<[u8]>>,
}
impl FileScanTask {
@@ -161,6 +174,13 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
.with_file_type(ctx.manifest_entry.content_type())
.with_partition_spec_id(ctx.partition_spec_id)
.with_equality_ids(ctx.manifest_entry.data_file.equality_ids.clone())
+ .with_key_metadata(
+ ctx.manifest_entry
+ .data_file
+ .key_metadata
+ .as_deref()
+ .map(Box::from),
+ )
.build()
}
}
@@ -184,4 +204,15 @@ pub struct FileScanTaskDeleteFile {
/// equality ids for equality deletes (null for anything other than equality-deletes)
#[builder(default)]
pub equality_ids: Option<Vec<i32>>,
+
+ /// Key metadata for encrypted delete files (Parquet Modular Encryption).
+ /// When present, the reader uses this to build `FileDecryptionProperties`.
+ ///
+ /// Same plaintext-DEK trust boundary as [`FileScanTask::key_metadata`]:
+ /// this is serialized into the scan plan and crosses the planner -> worker
+ /// channel in the clear for the standard encryption scheme.
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[builder(default)]
+ pub key_metadata: Option<Box<[u8]>>,
}