typetag serde for Storage

Author: CTTY

Cargo.lock

@@ -128,6 +128,7 @@ toml = "0.8"
 tracing = "0.1.41"
 tracing-subscriber = "0.3.20"
 typed-builder = "0.20"
+typetag = "0.2"
 url = "2.5.7"
 uuid = { version = "1.18", features = ["v7"] }
 volo = "0.10.6"

Cargo.toml

@@ -90,6 +90,7 @@ strum = { workspace = true, features = ["derive"] }
 thrift = { workspace = true }
 tokio = { workspace = true, optional = false, features = ["sync"] }
 typed-builder = { workspace = true }
+typetag = { workspace = true }
 url = { workspace = true }
 uuid = { workspace = true }
 zstd = { workspace = true }

crates/iceberg/Cargo.toml

@@ -523,6 +523,7 @@ mod tests {
     use bytes::Bytes;
     use futures::AsyncReadExt;
     use futures::io::AllowStdIo;
+    use serde::{Deserialize, Serialize};
     use tempfile::TempDir;
 
     use super::{FileIO, FileIOBuilder};
@@ -533,12 +534,17 @@ mod tests {
     use crate::{Error, ErrorKind, Result};
 
     // Test storage implementation that tracks write operations
-    #[derive(Debug, Clone)]
+    #[derive(Debug, Clone, Serialize, Deserialize)]
     struct TestStorage {
+        #[serde(skip, default = "default_written")]
         written: Arc<Mutex<Vec<String>>>,
         received_props: HashMap<String, String>,
     }
 
+    fn default_written() -> Arc<Mutex<Vec<String>>> {
+        Arc::new(Mutex::new(Vec::new()))
+    }
+
     #[allow(dead_code)]
     impl TestStorage {
         pub fn written(&self) -> MutexGuard<'_, Vec<String>> {
@@ -551,6 +557,7 @@ mod tests {
     }
 
     #[async_trait]
+    #[typetag::serde]
     impl Storage for TestStorage {
         async fn exists(&self, _path: &str) -> Result<bool> {
             Ok(true)

crates/iceberg/src/io/file_io.rs

@@ -36,6 +36,9 @@ use crate::Result;
 /// This trait defines the interface for all storage backends. Implementations
 /// provide access to different storage systems like S3, GCS, Azure, local filesystem, etc.
 ///
+/// The trait supports serialization via `typetag`, allowing storage instances to be
+/// serialized and deserialized across process boundaries.
+///
 /// # Example
 ///
 /// ```rust,ignore
@@ -51,6 +54,7 @@ use crate::Result;
 /// }
 /// ```
 #[async_trait]
+#[typetag::serde(tag = "type")]
 pub trait Storage: Debug + Send + Sync {
     /// Check if a file exists at the given path
     async fn exists(&self, path: &str) -> Result<bool>;

crates/iceberg/src/io/storage.rs

@@ -132,7 +132,7 @@ pub(crate) fn azdls_create_operator<'a>(
 ///   paths are expected to contain the `dfs` storage service.
 /// - `wasb[s]` is used to refer to files in Blob Storage directly; paths are
 ///   expected to contain the `blob` storage service.
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub(crate) enum AzureStorageScheme {
     Abfs,
     Abfss,
@@ -606,7 +606,9 @@ mod tests {
 }
 
 /// Azure Data Lake Storage implementation using OpenDAL
-#[derive(Debug, Clone)]
+///
+/// Stores configuration and creates operators on-demand.
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct OpenDALAzdlsStorage {
     /// Because Azdls accepts multiple possible schemes, we store the full
     /// passed scheme here to later validate schemes passed via paths.
@@ -625,6 +627,7 @@ impl OpenDALAzdlsStorage {
 }
 
 #[async_trait]
+#[typetag::serde]
 impl Storage for OpenDALAzdlsStorage {
     async fn exists(&self, path: &str) -> Result<bool> {
         let (op, relative_path) = self.create_operator(path)?;

crates/iceberg/src/io/storage_azdls.rs

@@ -22,6 +22,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use opendal::Operator;
 use opendal::services::FsConfig;
+use serde::{Deserialize, Serialize};
 
 use crate::Result;
 use crate::io::{
@@ -37,7 +38,9 @@ pub(crate) fn fs_config_build() -> Result<Operator> {
 }
 
 /// Filesystem storage implementation using OpenDAL
-#[derive(Debug, Clone)]
+///
+/// This storage is stateless and creates operators on-demand.
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct OpenDALFsStorage;
 
 impl OpenDALFsStorage {
@@ -52,6 +55,7 @@ impl OpenDALFsStorage {
 }
 
 #[async_trait]
+#[typetag::serde]
 impl Storage for OpenDALFsStorage {
     async fn exists(&self, path: &str) -> Result<bool> {
         let relative_path = self.extract_relative_path(path);
@@ -131,3 +135,27 @@ impl StorageFactory for OpenDALFsStorageFactory {
         Ok(Arc::new(OpenDALFsStorage))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::io::Storage;
+
+    #[test]
+    fn test_fs_storage_serialization() {
+        // Create a filesystem storage instance using the factory
+        let factory = OpenDALFsStorageFactory;
+        let storage = factory
+            .build(HashMap::new(), Extensions::default())
+            .unwrap();
+
+        // Serialize the storage
+        let serialized = serde_json::to_string(&storage).unwrap();
+
+        // Deserialize the storage
+        let deserialized: Box<dyn Storage> = serde_json::from_str(&serialized).unwrap();
+
+        // Verify the type is correct
+        assert!(format!("{:?}", deserialized).contains("OpenDALFsStorage"));
+    }
+}

crates/iceberg/src/io/storage_fs.rs

@@ -23,6 +23,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use opendal::Operator;
 use opendal::services::GcsConfig;
+use serde::{Deserialize, Serialize};
 use url::Url;
 
 use crate::io::{
@@ -112,7 +113,9 @@ pub(crate) fn gcs_config_build(cfg: &GcsConfig, path: &str) -> Result<Operator>
 }
 
 /// GCS storage implementation using OpenDAL
-#[derive(Debug, Clone)]
+///
+/// Stores configuration and creates operators on-demand.
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct OpenDALGcsStorage {
     config: Arc<GcsConfig>,
 }
@@ -136,6 +139,7 @@ impl OpenDALGcsStorage {
 }
 
 #[async_trait]
+#[typetag::serde]
 impl Storage for OpenDALGcsStorage {
     async fn exists(&self, path: &str) -> Result<bool> {
         let (op, relative_path) = self.create_operator(path)?;