From c814a41e201d75cdfb63f4f9df35f926dcc45152 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Wed, 15 Apr 2026 20:47:10 +0800 Subject: [PATCH 1/4] feat(storage/opendal): route oss:// URLs through the S3 backend --- crates/storage/opendal/src/lib.rs | 9 +++- crates/storage/opendal/src/oss.rs | 14 +++-- crates/storage/opendal/src/resolving.rs | 71 ++++++++++++++++++++++++- 3 files changed, 88 insertions(+), 6 deletions(-) diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 9e01552016..ff3069dba5 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -236,7 +236,14 @@ impl StorageFactory for OpenDalStorageFactory { config: gcs_config_parse(config.props().clone())?.into(), client_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(all(feature = "opendal-oss", feature = "opendal-s3"))] + OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::S3 { + 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(), client_config, diff --git a/crates/storage/opendal/src/oss.rs b/crates/storage/opendal/src/oss.rs index a1cf44549c..25405ccdba 100644 --- a/crates/storage/opendal/src/oss.rs +++ b/crates/storage/opendal/src/oss.rs @@ -15,9 +15,6 @@ // specific language governing permissions and limitations // under the License. -use std::collections::HashMap; - -use iceberg::io::{OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_ENDPOINT}; use iceberg::{Error, ErrorKind, Result}; use opendal::services::OssConfig; use opendal::{Configurator, Operator}; @@ -26,7 +23,16 @@ use url::Url; use crate::utils::from_opendal_error; /// Parse iceberg props to oss config. -pub(crate) fn oss_config_parse(mut m: HashMap) -> Result { +/// +/// Only reachable when the native OSS service is used (i.e. `opendal-s3` +/// is disabled); when `opendal-s3` is enabled, OSS is routed through the +/// S3 code path and this function is dead code. +#[cfg(not(feature = "opendal-s3"))] +pub(crate) fn oss_config_parse( + mut m: std::collections::HashMap, +) -> Result { + use iceberg::io::{OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_ENDPOINT}; + let mut cfg: OssConfig = OssConfig::default(); if let Some(endpoint) = m.remove(OSS_ENDPOINT) { cfg.endpoint = Some(endpoint); diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 51e76da472..e43460b51d 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -105,7 +105,19 @@ fn build_storage_for_scheme( client_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(all(feature = "opendal-oss", feature = "opendal-s3"))] + "oss" => { + let config = crate::s3::s3_config_parse(props.clone())?; + Ok(OpenDalStorage::S3 { + 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")))] "oss" => { let config = crate::oss::oss_config_parse(props.clone())?; Ok(OpenDalStorage::Oss { @@ -456,4 +468,61 @@ mod tests { "abfss and abfs should share one instance" ); } + + #[cfg(all(feature = "opendal-oss", 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("oss", &props, &None).unwrap(); + assert!( + matches!(storage, OpenDalStorage::S3 { .. }), + "OSS should route through S3 backend when opendal-s3 is enabled" + ); + } + + #[cfg(all(feature = "opendal-oss", 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("oss", &props).unwrap(); + assert!( + matches!(storage, OpenDalStorage::Oss { .. }), + "Without opendal-s3, OSS should use the native OSS backend" + ); + } } From 33e08ca20431df59c584519ce6ba1a40e4689164 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Tue, 26 May 2026 11:55:56 +0800 Subject: [PATCH 2/4] feat(storage/opendal): add native OSS fallback for RAM/OIDC auth --- crates/storage/opendal/src/lib.rs | 39 +++++++++++++++---- crates/storage/opendal/src/oss.rs | 14 ++----- crates/storage/opendal/src/resolving.rs | 51 ++++++++++++++----------- 3 files changed, 64 insertions(+), 40 deletions(-) diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index ff3069dba5..5fa6fd4e3f 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -151,6 +151,15 @@ fn parse_io_timeout_ms(value: &str) -> Result { }) } +/// Returns `true` when the property map contains S3-style configuration, +/// indicating that `oss://` paths should be routed through the S3 backend +/// (e.g. REST catalog vended credentials). +#[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] +fn has_s3_config(props: &HashMap) -> bool { + props.contains_key(iceberg::io::S3_ENDPOINT) + || props.contains_key(iceberg::io::S3_ACCESS_KEY_ID) +} + /// OpenDAL-based storage factory. /// /// Maps scheme to the corresponding OpenDalStorage storage variant. @@ -236,13 +245,22 @@ impl StorageFactory for OpenDalStorageFactory { config: gcs_config_parse(config.props().clone())?.into(), client_config, })), - // OSS is S3-API-compatible; route through S3 so `s3.*` props - // work for OSS-backed tables (mirrors pyiceberg/Java S3FileIO). + // OSS: if `s3.*` props are present (REST catalog vended credentials), + // route through S3 backend (OSS is S3-API-compatible). + // Otherwise fall back to native OSS backend (RAM/OIDC auth). #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] - OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::S3 { - config: s3_config_parse(config.props().clone())?.into(), - customized_credential_load: None, - })), + OpenDalStorageFactory::Oss => { + if has_s3_config(config.props()) { + Ok(Arc::new(OpenDalStorage::S3 { + config: s3_config_parse(config.props().clone())?.into(), + customized_credential_load: None, + })) + } else { + Ok(Arc::new(OpenDalStorage::Oss { + config: oss_config_parse(config.props().clone())?.into(), + })) + } + } #[cfg(all(feature = "opendal-oss", not(feature = "opendal-s3")))] OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::Oss { config: oss_config_parse(config.props().clone())?.into(), @@ -327,7 +345,10 @@ pub enum OpenDalStorage { #[serde(default)] client_config: OpenDalClientConfig, }, - /// OSS storage variant. + /// OSS storage variant (native OpenDAL OSS backend). + /// + /// Used when no `s3.*` configuration is present, allowing native + /// Aliyun authentication (RAM/OIDC/assume-role). #[cfg(feature = "opendal-oss")] Oss { /// OSS configuration. @@ -1094,7 +1115,9 @@ mod tests { // All S3-family schemes are accepted by the same storage instance. // Custom schemes for S3-compatible stores (e.g., `minio://`) are also // accepted because the path's scheme is used as-is for prefix matching. - for scheme in ["s3", "s3a", "s3n", "minio"] { + // `oss://` is included because OSS is routed through this backend when + // `s3.*` config is present (see `has_s3_config`). + for scheme in ["s3", "s3a", "s3n", "minio", "oss"] { assert_eq!( storage .relativize_path(&format!("{scheme}://my-bucket/path/to/file.parquet")) diff --git a/crates/storage/opendal/src/oss.rs b/crates/storage/opendal/src/oss.rs index 25405ccdba..a1cf44549c 100644 --- a/crates/storage/opendal/src/oss.rs +++ b/crates/storage/opendal/src/oss.rs @@ -15,6 +15,9 @@ // specific language governing permissions and limitations // under the License. +use std::collections::HashMap; + +use iceberg::io::{OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_ENDPOINT}; use iceberg::{Error, ErrorKind, Result}; use opendal::services::OssConfig; use opendal::{Configurator, Operator}; @@ -23,16 +26,7 @@ use url::Url; use crate::utils::from_opendal_error; /// Parse iceberg props to oss config. -/// -/// Only reachable when the native OSS service is used (i.e. `opendal-s3` -/// is disabled); when `opendal-s3` is enabled, OSS is routed through the -/// S3 code path and this function is dead code. -#[cfg(not(feature = "opendal-s3"))] -pub(crate) fn oss_config_parse( - mut m: std::collections::HashMap, -) -> Result { - use iceberg::io::{OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_ENDPOINT}; - +pub(crate) fn oss_config_parse(mut m: HashMap) -> Result { let mut cfg: OssConfig = OssConfig::default(); if let Some(endpoint) = m.remove(OSS_ENDPOINT) { cfg.endpoint = Some(endpoint); diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index e43460b51d..02e576079b 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -105,18 +105,24 @@ fn build_storage_for_scheme( client_config, }) } - // OSS is S3-API-compatible; route through S3 so `s3.*` props - // work for OSS-backed tables (mirrors pyiceberg/Java S3FileIO). + // OSS: if `s3.*` props are present (REST catalog vended credentials), + // route through S3 backend (OSS is S3-API-compatible). + // Otherwise fall back to native OSS backend (RAM/OIDC auth). #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] "oss" => { - let config = crate::s3::s3_config_parse(props.clone())?; - Ok(OpenDalStorage::S3 { - config: Arc::new(config), - customized_credential_load: customized_credential_load.clone(), - }) + if crate::has_s3_config(props) { + let config = crate::s3::s3_config_parse(props.clone())?; + Ok(OpenDalStorage::S3 { + config: Arc::new(config), + customized_credential_load: customized_credential_load.clone(), + }) + } else { + let config = crate::oss::oss_config_parse(props.clone())?; + Ok(OpenDalStorage::Oss { + config: Arc::new(config), + }) + } } - // 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")))] "oss" => { let config = crate::oss::oss_config_parse(props.clone())?; @@ -471,7 +477,7 @@ mod tests { #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] #[test] - fn oss_scheme_routes_through_s3_backend_when_available() { + fn oss_routes_through_s3_when_s3_config_present() { use iceberg::io::{S3_ACCESS_KEY_ID, S3_ENDPOINT}; let mut props = HashMap::new(); @@ -484,7 +490,19 @@ mod tests { let storage = build_storage_for_scheme("oss", &props, &None).unwrap(); assert!( matches!(storage, OpenDalStorage::S3 { .. }), - "OSS should route through S3 backend when opendal-s3 is enabled" + "OSS with s3.* config should route through S3 backend" + ); + } + + #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] + #[test] + fn oss_falls_back_to_native_without_s3_config() { + let props = HashMap::new(); + + let storage = build_storage_for_scheme("oss", &props, &None).unwrap(); + assert!( + matches!(storage, OpenDalStorage::Oss { .. }), + "OSS without s3.* config should use native OSS backend" ); } @@ -514,15 +532,4 @@ mod tests { 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("oss", &props).unwrap(); - assert!( - matches!(storage, OpenDalStorage::Oss { .. }), - "Without opendal-s3, OSS should use the native OSS backend" - ); - } } From 3272672cc39be4ec725ddf5abce8ec687f310323 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Fri, 14 Aug 2026 03:23:27 -0400 Subject: [PATCH 3/4] fix(storage/opendal): require an explicit s3 endpoint to route oss:// via s3 --- crates/storage/opendal/src/lib.rs | 42 +++++++++++++++++++++---- crates/storage/opendal/src/resolving.rs | 36 ++++++++++++++++++++- 2 files changed, 71 insertions(+), 7 deletions(-) diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 5fa6fd4e3f..762e6b8a81 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -151,13 +151,19 @@ fn parse_io_timeout_ms(value: &str) -> Result { }) } -/// Returns `true` when the property map contains S3-style configuration, -/// indicating that `oss://` paths should be routed through the S3 backend -/// (e.g. REST catalog vended credentials). +/// Returns `true` when `oss://` paths should go through the S3 backend rather +/// than the native OSS one (e.g. for REST catalog vended `s3.*` credentials). +/// +/// `s3.endpoint` is required, not merely one of the `s3.*` keys: without it +/// opendal falls back to `s3.amazonaws.com`, so an access key meant for Aliyun +/// would be sent to AWS. +/// +/// Note that one props map serves every scheme, so a map carrying both `s3.*` +/// (for `s3://` paths) and `oss.*` (for `oss://` paths) sends `oss://` to the +/// S3 endpoint. Configure the two in separate `FileIO`s. #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] -fn has_s3_config(props: &HashMap) -> bool { +fn routes_oss_via_s3(props: &HashMap) -> bool { props.contains_key(iceberg::io::S3_ENDPOINT) - || props.contains_key(iceberg::io::S3_ACCESS_KEY_ID) } /// OpenDAL-based storage factory. @@ -250,7 +256,7 @@ impl StorageFactory for OpenDalStorageFactory { // Otherwise fall back to native OSS backend (RAM/OIDC auth). #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] OpenDalStorageFactory::Oss => { - if has_s3_config(config.props()) { + if routes_oss_via_s3(config.props()) { Ok(Arc::new(OpenDalStorage::S3 { config: s3_config_parse(config.props().clone())?.into(), customized_credential_load: None, @@ -1023,6 +1029,30 @@ mod tests { ); } + #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] + #[test] + fn factory_routes_oss_like_the_resolver_does() { + use iceberg::io::{S3_ACCESS_KEY_ID, S3_ENDPOINT}; + + let via_s3 = StorageConfig::new() + .with_prop(S3_ENDPOINT, "https://oss-cn-shanghai.aliyuncs.com") + .with_prop(S3_ACCESS_KEY_ID, "AKIA_TEST"); + // `Storage` is only debuggable through the trait object, and the + // variant name is what distinguishes the two backends. + let built = format!("{:?}", OpenDalStorageFactory::Oss.build(&via_s3).unwrap()); + assert!(built.starts_with("S3"), "{built}"); + + // Credentials without an endpoint stay on the native backend — the + // same rule the resolver applies. + for props in [ + StorageConfig::new().with_prop(S3_ACCESS_KEY_ID, "AKIA_TEST"), + StorageConfig::new(), + ] { + let built = format!("{:?}", OpenDalStorageFactory::Oss.build(&props).unwrap()); + assert!(built.starts_with("Oss"), "{built}"); + } + } + #[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 02e576079b..067a558edc 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -110,7 +110,7 @@ fn build_storage_for_scheme( // Otherwise fall back to native OSS backend (RAM/OIDC auth). #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] "oss" => { - if crate::has_s3_config(props) { + if crate::routes_oss_via_s3(props) { let config = crate::s3::s3_config_parse(props.clone())?; Ok(OpenDalStorage::S3 { config: Arc::new(config), @@ -494,6 +494,40 @@ mod tests { ); } + #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] + #[test] + fn vended_credentials_shape_routes_through_s3() { + use iceberg::io::{S3_ACCESS_KEY_ID, S3_ENDPOINT, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN}; + + // What a REST catalog vends for an OSS warehouse: S3-compatible + // credentials plus the OSS endpoint, and no `oss.*` keys. + let props = HashMap::from([ + ( + S3_ENDPOINT.to_string(), + "https://oss-cn-hangzhou-internal.aliyuncs.com".to_string(), + ), + (S3_ACCESS_KEY_ID.to_string(), "STS.key".to_string()), + (S3_SECRET_ACCESS_KEY.to_string(), "secret".to_string()), + (S3_SESSION_TOKEN.to_string(), "token".to_string()), + ]); + + let storage = build_storage_for_scheme("oss", &props, &None).unwrap(); + assert!(matches!(storage, OpenDalStorage::S3 { .. })); + } + + #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] + #[test] + fn oss_needs_an_s3_endpoint_to_route_through_s3() { + use iceberg::io::S3_ACCESS_KEY_ID; + + // Credentials alone are not enough: opendal would default to + // `s3.amazonaws.com` and send Aliyun keys to AWS. + let props = HashMap::from([(S3_ACCESS_KEY_ID.to_string(), "AKIA_TEST".to_string())]); + + let storage = build_storage_for_scheme("oss", &props, &None).unwrap(); + assert!(matches!(storage, OpenDalStorage::Oss { .. })); + } + #[cfg(all(feature = "opendal-oss", feature = "opendal-s3"))] #[test] fn oss_falls_back_to_native_without_s3_config() { From e28deb46a04c465d3dafd005416ccb153f81d641 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Wed, 7 Oct 2026 22:25:42 -0400 Subject: [PATCH 4/4] fix(storage/opendal): carry the client config on the oss routes --- crates/storage/opendal/src/lib.rs | 2 ++ crates/storage/opendal/src/resolving.rs | 2 ++ 2 files changed, 4 insertions(+) diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 762e6b8a81..35718a3ed9 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -260,10 +260,12 @@ impl StorageFactory for OpenDalStorageFactory { Ok(Arc::new(OpenDalStorage::S3 { config: s3_config_parse(config.props().clone())?.into(), customized_credential_load: None, + client_config, })) } else { Ok(Arc::new(OpenDalStorage::Oss { config: oss_config_parse(config.props().clone())?.into(), + client_config, })) } } diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 067a558edc..b81f0aa3de 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -115,11 +115,13 @@ fn build_storage_for_scheme( Ok(OpenDalStorage::S3 { config: Arc::new(config), customized_credential_load: customized_credential_load.clone(), + client_config, }) } else { let config = crate::oss::oss_config_parse(props.clone())?; Ok(OpenDalStorage::Oss { config: Arc::new(config), + client_config, }) } }