From 685a65258b0ac1359cce94efa3d81e6904730c36 Mon Sep 17 00:00:00 2001 From: Dmitriy Tarasenko Date: Sun, 25 Oct 2026 18:39:00 +0200 Subject: [PATCH] feat(core): copy_with can replace user metadata and content type Adds user_metadata and content_type to CopyOptions with matching capability flags. S3 sends x-amz-metadata-directive: REPLACE, so a same-key copy updates metadata in one zero-body CopyObject instead of a read and rewrite. --- core/core/src/layers/correctness_check.rs | 55 +++++++++++++++++++ core/core/src/raw/ops.rs | 49 ++++++++++++++++- core/core/src/types/capability.rs | 4 ++ core/core/src/types/operator/operator.rs | 20 +++++-- .../src/types/operator/operator_futures.rs | 32 +++++++++++ core/core/src/types/options.rs | 23 ++++++++ core/services/s3/src/backend.rs | 55 +++++++++++++++++++ core/services/s3/src/copier.rs | 2 +- core/services/s3/src/core.rs | 25 +++++++++ core/tests/behavior/async_copy.rs | 36 ++++++++++++ 10 files changed, 293 insertions(+), 8 deletions(-) diff --git a/core/core/src/layers/correctness_check.rs b/core/core/src/layers/correctness_check.rs index 4919054b8d18..e6ef6db26809 100644 --- a/core/core/src/layers/correctness_check.rs +++ b/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 { diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs index a9678e553c4e..0d86f8b95a3d 100644 --- a/core/core/src/raw/ops.rs +++ b/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> { + 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), diff --git a/core/core/src/types/capability.rs b/core/core/src/types/capability.rs index 332a3ec04308..3cf6240764cc 100644 --- a/core/core/src/types/capability.rs +++ b/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. diff --git a/core/core/src/types/operator/operator.rs b/core/core/src/types/operator/operator.rs index 4e7d9b896495..4da583884838 100644 --- a/core/core/src/types/operator/operator.rs +++ b/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") diff --git a/core/core/src/types/operator/operator_futures.rs b/core/core/src/types/operator/operator_futures.rs index 924ff88cfbd4..c97e8af3d545 100644 --- a/core/core/src/types/operator/operator_futures.rs +++ b/core/core/src/types/operator/operator_futures.rs @@ -1763,6 +1763,22 @@ impl>> FutureCopy { 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) -> 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>> FutureCopier { 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) -> 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. diff --git a/core/core/src/types/options.rs b/core/core/src/types/options.rs index ec2e91c872dc..f9f4d4ab3580 100644 --- a/core/core/src/types/options.rs +++ b/core/core/src/types/options.rs @@ -1005,6 +1005,29 @@ pub struct CopyOptions { /// this option. pub source_version: Option, + /// 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>, + /// 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, + /// Asserted complete content length of the source object. /// /// OpenDAL may trust this value without reading source metadata. It can use diff --git a/core/services/s3/src/backend.rs b/core/services/s3/src/backend.rs index 655e522b6a03..f0ec2092caeb 100644 --- a/core/services/s3/src/backend.rs +++ b/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: @@ -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; diff --git a/core/services/s3/src/copier.rs b/core/services/s3/src/copier.rs index 777b0fb2fc0d..45f710c5132d 100644 --- a/core/services/s3/src/copier.rs +++ b/core/services/s3/src/copier.rs @@ -249,7 +249,7 @@ impl oio::MultipartCopy for S3Copier { async fn initiate_copy(&self) -> Result { 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() { diff --git a/core/services/s3/src/core.rs b/core/services/s3/src/core.rs index d4eb10cb11e9..9c7ee02c13a1 100644 --- a/core/services/s3/src/core.rs +++ b/core/services/s3/src/core.rs @@ -56,6 +56,7 @@ pub mod constants { pub const X_AMZ_COPY_SOURCE_IF_NONE_MATCH: &str = "x-amz-copy-source-if-none-match"; pub const X_AMZ_COPY_SOURCE_IF_MODIFIED_SINCE: &str = "x-amz-copy-source-if-modified-since"; pub const X_AMZ_COPY_SOURCE_IF_UNMODIFIED_SINCE: &str = "x-amz-copy-source-if-unmodified-since"; + pub const X_AMZ_METADATA_DIRECTIVE: &str = "x-amz-metadata-directive"; pub const X_AMZ_SERVER_SIDE_ENCRYPTION: &str = "x-amz-server-side-encryption"; pub const X_AMZ_SERVER_REQUEST_PAYER: (&str, &str) = ("x-amz-request-payer", "requester"); @@ -809,6 +810,19 @@ impl S3Core { req = req.header(constants::X_AMZ_COPY_SOURCE_IF_MATCH, source_if_match); } + // Without REPLACE, S3 copies the source metadata and ignores these headers. + if args.content_type().is_some() || args.user_metadata().is_some() { + req = req.header(constants::X_AMZ_METADATA_DIRECTIVE, "REPLACE"); + } + if let Some(mime) = args.content_type() { + req = req.header(CONTENT_TYPE, mime); + } + if let Some(user_metadata) = args.user_metadata() { + for (key, value) in user_metadata { + req = req.header(format!("{X_AMZ_META_PREFIX}{key}"), value); + } + } + // Set SSE headers. req = self.insert_sse_headers(req, true); @@ -1034,6 +1048,7 @@ impl S3Core { &self, ctx: &OperationContext, path: &str, + args: &OpCopy, ) -> Result> { let p = build_abs_path(&self.root, path); @@ -1041,6 +1056,16 @@ impl S3Core { let mut req = Request::post(&url); + // A multipart copy never inherits source metadata, so set it on creation. + if let Some(mime) = args.content_type() { + req = req.header(CONTENT_TYPE, mime); + } + if let Some(user_metadata) = args.user_metadata() { + for (key, value) in user_metadata { + req = req.header(format!("{X_AMZ_META_PREFIX}{key}"), value); + } + } + // Set storage class header. if let Some(v) = &self.default_storage_class { req = req.header(HeaderName::from_static(constants::X_AMZ_STORAGE_CLASS), v); diff --git a/core/tests/behavior/async_copy.rs b/core/tests/behavior/async_copy.rs index 1d819010ac08..e398c96cee75 100644 --- a/core/tests/behavior/async_copy.rs +++ b/core/tests/behavior/async_copy.rs @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +use std::collections::HashMap; + use anyhow::Result; use crate::*; @@ -94,6 +96,10 @@ pub fn tests(op: &Operator, tests: &mut Vec) { test_copy_with_source_version_to_same_file )) } + + if cap.read && cap.write && cap.stat && cap.copy && cap.copy_with_user_metadata { + tests.extend(async_trials!(op, test_copy_with_user_metadata_in_place)) + } } fn copy_multi_chunk_size(cap: Capability) -> Option<(usize, usize)> { @@ -874,3 +880,33 @@ pub async fn test_copier_with_if_not_exists_to_existing_file(op: Operator) -> Re op.delete(&target_path).await.expect("delete must succeed"); Ok(()) } + +/// Copy onto the same path with user_metadata should replace the metadata +/// and keep the content. +pub async fn test_copy_with_user_metadata_in_place(op: Operator) -> Result<()> { + let path = uuid::Uuid::new_v4().to_string(); + let (content, _) = gen_bytes(op.info().capability()); + op.write_with(&path, content.clone()) + .user_metadata([("old".to_string(), "value".to_string())]) + .await?; + + let new_metadata = HashMap::from([("mtime".to_string(), "1700000000".to_string())]); + op.copy_with(&path, &path) + .user_metadata(new_metadata.clone()) + .await?; + + let meta = op.stat(&path).await.expect("stat must succeed"); + let current = meta + .user_metadata() + .expect("user metadata must exist") + .into_iter() + .map(|(key, value)| (key.to_owned(), value.to_owned())) + .collect::>(); + assert_eq!(current, new_metadata); + + let current_content = op.read(&path).await.expect("read must succeed").to_bytes(); + assert_eq!(sha256_digest(current_content), sha256_digest(&content)); + + op.delete(&path).await.expect("delete must succeed"); + Ok(()) +}