feat(storage/opendal): route oss:// URLs through the S3 backend

crates/storage/opendal/src/lib.rs

@@ -153,7 +153,15 @@ impl StorageFactory for OpenDalStorageFactory {
             OpenDalStorageFactory::Gcs => Ok(Arc::new(OpenDalStorage::Gcs {
                 config: gcs_config_parse(config.props().clone())?.into(),
             })),
-            #[cfg(feature = "opendal-oss")]
+            // OSS is S3-API-compatible; route through S3 so `s3.*` props
+            // work for OSS-backed tables (mirrors pyiceberg/Java S3FileIO).
+            #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))]
+            OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::S3 {
+                configured_scheme: "oss".to_string(),
+                config: s3_config_parse(config.props().clone())?.into(),
+                customized_credential_load: None,
+            })),
+            #[cfg(all(feature = "opendal-oss", not(feature = "opendal-s3")))]
             OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::Oss {
                 config: oss_config_parse(config.props().clone())?.into(),
             })),

crates/storage/opendal/src/resolving.rs

@@ -104,7 +104,20 @@ fn build_storage_for_scheme(
                 config: Arc::new(config),
             })
         }
-        #[cfg(feature = "opendal-oss")]
+        // OSS is S3-API-compatible; route through S3 so `s3.*` props
+        // work for OSS-backed tables (mirrors pyiceberg/Java S3FileIO).
+        #[cfg(feature = "opendal-s3")]
+        Scheme::Oss => {
+            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(),
+            })
+        }
+        // Fallback: builds without `opendal-s3` but with `opendal-oss`
+        // still use the native OSS service (which consumes `oss.*` keys).
+        #[cfg(all(feature = "opendal-oss", not(feature = "opendal-s3")))]
         Scheme::Oss => {
             let config = crate::oss::oss_config_parse(props.clone())?;
             Ok(OpenDalStorage::Oss {
@@ -317,3 +330,77 @@ impl Storage for OpenDalResolvingStorage {
         ))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[cfg(feature = "opendal-s3")]
+    #[test]
+    fn oss_scheme_routes_through_s3_backend_when_available() {
+        use iceberg::io::{S3_ACCESS_KEY_ID, S3_ENDPOINT};
+
+        let mut props = HashMap::new();
+        props.insert(S3_ACCESS_KEY_ID.to_string(), "AKIA_TEST".to_string());
+        props.insert(
+            S3_ENDPOINT.to_string(),
+            "https://oss-cn-shanghai.aliyuncs.com".to_string(),
+        );
+
+        let storage = build_storage_for_scheme(SCHEME_OSS, &props, &None).unwrap();
+
+        match storage {
+            OpenDalStorage::S3 {
+                configured_scheme, ..
+            } => {
+                assert_eq!(
+                    configured_scheme, SCHEME_OSS,
+                    "S3 backend should advertise the `oss` scheme so `oss://` \
+                     URLs are accepted by OpenDalStorage::S3::create_operator"
+                );
+            }
+            other => panic!(
+                "expected OpenDalStorage::S3 for oss://, got {other:?}; OSS is \
+                 S3-API-compatible and should share the S3 code path"
+            ),
+        }
+    }
+
+    #[cfg(feature = "opendal-s3")]
+    #[test]
+    fn oss_resolve_uses_s3_backend_and_caches() {
+        use iceberg::io::{S3_ACCESS_KEY_ID, S3_ENDPOINT};
+
+        let mut props = HashMap::new();
+        props.insert(S3_ACCESS_KEY_ID.to_string(), "AKIA_TEST".to_string());
+        props.insert(
+            S3_ENDPOINT.to_string(),
+            "https://oss-cn-shanghai.aliyuncs.com".to_string(),
+        );
+
+        let resolving = OpenDalResolvingStorage {
+            props,
+            storages: RwLock::new(HashMap::new()),
+            #[cfg(feature = "opendal-s3")]
+            customized_credential_load: None,
+        };
+
+        let storage = resolving.resolve("oss://my-bucket/path/to/file").unwrap();
+        assert!(matches!(storage.as_ref(), OpenDalStorage::S3 { .. }));
+
+        // Second call should hit the cache.
+        let cached = resolving.resolve("oss://my-bucket/other").unwrap();
+        assert!(Arc::ptr_eq(&storage, &cached));
+    }
+
+    #[cfg(all(feature = "opendal-oss", not(feature = "opendal-s3")))]
+    #[test]
+    fn oss_scheme_falls_back_to_native_oss_backend() {
+        let props = HashMap::new();
+        let storage = build_storage_for_scheme(SCHEME_OSS, &props).unwrap();
+        assert!(
+            matches!(storage, OpenDalStorage::Oss { .. }),
+            "Without opendal-s3, OSS should use the native OSS backend"
+        );
+    }
+}