feat(services/s3): support default_region, server_side_encryption_bucket_key and skip_signature for Arrow object_store compatibility

core/src/services/s3/backend.rs

@@ -28,6 +28,7 @@ use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use constants::X_AMZ_META_PREFIX;
 use constants::X_AMZ_VERSION_ID;
+use http::HeaderValue;
 use http::Response;
 use http::StatusCode;
 use log::debug;
@@ -162,6 +163,16 @@ impl S3Builder {
         self
     }
 
+    /// Set default region to use when region detection fails.
+    /// This region will be used if region is not set explicitly and cannot be detected from environment.
+    pub fn default_region(mut self, region: &str) -> Self {
+        if !region.is_empty() {
+            self.config.default_region = Some(region.to_string())
+        }
+
+        self
+    }
+
     /// Set access_key_id of this backend.
     ///
     /// - If access_key_id is set, we will take user's input first.
@@ -600,6 +611,19 @@ impl S3Builder {
         self
     }
 
+    /// Enable the use of S3 bucket keys for server-side encryption.
+    /// This can reduce costs when using KMS encryption by using fewer KMS API calls.
+    pub fn enable_server_side_encryption_bucket_key(mut self) -> Self {
+        self.config.server_side_encryption_bucket_key_enabled = true;
+        self
+    }
+
+    /// Disable the use of S3 bucket keys for server-side encryption.
+    pub fn disable_server_side_encryption_bucket_key(mut self) -> Self {
+        self.config.server_side_encryption_bucket_key_enabled = false;
+        self
+    }
+
     /// Detect region of S3 bucket.
     ///
     /// # Args
@@ -783,6 +807,13 @@ impl Builder for S3Builder {
                 })?),
             };
 
+        let server_side_encryption_bucket_key_enabled =
+            if self.config.server_side_encryption_bucket_key_enabled {
+                Some(HeaderValue::from_static("true"))
+            } else {
+                None
+            };
+
         let checksum_algorithm = match self.config.checksum_algorithm.as_deref() {
             Some("crc32c") => Some(ChecksumAlgorithm::Crc32c),
             None => None,
@@ -809,12 +840,16 @@ impl Builder for S3Builder {
         }
 
         if cfg.region.is_none() {
-            return Err(Error::new(
-                ErrorKind::ConfigInvalid,
-                "region is missing. Please find it by S3::detect_region() or set them in env.",
-            )
-            .with_operation("Builder::build")
-            .with_context("service", Scheme::S3));
+            if let Some(ref default_region) = self.config.default_region {
+                cfg.region = Some(default_region.clone());
+            } else {
+                return Err(Error::new(
+                    ErrorKind::ConfigInvalid,
+                    "region is missing. Please find it by S3::detect_region(), set them in env, or set default_region.",
+                )
+                .with_operation("Builder::build")
+                .with_context("service", Scheme::S3));
+            }
         }
 
         let region = cfg.region.to_owned().unwrap();
@@ -998,6 +1033,7 @@ impl Builder for S3Builder {
                 server_side_encryption_customer_algorithm,
                 server_side_encryption_customer_key,
                 server_side_encryption_customer_key_md5,
+                server_side_encryption_bucket_key_enabled,
                 default_storage_class,
                 allow_anonymous: self.config.allow_anonymous,
                 disable_list_objects_v2: self.config.disable_list_objects_v2,
@@ -1162,6 +1198,7 @@ impl Access for S3Backend {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use serde_json;
 
     #[test]
     fn test_is_valid_bucket() {
@@ -1264,4 +1301,156 @@ mod tests {
             assert_eq!(region.as_deref(), expected, "{name}");
         }
     }
+
+    #[test]
+    fn test_default_region_config() {
+        let builder = S3Builder::default()
+            .bucket("test-bucket")
+            .default_region("us-west-2");
+
+        assert_eq!(builder.config.default_region, Some("us-west-2".to_string()));
+        assert_eq!(builder.config.region, None);
+
+        // Test that it works as a fallback
+        let builder_with_region = S3Builder::default()
+            .bucket("test-bucket")
+            .region("us-east-1")
+            .default_region("us-west-2");
+
+        assert_eq!(
+            builder_with_region.config.region,
+            Some("us-east-1".to_string())
+        );
+        assert_eq!(
+            builder_with_region.config.default_region,
+            Some("us-west-2".to_string())
+        );
+    }
+
+    #[test]
+    fn test_server_side_encryption_bucket_key_config() {
+        // Default should be false
+        let default_builder = S3Builder::default()
+            .bucket("test-bucket")
+            .region("us-east-1");
+        assert!(
+            !default_builder
+                .config
+                .server_side_encryption_bucket_key_enabled
+        );
+
+        // Test enable
+        let builder_enabled = S3Builder::default()
+            .bucket("test-bucket")
+            .region("us-east-1")
+            .enable_server_side_encryption_bucket_key();
+        assert!(
+            builder_enabled
+                .config
+                .server_side_encryption_bucket_key_enabled
+        );
+
+        // Test disable
+        let builder_disabled = S3Builder::default()
+            .bucket("test-bucket")
+            .region("us-east-1")
+            .disable_server_side_encryption_bucket_key();
+        assert!(
+            !builder_disabled
+                .config
+                .server_side_encryption_bucket_key_enabled
+        );
+    }
+
+    #[test]
+    fn test_default_region_fallback() {
+        // Test that default_region works as fallback when region is not set
+        let builder = S3Builder::default()
+            .bucket("test-bucket")
+            .default_region("us-west-2");
+
+        assert_eq!(builder.config.default_region, Some("us-west-2".to_string()));
+        assert_eq!(builder.config.region, None);
+    }
+
+    #[test]
+    fn test_config_aliases() {
+        // Test AWS-prefixed aliases using direct JSON with proper types
+        let config_json = r#"{
+            "bucket": "test-bucket",
+            "region": "us-east-1",
+            "aws_default_region": "us-west-1",
+            "aws_skip_signature": true,
+            "aws_sse_bucket_key_enabled": true
+        }"#;
+
+        let config: S3Config = serde_json::from_str(config_json).unwrap();
+
+        // Verify all AWS aliases work correctly
+        assert_eq!(config.bucket, "test-bucket");
+        assert_eq!(config.region, Some("us-east-1".to_string()));
+        assert_eq!(config.default_region, Some("us-west-1".to_string()));
+        assert!(config.allow_anonymous);
+        assert!(config.server_side_encryption_bucket_key_enabled);
+    }
+
+    #[test]
+    fn test_config_short_aliases() {
+        // Test short aliases (without aws_ prefix) using direct JSON with proper types
+        let config_json = r#"{
+            "bucket": "test-bucket",
+            "region": "us-east-1",
+            "default_region": "us-west-2",
+            "skip_signature": true,
+            "server_side_encryption_bucket_key_enabled": false
+        }"#;
+
+        let config: S3Config = serde_json::from_str(config_json).unwrap();
+
+        // Verify all short aliases work correctly
+        assert_eq!(config.bucket, "test-bucket");
+        assert_eq!(config.region, Some("us-east-1".to_string()));
+        assert_eq!(config.default_region, Some("us-west-2".to_string()));
+        assert!(config.allow_anonymous);
+        assert!(!config.server_side_encryption_bucket_key_enabled);
+    }
+
+    #[test]
+    fn test_default_region_alias() {
+        // Test aws_default_region alias
+        let config_json = r#"{
+            "bucket": "test-bucket",
+            "region": "us-east-1",
+            "aws_default_region": "us-west-2"
+        }"#;
+
+        let config: S3Config = serde_json::from_str(config_json).unwrap();
+        assert_eq!(config.default_region, Some("us-west-2".to_string()));
+    }
+
+    #[test]
+    fn test_skip_signature_alias() {
+        // Test aws_skip_signature alias
+        let config_json = r#"{
+            "bucket": "test-bucket",
+            "region": "us-east-1",
+            "aws_skip_signature": true
+        }"#;
+
+        let config: S3Config = serde_json::from_str(config_json).unwrap();
+        assert!(config.allow_anonymous);
+    }
+
+    #[test]
+    fn test_server_side_encryption_bucket_key_alias() {
+        // Test aws_sse_bucket_key_enabled alias
+        let config_json = r#"{
+            "bucket": "test-bucket",
+            "region": "us-east-1",
+            "aws_sse_bucket_key_enabled": true
+        }"#;
+
+        let config: S3Config = serde_json::from_str(config_json).unwrap();
+        assert!(config.server_side_encryption_bucket_key_enabled);
+    }
 }

core/src/services/s3/config.rs

@@ -62,6 +62,10 @@ pub struct S3Config {
     /// - If region is set, we will take user's input first.
     /// - If not, we will try to load it from environment.
     pub region: Option<String>,
+    /// Default region to use when region detection fails or when no region is explicitly set.
+    /// Falls back to this region if region detection from endpoint or environment fails.
+    #[serde(alias = "aws_default_region")]
+    pub default_region: Option<String>,
 
     /// access_key_id of this backend.
     ///
@@ -102,6 +106,7 @@ pub struct S3Config {
     pub disable_ec2_metadata: bool,
     /// Allow anonymous will allow opendal to send request without signing
     /// when credential is not loaded.
+    #[serde(alias = "aws_skip_signature", alias = "skip_signature")]
     pub allow_anonymous: bool,
     /// server_side_encryption for this backend.
     ///
@@ -131,6 +136,10 @@ pub struct S3Config {
     ///
     /// Value: MD5 digest of key specified in `server_side_encryption_customer_key`.
     pub server_side_encryption_customer_key_md5: Option<String>,
+    /// Enable or disable S3 bucket keys for server-side encryption with KMS.
+    /// This can reduce costs when using KMS encryption by using fewer KMS API calls.
+    #[serde(alias = "aws_sse_bucket_key_enabled")]
+    pub server_side_encryption_bucket_key_enabled: bool,
     /// default storage_class for this backend.
     ///
     /// Available values:

core/src/services/s3/core.rs

@@ -65,6 +65,8 @@ pub mod constants {
         "x-amz-server-side-encryption-customer-key-md5";
     pub const X_AMZ_SERVER_SIDE_ENCRYPTION_AWS_KMS_KEY_ID: &str =
         "x-amz-server-side-encryption-aws-kms-key-id";
+    pub const X_AMZ_SERVER_SIDE_ENCRYPTION_BUCKET_KEY_ENABLED: &str =
+        "x-amz-server-side-encryption-bucket-key-enabled";
     pub const X_AMZ_STORAGE_CLASS: &str = "x-amz-storage-class";
 
     pub const X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM: &str =
@@ -99,6 +101,7 @@ pub struct S3Core {
     pub server_side_encryption_customer_algorithm: Option<HeaderValue>,
     pub server_side_encryption_customer_key: Option<HeaderValue>,
     pub server_side_encryption_customer_key_md5: Option<HeaderValue>,
+    pub server_side_encryption_bucket_key_enabled: Option<HeaderValue>,
     pub default_storage_class: Option<HeaderValue>,
     pub allow_anonymous: bool,
     pub disable_list_objects_v2: bool,
@@ -234,6 +237,14 @@ impl S3Core {
                     v,
                 )
             }
+            if let Some(v) = &self.server_side_encryption_bucket_key_enabled {
+                req = req.header(
+                    HeaderName::from_static(
+                        constants::X_AMZ_SERVER_SIDE_ENCRYPTION_BUCKET_KEY_ENABLED,
+                    ),
+                    v,
+                )
+            }
         }
 
         if let Some(v) = &self.server_side_encryption_customer_algorithm {
@@ -343,6 +354,7 @@ impl S3Core {
                 req = req.header(format!("{X_AMZ_META_PREFIX}{key}"), value)
             }
         }
+
         req
     }