feat(core): copy_with can replace user metadata and content type

core/core/src/layers/correctness_check.rs

@@ -325,6 +325,20 @@ impl Service for CorrectnessService {
                 "if_version_not_match",
             ));
         }
+        if args.user_metadata().is_some() && !capability.copy_with_user_metadata {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::Copy,
+                "user_metadata",
+            ));
+        }
+        if args.content_type().is_some() && !capability.copy_with_content_type {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::Copy,
+                "content_type",
+            ));
+        }
         if args.source_version().is_some() && !capability.copy_with_source_version {
             return Err(new_unsupported_error(
                 scheme,
@@ -919,6 +933,47 @@ mod tests {
         .expect("supported compose options must be forwarded");
     }
 
+    #[tokio::test]
+    async fn test_copy_with_metadata_replacement() {
+        let op = new_test_operator(Capability {
+            copy: true,
+            ..Default::default()
+        });
+
+        let err = op
+            .copy_with("from", "to")
+            .user_metadata([("mtime".to_owned(), "42".to_owned())])
+            .await
+            .expect_err("user metadata must require a capability");
+        assert_eq!(err.kind(), ErrorKind::Unsupported);
+
+        let err = op
+            .copy_with("from", "to")
+            .content_type("text/plain")
+            .await
+            .expect_err("content type must require a capability");
+        assert_eq!(err.kind(), ErrorKind::Unsupported);
+
+        let op = new_test_operator(Capability {
+            copy: true,
+            copy_with_user_metadata: true,
+            copy_with_content_type: true,
+            ..Default::default()
+        });
+
+        let err = op
+            .copy("same", "same")
+            .await
+            .expect_err("a plain copy onto itself must fail");
+        assert_eq!(err.kind(), ErrorKind::IsSameFile);
+
+        op.copy_with("same", "same")
+            .user_metadata([("mtime".to_owned(), "42".to_owned())])
+            .content_type("text/plain")
+            .await
+            .expect("replacing metadata in place must be forwarded");
+    }
+
     #[tokio::test]
     async fn test_version_preconditions_are_forwarded() -> Result<()> {
         let op = new_test_operator(Capability {

core/core/src/raw/ops.rs

@@ -1269,6 +1269,8 @@ enum CopyField {
     IfVersionMatch,
     IfVersionNotMatch,
     SourceVersion,
+    ContentType,
+    UserMetadata,
 }
 
 impl OpCopy {
@@ -1321,13 +1323,19 @@ impl OpCopy {
             }
         }
 
+        let user_metadata = options.user_metadata.take().map(sorted_user_metadata);
         let fields = [
             options.if_match.as_deref().map(str::as_bytes),
             options.if_none_match.as_deref().map(str::as_bytes),
             options.if_version_match.as_deref().map(str::as_bytes),
             options.if_version_not_match.as_deref().map(str::as_bytes),
             options.source_version.as_deref().map(str::as_bytes),
+            options.content_type.as_deref().map(str::as_bytes),
+            None,
         ];
+        let mut lengths = fields.map(|value| value.map(<[u8]>::len));
+        lengths[CopyField::UserMetadata as usize] =
+            user_metadata.as_deref().map(user_metadata_encoded_len);
         Ok(Self {
             concurrent: options.concurrent.max(1),
             chunk: options.chunk,
@@ -1337,7 +1345,17 @@ impl OpCopy {
             } else {
                 0
             },
-            values: CompactValues::encode(&fields),
+            values: CompactValues::encode_with(&lengths, |field, output| match field {
+                field if field == CopyField::UserMetadata as usize => {
+                    write_user_metadata(
+                        user_metadata.as_deref().expect("present field has a value"),
+                        output,
+                    );
+                }
+                _ => {
+                    output.write(fields[field].expect("present field has a value"));
+                }
+            }),
         })
     }
 
@@ -1390,6 +1408,20 @@ impl OpCopy {
         string_value(&self.values, CopyField::SourceVersion as usize)
     }
 
+    /// Get the replacement destination Content-Type.
+    #[inline]
+    pub fn content_type(&self) -> Option<&str> {
+        string_value(&self.values, CopyField::ContentType as usize)
+    }
+
+    /// Get the replacement destination user metadata.
+    #[inline]
+    pub fn user_metadata(&self) -> Option<UserMetadata<'_>> {
+        self.values
+            .get(CopyField::UserMetadata as usize)
+            .map(UserMetadata::new)
+    }
+
     /// Get the concurrent tasks for the copy operation.
     pub fn concurrent(&self) -> usize {
         self.concurrent.max(1)
@@ -1749,6 +1781,21 @@ mod tests {
         assert_eq!(copy.if_version_match(), Some("destination-version"));
         assert_eq!(copy.source_version(), Some("source-version"));
         assert!(copy.is_conditional());
+        assert!(copy.user_metadata().is_none());
+
+        let copy = OpCopy::from_options(
+            &Capability::default(),
+            options::CopyOptions {
+                if_match: Some("etag".to_owned()),
+                content_type: Some("text/plain".to_owned()),
+                user_metadata: Some(HashMap::from([("mtime".to_owned(), "42".to_owned())])),
+                ..Default::default()
+            },
+        )
+        .unwrap();
+        assert_eq!(copy.if_match(), Some("etag"));
+        assert_eq!(copy.content_type(), Some("text/plain"));
+        assert_eq!(copy.user_metadata().unwrap().get("mtime"), Some("42"));
 
         let list: OpList = options::ListOptions {
             limit: Some(100),

core/core/src/types/capability.rs

@@ -186,6 +186,10 @@ pub struct Capability {
     pub copy_with_if_version_not_match: bool,
     /// Indicates if copy operations from a specific source version are supported.
     pub copy_with_source_version: bool,
+    /// Indicates if copy operations can replace the destination's user metadata.
+    pub copy_with_user_metadata: bool,
+    /// Indicates if copy operations can replace the destination's Content-Type.
+    pub copy_with_content_type: bool,
     /// Indicates if copy operations can be split into multiple server-side tasks.
     pub copy_can_multi: bool,
     /// Maximum size supported for segmented copy tasks.

core/core/src/types/operator/operator.rs

@@ -1347,8 +1347,8 @@ impl Operator {
     /// # Notes
     ///
     /// - `from` and `to` must be a file.
-    /// - If `from` and `to` are the same and `source_version` is not set,
-    ///   an `IsSameFile` error will occur.
+    /// - If `from` and `to` are the same and none of `source_version`,
+    ///   `user_metadata` or `content_type` is set, an `IsSameFile` error will occur.
     /// - `copy` is idempotent. For same `from` and `to` input, the result will be the same.
     ///
     /// # Options
@@ -1410,8 +1410,8 @@ impl Operator {
     /// # Notes
     ///
     /// - `from` and `to` must be a file.
-    /// - If `from` and `to` are the same and `source_version` is not set,
-    ///   an `IsSameFile` error will occur.
+    /// - If `from` and `to` are the same and none of `source_version`,
+    ///   `user_metadata` or `content_type` is set, an `IsSameFile` error will occur.
     /// - `copy` is idempotent. For same `from` and `to` input, the result will be the same.
     ///
     /// # Options
@@ -1497,7 +1497,11 @@ impl Operator {
             );
         }
 
-        if from == to && opts.source_version.is_none() {
+        if from == to
+            && opts.source_version.is_none()
+            && opts.user_metadata.is_none()
+            && opts.content_type.is_none()
+        {
             return Err(
                 Error::new(ErrorKind::IsSameFile, "from and to paths are same")
                     .with_operation("Operator::copy")
@@ -1543,7 +1547,11 @@ impl Operator {
             );
         }
 
-        if from == to && opts.source_version.is_none() {
+        if from == to
+            && opts.source_version.is_none()
+            && opts.user_metadata.is_none()
+            && opts.content_type.is_none()
+        {
             return Err(
                 Error::new(ErrorKind::IsSameFile, "from and to paths are same")
                     .with_operation("Operator::copier")

core/core/src/types/operator/operator_futures.rs

@@ -1763,6 +1763,22 @@ impl<F: Future<Output = Result<Metadata>>> FutureCopy<F> {
         self
     }
 
+    /// Replace the destination's user metadata with these entries.
+    ///
+    /// Refer to [`options::CopyOptions::user_metadata`] for the full contract.
+    pub fn user_metadata(mut self, data: impl IntoIterator<Item = (String, String)>) -> Self {
+        self.args.0.user_metadata = Some(HashMap::from_iter(data));
+        self
+    }
+
+    /// Replace the destination's Content-Type with this value.
+    ///
+    /// Refer to [`options::CopyOptions::content_type`] for the full contract.
+    pub fn content_type(mut self, v: &str) -> Self {
+        self.args.0.content_type = Some(v.to_string());
+        self
+    }
+
     /// Sets concurrent copy operations for this copy.
     ///
     /// Refer to [`options::CopyOptions::concurrent`] for more details.
@@ -1850,6 +1866,22 @@ impl<F: Future<Output = Result<Copier>>> FutureCopier<F> {
         self
     }
 
+    /// Replace the destination's user metadata with these entries.
+    ///
+    /// Refer to [`options::CopyOptions::user_metadata`] for the full contract.
+    pub fn user_metadata(mut self, data: impl IntoIterator<Item = (String, String)>) -> Self {
+        self.args.0.user_metadata = Some(HashMap::from_iter(data));
+        self
+    }
+
+    /// Replace the destination's Content-Type with this value.
+    ///
+    /// Refer to [`options::CopyOptions::content_type`] for the full contract.
+    pub fn content_type(mut self, v: &str) -> Self {
+        self.args.0.content_type = Some(v.to_string());
+        self
+    }
+
     /// Sets known source content length as a hint for this `copier`.
     ///
     /// Refer to [`options::CopyOptions::source_content_length_hint`] for more details.

core/core/src/types/options.rs

@@ -1005,6 +1005,29 @@ pub struct CopyOptions {
     /// this option.
     pub source_version: Option<String>,
 
+    /// Replaces the destination's user metadata with exactly these entries.
+    ///
+    /// When this option or [`Self::content_type`] is set, the destination does
+    /// not inherit metadata from the source. User metadata and Content-Type
+    /// take the values given here; anything left unset, including other system
+    /// metadata such as Cache-Control, takes the service default. Setting
+    /// either option also allows `from` and `to` to be the same path, which
+    /// updates metadata without rewriting the content. On S3 this is
+    /// `CopyObject` with `x-amz-metadata-directive: REPLACE`.
+    ///
+    /// The operation returns [`crate::ErrorKind::Unsupported`] when the
+    /// service does not advertise
+    /// [`crate::Capability::copy_with_user_metadata`].
+    pub user_metadata: Option<HashMap<String, String>>,
+    /// Replaces the destination's Content-Type with this value.
+    ///
+    /// Follows the same replacement rules as [`Self::user_metadata`].
+    ///
+    /// The operation returns [`crate::ErrorKind::Unsupported`] when the
+    /// service does not advertise
+    /// [`crate::Capability::copy_with_content_type`].
+    pub content_type: Option<String>,
+
     /// Asserted complete content length of the source object.
     ///
     /// OpenDAL may trust this value without reading source metadata. It can use

core/services/s3/src/backend.rs

@@ -1127,6 +1127,8 @@ impl Builder for S3Builder {
                     copy_with_if_not_exists: true,
                     copy_with_if_match: true,
                     copy_with_source_version: !is_s3_express,
+                    copy_with_user_metadata: true,
+                    copy_with_content_type: true,
                     // The min multipart size of S3 is 5 MiB.
                     //
                     // ref: <https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html>
@@ -1944,6 +1946,59 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_s3_copy_replaces_metadata() {
+        let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success);
+        let op = s3_general_operator(transport.clone());
+
+        op.copy_with("source", "source")
+            .user_metadata([("mtime".to_owned(), "42".to_owned())])
+            .content_type("text/plain")
+            .source_content_length_hint(7)
+            .await
+            .expect("same-key copy with new metadata must succeed");
+
+        let requests = transport.requests();
+        assert_eq!(requests.len(), 1);
+        assert_eq!(
+            header(&requests[0], constants::X_AMZ_METADATA_DIRECTIVE),
+            "REPLACE"
+        );
+        assert_eq!(header(&requests[0], "x-amz-meta-mtime"), "42");
+        assert_eq!(header(&requests[0], "content-type"), "text/plain");
+
+        op.copy_with("source", "target")
+            .source_content_length_hint(7)
+            .await
+            .expect("copy must succeed");
+        let requests = transport.requests();
+        assert!(
+            !requests[1]
+                .headers
+                .contains_key(constants::X_AMZ_METADATA_DIRECTIVE)
+        );
+    }
+
+    #[tokio::test]
+    async fn test_s3_multipart_copy_sets_metadata_on_create() {
+        let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success);
+        let op = s3_general_operator(transport.clone());
+
+        op.copy_with("source", "target")
+            .user_metadata([("mtime".to_owned(), "42".to_owned())])
+            .source_content_length_hint(5 * 1024 * 1024 + 1)
+            .chunk(5 * 1024 * 1024)
+            .await
+            .expect("multipart copy must succeed");
+
+        let requests = transport.requests();
+        let create = requests
+            .iter()
+            .find(|request| request.uri.query() == Some("uploads"))
+            .expect("CreateMultipartUpload request must be captured");
+        assert_eq!(header(create, "x-amz-meta-mtime"), "42");
+    }
+
     #[tokio::test]
     async fn test_s3_multipart_copy_without_hint_pins_every_part() {
         let source_content_length = 5 * 1024 * 1024 + 1;

core/services/s3/src/copier.rs

@@ -249,7 +249,7 @@ impl oio::MultipartCopy for S3Copier {
     async fn initiate_copy(&self) -> Result<String> {
         let resp = self
             .core
-            .s3_initiate_multipart_copy(&self.ctx, &self.to)
+            .s3_initiate_multipart_copy(&self.ctx, &self.to, &self.args)
             .await?;
 
         match resp.status() {