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

bindings/c/include/opendal.h

@@ -830,6 +830,30 @@ typedef struct opendal_copy_options {
    * If-Match condition; NULL means unset.
    */
   const char *if_match;
+  /**
+   * Source If-Match condition; NULL means unset.
+   */
+  const char *source_if_match;
+  /**
+   * Source If-None-Match condition; NULL means unset.
+   */
+  const char *source_if_none_match;
+  /**
+   * Whether `source_if_modified_since` has been set.
+   */
+  bool has_source_if_modified_since;
+  /**
+   * Source If-Modified-Since condition, in Unix milliseconds.
+   */
+  int64_t source_if_modified_since;
+  /**
+   * Whether `source_if_unmodified_since` has been set.
+   */
+  bool has_source_if_unmodified_since;
+  /**
+   * Source If-Unmodified-Since condition, in Unix milliseconds.
+   */
+  int64_t source_if_unmodified_since;
   /**
    * Source version; NULL means unset.
    */
@@ -2756,6 +2780,30 @@ void opendal_copy_options_set_if_not_exists(struct opendal_copy_options *opts, b
  */
 void opendal_copy_options_set_if_match(struct opendal_copy_options *opts, const char *if_match);
 
+/**
+ * \brief Set Source If-Match.
+ */
+void opendal_copy_options_set_source_if_match(struct opendal_copy_options *opts,
+                                              const char *source_if_match);
+
+/**
+ * \brief Set Source If-None-Match.
+ */
+void opendal_copy_options_set_source_if_none_match(struct opendal_copy_options *opts,
+                                                   const char *source_if_none_match);
+
+/**
+ * \brief Set Source If-Modified-Since, in Unix milliseconds.
+ */
+void opendal_copy_options_set_source_if_modified_since(struct opendal_copy_options *opts,
+                                                       int64_t source_if_modified_since);
+
+/**
+ * \brief Set Source If-Unmodified-Since, in Unix milliseconds.
+ */
+void opendal_copy_options_set_source_if_unmodified_since(struct opendal_copy_options *opts,
+                                                         int64_t source_if_unmodified_since);
+
 /**
  * \brief Set source version.
  */

bindings/c/src/types.rs

@@ -1313,6 +1313,18 @@ pub struct opendal_copy_options {
     pub if_not_exists: bool,
     /// If-Match condition; NULL means unset.
     pub if_match: *const c_char,
+    /// Source If-Match condition; NULL means unset.
+    pub source_if_match: *const c_char,
+    /// Source If-None-Match condition; NULL means unset.
+    pub source_if_none_match: *const c_char,
+    /// Whether `source_if_modified_since` has been set.
+    pub has_source_if_modified_since: bool,
+    /// Source If-Modified-Since condition, in Unix milliseconds.
+    pub source_if_modified_since: i64,
+    /// Whether `source_if_unmodified_since` has been set.
+    pub has_source_if_unmodified_since: bool,
+    /// Source If-Unmodified-Since condition, in Unix milliseconds.
+    pub source_if_unmodified_since: i64,
     /// Source version; NULL means unset.
     pub source_version: *const c_char,
     /// Whether `source_content_length_hint` has been set.
@@ -1364,6 +1376,52 @@ impl opendal_copy_options {
         }
     }
 
+    /// \brief Set Source If-Match.
+    #[no_mangle]
+    pub unsafe extern "C" fn opendal_copy_options_set_source_if_match(
+        opts: *mut opendal_copy_options,
+        source_if_match: *const c_char,
+    ) {
+        if !opts.is_null() {
+            (*opts).source_if_match = source_if_match;
+        }
+    }
+
+    /// \brief Set Source If-None-Match.
+    #[no_mangle]
+    pub unsafe extern "C" fn opendal_copy_options_set_source_if_none_match(
+        opts: *mut opendal_copy_options,
+        source_if_none_match: *const c_char,
+    ) {
+        if !opts.is_null() {
+            (*opts).source_if_none_match = source_if_none_match;
+        }
+    }
+
+    /// \brief Set Source If-Modified-Since, in Unix milliseconds.
+    #[no_mangle]
+    pub unsafe extern "C" fn opendal_copy_options_set_source_if_modified_since(
+        opts: *mut opendal_copy_options,
+        source_if_modified_since: i64,
+    ) {
+        if !opts.is_null() {
+            (*opts).has_source_if_modified_since = true;
+            (*opts).source_if_modified_since = source_if_modified_since;
+        }
+    }
+
+    /// \brief Set Source If-Unmodified-Since, in Unix milliseconds.
+    #[no_mangle]
+    pub unsafe extern "C" fn opendal_copy_options_set_source_if_unmodified_since(
+        opts: *mut opendal_copy_options,
+        source_if_unmodified_since: i64,
+    ) {
+        if !opts.is_null() {
+            (*opts).has_source_if_unmodified_since = true;
+            (*opts).source_if_unmodified_since = source_if_unmodified_since;
+        }
+    }
+
     /// \brief Set source version.
     #[no_mangle]
     pub unsafe extern "C" fn opendal_copy_options_set_source_version(
@@ -1416,6 +1474,12 @@ impl Default for opendal_copy_options {
         Self {
             if_not_exists: false,
             if_match: std::ptr::null(),
+            source_if_match: std::ptr::null(),
+            source_if_none_match: std::ptr::null(),
+            has_source_if_modified_since: false,
+            source_if_modified_since: 0,
+            has_source_if_unmodified_since: false,
+            source_if_unmodified_since: 0,
             source_version: std::ptr::null(),
             has_source_content_length_hint: false,
             source_content_length_hint: 0,
@@ -1431,6 +1495,16 @@ impl From<&opendal_copy_options> for options::CopyOptions {
         Self {
             if_not_exists: value.if_not_exists,
             if_match: unsafe { optional_cstr(value.if_match) },
+            source_if_match: unsafe { optional_cstr(value.source_if_match) },
+            source_if_none_match: unsafe { optional_cstr(value.source_if_none_match) },
+            source_if_modified_since: value
+                .has_source_if_modified_since
+                .then(|| Timestamp::from_millisecond(value.source_if_modified_since).ok())
+                .flatten(),
+            source_if_unmodified_since: value
+                .has_source_if_unmodified_since
+                .then(|| Timestamp::from_millisecond(value.source_if_unmodified_since).ok())
+                .flatten(),
             source_version: unsafe { optional_cstr(value.source_version) },
             source_content_length_hint: value
                 .has_source_content_length_hint

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.