feat(core): add if_not_exist in OpWrite (#5305)

core/src/raw/ops.rs

@@ -601,6 +601,7 @@ pub struct OpWrite {
     cache_control: Option<String>,
     executor: Option<Executor>,
     if_none_match: Option<String>,
+    if_not_exists: bool,
     user_metadata: Option<HashMap<String, String>>,
 }
 
@@ -697,6 +698,17 @@ impl OpWrite {
         self.if_none_match.as_deref()
     }
 
+    /// Set the If-Not-Exist of the option
+    pub fn with_if_not_exists(mut self, b: bool) -> Self {
+        self.if_not_exists = b;
+        self
+    }
+
+    /// Get If-Not-Exist from option
+    pub fn if_not_exists(&self) -> bool {
+        self.if_not_exists
+    }
+
     /// Merge given executor into option.
     ///
     /// If executor has already been set, this will do nothing.

core/src/services/s3/backend.rs

@@ -924,7 +924,7 @@ impl Access for S3Backend {
                 write_can_multi: true,
                 write_with_cache_control: true,
                 write_with_content_type: true,
-                write_with_if_none_match: true,
+                write_with_if_not_exists: true,
                 write_with_user_metadata: true,
 
                 // The min multipart size of S3 is 5 MiB.

core/src/services/s3/core.rs

@@ -476,8 +476,8 @@ impl S3Core {
             req = self.insert_checksum_header(req, &checksum);
         }
 
-        if let Some(if_none_match) = args.if_none_match() {
-            req = req.header(IF_NONE_MATCH, if_none_match);
+        if args.if_not_exists() {
+            req = req.header(IF_NONE_MATCH, "*");
         }
 
         // Set body

core/src/types/capability.rs

@@ -101,6 +101,8 @@ pub struct Capability {
     pub write_with_cache_control: bool,
     /// If operator supports write with if none match.
     pub write_with_if_none_match: bool,
+    /// If operator supports write with if not exist.
+    pub write_with_if_not_exists: bool,
     /// If operator supports write with user defined metadata
     pub write_with_user_metadata: bool,
     /// write_multi_max_size is the max size that services support in write_multi.

core/src/types/operator/operator.rs

@@ -1237,8 +1237,6 @@ impl Operator {
     ///
     /// This feature can be used to check if the file already exists.
     /// This prevents overwriting of existing objects with identical key names.
-    /// Users can use *(asterisk) to verify if a file already exists by matching with any ETag.
-    /// Note: S3 only support use *(asterisk).
     ///
     /// If file exists, an error with kind [`ErrorKind::ConditionNotMatch`] will be returned.
     ///
@@ -1247,13 +1245,34 @@ impl Operator {
     /// use opendal::Operator;
     /// # async fn test(op: Operator, etag: &str) -> Result<()> {
     /// let bs = b"hello, world!".to_vec();
-    /// let res = op.write_with("path/to/file", bs).if_none_match("*").await;
+    /// let res = op.write_with("path/to/file", bs).if_none_match(etag).await;
     /// assert!(res.is_err());
     /// assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
     /// # Ok(())
     /// # }
     /// ```
     ///
+    /// ## `if_not_exists`
+    ///
+    /// This feature allows to safely write a file only if it does not exist. It is designed
+    /// to be concurrency-safe, and can be used to a file lock. For storage services that
+    /// support the `if_not_exists` feature, only one write operation will succeed, while all
+    /// other attempts will fail.
+    ///
+    /// If the file already exists, an error with kind [`ErrorKind::ConditionNotMatch`] will
+    /// be returned.
+    ///
+    /// ```no_run
+    /// # use opendal::{ErrorKind, Result};
+    /// use opendal::Operator;
+    /// # async fn test(op: Operator, etag: &str) -> Result<()> {
+    /// let bs = b"hello, world!".to_vec();
+    /// let res = op.write_with("path/to/file", bs).if_not_exists(true).await;
+    /// assert!(res.is_err());
+    /// assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+    /// # Ok(())}
+    /// ```
+    ///
     /// # Examples
     ///
     /// ```

core/src/types/operator/operator_futures.rs

@@ -329,6 +329,11 @@ impl<F: Future<Output = Result<()>>> FutureWrite<F> {
         self.map(|(args, options, bs)| (args.with_if_none_match(s), options, bs))
     }
 
+    /// Set the If-Not-Exist for this operation.
+    pub fn if_not_exists(self, b: bool) -> Self {
+        self.map(|(args, options, bs)| (args.with_if_not_exists(b), options, bs))
+    }
+
     /// Set the user defined metadata of the op
     ///
     /// ## Notes

core/tests/behavior/async_write.rs

@@ -45,6 +45,7 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_write_with_content_type,
             test_write_with_content_disposition,
             test_write_with_if_none_match,
+            test_write_with_if_not_exists,
             test_write_with_user_metadata,
             test_writer_write,
             test_writer_write_with_overwrite,
@@ -637,9 +638,36 @@ pub async fn test_write_with_if_none_match(op: Operator) -> Result<()> {
     op.write(&path, content.clone())
         .await
         .expect("write must succeed");
+
+    let meta = op.stat(&path).await?;
+
+    let res = op
+        .write_with(&path, content.clone())
+        .if_none_match(meta.etag().expect("etag must exist"))
+        .await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
+
+/// Write an file with if_not_exists will get a ConditionNotMatch error if file exists.
+pub async fn test_write_with_if_not_exists(op: Operator) -> Result<()> {
+    if !op.info().full_capability().write_with_if_not_exists {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    let res = op
+        .write_with(&path, content.clone())
+        .if_not_exists(true)
+        .await;
+    assert!(res.is_ok());
+
     let res = op
         .write_with(&path, content.clone())
-        .if_none_match("*")
+        .if_not_exists(true)
         .await;
     assert!(res.is_err());
     assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);