diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 0ba3b6ed78..44dd2044a0 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1117,6 +1117,9 @@ pub const iceberg::io::S3_ENDPOINT: &str pub const iceberg::io::S3_MULTIPART_PART_SIZE_BYTES: &str pub const iceberg::io::S3_PATH_STYLE_ACCESS: &str pub const iceberg::io::S3_REGION: &str +pub const iceberg::io::S3_RETRY_MAX_WAIT_MS: &str +pub const iceberg::io::S3_RETRY_MIN_WAIT_MS: &str +pub const iceberg::io::S3_RETRY_NUM_RETRIES: &str pub const iceberg::io::S3_SECRET_ACCESS_KEY: &str pub const iceberg::io::S3_SESSION_TOKEN: &str pub const iceberg::io::S3_SSE_KEY: &str diff --git a/crates/iceberg/src/io/storage/config/s3.rs b/crates/iceberg/src/io/storage/config/s3.rs index 82f27e15dd..fcee3916e8 100644 --- a/crates/iceberg/src/io/storage/config/s3.rs +++ b/crates/iceberg/src/io/storage/config/s3.rs @@ -75,6 +75,13 @@ pub const S3_DISABLE_CONFIG_LOAD: &str = "s3.disable-config-load"; /// /// Only `iceberg-storage-opendal` reads this property, so [`S3Config`] has no matching field. pub const S3_MULTIPART_PART_SIZE_BYTES: &str = "s3.multipart.part-size-bytes"; +/// Number of times to retry an S3 operation, `0` to disable retrying. Retries back off exponentially, +/// with jitter, between [`S3_RETRY_MIN_WAIT_MS`] and [`S3_RETRY_MAX_WAIT_MS`]. +pub const S3_RETRY_NUM_RETRIES: &str = "s3.retry.num-retries"; +/// Minimum wait in milliseconds before retrying an S3 operation. +pub const S3_RETRY_MIN_WAIT_MS: &str = "s3.retry.min-wait-ms"; +/// Maximum wait in milliseconds between two retries of an S3 operation. +pub const S3_RETRY_MAX_WAIT_MS: &str = "s3.retry.max-wait-ms"; /// Amazon S3 storage configuration. /// diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index 03e38db6ab..ef24c0f820 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -130,3 +130,6 @@ pub fn iceberg_storage_opendal::OpenDalResolvingStorageFactory::serialize<__S>(& 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 +pub const iceberg_storage_opendal::OPENDAL_RETRY_MAX_WAIT_MS: &str +pub const iceberg_storage_opendal::OPENDAL_RETRY_MIN_WAIT_MS: &str +pub const iceberg_storage_opendal::OPENDAL_RETRY_NUM_RETRIES: &str diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index b1279b34b4..55b651ae1f 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -119,6 +119,18 @@ pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory}; /// affected and stays at its 60-second default. pub const OPENDAL_IO_TIMEOUT_MS: &str = "opendal.io-timeout-ms"; +/// Number of times to retry a failed operation on any backend, `0` to disable retrying. Unset keeps +/// OpenDAL's default. For S3, [`S3_RETRY_NUM_RETRIES`](iceberg::io::S3_RETRY_NUM_RETRIES) takes precedence. +pub const OPENDAL_RETRY_NUM_RETRIES: &str = "opendal.retry.num-retries"; + +/// Minimum wait in milliseconds before retrying a failed operation on any backend. Unset keeps +/// OpenDAL's default. For S3, [`S3_RETRY_MIN_WAIT_MS`](iceberg::io::S3_RETRY_MIN_WAIT_MS) takes precedence. +pub const OPENDAL_RETRY_MIN_WAIT_MS: &str = "opendal.retry.min-wait-ms"; + +/// Maximum wait in milliseconds between two retries on any backend. Unset keeps OpenDAL's default. +/// For S3, [`S3_RETRY_MAX_WAIT_MS`](iceberg::io::S3_RETRY_MAX_WAIT_MS) takes precedence. +pub const OPENDAL_RETRY_MAX_WAIT_MS: &str = "opendal.retry.max-wait-ms"; + /// Matches OpenDAL's `TimeoutLayer` default. const DEFAULT_IO_TIMEOUT_MS: NonZeroU64 = NonZeroU64::new(10_000).unwrap(); @@ -130,15 +142,35 @@ pub struct OpenDalClientConfig { #[property( key = OPENDAL_IO_TIMEOUT_MS, default = DEFAULT_IO_TIMEOUT_MS, - parse_with = parse_io_timeout_ms + parse_with = parse_positive_ms )] io_timeout_ms: NonZeroU64, + /// Retries on any backend. + #[property(key = OPENDAL_RETRY_NUM_RETRIES, default = None)] + retry_num_retries: Option, + /// Delay before the first retry on any backend in milliseconds. + #[property( + key = OPENDAL_RETRY_MIN_WAIT_MS, + default = None, + parse_with = parse_positive_ms + )] + retry_min_wait_ms: Option, + /// Upper bound of the delay between two retries on any backend in milliseconds. + #[property( + key = OPENDAL_RETRY_MAX_WAIT_MS, + default = None, + parse_with = parse_positive_ms + )] + retry_max_wait_ms: Option, } impl Default for OpenDalClientConfig { fn default() -> Self { Self { io_timeout_ms: DEFAULT_IO_TIMEOUT_MS, + retry_num_retries: None, + retry_min_wait_ms: None, + retry_max_wait_ms: None, } } } @@ -147,9 +179,30 @@ impl OpenDalClientConfig { pub(crate) fn io_timeout(&self) -> Duration { Duration::from_millis(self.io_timeout_ms.get()) } + + /// The retry policy, with jitter; OpenDAL's default when no key is set. + pub(crate) fn retry_layer(&self) -> RetryLayer { + let mut layer = RetryLayer::new(); + if self.retry_num_retries.is_none() + && self.retry_min_wait_ms.is_none() + && self.retry_max_wait_ms.is_none() + { + return layer; + } + if let Some(times) = self.retry_num_retries { + layer = layer.with_max_times(times); + } + if let Some(ms) = self.retry_min_wait_ms { + layer = layer.with_min_delay(Duration::from_millis(ms.get())); + } + if let Some(ms) = self.retry_max_wait_ms { + layer = layer.with_max_delay(Duration::from_millis(ms.get())); + } + layer.with_jitter() + } } -fn parse_io_timeout_ms(value: &str) -> Result { +fn parse_positive_ms(value: &str) -> Result { value.parse().map_err(|error| { Error::new( ErrorKind::DataInvalid, @@ -242,7 +295,7 @@ impl StorageFactory for OpenDalStorageFactory { config: s3_config_parse(config.props().clone())?.into(), multipart_part_size: s3_multipart_part_size_parse(config.props())?, customized_credential_load: customized_credential_load.clone(), - client_config, + client_config: s3_client_config(config.props(), client_config)?, })), #[cfg(feature = "opendal-gcs")] OpenDalStorageFactory::Gcs => Ok(Arc::new(OpenDalStorage::Gcs { @@ -513,7 +566,7 @@ impl OpenDalStorage { // benefits non-object-store backends. let operator = operator .layer(TimeoutLayer::new().with_io_timeout(self.client_config().io_timeout())) - .layer(RetryLayer::new()); + .layer(self.client_config().retry_layer()); Ok((operator, relative_path)) } @@ -927,6 +980,72 @@ mod tests { ); } + fn retry_config(props: &[(&str, &str)]) -> Result { + OpenDalClientConfig::from_properties( + &props + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect::>(), + ) + } + + fn retry_layer_debug(props: &[(&str, &str)]) -> Result { + Ok(format!("{:?}", retry_config(props)?.retry_layer())) + } + + #[test] + fn test_retry_defaults_match_opendal() { + // `RetryLayer` has no getters, so compare through `Debug`. + assert_eq!( + retry_layer_debug(&[]).unwrap(), + format!("{:?}", RetryLayer::new()) + ); + assert_ne!( + format!("{:?}", RetryLayer::new()), + format!("{:?}", RetryLayer::new().with_jitter()) + ); + } + + #[test] + fn test_retry_properties_reach_retry_layer() { + let expected = RetryLayer::new() + .with_max_times(5) + .with_min_delay(Duration::from_millis(100)) + .with_max_delay(Duration::from_millis(200)) + .with_jitter(); + assert_eq!( + retry_layer_debug(&[ + (OPENDAL_RETRY_NUM_RETRIES, "5"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + ]) + .unwrap(), + format!("{expected:?}") + ); + + // any one property switches jitter on and leaves the others at OpenDAL's defaults + assert_eq!( + retry_layer_debug(&[(OPENDAL_RETRY_NUM_RETRIES, "0")]).unwrap(), + format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter()) + ); + } + + #[test] + fn test_retry_property_parsing_rejects_invalid_values() { + for (key, invalid) in [ + (OPENDAL_RETRY_NUM_RETRIES, "-1"), + (OPENDAL_RETRY_NUM_RETRIES, "abc"), + (OPENDAL_RETRY_NUM_RETRIES, ""), + (OPENDAL_RETRY_MIN_WAIT_MS, "0"), + (OPENDAL_RETRY_MIN_WAIT_MS, "abc"), + (OPENDAL_RETRY_MAX_WAIT_MS, "0"), + (OPENDAL_RETRY_MAX_WAIT_MS, "12.5"), + ] { + let err = retry_config(&[(key, invalid)]).unwrap_err().to_string(); + assert!(err.contains(key), "{key}={invalid}: {err}"); + } + } + #[cfg(feature = "opendal-s3")] #[test] fn test_client_config_serde_round_trip() { diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 9123022de3..47459c42c8 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -97,7 +97,7 @@ fn build_storage_for_scheme( config: Arc::new(config), multipart_part_size: crate::s3::s3_multipart_part_size_parse(props)?, customized_credential_load: customized_credential_load.clone(), - client_config, + client_config: crate::s3::s3_client_config(props, client_config)?, }) } #[cfg(feature = "opendal-gcs")] diff --git a/crates/storage/opendal/src/s3.rs b/crates/storage/opendal/src/s3.rs index 8f1828dc46..b7e0166185 100644 --- a/crates/storage/opendal/src/s3.rs +++ b/crates/storage/opendal/src/s3.rs @@ -16,15 +16,18 @@ // under the License. use std::collections::HashMap; +use std::num::NonZeroU64; use std::sync::Arc; use iceberg::io::{ CLIENT_REGION, S3_ACCESS_KEY_ID, S3_ALLOW_ANONYMOUS, S3_ASSUME_ROLE_ARN, S3_ASSUME_ROLE_EXTERNAL_ID, S3_ASSUME_ROLE_SESSION_NAME, S3_DISABLE_CONFIG_LOAD, S3_DISABLE_EC2_METADATA, S3_ENDPOINT, S3_MULTIPART_PART_SIZE_BYTES, S3_PATH_STYLE_ACCESS, - S3_REGION, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SSE_KEY, S3_SSE_MD5, S3_SSE_TYPE, + S3_REGION, S3_RETRY_MAX_WAIT_MS, S3_RETRY_MIN_WAIT_MS, S3_RETRY_NUM_RETRIES, + S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN, S3_SSE_KEY, S3_SSE_MD5, S3_SSE_TYPE, }; use iceberg::{Error, ErrorKind, Result}; +use iceberg_property_macro::Properties; use opendal::services::S3Config; use opendal::{Configurator, Operator}; /// AWS credentials: access key ID, secret access key, and optional session token. @@ -35,6 +38,29 @@ use reqsign_core::{ProvideCredentialChain, ProvideCredentialDyn}; use url::Url; use crate::utils::{from_opendal_error, is_truthy}; +use crate::{OpenDalClientConfig, parse_positive_ms}; + +#[derive(Properties)] +struct S3RetryConfig { + #[property(key = S3_RETRY_NUM_RETRIES, default = None)] + num_retries: Option, + #[property(key = S3_RETRY_MIN_WAIT_MS, default = None, parse_with = parse_positive_ms)] + min_wait_ms: Option, + #[property(key = S3_RETRY_MAX_WAIT_MS, default = None, parse_with = parse_positive_ms)] + max_wait_ms: Option, +} + +/// Overlays the `s3.retry.*` properties onto the `opendal.retry.*` settings of `client_config`, per key. +pub(crate) fn s3_client_config( + props: &HashMap, + mut client_config: OpenDalClientConfig, +) -> Result { + let retry = S3RetryConfig::from_properties(props)?; + client_config.retry_num_retries = retry.num_retries.or(client_config.retry_num_retries); + client_config.retry_min_wait_ms = retry.min_wait_ms.or(client_config.retry_min_wait_ms); + client_config.retry_max_wait_ms = retry.max_wait_ms.or(client_config.retry_max_wait_ms); + Ok(client_config) +} /// S3 rejects a non-final part smaller than this. const MULTIPART_PART_SIZE_MIN: u64 = 5 * 1024 * 1024; @@ -227,12 +253,22 @@ impl CustomAwsCredentialLoader { #[cfg(test)] mod tests { use std::collections::HashMap; + use std::time::Duration; - use iceberg::io::{S3_MULTIPART_PART_SIZE_BYTES, S3_PATH_STYLE_ACCESS}; + use iceberg::Result; + use iceberg::io::{ + S3_MULTIPART_PART_SIZE_BYTES, S3_PATH_STYLE_ACCESS, S3_RETRY_MAX_WAIT_MS, + S3_RETRY_MIN_WAIT_MS, S3_RETRY_NUM_RETRIES, + }; + use opendal::layers::RetryLayer; use super::{ - MULTIPART_PART_SIZE_DEFAULT, MULTIPART_PART_SIZE_MAX, MULTIPART_PART_SIZE_MIN, Result, - s3_config_parse, s3_multipart_part_size_parse, + MULTIPART_PART_SIZE_DEFAULT, MULTIPART_PART_SIZE_MAX, MULTIPART_PART_SIZE_MIN, + s3_client_config, s3_config_parse, s3_multipart_part_size_parse, + }; + use crate::{ + OPENDAL_RETRY_MAX_WAIT_MS, OPENDAL_RETRY_MIN_WAIT_MS, OPENDAL_RETRY_NUM_RETRIES, + OpenDalClientConfig, }; fn parse_with(prop: Option<&str>) -> bool { @@ -277,4 +313,110 @@ mod tests { assert!(parse_part_size(Some(&(MULTIPART_PART_SIZE_MAX + 1).to_string())).is_err()); assert!(parse_part_size(Some("not-a-number")).is_err()); } + + fn s3_retry_layer(props: &[(&str, &str)]) -> Result { + let props = props + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect::>(); + let client_config = + s3_client_config(&props, OpenDalClientConfig::from_properties(&props)?)?; + Ok(format!("{:?}", client_config.retry_layer())) + } + + #[test] + fn test_s3_retry_properties_reach_retry_layer() { + let expected = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_secs(2)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + s3_retry_layer(&[ + (S3_RETRY_NUM_RETRIES, "10"), + (S3_RETRY_MIN_WAIT_MS, "2000"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(), + format!("{expected:?}") + ); + assert_eq!( + s3_retry_layer(&[(S3_RETRY_NUM_RETRIES, "0")]).unwrap(), + format!("{:?}", RetryLayer::new().with_max_times(0).with_jitter()) + ); + assert_eq!( + s3_retry_layer(&[]).unwrap(), + format!("{:?}", RetryLayer::new()) + ); + } + + #[test] + fn test_s3_retry_properties_win_over_the_generic_ones_per_key() { + let expected = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_secs(2)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + s3_retry_layer(&[ + (OPENDAL_RETRY_NUM_RETRIES, "5"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + (S3_RETRY_NUM_RETRIES, "10"), + (S3_RETRY_MIN_WAIT_MS, "2000"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(), + format!("{expected:?}") + ); + + // a key set at one layer only is taken from that layer + let mixed = RetryLayer::new() + .with_max_times(10) + .with_min_delay(Duration::from_millis(100)) + .with_max_delay(Duration::from_secs(20)) + .with_jitter(); + assert_eq!( + s3_retry_layer(&[ + (S3_RETRY_NUM_RETRIES, "10"), + (OPENDAL_RETRY_MIN_WAIT_MS, "100"), + (OPENDAL_RETRY_MAX_WAIT_MS, "200"), + (S3_RETRY_MAX_WAIT_MS, "20000"), + ]) + .unwrap(), + format!("{mixed:?}") + ); + } + + #[test] + fn test_s3_retry_property_parsing_rejects_invalid_values() { + for (key, invalid) in [ + (S3_RETRY_NUM_RETRIES, "-1"), + (S3_RETRY_NUM_RETRIES, "abc"), + (S3_RETRY_NUM_RETRIES, ""), + (S3_RETRY_MIN_WAIT_MS, "0"), + (S3_RETRY_MIN_WAIT_MS, "abc"), + (S3_RETRY_MAX_WAIT_MS, "0"), + (S3_RETRY_MAX_WAIT_MS, "12.5"), + ] { + let err = s3_retry_layer(&[(key, invalid)]).unwrap_err().to_string(); + assert!(err.contains(key), "{key}={invalid}: {err}"); + } + } + + #[cfg(feature = "opendal-memory")] + #[test] + fn test_other_backends_ignore_the_s3_retry_properties() { + use iceberg::io::{StorageConfig, StorageFactory}; + + use crate::OpenDalStorageFactory; + + let config = StorageConfig::new().with_props([ + (S3_RETRY_NUM_RETRIES.to_string(), "abc".to_string()), + (OPENDAL_RETRY_NUM_RETRIES.to_string(), "5".to_string()), + ]); + let storage = OpenDalStorageFactory::Memory.build(&config).unwrap(); + let json = serde_json::to_string(&storage).unwrap(); + assert!(json.contains(r#""retry_num_retries":5"#), "{json}"); + } }