feat(core): Add Copy Support

core/src/layers/logging.rs

@@ -322,6 +322,47 @@ impl<A: Accessor> LayeredAccessor for LoggingAccessor<A> {
             })
     }
 
+    async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        debug!(
+            target: LOGGING_TARGET,
+            "service={} operation={} from={} to={} -> started",
+            self.scheme,
+            Operation::Copy,
+            from,
+            to
+        );
+
+        self.inner
+            .copy(from, to, args)
+            .await
+            .map(|v| {
+                debug!(
+                    target: LOGGING_TARGET,
+                    "service={} operation={} from={} to={} -> finished",
+                    self.scheme,
+                    Operation::Copy,
+                    from,
+                    to
+                );
+                v
+            })
+            .map_err(|err| {
+                if let Some(lvl) = self.err_level(&err) {
+                    log!(
+                        target: LOGGING_TARGET,
+                        lvl,
+                        "service={} operation={} from={} to={} -> {}: {err:?}",
+                        self.scheme,
+                        Operation::Copy,
+                        from,
+                        to,
+                        self.err_status(&err)
+                    )
+                };
+                err
+            })
+    }
+
     async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         debug!(
             target: LOGGING_TARGET,
@@ -699,6 +740,46 @@ impl<A: Accessor> LayeredAccessor for LoggingAccessor<A> {
             })
     }
 
+    fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        debug!(
+            target: LOGGING_TARGET,
+            "service={} operation={} from={} to={} -> started",
+            self.scheme,
+            Operation::BlockingCopy,
+            from,
+            to,
+        );
+
+        self.inner
+            .blocking_copy(from, to, args)
+            .map(|v| {
+                debug!(
+                    target: LOGGING_TARGET,
+                    "service={} operation={} from={} to={} -> finished",
+                    self.scheme,
+                    Operation::BlockingCopy,
+                    from,
+                    to,
+                );
+                v
+            })
+            .map_err(|err| {
+                if let Some(lvl) = self.err_level(&err) {
+                    log!(
+                        target: LOGGING_TARGET,
+                        lvl,
+                        "service={} operation={} from={} to={} -> {}: {err:?}",
+                        self.scheme,
+                        Operation::BlockingCopy,
+                        from,
+                        to,
+                        self.err_status(&err)
+                    );
+                }
+                err
+            })
+    }
+
     fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         debug!(
             target: LOGGING_TARGET,

core/src/raw/accessor.rs

@@ -131,6 +131,24 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
         ))
     }
 
+    /// Invoke the `copy` operation on the specified `from` path and `to` path.
+    ///
+    /// Require [AccessorCapability::Read] and [AccessorCapability::Write]
+    ///
+    /// # Behaviour
+    ///
+    /// - `from` and `to` MUST be file path, DON'T NEED to check mode.
+    /// - Copy on existing file SHOULD succeed.
+    /// - Copy on existing file SHOULD overwrite and truncate.
+    async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        let (_, _, _) = (from, to, args);
+
+        Err(Error::new(
+            ErrorKind::Unsupported,
+            "operation is not supported",
+        ))
+    }
+
     /// Invoke the `stat` operation on the specified path.
     ///
     /// Require [`AccessorCapability::Read`]
@@ -263,6 +281,20 @@ pub trait Accessor: Send + Sync + Debug + Unpin + 'static {
         ))
     }
 
+    /// Invoke the `blocking_copy` operation on the specified `from` path and `to` path.
+    ///
+    /// This operation is the blocking version of [`Accessor::copy`]
+    ///
+    /// Require [`AccessorCapability::Read`], [`AccessorCapability::Write`] and [`AccessorCapability::Blocking`]
+    fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        let (_, _, _) = (from, to, args);
+
+        Err(Error::new(
+            ErrorKind::Unsupported,
+            "operation is not supported",
+        ))
+    }
+
     /// Invoke the `blocking_stat` operation on the specified path.
     ///
     /// This operation is the blocking version of [`Accessor::stat`]
@@ -369,6 +401,11 @@ impl<T: Accessor + ?Sized> Accessor for Arc<T> {
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
         self.as_ref().write(path, args).await
     }
+
+    async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        self.as_ref().copy(from, to, args).await
+    }
+
     async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         self.as_ref().stat(path, args).await
     }
@@ -399,6 +436,11 @@ impl<T: Accessor + ?Sized> Accessor for Arc<T> {
     fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
         self.as_ref().blocking_write(path, args)
     }
+
+    fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        self.as_ref().blocking_copy(from, to, args)
+    }
+
     fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         self.as_ref().blocking_stat(path, args)
     }

core/src/raw/layer.rs

@@ -159,6 +159,10 @@ pub trait LayeredAccessor: Send + Sync + Debug + Unpin + 'static {
 
     async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)>;
 
+    async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        self.inner().copy(from, to, args).await
+    }
+
     async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         self.inner().stat(path, args).await
     }
@@ -187,6 +191,10 @@ pub trait LayeredAccessor: Send + Sync + Debug + Unpin + 'static {
 
     fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)>;
 
+    fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        self.inner().blocking_copy(from, to, args)
+    }
+
     fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         self.inner().blocking_stat(path, args)
     }
@@ -225,6 +233,10 @@ impl<L: LayeredAccessor> Accessor for L {
         (self as &L).write(path, args).await
     }
 
+    async fn copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        (self as &L).copy(from, to, args).await
+    }
+
     async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         (self as &L).stat(path, args).await
     }
@@ -261,6 +273,10 @@ impl<L: LayeredAccessor> Accessor for L {
         (self as &L).blocking_write(path, args)
     }
 
+    fn blocking_copy(&self, from: &str, to: &str, args: OpCopy) -> Result<RpCopy> {
+        (self as &L).blocking_copy(from, to, args)
+    }
+
     fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
         (self as &L).blocking_stat(path, args)
     }

core/src/raw/operation.rs

@@ -30,6 +30,8 @@ pub enum Operation {
     Read,
     /// Operation for [`crate::raw::Accessor::write`]
     Write,
+    /// Operation for [`crate::raw::Accessor::copy`]
+    Copy,
     /// Operation for [`crate::raw::Accessor::stat`]
     Stat,
     /// Operation for [`crate::raw::Accessor::delete`]
@@ -48,6 +50,8 @@ pub enum Operation {
     BlockingRead,
     /// Operation for [`crate::raw::Accessor::blocking_write`]
     BlockingWrite,
+    /// Operation for [`crate::raw::Accessor::blocking_copy`]
+    BlockingCopy,
     /// Operation for [`crate::raw::Accessor::blocking_stat`]
     BlockingStat,
     /// Operation for [`crate::raw::Accessor::blocking_delete`]
@@ -84,6 +88,7 @@ impl From<Operation> for &'static str {
             Operation::Create => "create",
             Operation::Read => "read",
             Operation::Write => "write",
+            Operation::Copy => "copy",
             Operation::Stat => "stat",
             Operation::Delete => "delete",
             Operation::List => "list",
@@ -93,6 +98,7 @@ impl From<Operation> for &'static str {
             Operation::BlockingCreate => "blocking_create",
             Operation::BlockingRead => "blocking_read",
             Operation::BlockingWrite => "blocking_write",
+            Operation::BlockingCopy => "blocking_copy",
             Operation::BlockingStat => "blocking_stat",
             Operation::BlockingDelete => "blocking_delete",
             Operation::BlockingList => "blocking_list",

core/src/raw/rps.rs

@@ -227,6 +227,17 @@ impl RpWrite {
     }
 }
 
+/// Reply for `copy` operation.
+#[derive(Debug, Clone, Default)]
+pub struct RpCopy {}
+
+impl RpCopy {
+    /// Create a new reply for copy.
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use anyhow::Result;

core/src/services/fs/backend.rs

@@ -441,6 +441,17 @@ impl Accessor for FsBackend {
         Ok((RpWrite::new(), FsWriter::new(target_path, tmp_path, f)))
     }
 
+    async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+        tokio::fs::copy(
+            self.root.join(from.trim_end_matches('/')),
+            self.root.join(to.trim_end_matches('/')),
+        )
+        .await
+        .map_err(parse_io_error)?;
+
+        Ok(RpCopy::default())
+    }
+
     async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
         let p = self.root.join(path.trim_end_matches('/'));
 
@@ -627,6 +638,16 @@ impl Accessor for FsBackend {
         Ok((RpWrite::new(), FsWriter::new(target_path, tmp_path, f)))
     }
 
+    fn blocking_copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
+        std::fs::copy(
+            self.root.join(from.trim_end_matches('/')),
+            self.root.join(to.trim_end_matches('/')),
+        )
+        .map_err(parse_io_error)?;
+
+        Ok(RpCopy::default())
+    }
+
     fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
         let p = self.root.join(path.trim_end_matches('/'));
 

core/src/types/operator/blocking_operator.rs

@@ -447,6 +447,58 @@ impl BlockingOperator {
         self.write_with(path, OpWrite::new(), bs)
     }
 
+    /// Copy a file from `from` to `to`.
+    ///
+    /// # Notes
+    ///
+    /// - `from` and `to` must be a file.
+    /// - `to` will be overwritten if it exists.
+    /// - If `from` and `to` are the same, nothing will happen.
+    /// - `copy` is idempotent. For same `from` and `to` input, the result will be the same.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use std::io::Result;
+    /// # use opendal::BlockingOperator;
+    ///
+    /// # fn test(op: BlockingOperator) -> Result<()> {
+    /// op.copy("path/to/file", "path/to/file2")?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn copy(&self, from: &str, to: &str) -> Result<()> {
+        if from == to {
+            return Ok(());
+        }
+
+        let from = normalize_path(from);
+
+        if !validate_path(&from, EntryMode::FILE) {
+            return Err(
+                Error::new(ErrorKind::IsADirectory, "from path is a directory")
+                    .with_operation("Operator::copy")
+                    .with_context("service", self.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("Operator::copy")
+                    .with_context("service", self.info().scheme())
+                    .with_context("to", to),
+            );
+        }
+
+        self.inner().blocking_copy(&from, &to, OpCopy::new())?;
+
+        Ok(())
+    }
+
     /// Write data with option described in OpenDAL [rfc-0661](../../docs/rfcs/0661-path-in-accessor.md)
     ///
     /// # Notes

core/src/types/operator/metadata.rs

@@ -57,6 +57,11 @@ impl OperatorInfo {
         self.0.capabilities().contains(AccessorCapability::Write)
     }
 
+    /// Check if current backend supports [`Accessor::copy`] or not.
+    pub fn can_copy(&self) -> bool {
+        self.0.capabilities().contains(AccessorCapability::Read | AccessorCapability::Write)
+    }
+
     /// Check if current backend supports [`Accessor::list`] or not.
     pub fn can_list(&self) -> bool {
         self.0.capabilities().contains(AccessorCapability::List)