feat(storage): implement opendal resolving storage (#2231)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #2210

## What changes are included in this PR?
- Add OpenDalResolvingStorage
<!--
Provide a summary of the modifications in this PR. List the main changes
such as new features, bug fixes, refactoring, or any other updates.
-->

## Are these changes tested?
Added a new test
<!--
Specify what test covers (unit test, integration test, etc.).

If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->

Author: CTTY

crates/storage/opendal/src/lib.rs

@@ -90,6 +90,9 @@ cfg_if! {
     }
 }
 
+mod resolving;
+pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory};
+
 /// OpenDAL-based storage factory.
 ///
 /// Maps scheme to the corresponding OpenDalStorage storage variant.

crates/storage/opendal/src/resolving.rs

@@ -0,0 +1,319 @@
+// 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.
+
+//! Resolving storage that auto-detects the scheme from a path and delegates
+//! to the appropriate [`OpenDalStorage`] variant.
+
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::StreamExt;
+use futures::stream::BoxStream;
+use iceberg::io::{
+    FileMetadata, FileRead, FileWrite, InputFile, OutputFile, Storage, StorageConfig,
+    StorageFactory,
+};
+use iceberg::{Error, ErrorKind, Result};
+use opendal::Scheme;
+use serde::{Deserialize, Serialize};
+use url::Url;
+
+use crate::OpenDalStorage;
+#[cfg(feature = "opendal-s3")]
+use crate::s3::CustomAwsCredentialLoader;
+
+/// Schemes supported by OpenDalResolvingStorage
+pub const SCHEME_MEMORY: &str = "memory";
+pub const SCHEME_FILE: &str = "file";
+pub const SCHEME_S3: &str = "s3";
+pub const SCHEME_S3A: &str = "s3a";
+pub const SCHEME_S3N: &str = "s3n";
+pub const SCHEME_GS: &str = "gs";
+pub const SCHEME_GCS: &str = "gcs";
+pub const SCHEME_OSS: &str = "oss";
+pub const SCHEME_ABFSS: &str = "abfss";
+pub const SCHEME_ABFS: &str = "abfs";
+pub const SCHEME_WASBS: &str = "wasbs";
+pub const SCHEME_WASB: &str = "wasb";
+
+/// Parse a URL scheme string into an [`opendal::Scheme`].
+fn parse_scheme(scheme: &str) -> Result<Scheme> {
+    match scheme {
+        SCHEME_MEMORY => Ok(Scheme::Memory),
+        SCHEME_FILE | "" => Ok(Scheme::Fs),
+        SCHEME_S3 | SCHEME_S3A | SCHEME_S3N => Ok(Scheme::S3),
+        SCHEME_GS | SCHEME_GCS => Ok(Scheme::Gcs),
+        SCHEME_OSS => Ok(Scheme::Oss),
+        SCHEME_ABFSS | SCHEME_ABFS | SCHEME_WASBS | SCHEME_WASB => Ok(Scheme::Azdls),
+        s => s.parse::<Scheme>().map_err(|e| {
+            Error::new(
+                ErrorKind::FeatureUnsupported,
+                format!("Unsupported storage scheme: {s}: {e}"),
+            )
+        }),
+    }
+}
+
+/// Extract the scheme string from a path URL.
+fn extract_scheme(path: &str) -> Result<String> {
+    let url = Url::parse(path).map_err(|e| {
+        Error::new(
+            ErrorKind::DataInvalid,
+            format!("Invalid path: {path}, failed to parse URL: {e}"),
+        )
+    })?;
+    Ok(url.scheme().to_string())
+}
+
+/// Build an [`OpenDalStorage`] variant for the given scheme and config properties.
+fn build_storage_for_scheme(
+    scheme: &str,
+    props: &HashMap<String, String>,
+    #[cfg(feature = "opendal-s3")] customized_credential_load: &Option<CustomAwsCredentialLoader>,
+) -> Result<OpenDalStorage> {
+    match parse_scheme(scheme)? {
+        #[cfg(feature = "opendal-s3")]
+        Scheme::S3 => {
+            let config = crate::s3::s3_config_parse(props.clone())?;
+            Ok(OpenDalStorage::S3 {
+                configured_scheme: scheme.to_string(),
+                config: Arc::new(config),
+                customized_credential_load: customized_credential_load.clone(),
+            })
+        }
+        #[cfg(feature = "opendal-gcs")]
+        Scheme::Gcs => {
+            let config = crate::gcs::gcs_config_parse(props.clone())?;
+            Ok(OpenDalStorage::Gcs {
+                config: Arc::new(config),
+            })
+        }
+        #[cfg(feature = "opendal-oss")]
+        Scheme::Oss => {
+            let config = crate::oss::oss_config_parse(props.clone())?;
+            Ok(OpenDalStorage::Oss {
+                config: Arc::new(config),
+            })
+        }
+        #[cfg(feature = "opendal-azdls")]
+        Scheme::Azdls => {
+            let configured_scheme: crate::azdls::AzureStorageScheme = scheme.parse()?;
+            let config = crate::azdls::azdls_config_parse(props.clone())?;
+            Ok(OpenDalStorage::Azdls {
+                configured_scheme,
+                config: Arc::new(config),
+            })
+        }
+        #[cfg(feature = "opendal-fs")]
+        Scheme::Fs => Ok(OpenDalStorage::LocalFs),
+        #[cfg(feature = "opendal-memory")]
+        Scheme::Memory => Ok(OpenDalStorage::Memory(crate::memory::memory_config_build()?)),
+        unsupported => Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            format!("Unsupported storage scheme: {unsupported}"),
+        )),
+    }
+}
+
+/// A resolving storage factory that creates [`OpenDalResolvingStorage`] instances.
+///
+/// This factory accepts paths from any supported storage system and dynamically
+/// delegates operations to the appropriate [`OpenDalStorage`] variant based on
+/// the path scheme.
+///
+/// # Example
+///
+/// ```rust,ignore
+/// use std::sync::Arc;
+/// use iceberg::io::FileIOBuilder;
+/// use iceberg_storage_opendal::OpenDalResolvingStorageFactory;
+///
+/// let factory = OpenDalResolvingStorageFactory::new();
+/// let file_io = FileIOBuilder::new(Arc::new(factory))
+///     .with_prop("s3.region", "us-east-1")
+///     .build();
+/// ```
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct OpenDalResolvingStorageFactory {
+    /// Custom AWS credential loader for S3 storage.
+    #[cfg(feature = "opendal-s3")]
+    #[serde(skip)]
+    customized_credential_load: Option<CustomAwsCredentialLoader>,
+}
+
+impl Default for OpenDalResolvingStorageFactory {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl OpenDalResolvingStorageFactory {
+    /// Create a new resolving storage factory.
+    pub fn new() -> Self {
+        Self {
+            #[cfg(feature = "opendal-s3")]
+            customized_credential_load: None,
+        }
+    }
+
+    /// Set a custom AWS credential loader for S3 storage.
+    #[cfg(feature = "opendal-s3")]
+    pub fn with_s3_credential_loader(mut self, loader: CustomAwsCredentialLoader) -> Self {
+        self.customized_credential_load = Some(loader);
+        self
+    }
+}
+
+#[typetag::serde]
+impl StorageFactory for OpenDalResolvingStorageFactory {
+    fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>> {
+        Ok(Arc::new(OpenDalResolvingStorage {
+            props: config.props().clone(),
+            storages: RwLock::new(HashMap::new()),
+            #[cfg(feature = "opendal-s3")]
+            customized_credential_load: self.customized_credential_load.clone(),
+        }))
+    }
+}
+
+/// A resolving storage that auto-detects the scheme from a path and delegates
+/// to the appropriate [`OpenDalStorage`] variant.
+///
+/// Sub-storages are lazily created on first use for each scheme and cached
+/// for subsequent operations.
+#[derive(Debug, Serialize, Deserialize)]
+pub struct OpenDalResolvingStorage {
+    /// Configuration properties shared across all backends.
+    props: HashMap<String, String>,
+    /// Cache of scheme → storage mappings.
+    #[serde(skip, default)]
+    storages: RwLock<HashMap<String, Arc<OpenDalStorage>>>,
+    /// Custom AWS credential loader for S3 storage.
+    #[cfg(feature = "opendal-s3")]
+    #[serde(skip)]
+    customized_credential_load: Option<CustomAwsCredentialLoader>,
+}
+
+impl OpenDalResolvingStorage {
+    /// Resolve the storage for the given path by extracting the scheme and
+    /// returning the cached or newly-created [`OpenDalStorage`].
+    fn resolve(&self, path: &str) -> Result<Arc<OpenDalStorage>> {
+        let scheme = extract_scheme(path)?;
+
+        // Fast path: check read lock first.
+        {
+            let cache = self
+                .storages
+                .read()
+                .map_err(|_| Error::new(ErrorKind::Unexpected, "Storage cache lock poisoned"))?;
+            if let Some(storage) = cache.get(&scheme) {
+                return Ok(storage.clone());
+            }
+        }
+
+        // Slow path: build and insert under write lock.
+        let mut cache = self
+            .storages
+            .write()
+            .map_err(|_| Error::new(ErrorKind::Unexpected, "Storage cache lock poisoned"))?;
+
+        // Double-check after acquiring write lock.
+        if let Some(storage) = cache.get(&scheme) {
+            return Ok(storage.clone());
+        }
+
+        let storage = build_storage_for_scheme(
+            &scheme,
+            &self.props,
+            #[cfg(feature = "opendal-s3")]
+            &self.customized_credential_load,
+        )?;
+        let storage = Arc::new(storage);
+        cache.insert(scheme, storage.clone());
+        Ok(storage)
+    }
+}
+
+#[async_trait]
+#[typetag::serde]
+impl Storage for OpenDalResolvingStorage {
+    async fn exists(&self, path: &str) -> Result<bool> {
+        self.resolve(path)?.exists(path).await
+    }
+
+    async fn metadata(&self, path: &str) -> Result<FileMetadata> {
+        self.resolve(path)?.metadata(path).await
+    }
+
+    async fn read(&self, path: &str) -> Result<Bytes> {
+        self.resolve(path)?.read(path).await
+    }
+
+    async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> {
+        self.resolve(path)?.reader(path).await
+    }
+
+    async fn write(&self, path: &str, bs: Bytes) -> Result<()> {
+        self.resolve(path)?.write(path, bs).await
+    }
+
+    async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> {
+        self.resolve(path)?.writer(path).await
+    }
+
+    async fn delete(&self, path: &str) -> Result<()> {
+        self.resolve(path)?.delete(path).await
+    }
+
+    async fn delete_prefix(&self, path: &str) -> Result<()> {
+        self.resolve(path)?.delete_prefix(path).await
+    }
+
+    async fn delete_stream(&self, mut paths: BoxStream<'static, String>) -> Result<()> {
+        // Group paths by scheme so each resolved storage receives a batch,
+        // avoiding repeated operator creation per path.
+        let mut grouped: HashMap<String, Vec<String>> = HashMap::new();
+        while let Some(path) = paths.next().await {
+            let scheme = extract_scheme(&path)?;
+            grouped.entry(scheme).or_default().push(path);
+        }
+
+        for (_, paths) in grouped {
+            let storage = self.resolve(&paths[0])?;
+            storage
+                .delete_stream(futures::stream::iter(paths).boxed())
+                .await?;
+        }
+        Ok(())
+    }
+
+    fn new_input(&self, path: &str) -> Result<InputFile> {
+        Ok(InputFile::new(
+            Arc::new(self.resolve(path)?.as_ref().clone()),
+            path.to_string(),
+        ))
+    }
+
+    fn new_output(&self, path: &str) -> Result<OutputFile> {
+        Ok(OutputFile::new(
+            Arc::new(self.resolve(path)?.as_ref().clone()),
+            path.to_string(),
+        ))
+    }
+}