mbutrovich commented on code in PR #3263:
URL: https://github.com/apache/iceberg-rust/pull/3263#discussion_r4135238489
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -687,6 +811,73 @@ impl FileWrite for OpenDalWriter {
mod tests {
use super::*;
+ fn client_config(value: &str) -> Result<OpenDalClientConfig> {
+ 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_ms(), DEFAULT_IO_TIMEOUT_MS);
+ assert_eq!(client_config("45000").unwrap().io_timeout_ms(), 45_000);
+
+ for invalid in ["0", "-1", "12.5", "abc", ""] {
Review Comment:
Could this also cover the edges of the valid range? `1` is the smallest
accepted value, `u64::MAX` is the largest, and `18446744073709551616`
(`u64::MAX + 1`) should be rejected. On the head commit, `u64::MAX` parses, and
a write and read through `OpenDalStorage::Memory` configured with it both
succeed, because tokio's `timeout` falls back to a far-future deadline when
`Instant + Duration` overflows. `u64::MAX + 1` fails with `DataInvalid` and the
property key in the context. A test would keep both behaviors from regressing.
The suggestion is against the current `u64` field, so it would need `.get()` if
the field becomes `NonZeroU64`.
```suggestion
assert_eq!(unset.io_timeout_ms(), DEFAULT_IO_TIMEOUT_MS);
assert_eq!(client_config("45000").unwrap().io_timeout_ms(), 45_000);
assert_eq!(client_config("1").unwrap().io_timeout_ms(), 1);
assert_eq!(client_config(&u64::MAX.to_string()).unwrap().io_timeout_ms(),
u64::MAX);
for invalid in ["0", "-1", "12.5", "abc", "",
"18446744073709551616"] {
```
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -100,6 +102,55 @@ cfg_if! {
mod resolving;
pub use resolving::{OpenDalResolvingStorage, OpenDalResolvingStorageFactory};
+/// Deadline in milliseconds for one IO operation, and for every method call
on a returned
+/// reader, writer, lister or deleter. Honored by every [`OpenDalStorage`]
backend, where it
+/// defaults to 10000 to match OpenDAL's `TimeoutLayer`.
+///
+/// Each retry attempt is bounded separately, so it is a per-attempt budget,
not a total one.
+/// Control operations such as `stat` and `rename` are bounded by a separate,
fixed budget.
+pub const OPENDAL_IO_TIMEOUT_MS: &str = "opendal.io-timeout-ms";
+
+/// Matches the `opendal::layers::TimeoutLayer` default.
+const DEFAULT_IO_TIMEOUT_MS: u64 = 10_000;
+
+/// Backend-independent client settings, shared by every [`OpenDalStorage`]
variant.
+///
+/// Fields are private, so later settings are additive rather than breaking.
The
+/// container-level serde default lets an older payload deserialize as new
fields appear.
+#[derive(Clone, Debug, Properties, Serialize, Deserialize)]
+#[serde(default)]
+pub struct OpenDalClientConfig {
+ /// Per-attempt deadline for one IO operation, in milliseconds.
+ #[property(
+ key = OPENDAL_IO_TIMEOUT_MS,
+ default = DEFAULT_IO_TIMEOUT_MS,
+ parse_with = parse_io_timeout_ms,
+ getter
+ )]
+ io_timeout_ms: u64,
+}
+
+impl Default for OpenDalClientConfig {
+ fn default() -> Self {
+ Self {
+ io_timeout_ms: DEFAULT_IO_TIMEOUT_MS,
+ }
+ }
+}
+
+/// Parses one timeout value; the `Properties` derive adds the property-key
context.
+/// Zero is rejected: it would time every operation out before it starts.
+fn parse_io_timeout_ms(value: &str) -> Result<u64> {
+ match value.parse::<u64>() {
+ Ok(ms) if ms > 0 => Ok(ms),
+ _ => Err(Error::new(
+ ErrorKind::DataInvalid,
+ "Expected a positive integer number of milliseconds",
+ )
+ .with_context("value", format!("{value:?}"))),
+ }
+}
Review Comment:
`parse_io_timeout_ms` rejects zero, but the derived `Deserialize` doesn't go
through it. On the head commit,
`serde_json::from_str::<OpenDalClientConfig>(r#"{"io_timeout_ms":0}"#)`
succeeds with `io_timeout_ms() == 0`, and deserializing
`{"LocalFs":{"client_config":{"io_timeout_ms":0}}}` as an `OpenDalStorage` does
too. That's the path a serialized storage takes when it's rebuilt in another
process, so a zero can reach `TimeoutLayer` without any error.
What do you think about making the field a `NonZeroU64`?
`str::parse::<NonZeroU64>` rejects `"0"`, `"-1"`, `""`, and non-numeric input
by itself, and serde's `NonZeroU64` impl rejects zero on deserialize, so both
construction paths share one check. I tried the version below on the head
commit. The existing tests pass once they call `.get()`, clippy is clean, and
both zero payloads above fail with ``invalid value: integer `0`, expected a
nonzero u64``. It also needs `use std::num::NonZeroU64;` and `.get()` at the
`with_io_timeout` call in `create_operator`. The generated getter returns
`&NonZeroU64`, because
[`is_copy_type`](https://github.com/apache/iceberg-rust/blob/deab0692f6df211a396d6fc0ad21811f8b6c8491/crates/property-macro/src/properties.rs#L615-L662)
doesn't list the `NonZero` types.
```suggestion
const DEFAULT_IO_TIMEOUT_MS: NonZeroU64 = NonZeroU64::new(10_000).unwrap();
/// Backend-independent client settings, shared by every [`OpenDalStorage`]
variant.
///
/// Fields are private, so later settings are additive rather than breaking.
The
/// container-level serde default lets an older payload deserialize as new
fields appear.
#[derive(Clone, Debug, Properties, Serialize, Deserialize)]
#[serde(default)]
pub struct OpenDalClientConfig {
/// Per-attempt deadline for one IO operation, in milliseconds.
#[property(
key = OPENDAL_IO_TIMEOUT_MS,
default = DEFAULT_IO_TIMEOUT_MS,
parse_with = parse_io_timeout_ms,
getter
)]
io_timeout_ms: NonZeroU64,
}
impl Default for OpenDalClientConfig {
fn default() -> Self {
Self {
io_timeout_ms: DEFAULT_IO_TIMEOUT_MS,
}
}
}
/// Parses one timeout value; the `Properties` derive adds the property-key
context.
/// Zero is rejected: it would time every operation out before it starts.
fn parse_io_timeout_ms(value: &str) -> Result<NonZeroU64> {
value.parse().map_err(|_| {
Error::new(
ErrorKind::DataInvalid,
"Expected a positive integer number of milliseconds",
)
.with_context("value", format!("{value:?}"))
})
}
```
If you'd rather keep `u64`, routing deserialization through the same
validation with `#[serde(try_from = ...)]`, the pattern `FileScanTask` uses
since #3091, would also work. Either way, could
`test_client_config_serde_round_trip` add a case where a payload with a zero
timeout fails to deserialize?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
