feat(core): add conditional rename option

core/core/src/blocking/operator.rs

@@ -607,7 +607,7 @@ impl Operator {
     ///
     /// - `from` and `to` must be a file.
     /// - `to` will be overwritten if it exists.
-    /// - If `from` and `to` are the same, a `IsSameFile` error will occur.
+    /// - If `from` and `to` are the same, an `IsSameFile` error will occur.
     ///
     /// # Examples
     ///
@@ -622,10 +622,34 @@ impl Operator {
     /// # }
     /// ```
     pub fn rename(&self, from: &str, to: &str) -> Result<()> {
+        self.rename_options(from, to, options::RenameOptions::default())
+    }
+
+    /// Rename a file from `from` to `to` with additional options.
+    ///
+    /// # Options
+    ///
+    /// Visit [`options::RenameOptions`] for all available options.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use opendal_core::blocking;
+    /// use opendal_core::options::RenameOptions;
+    /// use opendal_core::Result;
+    ///
+    /// fn test(op: blocking::Operator) -> Result<()> {
+    ///     let mut opts = RenameOptions::default();
+    ///     opts.if_not_exists = true;
+    ///     op.rename_options("path/to/file", "path/to/file2", opts)?;
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn rename_options(&self, from: &str, to: &str, opts: options::RenameOptions) -> Result<()> {
         let op = self.op.clone();
         let from = from.to_string();
         let to = to.to_string();
-        self.spawn_block(async move { op.rename(&from, &to).await })?
+        self.spawn_block(async move { op.rename_options(&from, &to, opts).await })?
     }
 
     /// Delete given path.

core/core/src/docs/comparisons/vs_object_store.md

@@ -145,8 +145,8 @@ opendal has an idea called [`Capability`][crate::Capability], so it's services m
 | list      | list_with_delimiter  | -                                            |
 | -         | copy                 | -                                            |
 | -         | copy_if_not_exists   | -                                            |
-| -         | rename               | -                                            |
-| -         | rename_if_not_exists | -                                            |
+| rename    | rename               | -                                            |
+| rename_with(if_not_exists) | rename_if_not_exists | -                            |
 | presign   | -                    | get a presign URL of object                  |
 | multipart | multipart            | both support, but API is different           |
 | blocking  | -                    | opendal supports blocking API                |

core/core/src/layers/correctness_check.rs

@@ -248,6 +248,16 @@ impl Service for CorrectnessService {
         to: &str,
         args: OpRename,
     ) -> Result<RpRename> {
+        let capability = self.capability();
+        let scheme = self.info().scheme();
+        if args.if_not_exists() && !capability.rename_with_if_not_exists {
+            return Err(new_unsupported_error(
+                scheme,
+                Operation::Rename,
+                "if_not_exists",
+            ));
+        }
+
         self.inner.rename(ctx, from, to, args).await
     }
 
@@ -390,10 +400,7 @@ mod tests {
             _: &str,
             _: OpRename,
         ) -> Result<RpRename> {
-            Err(Error::new(
-                ErrorKind::Unsupported,
-                "operation is not supported",
-            ))
+            Ok(RpRename::default())
         }
 
         async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result<RpPresign> {
@@ -562,4 +569,23 @@ mod tests {
         let res = op.delete_with("path").version("version").await;
         assert!(res.is_ok())
     }
+
+    #[tokio::test]
+    async fn test_rename_with_if_not_exists() {
+        let op = new_test_operator(Capability {
+            rename: true,
+            ..Default::default()
+        });
+        let res = op.rename_with("from", "to").if_not_exists(true).await;
+        assert!(res.is_err());
+        assert_eq!(res.unwrap_err().kind(), ErrorKind::Unsupported);
+
+        let op = new_test_operator(Capability {
+            rename: true,
+            rename_with_if_not_exists: true,
+            ..Default::default()
+        });
+        let res = op.rename_with("from", "to").if_not_exists(true).await;
+        assert!(res.is_ok());
+    }
 }

core/core/src/raw/ops.rs

@@ -987,11 +987,32 @@ impl From<options::CopyOptions> for (OpCopy, OpCopier) {
 
 /// Args for `rename` operation.
 #[derive(Debug, Clone, Default)]
-pub struct OpRename {}
+pub struct OpRename {
+    if_not_exists: bool,
+}
 
 impl OpRename {
-    /// Create a new `OpMove`.
+    /// Create a new `OpRename`.
     pub fn new() -> Self {
         Self::default()
     }
+
+    /// Set the if_not_exists flag for the operation.
+    pub fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
+        self.if_not_exists = if_not_exists;
+        self
+    }
+
+    /// Get if_not_exists flag.
+    pub fn if_not_exists(&self) -> bool {
+        self.if_not_exists
+    }
+}
+
+impl From<options::RenameOptions> for OpRename {
+    fn from(value: options::RenameOptions) -> Self {
+        Self {
+            if_not_exists: value.if_not_exists,
+        }
+    }
 }

core/core/src/types/capability.rs

@@ -164,6 +164,8 @@ pub struct Capability {
 
     /// Indicates if rename operations are supported.
     pub rename: bool,
+    /// Indicates if conditional rename operations with if-not-exists are supported.
+    pub rename_with_if_not_exists: bool,
 
     /// Indicates if list operations are supported.
     pub list: bool,

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

@@ -1389,24 +1389,111 @@ impl Operator {
     /// # }
     /// ```
     pub async fn rename(&self, from: &str, to: &str) -> Result<()> {
+        self.rename_options(from, to, options::RenameOptions::default())
+            .await
+    }
+
+    /// Rename a file from `from` to `to` with additional options.
+    ///
+    /// # Notes
+    ///
+    /// - `from` and `to` must be a file.
+    /// - If `from` and `to` are the same, an `IsSameFile` error will occur.
+    ///
+    /// # Options
+    ///
+    /// Visit [`options::RenameOptions`] for all available options.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use opendal_core::Operator;
+    /// use opendal_core::Result;
+    ///
+    /// async fn test(op: Operator) -> Result<()> {
+    ///     op.rename_with("path/to/file", "path/to/file2")
+    ///         .if_not_exists(true)
+    ///         .await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn rename_with(
+        &self,
+        from: &str,
+        to: &str,
+    ) -> FutureRename<impl Future<Output = Result<()>>> {
+        let from = normalize_path(from);
+        let to = normalize_path(to);
+
+        OperatorFuture::new(
+            self.context().clone(),
+            self.service().clone(),
+            from,
+            (options::RenameOptions::default(), to),
+            Self::rename_inner,
+        )
+    }
+
+    /// Rename a file from `from` to `to` with additional options.
+    ///
+    /// # Options
+    ///
+    /// Visit [`options::RenameOptions`] for all available options.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use opendal_core::options::RenameOptions;
+    /// use opendal_core::Operator;
+    /// use opendal_core::Result;
+    ///
+    /// async fn test(op: Operator) -> Result<()> {
+    ///     let mut opts = RenameOptions::default();
+    ///     opts.if_not_exists = true;
+    ///     op.rename_options("path/to/file", "path/to/file2", opts)
+    ///         .await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn rename_options(
+        &self,
+        from: &str,
+        to: &str,
+        opts: impl Into<options::RenameOptions>,
+    ) -> Result<()> {
         let from = normalize_path(from);
+        let to = normalize_path(to);
+        let opts = opts.into();
 
+        Self::rename_inner(
+            self.context().clone(),
+            self.service().clone(),
+            from,
+            (opts, to),
+        )
+        .await
+    }
+
+    async fn rename_inner(
+        ctx: OperationContext,
+        srv: Servicer,
+        from: String,
+        (opts, to): (options::RenameOptions, String),
+    ) -> Result<()> {
         if !validate_path(&from, EntryMode::FILE) {
             return Err(
                 Error::new(ErrorKind::IsADirectory, "from path is a directory")
                     .with_operation(Operation::Rename)
-                    .with_context("service", self.info().scheme())
+                    .with_context("service", srv.info().scheme())
                     .with_context("from", from),
             );
         }
 
-        let to = normalize_path(to);
-
         if !validate_path(&to, EntryMode::FILE) {
             return Err(
                 Error::new(ErrorKind::IsADirectory, "to path is a directory")
                     .with_operation(Operation::Rename)
-                    .with_context("service", self.info().scheme())
+                    .with_context("service", srv.info().scheme())
                     .with_context("to", to),
             );
         }
@@ -1415,15 +1502,13 @@ impl Operator {
             return Err(
                 Error::new(ErrorKind::IsSameFile, "from and to paths are same")
                     .with_operation(Operation::Rename)
-                    .with_context("service", self.info().scheme())
+                    .with_context("service", srv.info().scheme())
                     .with_context("from", from)
                     .with_context("to", to),
             );
         }
 
-        self.srv
-            .rename(&self.ctx, &from, &to, OpRename::new())
-            .await?;
+        srv.rename(&ctx, &from, &to, opts.into()).await?;
 
         Ok(())
     }

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

@@ -1538,3 +1538,32 @@ impl<F: Future<Output = Result<Copier>>> FutureCopier<F> {
         self
     }
 }
+
+/// Future that generated by [`Operator::rename_with`].
+///
+/// Users can add more options by public functions provided by this struct.
+pub type FutureRename<F> = OperatorFuture<(options::RenameOptions, String), (), F>;
+
+impl<F: Future<Output = Result<()>>> FutureRename<F> {
+    /// Sets the condition that rename operation will succeed only if target does not exist.
+    ///
+    /// Refer to [`options::RenameOptions::if_not_exists`] for more details.
+    ///
+    /// ### Example
+    ///
+    /// ```
+    /// use opendal_core::Operator;
+    /// use opendal_core::Result;
+    ///
+    /// async fn test(op: Operator) -> Result<()> {
+    ///     op.rename_with("source/path", "target/path")
+    ///         .if_not_exists(true)
+    ///         .await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    pub fn if_not_exists(mut self, v: bool) -> Self {
+        self.args.0.if_not_exists = v;
+        self
+    }
+}

core/core/src/types/options.rs

@@ -601,3 +601,21 @@ pub struct CopyOptions {
     /// step. Services that cannot split copy operations can ignore it.
     pub chunk: Option<usize>,
 }
+
+/// Options for rename operations.
+#[derive(Debug, Clone, Default, Eq, PartialEq)]
+pub struct RenameOptions {
+    /// Sets the condition that rename operation will succeed only if target does not exist.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::rename_with_if_not_exists`] before using this feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If the target does not exist, the rename operation succeeds.
+    /// - If the target exists, the operation returns [`ErrorKind::ConditionNotMatch`].
+    /// - If the service does not support this condition, the operation returns
+    ///   [`ErrorKind::Unsupported`].
+    pub if_not_exists: bool,
+}

core/services/hdfs/src/backend.rs

@@ -171,6 +171,7 @@ impl Builder for HdfsBuilder {
                     list: true,
 
                     rename: true,
+                    rename_with_if_not_exists: true,
 
                     shared: true,
 
@@ -273,9 +274,9 @@ impl Service for HdfsBackend {
         _ctx: &OperationContext,
         from: &str,
         to: &str,
-        _args: OpRename,
+        args: OpRename,
     ) -> Result<RpRename> {
-        self.core.hdfs_rename(from, to)?;
+        self.core.hdfs_rename(from, to, &args)?;
         Ok(RpRename::new())
     }