feat(services/s3): support source-side conditional copy

core/core/src/layers/correctness_check.rs

@@ -217,6 +217,38 @@ impl Service for CorrectnessService {
         if args.if_match().is_some() && !capability.copy_with_if_match {
             return Err(new_unsupported_error(scheme, Operation::Copy, "if_match"));
         }
+        if args.source_if_match().is_some() && !capability.copy_with_source_if_match {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::Copy,
+                "source_if_match",
+            ));
+        }
+        if args.source_if_none_match().is_some() && !capability.copy_with_source_if_none_match {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::Copy,
+                "source_if_none_match",
+            ));
+        }
+        if args.source_if_modified_since().is_some()
+            && !capability.copy_with_source_if_modified_since
+        {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::Copy,
+                "source_if_modified_since",
+            ));
+        }
+        if args.source_if_unmodified_since().is_some()
+            && !capability.copy_with_source_if_unmodified_since
+        {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::Copy,
+                "source_if_unmodified_since",
+            ));
+        }
         if args.source_version().is_some() && !capability.copy_with_source_version {
             return Err(new_unsupported_error(
                 scheme,

core/core/src/raw/ops.rs

@@ -869,6 +869,10 @@ impl From<options::WriteOptions> for (OpWrite, OpWriter) {
 pub struct OpCopy {
     if_not_exists: bool,
     if_match: Option<String>,
+    source_if_match: Option<String>,
+    source_if_none_match: Option<String>,
+    source_if_modified_since: Option<Timestamp>,
+    source_if_unmodified_since: Option<Timestamp>,
     source_version: Option<String>,
 }
 
@@ -906,6 +910,62 @@ impl OpCopy {
         self.if_match.as_deref()
     }
 
+    /// Set the source_if_match condition for the operation.
+    ///
+    /// When set, the copy operation will only proceed if the source object's
+    /// ETag matches the given value.
+    pub fn with_source_if_match(mut self, source_if_match: impl Into<String>) -> Self {
+        self.source_if_match = Some(source_if_match.into());
+        self
+    }
+
+    /// Get source_if_match condition.
+    pub fn source_if_match(&self) -> Option<&str> {
+        self.source_if_match.as_deref()
+    }
+
+    /// Set the source_if_none_match condition for the operation.
+    ///
+    /// When set, the copy operation will only proceed if the source object's
+    /// ETag does not match the given value.
+    pub fn with_source_if_none_match(mut self, source_if_none_match: impl Into<String>) -> Self {
+        self.source_if_none_match = Some(source_if_none_match.into());
+        self
+    }
+
+    /// Get source_if_none_match condition.
+    pub fn source_if_none_match(&self) -> Option<&str> {
+        self.source_if_none_match.as_deref()
+    }
+
+    /// Set the source_if_modified_since condition for the operation.
+    ///
+    /// When set, the copy operation will only proceed if the source object has
+    /// been modified after the given timestamp.
+    pub fn with_source_if_modified_since(mut self, v: Timestamp) -> Self {
+        self.source_if_modified_since = Some(v);
+        self
+    }
+
+    /// Get source_if_modified_since condition.
+    pub fn source_if_modified_since(&self) -> Option<Timestamp> {
+        self.source_if_modified_since
+    }
+
+    /// Set the source_if_unmodified_since condition for the operation.
+    ///
+    /// When set, the copy operation will only proceed if the source object has
+    /// not been modified after the given timestamp.
+    pub fn with_source_if_unmodified_since(mut self, v: Timestamp) -> Self {
+        self.source_if_unmodified_since = Some(v);
+        self
+    }
+
+    /// Get source_if_unmodified_since condition.
+    pub fn source_if_unmodified_since(&self) -> Option<Timestamp> {
+        self.source_if_unmodified_since
+    }
+
     /// Set source version for the operation.
     ///
     /// When set, the copy operation will copy from the specified source version.
@@ -974,6 +1034,10 @@ impl From<options::CopyOptions> for (OpCopy, OpCopier) {
             OpCopy {
                 if_not_exists: value.if_not_exists,
                 if_match: value.if_match,
+                source_if_match: value.source_if_match,
+                source_if_none_match: value.source_if_none_match,
+                source_if_modified_since: value.source_if_modified_since,
+                source_if_unmodified_since: value.source_if_unmodified_since,
                 source_version: value.source_version,
             },
             OpCopier {

core/core/src/types/capability.rs

@@ -153,6 +153,14 @@ pub struct Capability {
     pub copy_with_if_not_exists: bool,
     /// Indicates if conditional copy operations with if-match are supported.
     pub copy_with_if_match: bool,
+    /// Indicates if conditional copy operations with source-side if-match are supported.
+    pub copy_with_source_if_match: bool,
+    /// Indicates if conditional copy operations with source-side if-none-match are supported.
+    pub copy_with_source_if_none_match: bool,
+    /// Indicates if conditional copy operations with source-side if-modified-since are supported.
+    pub copy_with_source_if_modified_since: bool,
+    /// Indicates if conditional copy operations with source-side if-unmodified-since are supported.
+    pub copy_with_source_if_unmodified_since: bool,
     /// Indicates if copy operations from a specific source version are supported.
     pub copy_with_source_version: bool,
     /// Indicates if copy operations can be split into multiple server-side tasks.

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

@@ -1450,6 +1450,42 @@ impl<F: Future<Output = Result<Metadata>>> FutureCopy<F> {
         self
     }
 
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object currently has the given ETag.
+    ///
+    /// Refer to [`options::CopyOptions::source_if_match`] for more details.
+    pub fn source_if_match(mut self, etag: &str) -> Self {
+        self.args.0.source_if_match = Some(etag.to_string());
+        self
+    }
+
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object's ETag does not match the given value.
+    ///
+    /// Refer to [`options::CopyOptions::source_if_none_match`] for more details.
+    pub fn source_if_none_match(mut self, etag: &str) -> Self {
+        self.args.0.source_if_none_match = Some(etag.to_string());
+        self
+    }
+
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object has been modified after the given timestamp.
+    ///
+    /// Refer to [`options::CopyOptions::source_if_modified_since`] for more details.
+    pub fn source_if_modified_since(mut self, v: Timestamp) -> Self {
+        self.args.0.source_if_modified_since = Some(v);
+        self
+    }
+
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object has not been modified after the given timestamp.
+    ///
+    /// Refer to [`options::CopyOptions::source_if_unmodified_since`] for more details.
+    pub fn source_if_unmodified_since(mut self, v: Timestamp) -> Self {
+        self.args.0.source_if_unmodified_since = Some(v);
+        self
+    }
+
     /// Sets source version for this copy operation.
     ///
     /// Refer to [`options::CopyOptions::source_version`] for more details.

core/core/src/types/options.rs

@@ -562,6 +562,61 @@ pub struct CopyOptions {
     ///   destination object's ETag matches the given value.
     pub if_match: Option<String>,
 
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object currently has the given ETag.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::copy_with_source_if_match`] before using this feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If supported, the copy operation will only succeed when the source
+    ///   object's ETag matches the given value.
+    pub source_if_match: Option<String>,
+
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object's ETag does not match the given value.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::copy_with_source_if_none_match`] before using this
+    /// feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If supported, the copy operation will only succeed when the source
+    ///   object's ETag does not match the given value.
+    pub source_if_none_match: Option<String>,
+
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object has been modified after the given timestamp.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::copy_with_source_if_modified_since`] before using
+    /// this feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If supported, the copy operation will only succeed when the source
+    ///   object has been modified after the given timestamp.
+    pub source_if_modified_since: Option<Timestamp>,
+
+    /// Sets the condition that copy operation will succeed only if the source
+    /// object has not been modified after the given timestamp.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::copy_with_source_if_unmodified_since`] before using
+    /// this feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If supported, the copy operation will only succeed when the source
+    ///   object has not been modified after the given timestamp.
+    pub source_if_unmodified_since: Option<Timestamp>,
+
     /// Copy from a specific source object version.
     ///
     /// ### Capability

core/layers/capability-check/src/lib.rs

@@ -175,6 +175,38 @@ impl Service for CapabilityCheckService {
         if args.if_match().is_some() && !capability.copy_with_if_match {
             return Err(new_unsupported_error(&info, Operation::Copy, "if_match"));
         }
+        if args.source_if_match().is_some() && !capability.copy_with_source_if_match {
+            return Err(new_unsupported_error(
+                &info,
+                Operation::Copy,
+                "source_if_match",
+            ));
+        }
+        if args.source_if_none_match().is_some() && !capability.copy_with_source_if_none_match {
+            return Err(new_unsupported_error(
+                &info,
+                Operation::Copy,
+                "source_if_none_match",
+            ));
+        }
+        if args.source_if_modified_since().is_some()
+            && !capability.copy_with_source_if_modified_since
+        {
+            return Err(new_unsupported_error(
+                &info,
+                Operation::Copy,
+                "source_if_modified_since",
+            ));
+        }
+        if args.source_if_unmodified_since().is_some()
+            && !capability.copy_with_source_if_unmodified_since
+        {
+            return Err(new_unsupported_error(
+                &info,
+                Operation::Copy,
+                "source_if_unmodified_since",
+            ));
+        }
         if args.source_version().is_some() && !capability.copy_with_source_version {
             return Err(new_unsupported_error(
                 &info,

core/services/s3/src/backend.rs

@@ -982,6 +982,10 @@ impl Builder for S3Builder {
                     copy_with_if_not_exists: true,
                     copy_with_if_match: true,
                     copy_with_source_version: true,
+                    copy_with_source_if_match: true,
+                    copy_with_source_if_none_match: true,
+                    copy_with_source_if_modified_since: true,
+                    copy_with_source_if_unmodified_since: true,
                     // The min multipart size of S3 is 5 MiB.
                     //
                     // ref: <https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html>

core/services/s3/src/core.rs

@@ -50,6 +50,10 @@ use opendal_core::*;
 pub mod constants {
     pub const X_AMZ_COPY_SOURCE: &str = "x-amz-copy-source";
     pub const X_AMZ_COPY_SOURCE_RANGE: &str = "x-amz-copy-source-range";
+    pub const X_AMZ_COPY_SOURCE_IF_MATCH: &str = "x-amz-copy-source-if-match";
+    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_SERVER_SIDE_ENCRYPTION: &str = "x-amz-server-side-encryption";
     pub const X_AMZ_SERVER_REQUEST_PAYER: (&str, &str) = ("x-amz-request-payer", "requester");
@@ -717,6 +721,27 @@ impl S3Core {
         if let Some(if_match) = args.if_match() {
             req = req.header(IF_MATCH, if_match);
         }
+        if let Some(source_if_match) = args.source_if_match() {
+            req = req.header(constants::X_AMZ_COPY_SOURCE_IF_MATCH, source_if_match);
+        }
+        if let Some(source_if_none_match) = args.source_if_none_match() {
+            req = req.header(
+                constants::X_AMZ_COPY_SOURCE_IF_NONE_MATCH,
+                source_if_none_match,
+            );
+        }
+        if let Some(v) = args.source_if_modified_since() {
+            req = req.header(
+                constants::X_AMZ_COPY_SOURCE_IF_MODIFIED_SINCE,
+                v.format_http_date(),
+            );
+        }
+        if let Some(v) = args.source_if_unmodified_since() {
+            req = req.header(
+                constants::X_AMZ_COPY_SOURCE_IF_UNMODIFIED_SINCE,
+                v.format_http_date(),
+            );
+        }
 
         // Set SSE headers.
         req = self.insert_sse_headers(req, true);
@@ -1186,6 +1211,27 @@ impl S3Core {
         if let Some(if_match) = args.if_match() {
             req = req.header(IF_MATCH, if_match);
         }
+        if let Some(source_if_match) = args.source_if_match() {
+            req = req.header(constants::X_AMZ_COPY_SOURCE_IF_MATCH, source_if_match);
+        }
+        if let Some(source_if_none_match) = args.source_if_none_match() {
+            req = req.header(
+                constants::X_AMZ_COPY_SOURCE_IF_NONE_MATCH,
+                source_if_none_match,
+            );
+        }
+        if let Some(v) = args.source_if_modified_since() {
+            req = req.header(
+                constants::X_AMZ_COPY_SOURCE_IF_MODIFIED_SINCE,
+                v.format_http_date(),
+            );
+        }
+        if let Some(v) = args.source_if_unmodified_since() {
+            req = req.header(
+                constants::X_AMZ_COPY_SOURCE_IF_UNMODIFIED_SINCE,
+                v.format_http_date(),
+            );
+        }
 
         // Set request payer header if enabled.
         req = self.insert_request_payer_header(req);