diff --git a/Cargo.lock b/Cargo.lock index 3802cc03ff..65b7182911 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4064,6 +4064,7 @@ dependencies = [ "reqsign-core", "reqwest 0.12.28", "serde", + "tempfile", "tokio", "typetag", "url", diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 19f6c26968..ed05fb6b3c 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -730,11 +730,13 @@ pub fn iceberg::io::FileIO::config(&self) -> &iceberg::io::StorageConfig pub async fn iceberg::io::FileIO::delete(&self, path: impl core::convert::AsRef) -> iceberg::Result<()> pub async fn iceberg::io::FileIO::delete_prefix(&self, path: impl core::convert::AsRef) -> iceberg::Result<()> pub async fn iceberg::io::FileIO::delete_stream(&self, paths: impl futures_core::stream::Stream + core::marker::Send + 'static) -> iceberg::Result<()> +pub fn iceberg::io::FileIO::deserialize_all(bytes: &[u8]) -> iceberg::Result pub async fn iceberg::io::FileIO::exists(&self, path: impl core::convert::AsRef) -> iceberg::Result pub fn iceberg::io::FileIO::new_input(&self, path: impl core::convert::AsRef) -> iceberg::Result pub fn iceberg::io::FileIO::new_output(&self, path: impl core::convert::AsRef) -> iceberg::Result pub fn iceberg::io::FileIO::new_with_fs() -> Self pub fn iceberg::io::FileIO::new_with_memory() -> Self +pub fn iceberg::io::FileIO::serialize_all(&self) -> iceberg::Result> impl core::clone::Clone for iceberg::io::FileIO pub fn iceberg::io::FileIO::clone(&self) -> iceberg::io::FileIO impl core::fmt::Debug for iceberg::io::FileIO diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index cd0a4434c4..87ac84784f 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -69,6 +69,26 @@ pub struct FileIO { storage: Arc>>, } +mod _serde { + use std::sync::Arc; + + use serde::{Deserialize, Serialize}; + + use super::{StorageConfig, StorageFactory}; + + #[derive(Serialize)] + pub(super) struct SerializableFileIO<'a> { + pub(super) config: &'a StorageConfig, + pub(super) factory: &'a Arc, + } + + #[derive(Deserialize)] + pub(super) struct DeserializedFileIO { + pub(super) config: StorageConfig, + pub(super) factory: Arc, + } +} + impl FileIO { /// Create a new FileIO backed by in-memory storage. /// @@ -92,6 +112,42 @@ impl FileIO { } } + /// Serializes all portable state of this `FileIO` into a byte vector. + /// + /// This includes the storage configuration and factory, but not the cached storage instance. + /// The storage cache is rebuilt lazily on first use after calling [`FileIO::deserialize_all`]. + /// + /// The serialized representation is not a stable format and may change between crate versions. + /// Applications should not rely on it for long-term storage or exchange it between incompatible + /// versions of this crate. + /// + /// All storage configuration properties are included in the serialized representation. These + /// properties may contain credentials or other sensitive values, so the returned bytes must be + /// protected in transit and at rest by the application embedding this crate. + /// + /// Storage factories are serialized through [`typetag`](https://docs.rs/typetag). Third-party + /// factories must use `#[typetag::serde]` on their [`StorageFactory`] implementation. + pub fn serialize_all(&self) -> Result> { + Ok(serde_json::to_vec(&_serde::SerializableFileIO { + config: &self.config, + factory: &self.factory, + })?) + } + + /// Deserializes a `FileIO` previously produced by [`FileIO::serialize_all`]. + /// + /// The receiving binary must use a compatible crate version and link the concrete factory + /// implementation so it is registered with `typetag`. Backend-specific requirements are + /// documented by each storage factory implementation. + pub fn deserialize_all(bytes: &[u8]) -> Result { + let _serde::DeserializedFileIO { config, factory } = serde_json::from_slice(bytes)?; + Ok(Self { + config, + factory, + storage: Arc::new(OnceLock::new()), + }) + } + /// Get the storage configuration. pub fn config(&self) -> &StorageConfig { &self.config @@ -544,4 +600,81 @@ mod tests { assert_eq!(file_io.config().get("key1"), Some(&"value1".to_string())); assert_eq!(file_io.config().get("key2"), Some(&"value2".to_string())); } + + #[tokio::test] + async fn test_memory_file_io_serialization_roundtrip() { + let file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory)) + .with_prop("test-property", "test-value") + .with_prop("s3.session-token", "test-token") + .build(); + + file_io + .new_output("memory://test/file.txt") + .unwrap() + .write("test".into()) + .await + .unwrap(); + assert!(file_io.storage.get().is_some()); + + let serialized = file_io.serialize_all().unwrap(); + let deserialized = FileIO::deserialize_all(&serialized).unwrap(); + assert!(deserialized.storage.get().is_none()); + assert_eq!( + deserialized.config().get("test-property"), + Some(&"test-value".to_string()) + ); + assert_eq!( + deserialized.config().get("s3.session-token"), + Some(&"test-token".to_string()) + ); + + deserialized + .new_output("memory://test/roundtrip.txt") + .unwrap() + .write("roundtrip".into()) + .await + .unwrap(); + assert_eq!( + deserialized + .new_input("memory://test/roundtrip.txt") + .unwrap() + .read() + .await + .unwrap(), + Bytes::from("roundtrip") + ); + assert!(deserialized.storage.get().is_some()); + } + + #[tokio::test] + async fn test_local_fs_file_io_serialization_roundtrip() { + let tmp_dir = TempDir::new().unwrap(); + let path = tmp_dir.path().join("roundtrip.txt"); + let path = path.to_str().unwrap(); + let file_io = FileIOBuilder::new(Arc::new(LocalFsStorageFactory)) + .with_prop("test-property", "test-value") + .build(); + + file_io + .new_output(path) + .unwrap() + .write("roundtrip".into()) + .await + .unwrap(); + assert!(file_io.storage.get().is_some()); + + let serialized = file_io.serialize_all().unwrap(); + let deserialized = FileIO::deserialize_all(&serialized).unwrap(); + assert!(deserialized.storage.get().is_none()); + assert_eq!( + deserialized.config().get("test-property"), + Some(&"test-value".to_string()) + ); + assert!(deserialized.exists(path).await.unwrap()); + assert_eq!( + deserialized.new_input(path).unwrap().read().await.unwrap(), + Bytes::from("roundtrip") + ); + assert!(deserialized.storage.get().is_some()); + } } diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index e43e7845b3..c6ea9f9b67 100644 --- a/crates/storage/opendal/Cargo.toml +++ b/crates/storage/opendal/Cargo.toml @@ -65,6 +65,7 @@ url = { workspace = true } async-trait = { workspace = true } iceberg_test_utils = { path = "../../test_utils", features = ["tests"] } reqwest = { workspace = true } +tempfile = { workspace = true } tokio = { workspace = true, features = ["macros"] } [lints] diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index d02507e7b5..2ed5a60cb1 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -104,6 +104,15 @@ pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory}; /// /// Maps scheme to the corresponding OpenDalStorage storage variant. /// Use this factory with `FileIOBuilder::new(factory)` to create FileIO instances. +/// +/// # Serialization +/// +/// The receiving binary must enable the feature corresponding to the serialized backend variant. +/// For example, deserializing `OpenDalStorageFactory::S3` requires the `opendal-s3` feature. +/// +/// Serialization fails when the `OpenDalStorageFactory::S3` variant contains a custom AWS +/// credential loader because the loader holds process-local state that cannot be reconstructed in +/// another process. Construct the factory without a custom loader before serializing it. #[derive(Clone, Debug, Serialize, Deserialize)] pub enum OpenDalStorageFactory { /// Memory storage factory. @@ -116,7 +125,11 @@ pub enum OpenDalStorageFactory { #[cfg(feature = "opendal-s3")] S3 { /// Custom AWS credential loader. - #[serde(skip)] + #[serde( + skip_deserializing, + skip_serializing_if = "Option::is_none", + serialize_with = "serialize_custom_credential_loader" + )] customized_credential_load: Option, }, /// GCS storage factory. @@ -133,6 +146,19 @@ pub enum OpenDalStorageFactory { Hf, } +#[cfg(feature = "opendal-s3")] +pub(crate) fn serialize_custom_credential_loader( + _loader: &Option, + _serializer: S, +) -> std::result::Result +where + S: serde::Serializer, +{ + Err(serde::ser::Error::custom( + "custom AWS credential loaders cannot be serialized", + )) +} + #[typetag::serde(name = "OpenDalStorageFactory")] impl StorageFactory for OpenDalStorageFactory { #[allow(unused_variables)] @@ -631,6 +657,37 @@ impl FileWrite for OpenDalWriter { mod tests { use super::*; + #[cfg(feature = "opendal-s3")] + #[derive(Debug)] + struct EmptyCredentialLoader; + + #[cfg(feature = "opendal-s3")] + impl ProvideCredential for EmptyCredentialLoader { + type Credential = AwsCredential; + + async fn provide_credential( + &self, + _ctx: &reqsign_core::Context, + ) -> reqsign_core::Result> { + Ok(None) + } + } + + #[cfg(feature = "opendal-s3")] + #[test] + fn test_s3_factory_custom_credential_loader_serialization_fails() { + let file_io = iceberg::io::FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: Some(CustomAwsCredentialLoader::new(EmptyCredentialLoader)), + })) + .build(); + + let err = file_io.serialize_all().unwrap_err(); + assert!( + err.to_string() + .contains("custom AWS credential loaders cannot be serialized") + ); + } + #[cfg(feature = "opendal-memory")] #[test] fn test_default_memory_operator() { diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 86993220a8..3b99b08fc0 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -140,6 +140,12 @@ fn build_storage_for_scheme( /// delegates operations to the appropriate [`OpenDalStorage`] variant based on /// the path scheme. /// +/// # Serialization +/// +/// Serialization fails when a custom S3 credential loader is configured because the loader holds +/// process-local state that cannot be reconstructed in another process. Construct the factory +/// without a custom loader before serializing it. +/// /// # Example /// /// ```rust,ignore @@ -156,7 +162,11 @@ fn build_storage_for_scheme( pub struct OpenDalResolvingStorageFactory { /// Custom AWS credential loader for S3 storage. #[cfg(feature = "opendal-s3")] - #[serde(skip)] + #[serde( + skip_deserializing, + skip_serializing_if = "Option::is_none", + serialize_with = "crate::serialize_custom_credential_loader" + )] customized_credential_load: Option, } @@ -326,6 +336,36 @@ impl Storage for OpenDalResolvingStorage { mod tests { use super::*; + #[cfg(feature = "opendal-s3")] + #[derive(Debug)] + struct EmptyCredentialLoader; + + #[cfg(feature = "opendal-s3")] + impl crate::s3::ProvideCredential for EmptyCredentialLoader { + type Credential = crate::s3::AwsCredential; + + async fn provide_credential( + &self, + _ctx: &reqsign_core::Context, + ) -> reqsign_core::Result> { + Ok(None) + } + } + + #[cfg(feature = "opendal-s3")] + #[test] + fn test_custom_credential_loader_serialization_fails() { + let factory = OpenDalResolvingStorageFactory::new() + .with_s3_credential_loader(CustomAwsCredentialLoader::new(EmptyCredentialLoader)); + let file_io = iceberg::io::FileIOBuilder::new(Arc::new(factory)).build(); + + let err = file_io.serialize_all().unwrap_err(); + assert!( + err.to_string() + .contains("custom AWS credential loaders cannot be serialized") + ); + } + /// Builds a resolving storage with empty props, suitable for `resolve()` /// calls that don't actually hit any backend. fn empty_resolving_storage() -> OpenDalResolvingStorage { diff --git a/crates/storage/opendal/tests/file_io_fs_test.rs b/crates/storage/opendal/tests/file_io_fs_test.rs new file mode 100644 index 0000000000..f9bb82db7e --- /dev/null +++ b/crates/storage/opendal/tests/file_io_fs_test.rs @@ -0,0 +1,51 @@ +// 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. + +//! Integration tests for FileIO OpenDAL filesystem storage. + +#[cfg(feature = "opendal-fs")] +mod tests { + use std::sync::Arc; + + use bytes::Bytes; + use iceberg::io::{FileIO, FileIOBuilder}; + use iceberg_storage_opendal::OpenDalStorageFactory; + use tempfile::TempDir; + + #[tokio::test] + async fn test_file_io_fs_serialization_roundtrip() { + let temp_dir = TempDir::new().unwrap(); + let path = temp_dir.path().join("serialization-roundtrip"); + let path = format!("file:/{}", path.display()); + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Fs)).build(); + let serialized = file_io.serialize_all().unwrap(); + let file_io = FileIO::deserialize_all(&serialized).unwrap(); + + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"roundtrip")) + .await + .unwrap(); + assert_eq!( + file_io.new_input(&path).unwrap().read().await.unwrap(), + Bytes::from_static(b"roundtrip") + ); + file_io.delete(&path).await.unwrap(); + assert!(!file_io.exists(&path).await.unwrap()); + } +} diff --git a/crates/storage/opendal/tests/file_io_gcs_test.rs b/crates/storage/opendal/tests/file_io_gcs_test.rs index b43cc3a535..c6ba770b17 100644 --- a/crates/storage/opendal/tests/file_io_gcs_test.rs +++ b/crates/storage/opendal/tests/file_io_gcs_test.rs @@ -47,6 +47,31 @@ mod tests { .build() } + fn roundtrip_file_io(file_io: &FileIO) -> FileIO { + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() + } + + #[tokio::test] + async fn test_file_io_gcs_serialization_roundtrip() { + let file_io = roundtrip_file_io(&get_file_io_gcs().await); + let path = format!("{}/serialization-roundtrip", get_gs_path()); + + let _ = file_io.delete(&path).await; + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"roundtrip")) + .await + .unwrap(); + assert_eq!( + file_io.new_input(&path).unwrap().read().await.unwrap(), + Bytes::from_static(b"roundtrip") + ); + file_io.delete(&path).await.unwrap(); + assert!(!file_io.exists(&path).await.unwrap()); + } + // Create a bucket against the emulated GCS storage server. async fn create_bucket(name: &str, server_endpoint: &str) -> anyhow::Result<()> { let mut bucket_data = HashMap::new(); diff --git a/crates/storage/opendal/tests/file_io_hf_test.rs b/crates/storage/opendal/tests/file_io_hf_test.rs index 44fe3420d4..f8d17e2ad1 100644 --- a/crates/storage/opendal/tests/file_io_hf_test.rs +++ b/crates/storage/opendal/tests/file_io_hf_test.rs @@ -68,13 +68,18 @@ mod tests { .build() } + fn roundtrip_file_io(file_io: &FileIO) -> FileIO { + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() + } + // --- bucket tests --- #[tokio::test] async fn test_hf_bucket_write_read_delete() { let token = require_env!(ENV_HF_TOKEN); let bucket = require_env!(ENV_HF_BUCKET); - let file_io = get_file_io(&token); + let file_io = roundtrip_file_io(&get_file_io(&token)); let path = format!( "hf://buckets/{}/{}", bucket, @@ -324,7 +329,7 @@ mod tests { #[tokio::test] async fn test_hf_resolving_storage() { let token = require_env!(ENV_HF_TOKEN); - let file_io = get_resolving_file_io(&token); + let file_io = roundtrip_file_io(&get_resolving_file_io(&token)); let bucket = require_env!(ENV_HF_BUCKET); let path = format!( diff --git a/crates/storage/opendal/tests/file_io_memory_test.rs b/crates/storage/opendal/tests/file_io_memory_test.rs new file mode 100644 index 0000000000..9c3c335a3f --- /dev/null +++ b/crates/storage/opendal/tests/file_io_memory_test.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. + +//! Integration tests for FileIO OpenDAL memory storage. + +#[cfg(feature = "opendal-memory")] +mod tests { + use std::sync::Arc; + + use bytes::Bytes; + use iceberg::io::{FileIO, FileIOBuilder}; + use iceberg_storage_opendal::OpenDalStorageFactory; + + #[tokio::test] + async fn test_file_io_memory_serialization_roundtrip() { + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Memory)).build(); + let serialized = file_io.serialize_all().unwrap(); + let file_io = FileIO::deserialize_all(&serialized).unwrap(); + let path = "memory://serialization-roundtrip"; + + file_io + .new_output(path) + .unwrap() + .write(Bytes::from_static(b"roundtrip")) + .await + .unwrap(); + assert_eq!( + file_io.new_input(path).unwrap().read().await.unwrap(), + Bytes::from_static(b"roundtrip") + ); + file_io.delete(path).await.unwrap(); + assert!(!file_io.exists(path).await.unwrap()); + } +} diff --git a/crates/storage/opendal/tests/file_io_s3_test.rs b/crates/storage/opendal/tests/file_io_s3_test.rs index d5858e18f4..6b57f86607 100644 --- a/crates/storage/opendal/tests/file_io_s3_test.rs +++ b/crates/storage/opendal/tests/file_io_s3_test.rs @@ -23,6 +23,7 @@ mod tests { use std::sync::Arc; + use bytes::Bytes; use futures::StreamExt; use iceberg::io::{ FileIO, FileIOBuilder, S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_PATH_STYLE_ACCESS, S3_REGION, @@ -52,6 +53,34 @@ mod tests { .build() } + fn roundtrip_file_io(file_io: &FileIO) -> FileIO { + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() + } + + #[tokio::test] + async fn test_file_io_s3_serialization_roundtrip() { + let file_io = roundtrip_file_io(&get_file_io().await); + let path = format!( + "s3://bucket1/{}", + normalize_test_name_with_parts!("test_file_io_s3_serialization_roundtrip") + ); + + let _ = file_io.delete(&path).await; + file_io + .new_output(&path) + .unwrap() + .write(Bytes::from_static(b"roundtrip")) + .await + .unwrap(); + assert_eq!( + file_io.new_input(&path).unwrap().read().await.unwrap(), + Bytes::from_static(b"roundtrip") + ); + file_io.delete(&path).await.unwrap(); + assert!(!file_io.exists(&path).await.unwrap()); + } + #[tokio::test] async fn test_file_io_s3_exists() { let file_io = get_file_io().await; diff --git a/crates/storage/opendal/tests/resolving_storage_test.rs b/crates/storage/opendal/tests/resolving_storage_test.rs index 1853a796dd..ba94b8332d 100644 --- a/crates/storage/opendal/tests/resolving_storage_test.rs +++ b/crates/storage/opendal/tests/resolving_storage_test.rs @@ -51,6 +51,11 @@ mod tests { .build() } + fn roundtrip_file_io(file_io: &iceberg::io::FileIO) -> iceberg::io::FileIO { + let serialized = file_io.serialize_all().unwrap(); + iceberg::io::FileIO::deserialize_all(&serialized).unwrap() + } + fn temp_fs_path(name: &str) -> String { let dir = std::env::temp_dir().join("iceberg_resolving_tests"); std::fs::create_dir_all(&dir).unwrap(); @@ -62,7 +67,7 @@ mod tests { #[tokio::test] async fn test_mixed_scheme_write_and_read() { - let file_io = get_resolving_file_io(); + let file_io = roundtrip_file_io(&get_resolving_file_io()); let s3_path = format!( "s3://bucket1/{}",