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

core/core/src/layers/correctness_check.rs

@@ -240,6 +240,38 @@ impl<A: Access> LayeredAccess for CorrectnessAccessor<A> {
                 "if_match",
             ));
         }
+        if args.source_if_match().is_some() && !capability.copy_with_source_if_match {
+            return Err(new_unsupported_error(
+                &self.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(
+                &self.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(
+                &self.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(
+                &self.info,
+                Operation::Copy,
+                "source_if_unmodified_since",
+            ));
+        }
 
         self.inner.copy(from, to, args, opts).await
     }

core/core/src/raw/ops.rs

@@ -895,6 +895,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>,
 }
 
 impl OpCopy {
@@ -930,6 +934,62 @@ impl OpCopy {
     pub fn if_match(&self) -> Option<&str> {
         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
+    }
 }
 
 /// Args for `copier` operation.
@@ -986,6 +1046,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,
             },
             OpCopier {
                 concurrent: value.concurrent.max(1),

core/core/src/types/capability.rs

@@ -158,6 +158,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 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_futures.rs

@@ -1437,6 +1437,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 concurrent copy operations for this copy.
     ///
     /// Refer to [`options::CopyOptions::concurrent`] for more details.

core/core/src/types/options.rs

@@ -560,6 +560,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>,
+
     /// Known content length of the source object.
     ///
     /// This is an execution hint that allows OpenDAL to avoid extra metadata

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

@@ -162,6 +162,38 @@ impl<A: Access> LayeredAccess for CapabilityAccessor<A> {
                 "if_not_exists",
             ));
         }
+        if args.source_if_match().is_some() && !capability.copy_with_source_if_match {
+            return Err(new_unsupported_error(
+                self.info.as_ref(),
+                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(
+                self.info.as_ref(),
+                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(
+                self.info.as_ref(),
+                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(
+                self.info.as_ref(),
+                Operation::Copy,
+                "source_if_unmodified_since",
+            ));
+        }
 
         self.inner.copy(from, to, args, opts).await
     }

core/services/s3/src/backend.rs

@@ -983,6 +983,10 @@ impl Builder for S3Builder {
                             copy_can_multi: true,
                             copy_with_if_not_exists: true,
                             copy_with_if_match: 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

@@ -49,6 +49,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");
@@ -669,6 +673,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);
@@ -1119,6 +1144,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);