laskoviymishka commented on code in PR #3179:
URL: https://github.com/apache/iceberg-rust/pull/3179#discussion_r4472847077
##########
crates/storage/opendal/src/s3.rs:
##########
@@ -36,6 +36,50 @@ 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;
+
+/// Java reads this property as an `int`, so a larger value fails there.
+const MULTIPART_PART_SIZE_MAX: u64 = i32::MAX as u64;
+
+/// Matches Java `S3FileIOProperties.MULTIPART_SIZE_DEFAULT`.
+const MULTIPART_PART_SIZE_DEFAULT: u64 = 32 * 1024 * 1024;
+
+pub(crate) fn default_multipart_part_size() -> u64 {
Review Comment:
nit: Could the reason this is a named fn go in a doc comment? Right now it
only lives in your PR reply: a bare `#[serde(default)]` would read payloads
without the field as 0, which OpenDAL [raises to its 5 MiB
floor](https://github.com/apache/opendal/blob/6f544a131be78e553d28e79f3cfe5ac948f2963e/core/core/src/types/context/write.rs#L78-L96)
instead of 32 MiB.
```suggestion
/// Default part size, also the serde default for
`OpenDalStorage::S3::multipart_part_size`.
/// A bare `#[serde(default)]` would read payloads without the field as 0,
which OpenDAL
/// raises to 5 MiB instead of 32 MiB.
pub(crate) fn default_multipart_part_size() -> u64 {
```
##########
crates/storage/opendal/tests/file_io_s3_test.rs:
##########
@@ -26,33 +26,80 @@ 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_object_store_endpoint,
normalize_test_name_with_parts, set_up};
+ use opendal::Configurator;
use reqsign_core::Context;
+ const OBJECT_STORE_ACCESS_KEY_ID: &str = "admin";
+ const OBJECT_STORE_SECRET_ACCESS_KEY: &str = "password";
+ const OBJECT_STORE_REGION: &str = "us-east-1";
+ const OBJECT_STORE_BUCKET: &str = "bucket1";
+
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 object_store_endpoint = get_object_store_endpoint();
+ let mut props = vec![
+ (S3_ENDPOINT, object_store_endpoint),
+ (S3_ACCESS_KEY_ID, OBJECT_STORE_ACCESS_KEY_ID.to_string()),
+ (
+ S3_SECRET_ACCESS_KEY,
+ OBJECT_STORE_SECRET_ACCESS_KEY.to_string(),
+ ),
+ (S3_REGION, OBJECT_STORE_REGION.to_string()),
+ (S3_PATH_STYLE_ACCESS, "true".to_string()),
+ ];
+ props.extend(extra);
+
FileIOBuilder::new(Arc::new(OpenDalStorageFactory::S3 {
customized_credential_load: None,
}))
- .with_props(vec![
- (S3_ENDPOINT, object_store_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()),
- ])
+ .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.
+ ///
+ /// A server-side encrypted object can report an ETag without the suffix,
so
+ /// this helper holds only for the object store fixture these tests run
against.
+ async fn upload_part_count(key: &str) -> usize {
+ let mut config = opendal::services::S3Config::default();
+ config.endpoint = Some(get_object_store_endpoint());
+ config.access_key_id = Some(OBJECT_STORE_ACCESS_KEY_ID.to_string());
+ config.secret_access_key =
Some(OBJECT_STORE_SECRET_ACCESS_KEY.to_string());
+ config.region = Some(OBJECT_STORE_REGION.to_string());
+ config.bucket = OBJECT_STORE_BUCKET.to_string();
+ // Each `#[tokio::test]` has its own runtime, and the first request
after
+ // an earlier one went away fails before the client reconnects. The
+ // storage under test carries the same layer.
+ let op = opendal::Operator::new(config.into_builder())
+ .unwrap()
+ .layer(opendal::layers::RetryLayer::new());
+ let meta = op.stat(key).await.unwrap();
+ let etag = meta
+ .etag()
+ .expect("object store reports an ETag")
+ .trim_matches('"');
+ match etag.rsplit_once('-') {
+ Some((_, parts)) => parts
+ .parse()
+ .unwrap_or_else(|e| panic!("unexpected ETag format {etag}:
{e}")),
+ None => 1,
Review Comment:
nit: Could `upload_part_count` return `None` for a bare-MD5 ETag, so the
single-request test asserts that directly? Today a PutObject and a one-part
multipart upload (`<md5>-1`) both come back as 1, so [this
assertion](https://github.com/apache/iceberg-rust/blob/2b4dea024ef2ebfe28d08da904afb265ee41b05b/crates/storage/opendal/tests/file_io_s3_test.rs#L233)
shows the write wasn't split, not that it went out as one request.
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -513,6 +519,24 @@ impl OpenDalStorage {
}
}
+ /// Bounds the S3 request size. Without a chunk size OpenDAL turns each
caller
+ /// buffer into one request, and `ParquetWriter` hands over a whole row
group
+ /// at a time. Other backends keep OpenDAL's defaults.
+ #[allow(unreachable_patterns)]
+ fn write_options(&self) -> WriteOptions {
+ match self {
+ #[cfg(feature = "opendal-s3")]
+ OpenDalStorage::S3 {
+ multipart_part_size,
+ ..
+ } => WriteOptions {
+ chunk:
Some(usize::try_from(*multipart_part_size).unwrap_or(usize::MAX)),
Review Comment:
nit: Could this line say why the cast can't fail? With
`MULTIPART_PART_SIZE_MAX` at `i32::MAX`, every value from
[`s3_multipart_part_size_parse`](https://github.com/apache/iceberg-rust/blob/2b4dea024ef2ebfe28d08da904afb265ee41b05b/crates/storage/opendal/src/s3.rs#L39-L81)
fits in `usize` on 32-bit too, so `unwrap_or(usize::MAX)` never fires on the
property path.
```suggestion
// The parser caps this at `i32::MAX`, so the cast can't
fail for parsed values.
chunk:
Some(usize::try_from(*multipart_part_size).unwrap_or(usize::MAX)),
```
--
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]
