laskoviymishka commented on code in PR #3179:
URL: https://github.com/apache/iceberg-rust/pull/3179#discussion_r4034217195
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -543,7 +567,7 @@ impl Storage for OpenDalStorage {
async fn write(&self, path: &str, bs: Bytes) -> Result<()> {
let (op, relative_path) = self.create_operator(&path)?;
- op.write(relative_path, bs)
+ op.write_options(relative_path, bs, self.write_options())
Review Comment:
I think there's a subtle issue here: `write_options()` sets `chunk`, but per
the OpenDAL docs `chunk` only takes effect on the streaming `writer_options`
path — `write_options()` is a single bulk write and won't split a complete
`Bytes` buffer into `UploadPart` calls no matter what `chunk` says.
If that's right, this line is misleading: the metadata writes that go
through `write()` (manifests, snapshot JSON, table metadata) silently ignore
the option, so we've only actually fixed the streaming `writer()` path. If it's
*not* right and OpenDAL does chunk here, then every small metadata write now
takes the 3x multipart round-trip (Create/Upload/Complete), which is a real
regression on commit-heavy workloads.
Could we confirm which it is, and then either drop `write_options()` from
`write()` if it's a no-op, or add a test through `OutputFile::write()` with a
small part size asserting the part count? The existing test only covers
`writer()`. wdyt?
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -233,6 +235,9 @@ pub enum OpenDalStorage {
S3 {
/// S3 configuration.
config: Arc<S3Config>,
+ /// Bytes carried by one multipart upload request.
+ #[serde(default = "default_multipart_part_size")]
+ multipart_part_size: u64,
Review Comment:
Adding `multipart_part_size` as a required public field to the `S3` variant
is a source-breaking change — any external crate constructing
`OpenDalStorage::S3 { config, customized_credential_load }` with an explicit
field list stops compiling.
The crate is 0.x and you've already regenerated `public-api.txt`, so this is
permitted, and `customized_credential_load` set a similar precedent. But since
users configure this through the `s3.multipart.part-size-bytes` property (same
as everything in `config`), I'd lean toward making the field `pub(crate)` and
dropping it from `public-api.txt` — that keeps the constructor surface stable
and matches how the rest of the config flows in.
If we'd rather keep it public, `#[non_exhaustive]` on the enum or a
changelog note calling out the break would at least make it intentional. wdyt?
##########
crates/storage/opendal/src/s3.rs:
##########
@@ -36,6 +36,36 @@ use url::Url;
use crate::utils::{from_opendal_error, is_truthy};
+/// S3 rejects a non-final part smaller than this.
+const MULTIPART_PART_SIZE_MIN: u64 = 5 * 1024 * 1024;
+
+/// Matches Java `S3FileIOProperties.MULTIPART_SIZE_DEFAULT`.
+pub(crate) fn default_multipart_part_size() -> u64 {
+ 32 * 1024 * 1024
Review Comment:
This default is a bare literal here and repeated in the test
(`assert_eq!(parse_part_size(None).unwrap(), 32 * 1024 * 1024)`), so they can
drift apart. I'd pull it into a `const DEFAULT_MULTIPART_PART_SIZE: u64 = 32 *
1024 * 1024;` next to `MULTIPART_PART_SIZE_MIN`, have the fn return it, and
assert against the const.
##########
crates/storage/opendal/src/s3.rs:
##########
@@ -36,6 +36,36 @@ use url::Url;
use crate::utils::{from_opendal_error, is_truthy};
+/// S3 rejects a non-final part smaller than this.
+const MULTIPART_PART_SIZE_MIN: u64 = 5 * 1024 * 1024;
+
+/// Matches Java `S3FileIOProperties.MULTIPART_SIZE_DEFAULT`.
+pub(crate) fn default_multipart_part_size() -> u64 {
+ 32 * 1024 * 1024
+}
+
+/// Parse iceberg props to s3 multipart upload part size.
+pub(crate) fn s3_multipart_part_size_parse(m: &HashMap<String, String>) ->
Result<u64> {
+ let Some(value) = m.get(S3_MULTIPART_PART_SIZE_BYTES) else {
+ return Ok(default_multipart_part_size());
+ };
+ let part_size = value.parse::<u64>().map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid {S3_MULTIPART_PART_SIZE_BYTES}: {value}: {e}"),
+ )
+ })?;
+ if part_size < MULTIPART_PART_SIZE_MIN {
Review Comment:
We validate the floor here but not the ceiling — S3 caps a part at 5 GiB, so
something like `6_000_000_000` passes this check and then fails with an opaque
S3 `InvalidArgument` mid-upload instead of a clean `DataInvalid`.
Adding a mirror check closes that, and it also makes the cast in
`write_options` (`usize::try_from(*multipart_part_size).unwrap_or(usize::MAX)`,
lib.rs:415) infallible — right now on a 32-bit target a value above ~4 GiB
silently clamps *up* to `usize::MAX` rather than erroring.
```rust
const MULTIPART_PART_SIZE_MAX: u64 = 5 * 1024 * 1024 * 1024;
```
Then the `try_from` can become `.expect("validated <= 5 GiB")`. Fix that and
this is good to land.
##########
crates/storage/opendal/tests/file_io_s3_test.rs:
##########
@@ -26,33 +26,62 @@ mod tests {
use bytes::Bytes;
use futures::StreamExt;
use iceberg::io::{
- FileIO, FileIOBuilder, S3_ACCESS_KEY_ID, S3_ENDPOINT,
S3_PATH_STYLE_ACCESS, S3_REGION,
- S3_SECRET_ACCESS_KEY,
+ FileIO, FileIOBuilder, S3_ACCESS_KEY_ID, S3_ENDPOINT,
S3_MULTIPART_PART_SIZE_BYTES,
+ S3_PATH_STYLE_ACCESS, S3_REGION, S3_SECRET_ACCESS_KEY,
};
use iceberg_storage_opendal::{
AwsCredential, CustomAwsCredentialLoader, OpenDalStorageFactory,
ProvideCredential,
};
use iceberg_test_utils::{get_minio_endpoint,
normalize_test_name_with_parts, set_up};
+ use opendal::Configurator;
use reqsign_core::Context;
async fn get_file_io() -> FileIO {
+ get_file_io_with_props(vec![]).await
+ }
+
+ async fn get_file_io_with_props(extra: Vec<(&'static str, String)>) ->
FileIO {
set_up();
let minio_endpoint = get_minio_endpoint();
- FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 {
- customized_credential_load: None,
- }))
- .with_props(vec![
+ let mut props = vec![
(S3_ENDPOINT, minio_endpoint),
(S3_ACCESS_KEY_ID, "admin".to_string()),
(S3_SECRET_ACCESS_KEY, "password".to_string()),
(S3_REGION, "us-east-1".to_string()),
(S3_PATH_STYLE_ACCESS, "true".to_string()),
- ])
+ ];
+ props.extend(extra);
+
+ FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 {
+ customized_credential_load: None,
+ }))
+ .with_props(props)
.build()
}
+ /// Number of upload parts S3 recorded for `key`. A multipart upload
reports
+ /// a `<md5>-<part-count>` ETag; a single-request upload reports a bare
MD5.
+ async fn upload_part_count(key: &str) -> usize {
+ let mut config = opendal::services::S3Config::default();
+ config.endpoint = Some(get_minio_endpoint());
+ config.access_key_id = Some("admin".to_string());
+ config.secret_access_key = Some("password".to_string());
+ config.region = Some("us-east-1".to_string());
+ config.bucket = "bucket1".to_string();
+ let op = opendal::Operator::new(config.into_builder()).unwrap();
+ let meta = op.stat(key).await.unwrap();
+ let etag = meta
+ .etag()
+ .expect("MinIO reports an ETag")
+ .trim_matches('"');
+ match etag.rsplit_once('-') {
Review Comment:
The `<md5>-<N>` ETag suffix is a MinIO/plain-S3 detail — with SSE-KMS/C the
suffix can be absent, and then this `None` arm silently returns 1, so a broken
multipart upload would read as a passing single-part assertion rather than
failing loudly. And `parts.parse().unwrap()` panics with no context if the
segment isn't numeric.
Since the test only runs against MinIO today it's fine in practice, but I'd
at least comment the MinIO assumption and `expect` on the parse so a surprise
format fails with a readable message. Not blocking.
##########
crates/storage/opendal/src/s3.rs:
##########
@@ -184,9 +214,9 @@ impl CustomAwsCredentialLoader {
mod tests {
use std::collections::HashMap;
- use iceberg::io::S3_PATH_STYLE_ACCESS;
+ use iceberg::io::{S3_MULTIPART_PART_SIZE_BYTES, S3_PATH_STYLE_ACCESS};
- use super::s3_config_parse;
+ use super::*;
Review Comment:
Small one — swapping the explicit import for `use super::*;` pulls every
`pub(crate)` symbol into the test module. I'd keep it explicit (`use
super::{default_multipart_part_size, s3_config_parse,
s3_multipart_part_size_parse};`) so the test's dependency surface stays visible.
##########
crates/storage/opendal/src/s3.rs:
##########
@@ -36,6 +36,36 @@ use url::Url;
use crate::utils::{from_opendal_error, is_truthy};
+/// S3 rejects a non-final part smaller than this.
+const MULTIPART_PART_SIZE_MIN: u64 = 5 * 1024 * 1024;
+
+/// Matches Java `S3FileIOProperties.MULTIPART_SIZE_DEFAULT`.
+pub(crate) fn default_multipart_part_size() -> u64 {
+ 32 * 1024 * 1024
+}
+
+/// Parse iceberg props to s3 multipart upload part size.
+pub(crate) fn s3_multipart_part_size_parse(m: &HashMap<String, String>) ->
Result<u64> {
+ let Some(value) = m.get(S3_MULTIPART_PART_SIZE_BYTES) else {
+ return Ok(default_multipart_part_size());
+ };
+ let part_size = value.parse::<u64>().map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Invalid {S3_MULTIPART_PART_SIZE_BYTES}: {value}: {e}"),
+ )
+ })?;
+ if part_size < MULTIPART_PART_SIZE_MIN {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Invalid {S3_MULTIPART_PART_SIZE_BYTES}: {part_size} is below
the S3 minimum part size of {MULTIPART_PART_SIZE_MIN}"
Review Comment:
Tiny thing: this prints raw bytes (`...minimum part size of 5242880`) with
no unit, which is hard to read. Something like `{part_size} bytes is below the
S3 minimum part size of {MULTIPART_PART_SIZE_MIN} (5 MiB)` reads a lot
friendlier.
##########
crates/iceberg/src/io/storage/config/s3.rs:
##########
@@ -64,6 +64,9 @@ pub const S3_ALLOW_ANONYMOUS: &str = "s3.allow-anonymous";
pub const S3_DISABLE_EC2_METADATA: &str = "s3.disable-ec2-metadata";
/// Option to skip loading configuration from config file and the env.
pub const S3_DISABLE_CONFIG_LOAD: &str = "s3.disable-config-load";
+/// Size in bytes of each part of a multipart upload. Must be at least 5 MiB.
+/// Defaults to 32 MiB, matching Java `S3FileIOProperties.MULTIPART_SIZE`.
+pub const S3_MULTIPART_PART_SIZE_BYTES: &str = "s3.multipart.part-size-bytes";
Review Comment:
While we're here — the `TimeoutLayer` default is 10s per IO op, and that
applies per `UploadPart`, so part size is really the knob for how much has to
transfer inside that window. Since that interaction is the whole motivation for
this property, I'd add a line to the doc comment noting it so a user on a slow
link knows reducing the part size is the lever. Small thing.
--
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]
