diff --git a/Cargo.lock b/Cargo.lock index 64c5c61875..8b5bb449e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2935,12 +2935,14 @@ dependencies = [ "cfg-if 1.0.4", "futures", "iceberg", + "iceberg-property-macro", "iceberg_test_utils", "opendal", "reqsign-aws-v4", "reqsign-core", "reqwest 0.12.28", "serde", + "serde_json", "tempfile", "tokio", "typetag", diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index c6ea9f9b67..d3403598e8 100644 --- a/crates/storage/opendal/Cargo.toml +++ b/crates/storage/opendal/Cargo.toml @@ -54,6 +54,7 @@ bytes = { workspace = true } cfg-if = { workspace = true } futures = { workspace = true } iceberg = { workspace = true } +iceberg-property-macro = { workspace = true } opendal = { workspace = true } reqsign-aws-v4 = { version = "3.0.0", optional = true } reqsign-core = { version = "3.0.0", optional = true } @@ -65,8 +66,9 @@ url = { workspace = true } async-trait = { workspace = true } iceberg_test_utils = { path = "../../test_utils", features = ["tests"] } reqwest = { workspace = true } +serde_json = { workspace = true } tempfile = { workspace = true } -tokio = { workspace = true, features = ["macros"] } +tokio = { workspace = true, features = ["macros", "test-util"] } [lints] workspace = true diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index d8c4ecdb38..c2e310335c 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -3,16 +3,24 @@ pub use iceberg_storage_opendal::AwsCredential pub use iceberg_storage_opendal::ProvideCredential pub enum iceberg_storage_opendal::OpenDalStorage pub iceberg_storage_opendal::OpenDalStorage::Azdls +pub iceberg_storage_opendal::OpenDalStorage::Azdls::client_config: iceberg_storage_opendal::OpenDalClientConfig pub iceberg_storage_opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::Gcs +pub iceberg_storage_opendal::OpenDalStorage::Gcs::client_config: iceberg_storage_opendal::OpenDalClientConfig pub iceberg_storage_opendal::OpenDalStorage::Gcs::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::Hf +pub iceberg_storage_opendal::OpenDalStorage::Hf::client_config: iceberg_storage_opendal::OpenDalClientConfig pub iceberg_storage_opendal::OpenDalStorage::Hf::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::LocalFs -pub iceberg_storage_opendal::OpenDalStorage::Memory(opendal_core::types::operator::operator::Operator) +pub iceberg_storage_opendal::OpenDalStorage::LocalFs::client_config: iceberg_storage_opendal::OpenDalClientConfig +pub iceberg_storage_opendal::OpenDalStorage::Memory +pub iceberg_storage_opendal::OpenDalStorage::Memory::client_config: iceberg_storage_opendal::OpenDalClientConfig +pub iceberg_storage_opendal::OpenDalStorage::Memory::operator: opendal_core::types::operator::operator::Operator pub iceberg_storage_opendal::OpenDalStorage::Oss +pub iceberg_storage_opendal::OpenDalStorage::Oss::client_config: iceberg_storage_opendal::OpenDalClientConfig pub iceberg_storage_opendal::OpenDalStorage::Oss::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::S3 +pub iceberg_storage_opendal::OpenDalStorage::S3::client_config: iceberg_storage_opendal::OpenDalClientConfig pub iceberg_storage_opendal::OpenDalStorage::S3::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::S3::customized_credential_load: core::option::Option impl core::clone::Clone for iceberg_storage_opendal::OpenDalStorage @@ -61,6 +69,19 @@ impl core::clone::Clone for iceberg_storage_opendal::CustomAwsCredentialLoader pub fn iceberg_storage_opendal::CustomAwsCredentialLoader::clone(&self) -> Self impl core::fmt::Debug for iceberg_storage_opendal::CustomAwsCredentialLoader pub fn iceberg_storage_opendal::CustomAwsCredentialLoader::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +pub struct iceberg_storage_opendal::OpenDalClientConfig +impl iceberg_storage_opendal::OpenDalClientConfig +pub fn iceberg_storage_opendal::OpenDalClientConfig::from_properties(properties: &std::collections::hash::map::HashMap) -> iceberg::error::Result +impl core::clone::Clone for iceberg_storage_opendal::OpenDalClientConfig +pub fn iceberg_storage_opendal::OpenDalClientConfig::clone(&self) -> iceberg_storage_opendal::OpenDalClientConfig +impl core::default::Default for iceberg_storage_opendal::OpenDalClientConfig +pub fn iceberg_storage_opendal::OpenDalClientConfig::default() -> Self +impl core::fmt::Debug for iceberg_storage_opendal::OpenDalClientConfig +pub fn iceberg_storage_opendal::OpenDalClientConfig::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl serde_core::ser::Serialize for iceberg_storage_opendal::OpenDalClientConfig +pub fn iceberg_storage_opendal::OpenDalClientConfig::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_opendal::OpenDalClientConfig where iceberg_storage_opendal::OpenDalClientConfig: core::default::Default +pub fn iceberg_storage_opendal::OpenDalClientConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg_storage_opendal::OpenDalResolvingStorage impl core::fmt::Debug for iceberg_storage_opendal::OpenDalResolvingStorage pub fn iceberg_storage_opendal::OpenDalResolvingStorage::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result @@ -96,3 +117,4 @@ impl serde_core::ser::Serialize for iceberg_storage_opendal::OpenDalResolvingSto pub fn iceberg_storage_opendal::OpenDalResolvingStorageFactory::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_opendal::OpenDalResolvingStorageFactory pub fn iceberg_storage_opendal::OpenDalResolvingStorageFactory::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> +pub const iceberg_storage_opendal::OPENDAL_IO_TIMEOUT_MS: &str diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index fe485cb915..f58d6f534c 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -26,7 +26,9 @@ mod utils; use std::collections::HashMap; use std::collections::hash_map::Entry; +use std::num::NonZeroU64; use std::sync::Arc; +use std::time::Duration; use async_trait::async_trait; use bytes::Bytes; @@ -38,6 +40,7 @@ use iceberg::io::{ StorageFactory, }; use iceberg::{Error, ErrorKind, Result}; +use iceberg_property_macro::Properties; use opendal::Operator; use opendal::layers::{RetryLayer, TimeoutLayer}; use serde::{Deserialize, Serialize}; @@ -100,6 +103,54 @@ cfg_if! { mod resolving; pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory}; +/// Timeout in milliseconds for each IO call on a reader, writer, lister or deleter, applied per +/// retry attempt via OpenDAL's `TimeoutLayer::with_io_timeout`. Defaults to 10 seconds. +/// +/// OpenDAL's separate `timeout`, which bounds whole control operations such as `stat`, is not +/// affected and stays at its 60-second default. +pub const OPENDAL_IO_TIMEOUT_MS: &str = "opendal.io-timeout-ms"; + +/// Matches OpenDAL's `TimeoutLayer` default. +const DEFAULT_IO_TIMEOUT_MS: NonZeroU64 = NonZeroU64::new(10_000).unwrap(); + +/// Backend-independent client settings shared by every [`OpenDalStorage`] variant. +#[derive(Clone, Debug, Properties, Serialize, Deserialize)] +#[serde(default)] +pub struct OpenDalClientConfig { + /// IO timeout in milliseconds. + #[property( + key = OPENDAL_IO_TIMEOUT_MS, + default = DEFAULT_IO_TIMEOUT_MS, + parse_with = parse_io_timeout_ms + )] + io_timeout_ms: NonZeroU64, +} + +impl Default for OpenDalClientConfig { + fn default() -> Self { + Self { + io_timeout_ms: DEFAULT_IO_TIMEOUT_MS, + } + } +} + +impl OpenDalClientConfig { + pub(crate) fn io_timeout(&self) -> Duration { + Duration::from_millis(self.io_timeout_ms.get()) + } +} + +fn parse_io_timeout_ms(value: &str) -> Result { + value.parse().map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + "Expected a positive integer number of milliseconds", + ) + .with_context("value", format!("{value:?}")) + .with_source(error) + }) +} + /// OpenDAL-based storage factory. /// /// Maps scheme to the corresponding OpenDalStorage storage variant. @@ -163,35 +214,42 @@ where impl StorageFactory for OpenDalStorageFactory { #[allow(unused_variables)] fn build(&self, config: &StorageConfig) -> Result> { + let client_config = OpenDalClientConfig::from_properties(config.props())?; match self { #[cfg(feature = "opendal-memory")] - OpenDalStorageFactory::Memory => { - Ok(Arc::new(OpenDalStorage::Memory(memory_config_build()?))) - } + OpenDalStorageFactory::Memory => Ok(Arc::new(OpenDalStorage::Memory { + operator: memory_config_build()?, + client_config, + })), #[cfg(feature = "opendal-fs")] - OpenDalStorageFactory::Fs => Ok(Arc::new(OpenDalStorage::LocalFs)), + OpenDalStorageFactory::Fs => Ok(Arc::new(OpenDalStorage::LocalFs { client_config })), #[cfg(feature = "opendal-s3")] OpenDalStorageFactory::S3 { customized_credential_load, } => Ok(Arc::new(OpenDalStorage::S3 { config: s3_config_parse(config.props().clone())?.into(), customized_credential_load: customized_credential_load.clone(), + client_config, })), #[cfg(feature = "opendal-gcs")] OpenDalStorageFactory::Gcs => Ok(Arc::new(OpenDalStorage::Gcs { config: gcs_config_parse(config.props().clone())?.into(), + client_config, })), #[cfg(feature = "opendal-oss")] OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::Oss { config: oss_config_parse(config.props().clone())?.into(), + client_config, })), #[cfg(feature = "opendal-azdls")] OpenDalStorageFactory::Azdls => Ok(Arc::new(OpenDalStorage::Azdls { config: azdls_config_parse(config.props().clone())?.into(), + client_config, })), #[cfg(feature = "opendal-hf")] OpenDalStorageFactory::Hf => Ok(Arc::new(OpenDalStorage::Hf { config: hf_config_parse(config.props().clone())?.into(), + client_config, })), #[cfg(all( not(feature = "opendal-memory"), @@ -217,14 +275,27 @@ fn default_memory_operator() -> Operator { } /// OpenDAL-based storage implementation. +/// +/// The serialized representation is not a stable format and may change between crate versions. #[derive(Clone, Debug, Serialize, Deserialize)] pub enum OpenDalStorage { /// Memory storage variant. #[cfg(feature = "opendal-memory")] - Memory(#[serde(skip, default = "self::default_memory_operator")] Operator), + Memory { + /// Pre-built memory operator. + #[serde(skip, default = "self::default_memory_operator")] + operator: Operator, + /// Backend-independent client settings. + #[serde(default)] + client_config: OpenDalClientConfig, + }, /// Local filesystem storage variant. #[cfg(feature = "opendal-fs")] - LocalFs, + LocalFs { + /// Backend-independent client settings. + #[serde(default)] + client_config: OpenDalClientConfig, + }, /// S3 storage variant. /// /// Accepts any S3-family URL (`s3://`, `s3a://`, `s3n://`); the scheme is @@ -236,18 +307,27 @@ pub enum OpenDalStorage { /// Custom AWS credential loader. #[serde(skip)] customized_credential_load: Option, + /// Backend-independent client settings. + #[serde(default)] + client_config: OpenDalClientConfig, }, /// GCS storage variant. #[cfg(feature = "opendal-gcs")] Gcs { /// GCS configuration. config: Arc, + /// Backend-independent client settings. + #[serde(default)] + client_config: OpenDalClientConfig, }, /// OSS storage variant. #[cfg(feature = "opendal-oss")] Oss { /// OSS configuration. config: Arc, + /// Backend-independent client settings. + #[serde(default)] + client_config: OpenDalClientConfig, }, /// Azure Data Lake Storage variant. /// @@ -259,6 +339,9 @@ pub enum OpenDalStorage { Azdls { /// Azure DLS configuration. config: Arc, + /// Backend-independent client settings. + #[serde(default)] + client_config: OpenDalClientConfig, }, /// HuggingFace Hub storage variant. /// @@ -269,6 +352,9 @@ pub enum OpenDalStorage { Hf { /// HuggingFace Hub configuration (token + endpoint). config: Arc, + /// Backend-independent client settings. + #[serde(default)] + client_config: OpenDalClientConfig, }, } @@ -293,7 +379,7 @@ impl OpenDalStorage { let path = path.as_ref(); let (operator, relative_path): (Operator, &str) = match self { #[cfg(feature = "opendal-memory")] - OpenDalStorage::Memory(op) => { + OpenDalStorage::Memory { operator: op, .. } => { if let Some(stripped) = path.strip_prefix("memory:/") { (op.clone(), stripped) } else { @@ -301,7 +387,7 @@ impl OpenDalStorage { } } #[cfg(feature = "opendal-fs")] - OpenDalStorage::LocalFs => { + OpenDalStorage::LocalFs { .. } => { let op = fs_config_build()?; if let Some(stripped) = path.strip_prefix("file:/") { (op, stripped) @@ -313,6 +399,7 @@ impl OpenDalStorage { OpenDalStorage::S3 { config, customized_credential_load, + .. } => { let op = s3_config_build(config, customized_credential_load, path)?; let op_info = op.info(); @@ -336,7 +423,7 @@ impl OpenDalStorage { } } #[cfg(feature = "opendal-gcs")] - OpenDalStorage::Gcs { config } => { + OpenDalStorage::Gcs { config, .. } => { let operator = gcs_config_build(config, path)?; let prefix = format!("gs://{}/", operator.info().name()); if path.starts_with(&prefix) { @@ -349,7 +436,7 @@ impl OpenDalStorage { } } #[cfg(feature = "opendal-oss")] - OpenDalStorage::Oss { config } => { + OpenDalStorage::Oss { config, .. } => { let op = oss_config_build(config, path)?; let prefix = format!("oss://{}/", op.info().name()); if path.starts_with(&prefix) { @@ -362,9 +449,9 @@ impl OpenDalStorage { } } #[cfg(feature = "opendal-azdls")] - OpenDalStorage::Azdls { config } => azdls_create_operator(path, config)?, + OpenDalStorage::Azdls { config, .. } => azdls_create_operator(path, config)?, #[cfg(feature = "opendal-hf")] - OpenDalStorage::Hf { config } => hf_config_build(config, path)?, + OpenDalStorage::Hf { config, .. } => hf_config_build(config, path)?, #[cfg(all( not(feature = "opendal-s3"), not(feature = "opendal-fs"), @@ -391,10 +478,41 @@ impl OpenDalStorage { // Transient errors are common for object stores; we retry temporary // failures with exponential backoff. The retry behavior also // benefits non-object-store backends. - let operator = operator.layer(TimeoutLayer::new()).layer(RetryLayer::new()); + let operator = operator + .layer(TimeoutLayer::new().with_io_timeout(self.client_config().io_timeout())) + .layer(RetryLayer::new()); Ok((operator, relative_path)) } + pub(crate) fn client_config(&self) -> &OpenDalClientConfig { + match self { + #[cfg(feature = "opendal-memory")] + OpenDalStorage::Memory { client_config, .. } => client_config, + #[cfg(feature = "opendal-fs")] + OpenDalStorage::LocalFs { client_config } => client_config, + #[cfg(feature = "opendal-s3")] + OpenDalStorage::S3 { client_config, .. } => client_config, + #[cfg(feature = "opendal-gcs")] + OpenDalStorage::Gcs { client_config, .. } => client_config, + #[cfg(feature = "opendal-oss")] + OpenDalStorage::Oss { client_config, .. } => client_config, + #[cfg(feature = "opendal-azdls")] + OpenDalStorage::Azdls { client_config, .. } => client_config, + #[cfg(feature = "opendal-hf")] + OpenDalStorage::Hf { client_config, .. } => client_config, + #[cfg(all( + not(feature = "opendal-memory"), + not(feature = "opendal-s3"), + not(feature = "opendal-fs"), + not(feature = "opendal-gcs"), + not(feature = "opendal-oss"), + not(feature = "opendal-azdls"), + not(feature = "opendal-hf"), + ))] + _ => unreachable!(), + } + } + /// Returns a cache key used by `delete_stream` to group paths by storage operator. /// /// For most backends the URL host (bucket name) is sufficient. For HF the host @@ -419,9 +537,11 @@ impl OpenDalStorage { pub(crate) fn relativize_path<'a>(&self, path: &'a str) -> Result<&'a str> { match self { #[cfg(feature = "opendal-memory")] - OpenDalStorage::Memory(_) => Ok(path.strip_prefix("memory:/").unwrap_or(&path[1..])), + OpenDalStorage::Memory { .. } => { + Ok(path.strip_prefix("memory:/").unwrap_or(&path[1..])) + } #[cfg(feature = "opendal-fs")] - OpenDalStorage::LocalFs => Ok(path.strip_prefix("file:/").unwrap_or(&path[1..])), + OpenDalStorage::LocalFs { .. } => Ok(path.strip_prefix("file:/").unwrap_or(&path[1..])), #[cfg(feature = "opendal-s3")] OpenDalStorage::S3 { .. } => { let url = url::Url::parse(path)?; @@ -480,7 +600,7 @@ impl OpenDalStorage { } } #[cfg(feature = "opendal-azdls")] - OpenDalStorage::Azdls { config } => { + OpenDalStorage::Azdls { config, .. } => { let azure_path = path.parse::()?; match_path_with_config(&azure_path, config)?; let relative_path_len = azure_path.path.len(); @@ -687,6 +807,164 @@ impl FileWrite for OpenDalWriter { mod tests { use super::*; + fn client_config(value: &str) -> Result { + OpenDalClientConfig::from_properties(&HashMap::from([( + OPENDAL_IO_TIMEOUT_MS.to_string(), + value.to_string(), + )])) + } + + #[test] + fn test_io_timeout_parsing() { + let unset = OpenDalClientConfig::from_properties(&HashMap::new()).unwrap(); + assert_eq!( + unset.io_timeout(), + Duration::from_millis(DEFAULT_IO_TIMEOUT_MS.get()) + ); + + let max = u64::MAX.to_string(); + for (valid, ms) in [("45000", 45_000), ("1", 1), (max.as_str(), u64::MAX)] { + assert_eq!( + client_config(valid).unwrap().io_timeout(), + Duration::from_millis(ms), + "{valid}" + ); + } + + for invalid in ["0", "-1", "12.5", "abc", "", "18446744073709551616"] { + let err = client_config(invalid).unwrap_err().to_string(); + assert!(err.contains(OPENDAL_IO_TIMEOUT_MS), "{invalid}"); + assert!(err.contains(&format!("value: {invalid:?}")), "{err}"); + let reason = invalid.parse::().unwrap_err().to_string(); + assert!(err.contains(&reason), "{err}"); + } + } + + #[test] + fn test_default_timeouts_match_opendal() { + // `TimeoutLayer` has no getters, so compare through `Debug`. + let opendal_default = format!("{:?}", TimeoutLayer::new()); + let layer = |timeout, io_timeout| { + format!( + "{:?}", + TimeoutLayer::new() + .with_timeout(timeout) + .with_io_timeout(io_timeout) + ) + }; + let control = Duration::from_secs(60); + let io = Duration::from_millis(DEFAULT_IO_TIMEOUT_MS.get()); + assert_eq!(opendal_default, layer(control, io)); + assert_ne!(opendal_default, layer(control + Duration::from_secs(1), io)); + assert_ne!( + opendal_default, + layer(control, io + Duration::from_millis(1)) + ); + } + + #[cfg(feature = "opendal-s3")] + #[test] + fn test_client_config_serde_round_trip() { + let storage = OpenDalStorage::S3 { + config: Arc::new(S3Config::default()), + customized_credential_load: None, + client_config: client_config("45000").unwrap(), + }; + + let mut value = serde_json::to_value(&storage).unwrap(); + let restored: OpenDalStorage = serde_json::from_value(value.clone()).unwrap(); + assert_eq!( + restored.client_config().io_timeout(), + Duration::from_secs(45) + ); + + let mut zero = value.clone(); + zero["S3"]["client_config"]["io_timeout_ms"] = 0.into(); + assert!(serde_json::from_value::(zero).is_err()); + + value["S3"].as_object_mut().unwrap().remove("client_config"); + let restored: OpenDalStorage = serde_json::from_value(value).unwrap(); + assert_eq!( + restored.client_config().io_timeout(), + Duration::from_millis(DEFAULT_IO_TIMEOUT_MS.get()) + ); + } + + #[cfg(all(feature = "opendal-fs", feature = "opendal-memory"))] + #[test] + fn test_old_unit_variant_forms_are_rejected() { + for old in [r#""LocalFs""#, r#""Memory""#, r#"{"Memory":null}"#] { + assert!( + serde_json::from_str::(old).is_err(), + "{old}" + ); + } + } + + #[cfg(feature = "opendal-memory")] + #[test] + fn test_factory_rejects_invalid_io_timeout() { + let config = StorageConfig::new().with_prop(OPENDAL_IO_TIMEOUT_MS, "nope"); + + let err = OpenDalStorageFactory::Memory.build(&config).unwrap_err(); + assert!(err.to_string().contains(OPENDAL_IO_TIMEOUT_MS)); + } + + #[test] + fn test_factory_propagates_io_timeout() { + let config = StorageConfig::new().with_prop(OPENDAL_IO_TIMEOUT_MS, "45000"); + let factories = [ + #[cfg(feature = "opendal-memory")] + OpenDalStorageFactory::Memory, + #[cfg(feature = "opendal-fs")] + OpenDalStorageFactory::Fs, + #[cfg(feature = "opendal-s3")] + OpenDalStorageFactory::S3 { + customized_credential_load: None, + }, + #[cfg(feature = "opendal-gcs")] + OpenDalStorageFactory::Gcs, + #[cfg(feature = "opendal-oss")] + OpenDalStorageFactory::Oss, + #[cfg(feature = "opendal-azdls")] + OpenDalStorageFactory::Azdls, + #[cfg(feature = "opendal-hf")] + OpenDalStorageFactory::Hf, + ]; + for factory in factories { + // `build` returns `dyn Storage`, so read the config back from its serialized form. + let storage = factory.build(&config).unwrap(); + let json = serde_json::to_value(&*storage).unwrap(); + let client_config = json + .as_object() + .unwrap() + .values() + .find_map(|variant| variant.get("client_config")) + .unwrap_or_else(|| panic!("{json}")); + assert_eq!(client_config["io_timeout_ms"], 45_000, "{factory:?}"); + } + } + + #[cfg(feature = "opendal-memory")] + #[tokio::test(start_paused = true)] + async fn test_io_timeout_reaches_timeout_layer() { + use opendal::layers::ConcurrentLimitLayer; + + // A zero concurrency limit stalls every IO call; paused time skips the waits. + let storage = OpenDalStorage::Memory { + operator: default_memory_operator().layer(ConcurrentLimitLayer::new(0)), + client_config: client_config("45000").unwrap(), + }; + + let err = storage + .read("memory:/stalled") + .await + .unwrap_err() + .to_string(); + assert!(err.contains("io operation timeout reached"), "{err}"); + assert!(err.contains("{ timeout: 45 }"), "{err}"); + } + #[cfg(feature = "opendal-s3")] #[derive(Debug)] struct EmptyCredentialLoader; @@ -732,7 +1010,10 @@ mod tests { // Note: the memory service does report a content length, so this only pins the happy // path. The counter in `OpenDalWriter` is what covers services that don't, such as S3. - let storage = Arc::new(OpenDalStorage::Memory(default_memory_operator())); + let storage = Arc::new(OpenDalStorage::Memory { + operator: default_memory_operator(), + client_config: OpenDalClientConfig::default(), + }); let path = "memory:///stored-size"; for plaintext in [ Bytes::new(), @@ -760,7 +1041,10 @@ mod tests { #[cfg(feature = "opendal-memory")] #[test] fn test_relativize_path_memory() { - let storage = OpenDalStorage::Memory(default_memory_operator()); + let storage = OpenDalStorage::Memory { + operator: default_memory_operator(), + client_config: OpenDalClientConfig::default(), + }; assert_eq!( storage.relativize_path("memory:/path/to/file").unwrap(), @@ -776,7 +1060,9 @@ mod tests { #[cfg(feature = "opendal-fs")] #[test] fn test_relativize_path_fs() { - let storage = OpenDalStorage::LocalFs; + let storage = OpenDalStorage::LocalFs { + client_config: OpenDalClientConfig::default(), + }; assert_eq!( storage @@ -796,6 +1082,7 @@ mod tests { let storage = OpenDalStorage::S3 { config: Arc::new(S3Config::default()), customized_credential_load: None, + client_config: OpenDalClientConfig::default(), }; // All S3-family schemes are accepted by the same storage instance. @@ -816,6 +1103,7 @@ mod tests { fn test_relativize_path_gcs() { let storage = OpenDalStorage::Gcs { config: Arc::new(GcsConfig::default()), + client_config: OpenDalClientConfig::default(), }; assert_eq!( @@ -831,6 +1119,7 @@ mod tests { fn test_relativize_path_gcs_invalid_scheme() { let storage = OpenDalStorage::Gcs { config: Arc::new(GcsConfig::default()), + client_config: OpenDalClientConfig::default(), }; assert!( @@ -845,6 +1134,7 @@ mod tests { fn test_relativize_path_oss() { let storage = OpenDalStorage::Oss { config: Arc::new(OssConfig::default()), + client_config: OpenDalClientConfig::default(), }; assert_eq!( @@ -860,6 +1150,7 @@ mod tests { fn test_relativize_path_oss_invalid_scheme() { let storage = OpenDalStorage::Oss { config: Arc::new(OssConfig::default()), + client_config: OpenDalClientConfig::default(), }; assert!( @@ -878,6 +1169,7 @@ mod tests { endpoint: Some("https://myaccount.dfs.core.windows.net".to_string()), ..Default::default() }), + client_config: OpenDalClientConfig::default(), }; assert_eq!( diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 3b99b08fc0..51e76da472 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -33,9 +33,9 @@ use iceberg::{Error, ErrorKind, Result}; use serde::{Deserialize, Serialize}; use url::Url; -use crate::OpenDalStorage; #[cfg(feature = "opendal-s3")] use crate::s3::CustomAwsCredentialLoader; +use crate::{OpenDalClientConfig, OpenDalStorage}; /// Schemes supported by OpenDalResolvingStorage pub const SCHEME_MEMORY: &str = "memory"; @@ -86,6 +86,7 @@ fn build_storage_for_scheme( props: &HashMap, #[cfg(feature = "opendal-s3")] customized_credential_load: &Option, ) -> Result { + let client_config = OpenDalClientConfig::from_properties(props)?; match scheme { #[cfg(feature = "opendal-s3")] "s3" => { @@ -93,6 +94,7 @@ fn build_storage_for_scheme( Ok(OpenDalStorage::S3 { config: Arc::new(config), customized_credential_load: customized_credential_load.clone(), + client_config, }) } #[cfg(feature = "opendal-gcs")] @@ -100,6 +102,7 @@ fn build_storage_for_scheme( let config = crate::gcs::gcs_config_parse(props.clone())?; Ok(OpenDalStorage::Gcs { config: Arc::new(config), + client_config, }) } #[cfg(feature = "opendal-oss")] @@ -107,6 +110,7 @@ fn build_storage_for_scheme( let config = crate::oss::oss_config_parse(props.clone())?; Ok(OpenDalStorage::Oss { config: Arc::new(config), + client_config, }) } #[cfg(feature = "opendal-azdls")] @@ -114,17 +118,22 @@ fn build_storage_for_scheme( let config = crate::azdls::azdls_config_parse(props.clone())?; Ok(OpenDalStorage::Azdls { config: Arc::new(config), + client_config, }) } #[cfg(feature = "opendal-fs")] - "file" => Ok(OpenDalStorage::LocalFs), + "file" => Ok(OpenDalStorage::LocalFs { client_config }), #[cfg(feature = "opendal-memory")] - "memory" => Ok(OpenDalStorage::Memory(crate::memory::memory_config_build()?)), + "memory" => Ok(OpenDalStorage::Memory { + operator: crate::memory::memory_config_build()?, + client_config, + }), #[cfg(feature = "opendal-hf")] "hf" => { let config = crate::hf::hf_config_parse(props.clone())?; Ok(OpenDalStorage::Hf { config: Arc::new(config), + client_config, }) } unsupported => Err(Error::new( @@ -334,7 +343,10 @@ impl Storage for OpenDalResolvingStorage { #[cfg(test)] mod tests { + use std::time::Duration; + use super::*; + use crate::OPENDAL_IO_TIMEOUT_MS; #[cfg(feature = "opendal-s3")] #[derive(Debug)] @@ -377,6 +389,39 @@ mod tests { } } + #[test] + fn test_resolve_propagates_io_timeout() { + let mut storage = empty_resolving_storage(); + storage + .props + .insert(OPENDAL_IO_TIMEOUT_MS.to_string(), "45000".to_string()); + + let paths: &[&str] = &[ + #[cfg(feature = "opendal-memory")] + "memory:/key", + #[cfg(feature = "opendal-fs")] + "file:/key", + #[cfg(feature = "opendal-s3")] + "s3://bucket/key", + #[cfg(feature = "opendal-gcs")] + "gs://bucket/key", + #[cfg(feature = "opendal-oss")] + "oss://bucket/key", + #[cfg(feature = "opendal-azdls")] + "abfss://myfs@myaccount.dfs.core.windows.net/key", + #[cfg(feature = "opendal-hf")] + "hf://datasets/user/repo/key", + ]; + for path in paths { + let resolved = storage.resolve(path).unwrap(); + assert_eq!( + resolved.client_config().io_timeout(), + Duration::from_secs(45), + "{path}" + ); + } + } + #[cfg(feature = "opendal-s3")] #[test] fn test_resolve_s3_aliases_share_instance() {