fix!(io): bound S3 write request size with a configurable part size

crates/iceberg/public-api.txt

@@ -1086,6 +1086,7 @@ pub const iceberg::io::S3_ASSUME_ROLE_SESSION_NAME: &str
 pub const iceberg::io::S3_DISABLE_CONFIG_LOAD: &str
 pub const iceberg::io::S3_DISABLE_EC2_METADATA: &str
 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_SECRET_ACCESS_KEY: &str

crates/iceberg/src/io/storage/config/s3.rs

@@ -65,6 +65,16 @@ 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. Defaults to 32 MiB, matching Java
+/// `S3FileIOProperties.MULTIPART_SIZE_DEFAULT`. Must be at least 5 MiB, the S3 minimum,
+/// and at most `i32::MAX`, since Java reads the property as an `int`.
+///
+/// Each part travels as one request, so the part size bounds how much data has to
+/// transfer within the per-operation IO timeout. Lower it on a slow connection; if a
+/// 5 MiB part still cannot finish in time, raise `opendal.io-timeout-ms` instead.
+///
+/// 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";
 
 /// Amazon S3 storage configuration.
 ///

crates/storage/opendal/public-api.txt

@@ -19,10 +19,11 @@ pub iceberg_storage_opendal::OpenDalStorage::Memory::operator: opendal_core::typ
 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<opendal_service_oss::config::OssConfig>
-pub iceberg_storage_opendal::OpenDalStorage::S3
+#[non_exhaustive] 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<opendal_service_s3::config::S3Config>
 pub iceberg_storage_opendal::OpenDalStorage::S3::customized_credential_load: core::option::Option<iceberg_storage_opendal::CustomAwsCredentialLoader>
+pub iceberg_storage_opendal::OpenDalStorage::S3::multipart_part_size: u64
 impl core::clone::Clone for iceberg_storage_opendal::OpenDalStorage
 pub fn iceberg_storage_opendal::OpenDalStorage::clone(&self) -> iceberg_storage_opendal::OpenDalStorage
 impl core::fmt::Debug for iceberg_storage_opendal::OpenDalStorage

crates/storage/opendal/src/lib.rs

@@ -43,6 +43,7 @@ use iceberg::{Error, ErrorKind, Result};
 use iceberg_property_macro::Properties;
 use opendal::Operator;
 use opendal::layers::{RetryLayer, TimeoutLayer};
+use opendal::options::WriteOptions;
 use serde::{Deserialize, Serialize};
 use utils::from_opendal_error;
 
@@ -228,6 +229,7 @@ impl StorageFactory for OpenDalStorageFactory {
                 customized_credential_load,
             } => Ok(Arc::new(OpenDalStorage::S3 {
                 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,
             })),
@@ -301,9 +303,13 @@ pub enum OpenDalStorage {
     /// Accepts any S3-family URL (`s3://`, `s3a://`, `s3n://`); the scheme is
     /// derived from the path at call time.
     #[cfg(feature = "opendal-s3")]
+    #[non_exhaustive]
     S3 {
         /// S3 configuration.
         config: Arc<S3Config>,
+        /// Bytes carried by one multipart upload request.
+        #[serde(default = "default_multipart_part_size")]
+        multipart_part_size: u64,
         /// Custom AWS credential loader.
         #[serde(skip)]
         customized_credential_load: Option<CustomAwsCredentialLoader>,
@@ -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)),
+                ..WriteOptions::default()
+            },
+            _ => WriteOptions::default(),
+        }
+    }
+
     /// 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
@@ -662,7 +686,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())
             .await
             .map_err(from_opendal_error)?;
         Ok(())
@@ -671,7 +695,9 @@ impl Storage for OpenDalStorage {
     async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> {
         let (op, relative_path) = self.create_operator(&path)?;
         Ok(Box::new(OpenDalWriter::new(
-            op.writer(relative_path).await.map_err(from_opendal_error)?,
+            op.writer_options(relative_path, self.write_options())
+                .await
+                .map_err(from_opendal_error)?,
         )))
     }
 
@@ -866,6 +892,7 @@ mod tests {
     fn test_client_config_serde_round_trip() {
         let storage = OpenDalStorage::S3 {
             config: Arc::new(S3Config::default()),
+            multipart_part_size: default_multipart_part_size(),
             customized_credential_load: None,
             client_config: client_config("45000").unwrap(),
         };
@@ -1080,6 +1107,7 @@ mod tests {
     fn test_relativize_path_s3() {
         let storage = OpenDalStorage::S3 {
             config: Arc::new(S3Config::default()),
+            multipart_part_size: default_multipart_part_size(),
             customized_credential_load: None,
             client_config: OpenDalClientConfig::default(),
         };

crates/storage/opendal/src/resolving.rs

@@ -93,6 +93,7 @@ fn build_storage_for_scheme(
             let config = crate::s3::s3_config_parse(props.clone())?;
             Ok(OpenDalStorage::S3 {
                 config: Arc::new(config),
+                multipart_part_size: crate::s3::s3_multipart_part_size_parse(props)?,
                 customized_credential_load: customized_credential_load.clone(),
                 client_config,
             })

crates/storage/opendal/src/s3.rs

@@ -21,8 +21,8 @@ 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_PATH_STYLE_ACCESS, S3_REGION, S3_SECRET_ACCESS_KEY,
-    S3_SESSION_TOKEN, S3_SSE_KEY, S3_SSE_MD5, S3_SSE_TYPE,
+    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,
 };
 use iceberg::{Error, ErrorKind, Result};
 use opendal::services::S3Config;
@@ -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 {
+    MULTIPART_PART_SIZE_DEFAULT
+}
+
+/// 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} bytes is below the S3 minimum part size of {} MiB",
+                MULTIPART_PART_SIZE_MIN / (1024 * 1024)
+            ),
+        ));
+    }
+    if part_size > MULTIPART_PART_SIZE_MAX {
+        return Err(Error::new(
+            ErrorKind::DataInvalid,
+            format!(
+                "Invalid {S3_MULTIPART_PART_SIZE_BYTES}: {part_size} bytes is above {MULTIPART_PART_SIZE_MAX} bytes, the largest value Java accepts"
+            ),
+        ));
+    }
+    Ok(part_size)
+}
+
 /// Parse iceberg props to s3 config.
 pub(crate) fn s3_config_parse(mut m: HashMap<String, String>) -> Result<S3Config> {
     let mut cfg = S3Config::default();
@@ -184,9 +228,12 @@ 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::{
+        MULTIPART_PART_SIZE_DEFAULT, MULTIPART_PART_SIZE_MAX, MULTIPART_PART_SIZE_MIN, Result,
+        s3_config_parse, s3_multipart_part_size_parse,
+    };
 
     fn parse_with(prop: Option<&str>) -> bool {
         let mut props = HashMap::new();
@@ -203,4 +250,31 @@ mod tests {
         assert!(parse_with(Some("false")));
         assert!(!parse_with(Some("true")));
     }
+
+    fn parse_part_size(prop: Option<&str>) -> Result<u64> {
+        let mut props = HashMap::new();
+        if let Some(v) = prop {
+            props.insert(S3_MULTIPART_PART_SIZE_BYTES.to_string(), v.to_string());
+        }
+        s3_multipart_part_size_parse(&props)
+    }
+
+    #[test]
+    fn s3_multipart_part_size_parse_defaults_and_overrides() {
+        // Match Iceberg S3FileIOProperties.MULTIPART_SIZE_DEFAULT = 32 MiB.
+        assert_eq!(MULTIPART_PART_SIZE_DEFAULT, 32 * 1024 * 1024);
+        assert_eq!(parse_part_size(None).unwrap(), MULTIPART_PART_SIZE_DEFAULT);
+        assert_eq!(parse_part_size(Some("8388608")).unwrap(), 8 * 1024 * 1024);
+    }
+
+    #[test]
+    fn s3_multipart_part_size_parse_rejects_invalid() {
+        // Match Iceberg S3FileIOProperties.MULTIPART_SIZE_MIN = 5 MiB.
+        assert_eq!(MULTIPART_PART_SIZE_MIN, 5 * 1024 * 1024);
+        assert!(parse_part_size(Some(&MULTIPART_PART_SIZE_MIN.to_string())).is_ok());
+        assert!(parse_part_size(Some(&(MULTIPART_PART_SIZE_MIN - 1).to_string())).is_err());
+        assert!(parse_part_size(Some(&MULTIPART_PART_SIZE_MAX.to_string())).is_ok());
+        assert!(parse_part_size(Some(&(MULTIPART_PART_SIZE_MAX + 1).to_string())).is_err());
+        assert!(parse_part_size(Some("not-a-number")).is_err());
+    }
 }