feat(services/s3): add S3 tagging write support

bindings/java/src/lib.rs

@@ -232,6 +232,7 @@ fn make_write_options<'a>(
         if_none_match: convert::read_string_field(env, options, "ifNoneMatch")?,
         if_not_exists: convert::read_bool_field(env, options, "ifNotExists").unwrap_or_default(),
         user_metadata: convert::read_map_field(env, options, "userMetadata")?,
+        tags: None, // Tags not yet supported in Java bindings
         concurrent,
         chunk: convert::read_jlong_field_to_usize(env, options, "chunk")?,
     })

bindings/nodejs/src/options.rs

@@ -496,6 +496,7 @@ impl From<WriteOptions> for opendal::options::WriteOptions {
             cache_control: value.cache_control,
             content_encoding: value.content_encoding,
             user_metadata: value.user_metadata,
+            tags: None, // Tags not yet supported in Node.js bindings
             if_match: value.if_match,
             if_none_match: value.if_none_match,
             if_not_exists: value.if_not_exists.unwrap_or_default(),

bindings/python/src/options.rs

@@ -112,6 +112,7 @@ impl From<WriteOptions> for ocore::options::WriteOptions {
             cache_control: opts.cache_control,
             content_encoding: opts.content_encoding,
             user_metadata: opts.user_metadata,
+            tags: None, // Tags not yet supported in Python bindings
             if_match: opts.if_match,
             if_none_match: opts.if_none_match,
             if_not_exists: opts.if_not_exists.unwrap_or(false),

core/src/raw/ops.rs

@@ -694,6 +694,7 @@ pub struct OpWrite {
     if_none_match: Option<String>,
     if_not_exists: bool,
     user_metadata: Option<HashMap<String, String>>,
+    tags: Option<HashMap<String, String>>,
 }
 
 impl OpWrite {
@@ -821,6 +822,17 @@ impl OpWrite {
     pub fn user_metadata(&self) -> Option<&HashMap<String, String>> {
         self.user_metadata.as_ref()
     }
+
+    /// Set the tags of the op
+    pub fn with_tags(mut self, tags: HashMap<String, String>) -> Self {
+        self.tags = Some(tags);
+        self
+    }
+
+    /// Get tags from the op
+    pub fn tags(&self) -> Option<&HashMap<String, String>> {
+        self.tags.as_ref()
+    }
 }
 
 /// Args for `writer` operation.
@@ -872,6 +884,7 @@ impl From<options::WriteOptions> for (OpWrite, OpWriter) {
                 if_none_match: value.if_none_match,
                 if_not_exists: value.if_not_exists,
                 user_metadata: value.user_metadata,
+                tags: value.tags,
             },
             OpWriter { chunk: value.chunk },
         )

core/src/services/s3/backend.rs

@@ -600,6 +600,15 @@ impl S3Builder {
         self
     }
 
+    /// Disable tagging objects during write operations.
+    ///
+    /// This can be desirable if not supported by the backing store or to reduce
+    /// request overhead when tags are not needed.
+    pub fn disable_tagging(mut self) -> Self {
+        self.config.disable_tagging = true;
+        self
+    }
+
     /// Detect region of S3 bucket.
     ///
     /// # Args
@@ -945,6 +954,7 @@ impl Builder for S3Builder {
                             write_with_if_match: !self.config.disable_write_with_if_match,
                             write_with_if_not_exists: true,
                             write_with_user_metadata: true,
+                            write_with_tags: !self.config.disable_tagging,
 
                             // The min multipart size of S3 is 5 MiB.
                             //
@@ -1002,6 +1012,7 @@ impl Builder for S3Builder {
                 allow_anonymous: self.config.allow_anonymous,
                 disable_list_objects_v2: self.config.disable_list_objects_v2,
                 enable_request_payer: self.config.enable_request_payer,
+                disable_tagging: self.config.disable_tagging,
                 signer,
                 loader,
                 credential_loaded: AtomicBool::new(false),

core/src/services/s3/config.rs

@@ -196,6 +196,12 @@ pub struct S3Config {
 
     /// Indicates whether the client agrees to pay for the requests made to the S3 bucket.
     pub enable_request_payer: bool,
+
+    /// Disable tagging objects during write operations.
+    ///
+    /// This can be desirable if not supported by the backing store or to reduce
+    /// request overhead when tags are not needed.
+    pub disable_tagging: bool,
 }
 
 impl Debug for S3Config {

core/src/services/s3/core.rs

@@ -103,6 +103,7 @@ pub struct S3Core {
     pub allow_anonymous: bool,
     pub disable_list_objects_v2: bool,
     pub enable_request_payer: bool,
+    pub disable_tagging: bool,
 
     pub signer: AwsV4Signer,
     pub loader: Box<dyn AwsCredentialLoad>,
@@ -343,6 +344,31 @@ impl S3Core {
                 req = req.header(format!("{X_AMZ_META_PREFIX}{key}"), value)
             }
         }
+
+        // Set object tagging headers if tagging is enabled and tags are provided.
+        if !self.disable_tagging {
+            if let Some(tags) = args.tags() {
+                if !tags.is_empty() {
+                    let tags_string = tags
+                        .iter()
+                        .map(|(k, v)| {
+                            format!(
+                                "{}={}",
+                                k.replace('&', "%26")
+                                    .replace('=', "%3D")
+                                    .replace('+', "%2B"),
+                                v.replace('&', "%26")
+                                    .replace('=', "%3D")
+                                    .replace('+', "%2B")
+                            )
+                        })
+                        .collect::<Vec<String>>()
+                        .join("&");
+                    req = req.header("x-amz-tagging", tags_string);
+                }
+            }
+        }
+
         req
     }
 
@@ -1738,4 +1764,106 @@ mod tests {
             },]
         );
     }
+
+    #[test]
+    fn test_disable_tagging_functionality() {
+        use crate::raw::{AccessorInfo, OpWrite};
+        use reqsign::{AwsConfig, AwsDefaultLoader, AwsV4Signer};
+        use std::collections::HashMap;
+        use std::sync::atomic::AtomicBool;
+        use std::sync::Arc;
+
+        // Test with tagging enabled (default)
+        let info = Arc::new(AccessorInfo::default());
+        let client = reqwest::Client::new();
+        let config = AwsConfig::default();
+        let loader1 = Box::new(AwsDefaultLoader::new(client.clone(), config.clone()));
+        let loader2 = Box::new(AwsDefaultLoader::new(client, config));
+        let signer1 = AwsV4Signer::new("s3", "us-east-1");
+        let signer2 = AwsV4Signer::new("s3", "us-east-1");
+
+        let core_with_tagging = S3Core {
+            info: info.clone(),
+            bucket: "test-bucket".to_string(),
+            endpoint: "https://s3.amazonaws.com".to_string(),
+            root: "/".to_string(),
+            server_side_encryption: None,
+            server_side_encryption_aws_kms_key_id: None,
+            server_side_encryption_customer_algorithm: None,
+            server_side_encryption_customer_key: None,
+            server_side_encryption_customer_key_md5: None,
+            default_storage_class: None,
+            allow_anonymous: false,
+            disable_list_objects_v2: false,
+            enable_request_payer: false,
+            disable_tagging: false, // Tagging enabled
+            signer: signer1,
+            loader: loader1,
+            credential_loaded: AtomicBool::new(false),
+            checksum_algorithm: None,
+        };
+
+        // Test with tagging disabled
+        let core_without_tagging = S3Core {
+            info,
+            bucket: "test-bucket".to_string(),
+            endpoint: "https://s3.amazonaws.com".to_string(),
+            root: "/".to_string(),
+            server_side_encryption: None,
+            server_side_encryption_aws_kms_key_id: None,
+            server_side_encryption_customer_algorithm: None,
+            server_side_encryption_customer_key: None,
+            server_side_encryption_customer_key_md5: None,
+            default_storage_class: None,
+            allow_anonymous: false,
+            disable_list_objects_v2: false,
+            enable_request_payer: false,
+            disable_tagging: true, // Tagging disabled
+            signer: signer2,
+            loader: loader2,
+            credential_loaded: AtomicBool::new(false),
+            checksum_algorithm: None,
+        };
+
+        // Create test tags
+        let mut tags = HashMap::new();
+        tags.insert("env".to_string(), "test".to_string());
+        tags.insert("team".to_string(), "engineering".to_string());
+
+        let op_write = OpWrite::default().with_tags(tags.clone());
+
+        // Test with tagging enabled - should include x-amz-tagging header
+        let req_builder = http::request::Builder::new();
+        let req_with_tags =
+            core_with_tagging.insert_metadata_headers(req_builder, Some(1024), &op_write);
+        let req_with_tags = req_with_tags.body(()).unwrap();
+
+        let tagging_header = req_with_tags.headers().get("x-amz-tagging");
+        assert!(
+            tagging_header.is_some(),
+            "Expected x-amz-tagging header when tagging is enabled"
+        );
+
+        let tagging_value = tagging_header.unwrap().to_str().unwrap();
+        assert!(
+            tagging_value.contains("env=test"),
+            "Expected tag env=test in tagging header"
+        );
+        assert!(
+            tagging_value.contains("team=engineering"),
+            "Expected tag team=engineering in tagging header"
+        );
+
+        // Test with tagging disabled - should not include x-amz-tagging header
+        let req_builder = http::request::Builder::new();
+        let req_without_tags =
+            core_without_tagging.insert_metadata_headers(req_builder, Some(1024), &op_write);
+        let req_without_tags = req_without_tags.body(()).unwrap();
+
+        let tagging_header = req_without_tags.headers().get("x-amz-tagging");
+        assert!(
+            tagging_header.is_none(),
+            "Expected no x-amz-tagging header when tagging is disabled"
+        );
+    }
 }

core/src/types/capability.rs

@@ -126,6 +126,8 @@ pub struct Capability {
     pub write_with_if_not_exists: bool,
     /// Indicates if custom user metadata can be attached during write operations.
     pub write_with_user_metadata: bool,
+    /// Indicates if object tags can be attached during write operations.
+    pub write_with_tags: bool,
     /// Maximum size supported for multipart uploads.
     /// For example, AWS S3 supports up to 5GiB per part in multipart uploads.
     pub write_multi_max_size: Option<usize>,

core/src/types/options.rs

@@ -499,6 +499,27 @@ pub struct WriteOptions {
     /// - Better utilize available bandwidth
     /// - Trade memory for performance
     pub concurrent: usize,
+
+    /// Sets object tags for this write request.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::write_with_tags`] before using this feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If supported, the tags will be attached to the object during write
+    /// - Accepts key-value pairs where both key and value are strings
+    /// - Services may have limitations for tags, for example:
+    ///   - Maximum number of tags per object
+    ///   - Key and value length restrictions
+    ///   - Character restrictions for keys and values
+    /// - If not supported or disabled, the tags will be ignored
+    ///
+    /// Object tags provide a way to categorize and manage objects using metadata.
+    /// Tags can be used for access control, cost allocation, and lifecycle management.
+    pub tags: Option<HashMap<String, String>>,
+
     /// Sets chunk size for buffered writes.
     ///
     /// ### Capability