use std::sync::Arc;
use super::ManifestList;
use crate::encryption::{EncryptedInputFile, EncryptionManager};
use crate::error::Result;
use crate::io::FileIO;
use crate::spec::{SnapshotRef, TableMetadataRef};
use crate::{Error, ErrorKind};
pub struct ManifestListReader {
snapshot: SnapshotRef,
file_io: FileIO,
table_metadata: TableMetadataRef,
encryption_manager: Option<Arc<EncryptionManager>>,
}
impl ManifestListReader {
pub(crate) fn new(
snapshot: SnapshotRef,
file_io: FileIO,
table_metadata: TableMetadataRef,
encryption_manager: Option<Arc<EncryptionManager>>,
) -> Self {
Self {
snapshot,
file_io,
table_metadata,
encryption_manager,
}
}
pub async fn load(&self) -> Result<ManifestList> {
let manifest_list_content = match self.snapshot.encryption_key_id() {
Some(key_id) => {
let em = self.encryption_manager.as_ref().ok_or_else(|| {
Error::new(
ErrorKind::PreconditionFailed,
"Snapshot has encryption_key_id but no EncryptionManager configured on Table",
)
})?;
let key_metadata = em.decrypt_manifest_list_key_metadata(key_id).await?;
let input = self.file_io.new_input(self.snapshot.manifest_list())?;
EncryptedInputFile::new(input, key_metadata).read().await?
}
None => {
self.file_io
.new_input(self.snapshot.manifest_list())?
.read()
.await?
}
};
ManifestList::parse_with_version(
&manifest_list_content,
self.table_metadata.format_version(),
)
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use super::ManifestListReader;
use crate::encryption::kms::MemoryKeyManagementClient;
use crate::encryption::{EncryptionManager, SensitiveBytes};
use crate::io::FileIO;
use crate::spec::{TableMetadata, TableMetadataRef};
fn encryption_test_metadata() -> TableMetadata {
let path = format!(
"{}/testdata/table_metadata/TableMetadataV3ValidEncryption.json",
env!("CARGO_MANIFEST_DIR"),
);
let json = std::fs::read_to_string(path).unwrap();
serde_json::from_str(&json).unwrap()
}
#[tokio::test]
async fn load_manifest_list_errors_when_encrypted_but_no_manager_configured() {
let mut metadata = encryption_test_metadata();
let manifest_list_path = format!(
"{}/testdata/manifests_lists/manifest-list-v3-encrypted.avro",
env!("CARGO_MANIFEST_DIR"),
);
let snapshot = metadata.snapshots.get_mut(&1).unwrap();
let mut patched = snapshot.as_ref().clone();
patched.manifest_list = manifest_list_path;
*snapshot = Arc::new(patched);
let snapshot_ref = metadata.current_snapshot().unwrap().clone();
let metadata_ref = TableMetadataRef::new(metadata);
let err = ManifestListReader::new(snapshot_ref, FileIO::new_with_fs(), metadata_ref, None)
.load()
.await
.unwrap_err();
assert_eq!(err.kind(), crate::ErrorKind::PreconditionFailed);
}
#[tokio::test]
async fn load_manifest_list_decrypts_roundtrip() {
let mut metadata = encryption_test_metadata();
let manifest_list_path = format!(
"{}/testdata/manifests_lists/manifest-list-v3-encrypted.avro",
env!("CARGO_MANIFEST_DIR"),
);
let snapshot = metadata.snapshots.get_mut(&1).unwrap();
let mut patched = snapshot.as_ref().clone();
patched.manifest_list = manifest_list_path;
*snapshot = Arc::new(patched);
let kms = MemoryKeyManagementClient::new();
kms.add_master_key_bytes(
"master-1",
SensitiveBytes::new([
0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d,
0x0e, 0x0f,
]),
)
.unwrap();
let mgr = EncryptionManager::builder()
.kms_client(Arc::new(kms) as Arc<dyn crate::encryption::kms::KeyManagementClient>)
.table_key_id("master-1")
.encryption_keys(metadata.encryption_keys.clone())
.build();
let snapshot_ref = metadata.current_snapshot().unwrap().clone();
let metadata_ref = TableMetadataRef::new(metadata);
let manifest_list = ManifestListReader::new(
snapshot_ref,
FileIO::new_with_fs(),
metadata_ref,
Some(Arc::new(mgr)),
)
.load()
.await
.unwrap();
assert_eq!(manifest_list.entries().len(), 0);
}
}