From db4f8a78f9df70bd72025dc3a467ca7b1fbdf9ee Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Fri, 28 Aug 2026 16:19:37 +0800 Subject: [PATCH 1/7] feat(io): make FileIO serializable --- crates/iceberg/src/io/file_io.rs | 42 +++++++++++++++++++++++++++++++- 1 file changed, 41 insertions(+), 1 deletion(-) diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index cd0a4434c4..6a2424d40d 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -20,6 +20,7 @@ use std::sync::{Arc, OnceLock}; use bytes::Bytes; use futures::{Stream, StreamExt}; +use serde::{Deserialize, Serialize}; use super::storage::{ LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageFactory, @@ -59,13 +60,14 @@ use crate::Result; /// .with_prop("key", "value") /// .build(); /// ``` -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Serialize, Deserialize)] pub struct FileIO { /// Storage configuration containing properties config: StorageConfig, /// Factory for creating storage instances factory: Arc, /// Cached storage instance (lazily initialized) + #[serde(skip, default)] storage: Arc>>, } @@ -544,4 +546,42 @@ 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_file_io_serialization_roundtrip_issue_3088() { + let file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory)) + .with_prop("key", "value") + .build(); + + // Initialize the storage cache before serializing. The cache is process-local and should + // be rebuilt from the factory and configuration after deserialization. + file_io + .new_output("memory://test/file.txt") + .unwrap() + .write("test".into()) + .await + .unwrap(); + + let serialized = serde_json::to_string(&file_io).unwrap(); + let deserialized: FileIO = serde_json::from_str(&serialized).unwrap(); + + assert_eq!(deserialized.config().get("key"), Some(&"value".to_string())); + assert!(!deserialized.exists("memory://test/file.txt").await.unwrap()); + + 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") + ); + } } From 72e1ce37b4878c1b3fdada58e73148a064d91fe8 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Fri, 28 Aug 2026 16:34:16 +0800 Subject: [PATCH 2/7] chore: update public API snapshot --- crates/iceberg/public-api.txt | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 19f6c26968..4bc8f20ae8 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -739,6 +739,10 @@ 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 pub fn iceberg::io::FileIO::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg::io::FileIO +pub fn iceberg::io::FileIO::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::io::FileIO +pub fn iceberg::io::FileIO::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::io::FileIOBuilder impl iceberg::io::FileIOBuilder pub fn iceberg::io::FileIOBuilder::build(self) -> iceberg::io::FileIO From 982669d158793fa759b8f1ca650144814137eb3d Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Fri, 28 Aug 2026 17:39:06 +0800 Subject: [PATCH 3/7] test(io): simplify FileIO serde test name --- crates/iceberg/src/io/file_io.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index 6a2424d40d..6f7d2bc8ba 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -548,7 +548,7 @@ mod tests { } #[tokio::test] - async fn test_file_io_serialization_roundtrip_issue_3088() { + async fn test_file_io_serialization_roundtrip() { let file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory)) .with_prop("key", "value") .build(); From 1bd015b4cdde2cf2e1e1652eb5413ee325af44c0 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Sat, 29 Aug 2026 10:22:13 +0800 Subject: [PATCH 4/7] test(io): expand FileIO serde coverage --- Cargo.lock | 1 + crates/iceberg/src/io/file_io.rs | 89 ++++++++- crates/storage/opendal/Cargo.toml | 1 + crates/storage/opendal/src/lib.rs | 7 + crates/storage/opendal/src/resolving.rs | 6 + .../tests/file_io_serialization_test.rs | 188 ++++++++++++++++++ 6 files changed, 284 insertions(+), 8 deletions(-) create mode 100644 crates/storage/opendal/tests/file_io_serialization_test.rs diff --git a/Cargo.lock b/Cargo.lock index 3802cc03ff..fda8960910 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4064,6 +4064,7 @@ dependencies = [ "reqsign-core", "reqwest 0.12.28", "serde", + "serde_json", "tokio", "typetag", "url", diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index 6f7d2bc8ba..74ee414414 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -42,6 +42,20 @@ use crate::Result; /// OSS, Azure, etc.), use the /// [`iceberg-storage-opendal`](https://crates.io/crates/iceberg-storage-opendal) crate. /// +/// # Serialization +/// +/// `FileIO` serializes its storage configuration and factory, but not its cached storage +/// instance. The storage cache is rebuilt lazily on first use after deserialization. +/// +/// All storage configuration properties are included in the serialized representation. These +/// properties may contain credentials or other sensitive values, so serialized `FileIO` data +/// must be protected in transit and at rest by the application embedding this crate. +/// +/// Storage factories are serialized through [`typetag`](https://docs.rs/typetag). The receiving +/// binary must link the concrete factory implementation so it is registered for deserialization. +/// Third-party factories must use `#[typetag::serde]` on their [`StorageFactory`] +/// implementation. +/// /// # Example /// /// ```rust,ignore @@ -396,6 +410,7 @@ mod tests { use bytes::Bytes; use futures::AsyncReadExt; use futures::io::AllowStdIo; + use serde_json::json; use tempfile::TempDir; use super::{FileIO, FileIOBuilder}; @@ -548,25 +563,42 @@ mod tests { } #[tokio::test] - async fn test_file_io_serialization_roundtrip() { + async fn test_memory_file_io_serialization_roundtrip() { let file_io = FileIOBuilder::new(Arc::new(MemoryStorageFactory)) - .with_prop("key", "value") + .with_prop("test-property", "test-value") + .with_prop("s3.session-token", "test-token") .build(); - // Initialize the storage cache before serializing. The cache is process-local and should - // be rebuilt from the factory and configuration after deserialization. file_io .new_output("memory://test/file.txt") .unwrap() .write("test".into()) .await .unwrap(); + assert!(file_io.storage.get().is_some()); - let serialized = serde_json::to_string(&file_io).unwrap(); - let deserialized: FileIO = serde_json::from_str(&serialized).unwrap(); + let serialized = serde_json::to_value(&file_io).unwrap(); + assert_eq!( + serialized, + json!({ + "config": {"props": { + "s3.session-token": "test-token", + "test-property": "test-value" + }}, + "factory": {"type": "MemoryStorageFactory"} + }) + ); - assert_eq!(deserialized.config().get("key"), Some(&"value".to_string())); - assert!(!deserialized.exists("memory://test/file.txt").await.unwrap()); + let deserialized: FileIO = serde_json::from_value(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") @@ -583,5 +615,46 @@ mod tests { .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 = serde_json::to_value(&file_io).unwrap(); + assert_eq!( + serialized, + json!({ + "config": {"props": {"test-property": "test-value"}}, + "factory": {"type": "LocalFsStorageFactory"} + }) + ); + + let deserialized: FileIO = serde_json::from_value(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..0258c106da 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 } +serde_json = { 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..e87c741fdb 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -104,6 +104,13 @@ 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 custom AWS credential loader on the [`OpenDalStorageFactory::S3`] variant is +/// process-local and is not serialized. After deserialization, S3 storage uses OpenDAL's normal +/// credential resolution. Applications that require a custom loader must reconstruct the factory +/// in the receiving process. #[derive(Clone, Debug, Serialize, Deserialize)] pub enum OpenDalStorageFactory { /// Memory storage factory. diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 86993220a8..ef188bc480 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 +/// +/// The custom S3 credential loader is process-local and is not serialized. After +/// deserialization, S3 storage uses OpenDAL's normal credential resolution. Applications that +/// require a custom loader must reconstruct the factory in the receiving process. +/// /// # Example /// /// ```rust,ignore diff --git a/crates/storage/opendal/tests/file_io_serialization_test.rs b/crates/storage/opendal/tests/file_io_serialization_test.rs new file mode 100644 index 0000000000..83bafe82b0 --- /dev/null +++ b/crates/storage/opendal/tests/file_io_serialization_test.rs @@ -0,0 +1,188 @@ +// 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::sync::Arc; + +use iceberg::io::{FileIO, FileIOBuilder, StorageFactory}; +use iceberg_storage_opendal::OpenDalResolvingStorageFactory; +#[cfg(any( + feature = "opendal-memory", + feature = "opendal-fs", + feature = "opendal-s3", + feature = "opendal-gcs", + feature = "opendal-oss", + feature = "opendal-azdls", + feature = "opendal-hf" +))] +use iceberg_storage_opendal::OpenDalStorageFactory; +use serde_json::{Value, json}; + +fn assert_file_io_roundtrip(factory: Arc, expected_factory: Value) { + let file_io = FileIOBuilder::new(factory) + .with_prop("test-property", "test-value") + .build(); + + let serialized = serde_json::to_value(&file_io).unwrap(); + assert_eq!( + serialized, + json!({ + "config": {"props": {"test-property": "test-value"}}, + "factory": expected_factory + }) + ); + + let deserialized: FileIO = serde_json::from_value(serialized.clone()).unwrap(); + assert_eq!( + deserialized.config().get("test-property"), + Some(&"test-value".to_string()) + ); + assert_eq!(serde_json::to_value(deserialized).unwrap(), serialized); +} + +#[test] +fn test_resolving_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalResolvingStorageFactory::new()), + json!({"type": "OpenDalResolvingStorageFactory"}), + ); +} + +#[cfg(feature = "opendal-memory")] +#[test] +fn test_memory_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalStorageFactory::Memory), + json!({"type": "OpenDalStorageFactory", "Memory": null}), + ); +} + +#[cfg(feature = "opendal-fs")] +#[test] +fn test_fs_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalStorageFactory::Fs), + json!({"type": "OpenDalStorageFactory", "Fs": null}), + ); +} + +#[cfg(feature = "opendal-s3")] +#[test] +fn test_s3_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: None, + }), + json!({ + "type": "OpenDalStorageFactory", + "S3": {} + }), + ); +} + +#[cfg(feature = "opendal-gcs")] +#[test] +fn test_gcs_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalStorageFactory::Gcs), + json!({"type": "OpenDalStorageFactory", "Gcs": null}), + ); +} + +#[cfg(feature = "opendal-oss")] +#[test] +fn test_oss_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalStorageFactory::Oss), + json!({"type": "OpenDalStorageFactory", "Oss": null}), + ); +} + +#[cfg(feature = "opendal-azdls")] +#[test] +fn test_azdls_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalStorageFactory::Azdls), + json!({"type": "OpenDalStorageFactory", "Azdls": null}), + ); +} + +#[cfg(feature = "opendal-hf")] +#[test] +fn test_hf_factory_serialization_roundtrip() { + assert_file_io_roundtrip( + Arc::new(OpenDalStorageFactory::Hf), + json!({"type": "OpenDalStorageFactory", "Hf": null}), + ); +} + +#[cfg(feature = "opendal-s3")] +mod credential_loader_tests { + use iceberg_storage_opendal::{AwsCredential, CustomAwsCredentialLoader, ProvideCredential}; + use reqsign_core::Context; + + use super::*; + + #[derive(Debug)] + struct EmptyCredentialLoader; + + impl ProvideCredential for EmptyCredentialLoader { + type Credential = AwsCredential; + + async fn provide_credential( + &self, + _ctx: &Context, + ) -> reqsign_core::Result> { + Ok(None) + } + } + + fn loader() -> CustomAwsCredentialLoader { + CustomAwsCredentialLoader::new(EmptyCredentialLoader) + } + + #[test] + fn test_s3_factory_does_not_serialize_custom_credential_loader() { + let with_loader = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: Some(loader()), + })) + .build(); + let without_loader = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { + customized_credential_load: None, + })) + .build(); + + assert_eq!( + serde_json::to_value(with_loader).unwrap(), + serde_json::to_value(without_loader).unwrap() + ); + } + + #[test] + fn test_resolving_factory_does_not_serialize_custom_credential_loader() { + let with_loader = FileIOBuilder::new(Arc::new( + OpenDalResolvingStorageFactory::new().with_s3_credential_loader(loader()), + )) + .build(); + let without_loader = + FileIOBuilder::new(Arc::new(OpenDalResolvingStorageFactory::new())).build(); + + assert_eq!( + serde_json::to_value(with_loader).unwrap(), + serde_json::to_value(without_loader).unwrap() + ); + } +} From adb59b98970b82b60b18aa6d8392e9a19e363503 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Mon, 31 Aug 2026 10:58:38 +0800 Subject: [PATCH 5/7] fix(io): reject unserializable credential loaders --- Cargo.lock | 1 + crates/iceberg/src/io/file_io.rs | 32 +-- crates/storage/opendal/Cargo.toml | 1 + crates/storage/opendal/src/lib.rs | 60 +++++- crates/storage/opendal/src/resolving.rs | 42 +++- .../storage/opendal/tests/file_io_fs_test.rs | 51 +++++ .../storage/opendal/tests/file_io_gcs_test.rs | 25 +++ .../storage/opendal/tests/file_io_hf_test.rs | 9 +- .../opendal/tests/file_io_memory_test.rs | 48 +++++ .../storage/opendal/tests/file_io_s3_test.rs | 29 +++ .../tests/file_io_serialization_test.rs | 188 ------------------ .../opendal/tests/resolving_storage_test.rs | 7 +- 12 files changed, 269 insertions(+), 224 deletions(-) create mode 100644 crates/storage/opendal/tests/file_io_fs_test.rs create mode 100644 crates/storage/opendal/tests/file_io_memory_test.rs delete mode 100644 crates/storage/opendal/tests/file_io_serialization_test.rs diff --git a/Cargo.lock b/Cargo.lock index fda8960910..5f8a49651d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4065,6 +4065,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", + "tempfile", "tokio", "typetag", "url", diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index 74ee414414..487f8278f2 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -47,6 +47,10 @@ use crate::Result; /// `FileIO` serializes its storage configuration and factory, but not its cached storage /// instance. The storage cache is rebuilt lazily on first use after deserialization. /// +/// 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 serialized `FileIO` data /// must be protected in transit and at rest by the application embedding this crate. @@ -410,7 +414,6 @@ mod tests { use bytes::Bytes; use futures::AsyncReadExt; use futures::io::AllowStdIo; - use serde_json::json; use tempfile::TempDir; use super::{FileIO, FileIOBuilder}; @@ -577,19 +580,8 @@ mod tests { .unwrap(); assert!(file_io.storage.get().is_some()); - let serialized = serde_json::to_value(&file_io).unwrap(); - assert_eq!( - serialized, - json!({ - "config": {"props": { - "s3.session-token": "test-token", - "test-property": "test-value" - }}, - "factory": {"type": "MemoryStorageFactory"} - }) - ); - - let deserialized: FileIO = serde_json::from_value(serialized).unwrap(); + let serialized = serde_json::to_vec(&file_io).unwrap(); + let deserialized: FileIO = serde_json::from_slice(&serialized).unwrap(); assert!(deserialized.storage.get().is_none()); assert_eq!( deserialized.config().get("test-property"), @@ -635,16 +627,8 @@ mod tests { .unwrap(); assert!(file_io.storage.get().is_some()); - let serialized = serde_json::to_value(&file_io).unwrap(); - assert_eq!( - serialized, - json!({ - "config": {"props": {"test-property": "test-value"}}, - "factory": {"type": "LocalFsStorageFactory"} - }) - ); - - let deserialized: FileIO = serde_json::from_value(serialized).unwrap(); + let serialized = serde_json::to_vec(&file_io).unwrap(); + let deserialized: FileIO = serde_json::from_slice(&serialized).unwrap(); assert!(deserialized.storage.get().is_none()); assert_eq!( deserialized.config().get("test-property"), diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index 0258c106da..3a35079ce4 100644 --- a/crates/storage/opendal/Cargo.toml +++ b/crates/storage/opendal/Cargo.toml @@ -66,6 +66,7 @@ async-trait = { workspace = true } iceberg_test_utils = { path = "../../test_utils", features = ["tests"] } reqwest = { workspace = true } serde_json = { 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 e87c741fdb..6ef0e83e68 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -107,10 +107,9 @@ pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory}; /// /// # Serialization /// -/// The custom AWS credential loader on the [`OpenDalStorageFactory::S3`] variant is -/// process-local and is not serialized. After deserialization, S3 storage uses OpenDAL's normal -/// credential resolution. Applications that require a custom loader must reconstruct the factory -/// in the receiving process. +/// 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. @@ -123,7 +122,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. @@ -140,6 +143,22 @@ 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, +{ + match loader { + Some(_) => Err(serde::ser::Error::custom( + "custom AWS credential loaders cannot be serialized", + )), + None => serializer.serialize_none(), + } +} + #[typetag::serde(name = "OpenDalStorageFactory")] impl StorageFactory for OpenDalStorageFactory { #[allow(unused_variables)] @@ -638,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 = serde_json::to_value(file_io).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 ef188bc480..3df6bde367 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -142,9 +142,9 @@ fn build_storage_for_scheme( /// /// # Serialization /// -/// The custom S3 credential loader is process-local and is not serialized. After -/// deserialization, S3 storage uses OpenDAL's normal credential resolution. Applications that -/// require a custom loader must reconstruct the factory in the receiving process. +/// 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 /// @@ -162,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, } @@ -332,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 = serde_json::to_value(file_io).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..aeab2acb63 --- /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 = serde_json::to_vec(&file_io).unwrap(); + let file_io: FileIO = serde_json::from_slice(&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..f6e8e94377 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 = serde_json::to_vec(file_io).unwrap(); + serde_json::from_slice(&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..6496f857cd 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 = serde_json::to_vec(file_io).unwrap(); + serde_json::from_slice(&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..913e3519bb --- /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 = serde_json::to_vec(&file_io).unwrap(); + let file_io: FileIO = serde_json::from_slice(&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..5c6fd99814 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 = serde_json::to_vec(file_io).unwrap(); + serde_json::from_slice(&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/file_io_serialization_test.rs b/crates/storage/opendal/tests/file_io_serialization_test.rs deleted file mode 100644 index 83bafe82b0..0000000000 --- a/crates/storage/opendal/tests/file_io_serialization_test.rs +++ /dev/null @@ -1,188 +0,0 @@ -// 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::sync::Arc; - -use iceberg::io::{FileIO, FileIOBuilder, StorageFactory}; -use iceberg_storage_opendal::OpenDalResolvingStorageFactory; -#[cfg(any( - feature = "opendal-memory", - feature = "opendal-fs", - feature = "opendal-s3", - feature = "opendal-gcs", - feature = "opendal-oss", - feature = "opendal-azdls", - feature = "opendal-hf" -))] -use iceberg_storage_opendal::OpenDalStorageFactory; -use serde_json::{Value, json}; - -fn assert_file_io_roundtrip(factory: Arc, expected_factory: Value) { - let file_io = FileIOBuilder::new(factory) - .with_prop("test-property", "test-value") - .build(); - - let serialized = serde_json::to_value(&file_io).unwrap(); - assert_eq!( - serialized, - json!({ - "config": {"props": {"test-property": "test-value"}}, - "factory": expected_factory - }) - ); - - let deserialized: FileIO = serde_json::from_value(serialized.clone()).unwrap(); - assert_eq!( - deserialized.config().get("test-property"), - Some(&"test-value".to_string()) - ); - assert_eq!(serde_json::to_value(deserialized).unwrap(), serialized); -} - -#[test] -fn test_resolving_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalResolvingStorageFactory::new()), - json!({"type": "OpenDalResolvingStorageFactory"}), - ); -} - -#[cfg(feature = "opendal-memory")] -#[test] -fn test_memory_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalStorageFactory::Memory), - json!({"type": "OpenDalStorageFactory", "Memory": null}), - ); -} - -#[cfg(feature = "opendal-fs")] -#[test] -fn test_fs_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalStorageFactory::Fs), - json!({"type": "OpenDalStorageFactory", "Fs": null}), - ); -} - -#[cfg(feature = "opendal-s3")] -#[test] -fn test_s3_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalStorageFactory::S3 { - customized_credential_load: None, - }), - json!({ - "type": "OpenDalStorageFactory", - "S3": {} - }), - ); -} - -#[cfg(feature = "opendal-gcs")] -#[test] -fn test_gcs_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalStorageFactory::Gcs), - json!({"type": "OpenDalStorageFactory", "Gcs": null}), - ); -} - -#[cfg(feature = "opendal-oss")] -#[test] -fn test_oss_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalStorageFactory::Oss), - json!({"type": "OpenDalStorageFactory", "Oss": null}), - ); -} - -#[cfg(feature = "opendal-azdls")] -#[test] -fn test_azdls_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalStorageFactory::Azdls), - json!({"type": "OpenDalStorageFactory", "Azdls": null}), - ); -} - -#[cfg(feature = "opendal-hf")] -#[test] -fn test_hf_factory_serialization_roundtrip() { - assert_file_io_roundtrip( - Arc::new(OpenDalStorageFactory::Hf), - json!({"type": "OpenDalStorageFactory", "Hf": null}), - ); -} - -#[cfg(feature = "opendal-s3")] -mod credential_loader_tests { - use iceberg_storage_opendal::{AwsCredential, CustomAwsCredentialLoader, ProvideCredential}; - use reqsign_core::Context; - - use super::*; - - #[derive(Debug)] - struct EmptyCredentialLoader; - - impl ProvideCredential for EmptyCredentialLoader { - type Credential = AwsCredential; - - async fn provide_credential( - &self, - _ctx: &Context, - ) -> reqsign_core::Result> { - Ok(None) - } - } - - fn loader() -> CustomAwsCredentialLoader { - CustomAwsCredentialLoader::new(EmptyCredentialLoader) - } - - #[test] - fn test_s3_factory_does_not_serialize_custom_credential_loader() { - let with_loader = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { - customized_credential_load: Some(loader()), - })) - .build(); - let without_loader = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 { - customized_credential_load: None, - })) - .build(); - - assert_eq!( - serde_json::to_value(with_loader).unwrap(), - serde_json::to_value(without_loader).unwrap() - ); - } - - #[test] - fn test_resolving_factory_does_not_serialize_custom_credential_loader() { - let with_loader = FileIOBuilder::new(Arc::new( - OpenDalResolvingStorageFactory::new().with_s3_credential_loader(loader()), - )) - .build(); - let without_loader = - FileIOBuilder::new(Arc::new(OpenDalResolvingStorageFactory::new())).build(); - - assert_eq!( - serde_json::to_value(with_loader).unwrap(), - serde_json::to_value(without_loader).unwrap() - ); - } -} diff --git a/crates/storage/opendal/tests/resolving_storage_test.rs b/crates/storage/opendal/tests/resolving_storage_test.rs index 1853a796dd..0277b737e8 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 = serde_json::to_vec(file_io).unwrap(); + serde_json::from_slice(&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/{}", From 5b50b1f0f21cf994709991596e765877eeb2c1a2 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:29:22 +0800 Subject: [PATCH 6/7] fix(io): make FileIO serialization explicit --- Cargo.lock | 1 - crates/iceberg/public-api.txt | 6 +- crates/iceberg/src/io/file_io.rs | 77 +++++++++++++------ crates/storage/opendal/Cargo.toml | 1 - crates/storage/opendal/src/lib.rs | 20 ++--- crates/storage/opendal/src/resolving.rs | 2 +- .../storage/opendal/tests/file_io_fs_test.rs | 4 +- .../storage/opendal/tests/file_io_gcs_test.rs | 4 +- .../storage/opendal/tests/file_io_hf_test.rs | 4 +- .../opendal/tests/file_io_memory_test.rs | 4 +- .../storage/opendal/tests/file_io_s3_test.rs | 4 +- .../opendal/tests/resolving_storage_test.rs | 4 +- 12 files changed, 78 insertions(+), 53 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5f8a49651d..65b7182911 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4064,7 +4064,6 @@ dependencies = [ "reqsign-core", "reqwest 0.12.28", "serde", - "serde_json", "tempfile", "tokio", "typetag", diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 4bc8f20ae8..ed05fb6b3c 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -730,19 +730,17 @@ 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 pub fn iceberg::io::FileIO::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result -impl serde_core::ser::Serialize for iceberg::io::FileIO -pub fn iceberg::io::FileIO::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::io::FileIO -pub fn iceberg::io::FileIO::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::io::FileIOBuilder impl iceberg::io::FileIOBuilder pub fn iceberg::io::FileIOBuilder::build(self) -> iceberg::io::FileIO diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index 487f8278f2..e741f7c2d9 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -42,24 +42,6 @@ use crate::Result; /// OSS, Azure, etc.), use the /// [`iceberg-storage-opendal`](https://crates.io/crates/iceberg-storage-opendal) crate. /// -/// # Serialization -/// -/// `FileIO` serializes its storage configuration and factory, but not its cached storage -/// instance. The storage cache is rebuilt lazily on first use after deserialization. -/// -/// 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 serialized `FileIO` data -/// must be protected in transit and at rest by the application embedding this crate. -/// -/// Storage factories are serialized through [`typetag`](https://docs.rs/typetag). The receiving -/// binary must link the concrete factory implementation so it is registered for deserialization. -/// Third-party factories must use `#[typetag::serde]` on their [`StorageFactory`] -/// implementation. -/// /// # Example /// /// ```rust,ignore @@ -78,17 +60,28 @@ use crate::Result; /// .with_prop("key", "value") /// .build(); /// ``` -#[derive(Clone, Debug, Serialize, Deserialize)] +#[derive(Clone, Debug)] pub struct FileIO { /// Storage configuration containing properties config: StorageConfig, /// Factory for creating storage instances factory: Arc, /// Cached storage instance (lazily initialized) - #[serde(skip, default)] storage: Arc>>, } +#[derive(Serialize)] +struct SerializableFileIO<'a> { + config: &'a StorageConfig, + factory: &'a Arc, +} + +#[derive(Deserialize)] +struct DeserializedFileIO { + config: StorageConfig, + factory: Arc, +} + impl FileIO { /// Create a new FileIO backed by in-memory storage. /// @@ -112,6 +105,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(&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 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 @@ -580,8 +609,8 @@ mod tests { .unwrap(); assert!(file_io.storage.get().is_some()); - let serialized = serde_json::to_vec(&file_io).unwrap(); - let deserialized: FileIO = serde_json::from_slice(&serialized).unwrap(); + 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"), @@ -627,8 +656,8 @@ mod tests { .unwrap(); assert!(file_io.storage.get().is_some()); - let serialized = serde_json::to_vec(&file_io).unwrap(); - let deserialized: FileIO = serde_json::from_slice(&serialized).unwrap(); + 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"), diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index 3a35079ce4..c6ea9f9b67 100644 --- a/crates/storage/opendal/Cargo.toml +++ b/crates/storage/opendal/Cargo.toml @@ -65,7 +65,6 @@ url = { workspace = true } async-trait = { workspace = true } iceberg_test_utils = { path = "../../test_utils", features = ["tests"] } reqwest = { workspace = true } -serde_json = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["macros"] } diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 6ef0e83e68..2ed5a60cb1 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -107,7 +107,10 @@ pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory}; /// /// # Serialization /// -/// Serialization fails when the [`OpenDalStorageFactory::S3`] variant contains a custom AWS +/// 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)] @@ -145,18 +148,15 @@ pub enum OpenDalStorageFactory { #[cfg(feature = "opendal-s3")] pub(crate) fn serialize_custom_credential_loader( - loader: &Option, - serializer: S, + _loader: &Option, + _serializer: S, ) -> std::result::Result where S: serde::Serializer, { - match loader { - Some(_) => Err(serde::ser::Error::custom( - "custom AWS credential loaders cannot be serialized", - )), - None => serializer.serialize_none(), - } + Err(serde::ser::Error::custom( + "custom AWS credential loaders cannot be serialized", + )) } #[typetag::serde(name = "OpenDalStorageFactory")] @@ -681,7 +681,7 @@ mod tests { })) .build(); - let err = serde_json::to_value(file_io).unwrap_err(); + let err = file_io.serialize_all().unwrap_err(); assert!( err.to_string() .contains("custom AWS credential loaders cannot be serialized") diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 3df6bde367..3b99b08fc0 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -359,7 +359,7 @@ mod tests { .with_s3_credential_loader(CustomAwsCredentialLoader::new(EmptyCredentialLoader)); let file_io = iceberg::io::FileIOBuilder::new(Arc::new(factory)).build(); - let err = serde_json::to_value(file_io).unwrap_err(); + let err = file_io.serialize_all().unwrap_err(); assert!( err.to_string() .contains("custom AWS credential loaders cannot be serialized") diff --git a/crates/storage/opendal/tests/file_io_fs_test.rs b/crates/storage/opendal/tests/file_io_fs_test.rs index aeab2acb63..f9bb82db7e 100644 --- a/crates/storage/opendal/tests/file_io_fs_test.rs +++ b/crates/storage/opendal/tests/file_io_fs_test.rs @@ -32,8 +32,8 @@ mod tests { 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 = serde_json::to_vec(&file_io).unwrap(); - let file_io: FileIO = serde_json::from_slice(&serialized).unwrap(); + let serialized = file_io.serialize_all().unwrap(); + let file_io = FileIO::deserialize_all(&serialized).unwrap(); file_io .new_output(&path) diff --git a/crates/storage/opendal/tests/file_io_gcs_test.rs b/crates/storage/opendal/tests/file_io_gcs_test.rs index f6e8e94377..c6ba770b17 100644 --- a/crates/storage/opendal/tests/file_io_gcs_test.rs +++ b/crates/storage/opendal/tests/file_io_gcs_test.rs @@ -48,8 +48,8 @@ mod tests { } fn roundtrip_file_io(file_io: &FileIO) -> FileIO { - let serialized = serde_json::to_vec(file_io).unwrap(); - serde_json::from_slice(&serialized).unwrap() + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() } #[tokio::test] diff --git a/crates/storage/opendal/tests/file_io_hf_test.rs b/crates/storage/opendal/tests/file_io_hf_test.rs index 6496f857cd..f8d17e2ad1 100644 --- a/crates/storage/opendal/tests/file_io_hf_test.rs +++ b/crates/storage/opendal/tests/file_io_hf_test.rs @@ -69,8 +69,8 @@ mod tests { } fn roundtrip_file_io(file_io: &FileIO) -> FileIO { - let serialized = serde_json::to_vec(file_io).unwrap(); - serde_json::from_slice(&serialized).unwrap() + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() } // --- bucket tests --- diff --git a/crates/storage/opendal/tests/file_io_memory_test.rs b/crates/storage/opendal/tests/file_io_memory_test.rs index 913e3519bb..9c3c335a3f 100644 --- a/crates/storage/opendal/tests/file_io_memory_test.rs +++ b/crates/storage/opendal/tests/file_io_memory_test.rs @@ -28,8 +28,8 @@ mod tests { #[tokio::test] async fn test_file_io_memory_serialization_roundtrip() { let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Memory)).build(); - let serialized = serde_json::to_vec(&file_io).unwrap(); - let file_io: FileIO = serde_json::from_slice(&serialized).unwrap(); + let serialized = file_io.serialize_all().unwrap(); + let file_io = FileIO::deserialize_all(&serialized).unwrap(); let path = "memory://serialization-roundtrip"; file_io diff --git a/crates/storage/opendal/tests/file_io_s3_test.rs b/crates/storage/opendal/tests/file_io_s3_test.rs index 5c6fd99814..6b57f86607 100644 --- a/crates/storage/opendal/tests/file_io_s3_test.rs +++ b/crates/storage/opendal/tests/file_io_s3_test.rs @@ -54,8 +54,8 @@ mod tests { } fn roundtrip_file_io(file_io: &FileIO) -> FileIO { - let serialized = serde_json::to_vec(file_io).unwrap(); - serde_json::from_slice(&serialized).unwrap() + let serialized = file_io.serialize_all().unwrap(); + FileIO::deserialize_all(&serialized).unwrap() } #[tokio::test] diff --git a/crates/storage/opendal/tests/resolving_storage_test.rs b/crates/storage/opendal/tests/resolving_storage_test.rs index 0277b737e8..ba94b8332d 100644 --- a/crates/storage/opendal/tests/resolving_storage_test.rs +++ b/crates/storage/opendal/tests/resolving_storage_test.rs @@ -52,8 +52,8 @@ mod tests { } fn roundtrip_file_io(file_io: &iceberg::io::FileIO) -> iceberg::io::FileIO { - let serialized = serde_json::to_vec(file_io).unwrap(); - serde_json::from_slice(&serialized).unwrap() + let serialized = file_io.serialize_all().unwrap(); + iceberg::io::FileIO::deserialize_all(&serialized).unwrap() } fn temp_fs_path(name: &str) -> String { From 4b4263395325e75cf801a172442b4754a69e3746 Mon Sep 17 00:00:00 2001 From: Ray Liu <257669749+blackmwk@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:51:21 +0800 Subject: [PATCH 7/7] refactor(io): isolate FileIO serde helpers --- crates/iceberg/src/io/file_io.rs | 31 +++++++++++++++++++------------ 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index e741f7c2d9..87ac84784f 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -20,7 +20,6 @@ use std::sync::{Arc, OnceLock}; use bytes::Bytes; use futures::{Stream, StreamExt}; -use serde::{Deserialize, Serialize}; use super::storage::{ LocalFsStorageFactory, MemoryStorageFactory, Storage, StorageConfig, StorageFactory, @@ -70,16 +69,24 @@ pub struct FileIO { storage: Arc>>, } -#[derive(Serialize)] -struct SerializableFileIO<'a> { - config: &'a StorageConfig, - factory: &'a Arc, -} +mod _serde { + use std::sync::Arc; -#[derive(Deserialize)] -struct DeserializedFileIO { - config: StorageConfig, - factory: 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 { @@ -121,7 +128,7 @@ impl FileIO { /// 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(&SerializableFileIO { + Ok(serde_json::to_vec(&_serde::SerializableFileIO { config: &self.config, factory: &self.factory, })?) @@ -133,7 +140,7 @@ impl FileIO { /// 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 DeserializedFileIO { config, factory } = serde_json::from_slice(bytes)?; + let _serde::DeserializedFileIO { config, factory } = serde_json::from_slice(bytes)?; Ok(Self { config, factory,