feat(io): make FileIO serializable

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<str>) -> iceberg::Result<()>
 pub async fn iceberg::io::FileIO::delete_prefix(&self, path: impl core::convert::AsRef<str>) -> iceberg::Result<()>
 pub async fn iceberg::io::FileIO::delete_stream(&self, paths: impl futures_core::stream::Stream<Item = alloc::string::String> + core::marker::Send + 'static) -> iceberg::Result<()>
+pub fn iceberg::io::FileIO::deserialize_all(bytes: &[u8]) -> iceberg::Result<Self>
 pub async fn iceberg::io::FileIO::exists(&self, path: impl core::convert::AsRef<str>) -> iceberg::Result<bool>
 pub fn iceberg::io::FileIO::new_input(&self, path: impl core::convert::AsRef<str>) -> iceberg::Result<iceberg::io::InputFile>
 pub fn iceberg::io::FileIO::new_output(&self, path: impl core::convert::AsRef<str>) -> iceberg::Result<iceberg::io::OutputFile>
 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<alloc::vec::Vec<u8>>
 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

crates/iceberg/src/io/file_io.rs

@@ -69,6 +69,26 @@ pub struct FileIO {
     storage: Arc<OnceLock<Arc<dyn Storage>>>,
 }
 
+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<dyn StorageFactory>,
+    }
+
+    #[derive(Deserialize)]
+    pub(super) struct DeserializedFileIO {
+        pub(super) config: StorageConfig,
+        pub(super) factory: Arc<dyn StorageFactory>,
+    }
+}
+
 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<Vec<u8>> {
+        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<Self> {
+        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());
+    }
 }

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]

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<CustomAwsCredentialLoader>,
     },
     /// GCS storage factory.
@@ -133,6 +146,19 @@ pub enum OpenDalStorageFactory {
     Hf,
 }
 
+#[cfg(feature = "opendal-s3")]
+pub(crate) fn serialize_custom_credential_loader<S>(
+    _loader: &Option<CustomAwsCredentialLoader>,
+    _serializer: S,
+) -> std::result::Result<S::Ok, S::Error>
+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<Option<AwsCredential>> {
+            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() {

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<CustomAwsCredentialLoader>,
 }
 
@@ -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<Option<Self::Credential>> {
+            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 {

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());
+    }
+}