You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/05/24 11:35:48 UTC
(opendal) branch main updated: feat(core): return metadata from copy (#7606)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 320068740 feat(core): return metadata from copy (#7606)
320068740 is described below
commit 320068740d3cc78bdcbc5cbc9ebe1b9901cbc1ca
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Sun May 24 19:35:43 2026 +0800
feat(core): return metadata from copy (#7606)
---
bindings/cpp/src/async.rs | 1 +
bindings/cpp/src/lib.rs | 3 +-
bindings/dotnet/src/operator.rs | 2 +
bindings/haskell/src/lib.rs | 2 +-
bindings/java/src/operator.rs | 3 +-
bindings/nodejs/src/lib.rs | 6 ++-
bindings/ocaml/src/operator/mod.rs | 2 +-
bindings/python/src/operator.rs | 10 ++++-
bindings/ruby/src/operator.rs | 1 +
core/core/src/blocking/operator.rs | 9 ++++-
core/core/src/layers/error_context.rs | 9 +++++
core/core/src/raw/enum_utils.rs | 7 ++++
core/core/src/raw/oio/copy/api.rs | 45 ++++++++++++++++++---
core/core/src/raw/oio/copy/block_copy.rs | 31 +++++++++++----
core/core/src/raw/oio/copy/multipart_copy.rs | 27 +++++++++----
core/core/src/types/copy.rs | 31 +++++++++++++--
core/core/src/types/operator/operator.rs | 26 ++++++------
core/core/src/types/operator/operator_futures.rs | 4 +-
core/layers/await-tree/src/lib.rs | 6 +++
core/layers/concurrent-limit/src/lib.rs | 4 ++
core/layers/fastrace/src/lib.rs | 6 +++
core/layers/hotpath/src/lib.rs | 5 +++
core/layers/logging/src/lib.rs | 4 ++
core/layers/observe-metrics-common/src/lib.rs | 10 +++++
core/layers/retry/src/lib.rs | 35 +++++++++++++++++
core/layers/tail-cut/src/lib.rs | 13 ++++++
core/layers/timeout/src/lib.rs | 9 +++++
core/services/azblob/src/copier.rs | 13 +++---
core/services/azblob/src/writer.rs | 2 +-
core/services/gcs/src/copier.rs | 12 ++++++
core/services/gcs/src/core.rs | 50 ++++++++++++++++--------
core/services/s3/src/copier.rs | 31 +++++++++++++--
core/services/tos/src/copier.rs | 22 +++++++++--
integrations/dav-server/src/fs.rs | 1 +
34 files changed, 361 insertions(+), 81 deletions(-)
diff --git a/bindings/cpp/src/async.rs b/bindings/cpp/src/async.rs
index a743dd5f5..368258f6e 100644
--- a/bindings/cpp/src/async.rs
+++ b/bindings/cpp/src/async.rs
@@ -221,6 +221,7 @@ unsafe fn operator_copy(op: ffi::OperatorPtr, from: String, to: String) -> RustF
RustFutureWrite::fallible(async move {
op.0.copy(&from, &to)
.await
+ .map(|_| ())
.map_err(|e| CxxAsyncException::new(e.to_string().into_boxed_str()))
})
}
diff --git a/bindings/cpp/src/lib.rs b/bindings/cpp/src/lib.rs
index db71710ea..0067f1255 100644
--- a/bindings/cpp/src/lib.rs
+++ b/bindings/cpp/src/lib.rs
@@ -221,7 +221,8 @@ impl Operator {
}
fn copy(&self, src: &str, dst: &str) -> Result<()> {
- Ok(self.0.copy(src, dst)?)
+ self.0.copy(src, dst)?;
+ Ok(())
}
fn rename(&self, src: &str, dst: &str) -> Result<()> {
diff --git a/bindings/dotnet/src/operator.rs b/bindings/dotnet/src/operator.rs
index a66f6b9f9..043f752e6 100644
--- a/bindings/dotnet/src/operator.rs
+++ b/bindings/dotnet/src/operator.rs
@@ -693,6 +693,7 @@ fn operator_copy_inner(
executor
.block_on(op.copy(source_path, target_path))
+ .map(|_| ())
.map_err(OpenDALError::from_opendal_error)
}
@@ -738,6 +739,7 @@ fn operator_copy_async_inner(
let result = op
.copy(&source_path, &target_path)
.await
+ .map(|_| ())
.map_err(OpenDALError::from_opendal_error);
callback(
diff --git a/bindings/haskell/src/lib.rs b/bindings/haskell/src/lib.rs
index f196002a0..c1178929c 100644
--- a/bindings/haskell/src/lib.rs
+++ b/bindings/haskell/src/lib.rs
@@ -353,7 +353,7 @@ pub unsafe extern "C" fn blocking_copy(
};
let res = match op.copy(path_from_str, path_to_str) {
- Ok(()) => FFIResult::ok(()),
+ Ok(_) => FFIResult::ok(()),
Err(e) => FFIResult::err_with_source("Failed to copy", e),
};
diff --git a/bindings/java/src/operator.rs b/bindings/java/src/operator.rs
index 76b5ba5fe..c91824d43 100644
--- a/bindings/java/src/operator.rs
+++ b/bindings/java/src/operator.rs
@@ -224,7 +224,8 @@ fn intern_copy(
let source_path = jstring_to_string(env, &source_path)?;
let target_path = jstring_to_string(env, &target_path)?;
- Ok(op.copy(&source_path, &target_path)?)
+ op.copy(&source_path, &target_path)?;
+ Ok(())
}
/// # Safety
diff --git a/bindings/nodejs/src/lib.rs b/bindings/nodejs/src/lib.rs
index 1c22e8842..3f113a9c6 100644
--- a/bindings/nodejs/src/lib.rs
+++ b/bindings/nodejs/src/lib.rs
@@ -422,6 +422,7 @@ impl Operator {
self.async_op
.copy(&from, &to)
.await
+ .map(|_| ())
.map_err(format_napi_error)
}
@@ -433,7 +434,10 @@ impl Operator {
/// ```
#[napi]
pub fn copy_sync(&self, from: String, to: String) -> Result<()> {
- self.blocking_op.copy(&from, &to).map_err(format_napi_error)
+ self.blocking_op
+ .copy(&from, &to)
+ .map(|_| ())
+ .map_err(format_napi_error)
}
/// Rename file according to given `from` and `to` path.
diff --git a/bindings/ocaml/src/operator/mod.rs b/bindings/ocaml/src/operator/mod.rs
index ded6da0d2..c6646c2b1 100644
--- a/bindings/ocaml/src/operator/mod.rs
+++ b/bindings/ocaml/src/operator/mod.rs
@@ -124,7 +124,7 @@ pub fn blocking_writer(
#[ocaml::func]
#[ocaml::sig("operator -> string -> string -> (unit, string) Result.t ")]
pub fn blocking_copy(operator: &mut Operator, from: String, to: String) -> Result<(), String> {
- map_res_error(operator.0.copy(from.as_str(), to.as_str()))
+ map_res_error(operator.0.copy(from.as_str(), to.as_str())).map(|_| ())
}
#[ocaml::func]
diff --git a/bindings/python/src/operator.rs b/bindings/python/src/operator.rs
index e21fbbc0a..af45e25a9 100644
--- a/bindings/python/src/operator.rs
+++ b/bindings/python/src/operator.rs
@@ -460,7 +460,10 @@ impl Operator {
pub fn copy(&self, source: PathBuf, target: PathBuf) -> PyResult<()> {
let source = source.to_string_lossy().to_string();
let target = target.to_string_lossy().to_string();
- self.core.copy(&source, &target).map_err(format_pyerr)
+ self.core
+ .copy(&source, &target)
+ .map(|_| ())
+ .map_err(format_pyerr)
}
/// Rename (move) a file from one path to another.
@@ -1192,7 +1195,10 @@ impl AsyncOperator {
let source = source.to_string_lossy().to_string();
let target = target.to_string_lossy().to_string();
future_into_py(py, async move {
- this.copy(&source, &target).await.map_err(format_pyerr)
+ this.copy(&source, &target)
+ .await
+ .map(|_| ())
+ .map_err(format_pyerr)
})
}
diff --git a/bindings/ruby/src/operator.rs b/bindings/ruby/src/operator.rs
index c9ae692c9..2af34969a 100644
--- a/bindings/ruby/src/operator.rs
+++ b/bindings/ruby/src/operator.rs
@@ -223,6 +223,7 @@ impl Operator {
rb_self
.blocking_op
.copy(&from, &to)
+ .map(|_| ())
.map_err(|err| Error::new(ruby.exception_runtime_error(), err.to_string()))
}
diff --git a/core/core/src/blocking/operator.rs b/core/core/src/blocking/operator.rs
index bc9b978cf..ab905124a 100644
--- a/core/core/src/blocking/operator.rs
+++ b/core/core/src/blocking/operator.rs
@@ -515,12 +515,17 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn copy(&self, from: &str, to: &str) -> Result<()> {
+ pub fn copy(&self, from: &str, to: &str) -> Result<Metadata> {
self.copy_options(from, to, options::CopyOptions::default())
}
/// Copy a file from `from` to `to` with additional options.
- pub fn copy_options(&self, from: &str, to: &str, opts: options::CopyOptions) -> Result<()> {
+ pub fn copy_options(
+ &self,
+ from: &str,
+ to: &str,
+ opts: options::CopyOptions,
+ ) -> Result<Metadata> {
let op = self.op.clone();
let from = from.to_string();
let to = to.to_string();
diff --git a/core/core/src/layers/error_context.rs b/core/core/src/layers/error_context.rs
index c3593f452..1c943ca86 100644
--- a/core/core/src/layers/error_context.rs
+++ b/core/core/src/layers/error_context.rs
@@ -335,6 +335,15 @@ impl<T: oio::Copy> oio::Copy for ErrorContextWrapper<T> {
})
}
+ async fn close(&mut self) -> Result<Metadata> {
+ self.inner.close().await.map_err(|err| {
+ err.with_operation(Operation::Copy)
+ .with_context("service", self.scheme)
+ .with_context("path", &self.path)
+ .with_context("copied", self.processed.to_string())
+ })
+ }
+
async fn abort(&mut self) -> Result<()> {
self.inner.abort().await.map_err(|err| {
err.with_operation(Operation::Copy)
diff --git a/core/core/src/raw/enum_utils.rs b/core/core/src/raw/enum_utils.rs
index 59630a741..589c30d8f 100644
--- a/core/core/src/raw/enum_utils.rs
+++ b/core/core/src/raw/enum_utils.rs
@@ -91,6 +91,13 @@ impl<ONE: oio::Copy, TWO: oio::Copy> oio::Copy for TwoWays<ONE, TWO> {
}
}
+ async fn close(&mut self) -> Result<Metadata> {
+ match self {
+ Self::One(v) => v.close().await,
+ Self::Two(v) => v.close().await,
+ }
+ }
+
async fn abort(&mut self) -> Result<()> {
match self {
Self::One(v) => v.abort().await,
diff --git a/core/core/src/raw/oio/copy/api.rs b/core/core/src/raw/oio/copy/api.rs
index a02db85a6..0bd81aff8 100644
--- a/core/core/src/raw/oio/copy/api.rs
+++ b/core/core/src/raw/oio/copy/api.rs
@@ -32,6 +32,9 @@ pub trait Copy: Unpin + Send + Sync {
/// `Ok(None)` means the copy operation has completed.
fn next(&mut self) -> impl Future<Output = Result<Option<usize>>> + MaybeSend;
+ /// Close the copier and return metadata from the server-side completion response.
+ fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend;
+
/// Abort the pending copy operation.
fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend;
}
@@ -41,6 +44,10 @@ impl Copy for () {
Ok(None)
}
+ async fn close(&mut self) -> Result<Metadata> {
+ Ok(Metadata::default())
+ }
+
async fn abort(&mut self) -> Result<()> {
Ok(())
}
@@ -48,7 +55,8 @@ impl Copy for () {
/// OneShotCopier drives a single asynchronous copy step.
pub struct OneShotCopier {
- fut: Option<BoxedStaticFuture<Result<Option<usize>>>>,
+ fut: Option<BoxedStaticFuture<Result<Metadata>>>,
+ meta: Option<Metadata>,
}
/// # Safety
@@ -63,28 +71,42 @@ unsafe impl Send for OneShotCopier {}
impl OneShotCopier {
/// Create a new one-shot copier.
- pub fn new(fut: impl Future<Output = Result<Option<usize>>> + MaybeSend + 'static) -> Self {
+ pub fn new(fut: impl Future<Output = Result<Metadata>> + MaybeSend + 'static) -> Self {
Self {
fut: Some(Box::pin(fut)),
+ meta: None,
}
}
/// Create a one-shot copier that has already completed.
pub fn completed() -> Self {
- Self { fut: None }
+ Self {
+ fut: None,
+ meta: Some(Metadata::default()),
+ }
}
}
impl Copy for OneShotCopier {
async fn next(&mut self) -> Result<Option<usize>> {
- match self.fut.take() {
- Some(fut) => fut.await,
- None => Ok(None),
+ if self.meta.is_none() {
+ self.close().await?;
+ }
+
+ Ok(None)
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ if let Some(fut) = self.fut.take() {
+ self.meta = Some(fut.await?);
}
+
+ Ok(self.meta.clone().unwrap_or_default())
}
async fn abort(&mut self) -> Result<()> {
self.fut = None;
+ self.meta = None;
Ok(())
}
}
@@ -94,6 +116,9 @@ pub trait CopyDyn: Unpin + Send + Sync {
/// The dyn version of [`Copy::next`].
fn next_dyn(&mut self) -> BoxedFuture<'_, Result<Option<usize>>>;
+ /// The dyn version of [`Copy::close`].
+ fn close_dyn(&mut self) -> BoxedFuture<'_, Result<Metadata>>;
+
/// The dyn version of [`Copy::abort`].
fn abort_dyn(&mut self) -> BoxedFuture<'_, Result<()>>;
}
@@ -103,6 +128,10 @@ impl<T: Copy + ?Sized> CopyDyn for T {
Box::pin(self.next())
}
+ fn close_dyn(&mut self) -> BoxedFuture<'_, Result<Metadata>> {
+ Box::pin(self.close())
+ }
+
fn abort_dyn(&mut self) -> BoxedFuture<'_, Result<()>> {
Box::pin(self.abort())
}
@@ -113,6 +142,10 @@ impl<T: CopyDyn + ?Sized> Copy for Box<T> {
self.deref_mut().next_dyn().await
}
+ async fn close(&mut self) -> Result<Metadata> {
+ self.deref_mut().close_dyn().await
+ }
+
async fn abort(&mut self) -> Result<()> {
self.deref_mut().abort_dyn().await
}
diff --git a/core/core/src/raw/oio/copy/block_copy.rs b/core/core/src/raw/oio/copy/block_copy.rs
index d04d2505a..46aab7817 100644
--- a/core/core/src/raw/oio/copy/block_copy.rs
+++ b/core/core/src/raw/oio/copy/block_copy.rs
@@ -41,7 +41,7 @@ pub trait BlockCopy: Send + Sync + Unpin + 'static {
///
/// BlockCopier will call this API when the source object can be copied
/// without starting block copy.
- fn copy_once(&self) -> impl Future<Output = Result<()>> + MaybeSend;
+ fn copy_once(&self) -> impl Future<Output = Result<Metadata>> + MaybeSend;
/// copy_block copies one source range into one block.
fn copy_block(
@@ -51,7 +51,10 @@ pub trait BlockCopy: Send + Sync + Unpin + 'static {
) -> impl Future<Output = Result<()>> + MaybeSend;
/// complete_block completes the block copy with the ordered block id list.
- fn complete_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output = Result<()>> + MaybeSend;
+ fn complete_block(
+ &self,
+ block_ids: Vec<Uuid>,
+ ) -> impl Future<Output = Result<Metadata>> + MaybeSend;
/// abort_block cancels the pending block copy and purges intermediate state.
fn abort_block(&self, block_ids: Vec<Uuid>) -> impl Future<Output = Result<()>> + MaybeSend;
@@ -97,6 +100,7 @@ pub struct BlockCopier<C: BlockCopy> {
block_size: u64,
concurrent: usize,
completed: bool,
+ metadata: Option<Metadata>,
tasks: ConcurrentTasks<CopyInput<C>, CopiedBlock>,
}
@@ -127,6 +131,7 @@ impl<C: BlockCopy> BlockCopier<C> {
block_size,
concurrent,
completed: false,
+ metadata: None,
tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
Box::pin(async move {
@@ -227,7 +232,7 @@ where
let source_size = self.source_size().await?;
if self.block_ids.is_empty() && source_size <= self.copy_once_threshold {
- self.copier.copy_once().await?;
+ self.metadata = Some(self.copier.copy_once().await?);
self.completed = true;
return Ok(None);
}
@@ -265,15 +270,24 @@ where
.iter()
.map(|(_, block_id)| *block_id)
.collect();
- self.copier.complete_block(block_ids).await?;
+ self.metadata = Some(self.copier.complete_block(block_ids).await?);
self.completed = true;
Ok(None)
}
+ async fn close(&mut self) -> Result<Metadata> {
+ while !self.completed {
+ self.next().await?;
+ }
+
+ Ok(self.metadata.clone().unwrap_or_default())
+ }
+
async fn abort(&mut self) -> Result<()> {
self.tasks.clear();
if self.scheduled_block_ids.is_empty() {
self.completed = true;
+ self.metadata = None;
return Ok(());
}
@@ -281,6 +295,7 @@ where
.abort_block(self.scheduled_block_ids.clone())
.await?;
self.completed = true;
+ self.metadata = None;
Ok(())
}
}
@@ -311,8 +326,8 @@ mod tests {
Ok(Metadata::default().with_content_length(4))
}
- async fn copy_once(&self) -> Result<()> {
- Ok(())
+ async fn copy_once(&self) -> Result<Metadata> {
+ Ok(Metadata::default())
}
async fn copy_block(&self, block_id: Uuid, range: BytesRange) -> Result<()> {
@@ -329,13 +344,13 @@ mod tests {
Ok(())
}
- async fn complete_block(&self, block_ids: Vec<Uuid>) -> Result<()> {
+ async fn complete_block(&self, block_ids: Vec<Uuid>) -> Result<Metadata> {
let mut state = self.state.lock().expect("test state mutex poisoned");
state.completed_ranges = block_ids
.into_iter()
.map(|block_id| state.ranges[&block_id])
.collect();
- Ok(())
+ Ok(Metadata::default())
}
async fn abort_block(&self, _: Vec<Uuid>) -> Result<()> {
diff --git a/core/core/src/raw/oio/copy/multipart_copy.rs b/core/core/src/raw/oio/copy/multipart_copy.rs
index 445f55cb8..d4a85c198 100644
--- a/core/core/src/raw/oio/copy/multipart_copy.rs
+++ b/core/core/src/raw/oio/copy/multipart_copy.rs
@@ -41,7 +41,7 @@ pub trait MultipartCopy: Send + Sync + Unpin + 'static {
///
/// MultipartCopier will call this API when the source object can be copied
/// without starting a multipart copy.
- fn copy_once(&self) -> impl Future<Output = Result<()>> + MaybeSend;
+ fn copy_once(&self) -> impl Future<Output = Result<Metadata>> + MaybeSend;
/// initiate_copy starts a multipart copy and returns the upload id.
fn initiate_copy(&self) -> impl Future<Output = Result<String>> + MaybeSend;
@@ -61,7 +61,7 @@ pub trait MultipartCopy: Send + Sync + Unpin + 'static {
&self,
upload_id: &str,
parts: &[MultipartPart],
- ) -> impl Future<Output = Result<()>> + MaybeSend;
+ ) -> impl Future<Output = Result<Metadata>> + MaybeSend;
/// abort_copy cancels the pending multipart copy and purges intermediate state.
fn abort_copy(&self, upload_id: &str) -> impl Future<Output = Result<()>> + MaybeSend;
@@ -106,6 +106,7 @@ pub struct MultipartCopier<C: MultipartCopy> {
part_size: u64,
concurrent: usize,
completed: bool,
+ metadata: Option<Metadata>,
tasks: ConcurrentTasks<CopyInput<C>, CopiedPart>,
}
@@ -136,6 +137,7 @@ impl<C: MultipartCopy> MultipartCopier<C> {
part_size,
concurrent,
completed: false,
+ metadata: None,
tasks: ConcurrentTasks::new(executor, concurrent, 8192, |input| {
Box::pin(async move {
@@ -279,7 +281,7 @@ where
let source_size = self.source_size().await?;
if self.upload_id.is_none() && source_size <= self.copy_once_threshold {
- self.copier.copy_once().await?;
+ self.metadata = Some(self.copier.copy_once().await?);
self.completed = true;
return Ok(None);
}
@@ -321,11 +323,19 @@ where
}
self.parts.sort_by_key(|part| part.part_number);
- self.copier.complete_copy(&upload_id, &self.parts).await?;
+ self.metadata = Some(self.copier.complete_copy(&upload_id, &self.parts).await?);
self.completed = true;
Ok(None)
}
+ async fn close(&mut self) -> Result<Metadata> {
+ while !self.completed {
+ self.next().await?;
+ }
+
+ Ok(self.metadata.clone().unwrap_or_default())
+ }
+
async fn abort(&mut self) -> Result<()> {
self.tasks.clear();
let Some(upload_id) = self.upload_id.take() else {
@@ -334,6 +344,7 @@ where
self.copier.abort_copy(&upload_id).await?;
self.completed = true;
+ self.metadata = None;
Ok(())
}
}
@@ -370,9 +381,9 @@ mod tests {
Ok(Metadata::default().with_content_length(self.source_size))
}
- async fn copy_once(&self) -> Result<()> {
+ async fn copy_once(&self) -> Result<Metadata> {
self.copy_once_calls.fetch_add(1, Ordering::Relaxed);
- Ok(())
+ Ok(Metadata::default())
}
async fn initiate_copy(&self) -> Result<String> {
@@ -394,8 +405,8 @@ mod tests {
})
}
- async fn complete_copy(&self, _: &str, _: &[MultipartPart]) -> Result<()> {
- Ok(())
+ async fn complete_copy(&self, _: &str, _: &[MultipartPart]) -> Result<Metadata> {
+ Ok(Metadata::default())
}
async fn abort_copy(&self, _: &str) -> Result<()> {
diff --git a/core/core/src/types/copy.rs b/core/core/src/types/copy.rs
index 759db4a85..9a1384bce 100644
--- a/core/core/src/types/copy.rs
+++ b/core/core/src/types/copy.rs
@@ -33,6 +33,7 @@ use crate::*;
/// - `None` means the copy operation has completed.
pub struct Copier {
copier: Option<oio::Copier>,
+ metadata: Option<Metadata>,
fut: Option<BoxedStaticFuture<(oio::Copier, Result<Option<usize>>)>>,
errored: bool,
@@ -65,6 +66,7 @@ impl Copier {
Ok(Self {
copier: Some(copier),
+ metadata: None,
fut: None,
errored: false,
@@ -84,6 +86,7 @@ impl Copier {
match copier.next_dyn().await {
Ok(Some(n)) => Ok(Some(n)),
Ok(None) => {
+ self.metadata = Some(copier.close_dyn().await?);
self.copier = None;
Ok(None)
}
@@ -102,6 +105,29 @@ impl Copier {
copier.abort_dyn().await
}
+
+ /// Close the copier and return metadata from the server-side completion response.
+ pub async fn close(&mut self) -> Result<Metadata> {
+ if self.errored {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "copier has already failed and can't be closed",
+ ));
+ }
+
+ if let Some(metadata) = self.metadata.clone() {
+ return Ok(metadata);
+ }
+
+ while self.metadata.is_none() {
+ match self.next().await? {
+ Some(_) => continue,
+ None => break,
+ }
+ }
+
+ Ok(self.metadata.clone().unwrap_or_default())
+ }
}
impl Stream for Copier {
@@ -127,10 +153,7 @@ impl Stream for Copier {
return match res {
Ok(Some(n)) => Poll::Ready(Some(Ok(n))),
- Ok(None) => {
- self.copier = None;
- Poll::Ready(None)
- }
+ Ok(None) => Poll::Ready(None),
Err(err) => {
self.errored = true;
Poll::Ready(Some(Err(err)))
diff --git a/core/core/src/types/operator/operator.rs b/core/core/src/types/operator/operator.rs
index 3bbe21d6b..e2c73e844 100644
--- a/core/core/src/types/operator/operator.rs
+++ b/core/core/src/types/operator/operator.rs
@@ -989,7 +989,7 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub async fn copy(&self, from: &str, to: &str) -> Result<()> {
+ pub async fn copy(&self, from: &str, to: &str) -> Result<Metadata> {
self.copy_options(from, to, options::CopyOptions::default())
.await
}
@@ -1027,7 +1027,11 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
- pub fn copy_with(&self, from: &str, to: &str) -> FutureCopy<impl Future<Output = Result<()>>> {
+ pub fn copy_with(
+ &self,
+ from: &str,
+ to: &str,
+ ) -> FutureCopy<impl Future<Output = Result<Metadata>>> {
let from = normalize_path(from);
let to = normalize_path(to);
@@ -1089,7 +1093,7 @@ impl Operator {
from: &str,
to: &str,
opts: impl Into<options::CopyOptions>,
- ) -> Result<()> {
+ ) -> Result<Metadata> {
let from = normalize_path(from);
let to = normalize_path(to);
let opts = opts.into();
@@ -1115,7 +1119,7 @@ impl Operator {
acc: Accessor,
from: String,
(opts, to): (options::CopyOptions, String),
- ) -> Result<()> {
+ ) -> Result<Metadata> {
if !validate_path(&from, EntryMode::FILE) {
return Err(
Error::new(ErrorKind::IsADirectory, "from path is a directory")
@@ -1145,15 +1149,11 @@ impl Operator {
}
let mut copier = Self::copier_inner(acc, from, (opts, to)).await?;
-
- loop {
- match copier.next().await {
- Ok(Some(_)) => continue,
- Ok(None) => return Ok(()),
- Err(err) => {
- let _ = copier.abort().await;
- return Err(err);
- }
+ match copier.close().await {
+ Ok(meta) => Ok(meta),
+ Err(err) => {
+ let _ = copier.abort().await;
+ Err(err)
}
}
}
diff --git a/core/core/src/types/operator/operator_futures.rs b/core/core/src/types/operator/operator_futures.rs
index 34e227117..093395fb3 100644
--- a/core/core/src/types/operator/operator_futures.rs
+++ b/core/core/src/types/operator/operator_futures.rs
@@ -1411,9 +1411,9 @@ impl<F: Future<Output = Result<Lister>>> FutureLister<F> {
/// Future that generated by [`Operator::copy_with`].
///
/// Users can add more options by public functions provided by this struct.
-pub type FutureCopy<F> = OperatorFuture<(options::CopyOptions, String), (), F>;
+pub type FutureCopy<F> = OperatorFuture<(options::CopyOptions, String), Metadata, F>;
-impl<F: Future<Output = Result<()>>> FutureCopy<F> {
+impl<F: Future<Output = Result<Metadata>>> FutureCopy<F> {
/// Sets the condition that copy operation will succeed only if target does not exist.
///
/// Refer to [`options::CopyOptions::if_not_exists`] for more details.
diff --git a/core/layers/await-tree/src/lib.rs b/core/layers/await-tree/src/lib.rs
index bfbe4d65f..457eca195 100644
--- a/core/layers/await-tree/src/lib.rs
+++ b/core/layers/await-tree/src/lib.rs
@@ -222,6 +222,12 @@ impl<C: oio::Copy> oio::Copy for AwaitTreeWrapper<C> {
.instrument_await(format!("opendal::{}", Operation::Copy.into_static()))
}
+ fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend {
+ self.inner
+ .close()
+ .instrument_await(format!("opendal::{}", Operation::Copy.into_static()))
+ }
+
fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
self.inner
.abort()
diff --git a/core/layers/concurrent-limit/src/lib.rs b/core/layers/concurrent-limit/src/lib.rs
index ea571b723..00707a0cd 100644
--- a/core/layers/concurrent-limit/src/lib.rs
+++ b/core/layers/concurrent-limit/src/lib.rs
@@ -385,6 +385,10 @@ impl<C: oio::Copy, P: Send + Sync + 'static + Unpin> oio::Copy for ConcurrentLim
self.inner.next().await
}
+ async fn close(&mut self) -> Result<Metadata> {
+ self.inner.close().await
+ }
+
async fn abort(&mut self) -> Result<()> {
self.inner.abort().await
}
diff --git a/core/layers/fastrace/src/lib.rs b/core/layers/fastrace/src/lib.rs
index bc7967067..12df74d9b 100644
--- a/core/layers/fastrace/src/lib.rs
+++ b/core/layers/fastrace/src/lib.rs
@@ -305,6 +305,12 @@ impl<C: oio::Copy> oio::Copy for FastraceWrapper<C> {
self.inner.next()
}
+ fn close(&mut self) -> impl Future<Output = Result<Metadata>> + MaybeSend {
+ let _g = self.span.set_local_parent();
+ let _span = LocalSpan::enter_with_local_parent(Operation::Copy.into_static());
+ self.inner.close()
+ }
+
fn abort(&mut self) -> impl Future<Output = Result<()>> + MaybeSend {
let _g = self.span.set_local_parent();
let _span = LocalSpan::enter_with_local_parent(Operation::Copy.into_static());
diff --git a/core/layers/hotpath/src/lib.rs b/core/layers/hotpath/src/lib.rs
index 58bf0e5cc..2b1269f1b 100644
--- a/core/layers/hotpath/src/lib.rs
+++ b/core/layers/hotpath/src/lib.rs
@@ -47,6 +47,7 @@ const LABEL_LISTER_NEXT: &str = "opendal.lister.next";
const LABEL_DELETER_DELETE: &str = "opendal.deleter.delete";
const LABEL_DELETER_CLOSE: &str = "opendal.deleter.close";
const LABEL_COPIER_NEXT: &str = "opendal.copier.next";
+const LABEL_COPIER_CLOSE: &str = "opendal.copier.close";
const LABEL_COPIER_ABORT: &str = "opendal.copier.abort";
const LABEL_HTTP_FETCH: &str = "opendal.http.fetch";
const LABEL_HTTP_BODY_POLL: &str = "opendal.http.body.poll";
@@ -224,6 +225,10 @@ impl<C: oio::Copy> oio::Copy for HotpathWrapper<C> {
hotpath::measure_async(LABEL_COPIER_NEXT, self.inner.next()).await
}
+ async fn close(&mut self) -> Result<Metadata> {
+ hotpath::measure_async(LABEL_COPIER_CLOSE, self.inner.close()).await
+ }
+
async fn abort(&mut self) -> Result<()> {
hotpath::measure_async(LABEL_COPIER_ABORT, self.inner.abort()).await
}
diff --git a/core/layers/logging/src/lib.rs b/core/layers/logging/src/lib.rs
index 6317927dd..4b3992734 100644
--- a/core/layers/logging/src/lib.rs
+++ b/core/layers/logging/src/lib.rs
@@ -909,6 +909,10 @@ impl<C: oio::Copy, I: LoggingInterceptor> oio::Copy for LoggingCopier<C, I> {
}
}
+ async fn close(&mut self) -> Result<Metadata> {
+ self.inner.close().await
+ }
+
async fn abort(&mut self) -> Result<()> {
match self.inner.abort().await {
Ok(_) => {
diff --git a/core/layers/observe-metrics-common/src/lib.rs b/core/layers/observe-metrics-common/src/lib.rs
index 2b12826e9..1c381056b 100644
--- a/core/layers/observe-metrics-common/src/lib.rs
+++ b/core/layers/observe-metrics-common/src/lib.rs
@@ -1179,6 +1179,16 @@ impl<C: oio::Copy, I: MetricsIntercept> oio::Copy for MetricsWrapper<C, I> {
.inspect_err(|err| self.record_error(err))
}
+ async fn close(&mut self) -> Result<Metadata> {
+ let result = self
+ .inner
+ .close()
+ .await
+ .inspect_err(|err| self.record_error(err));
+ self.completed = true;
+ result
+ }
+
async fn abort(&mut self) -> Result<()> {
let result = self
.inner
diff --git a/core/layers/retry/src/lib.rs b/core/layers/retry/src/lib.rs
index ef1ed0142..eb4a4fc28 100644
--- a/core/layers/retry/src/lib.rs
+++ b/core/layers/retry/src/lib.rs
@@ -789,6 +789,37 @@ impl<C: oio::Copy, I: RetryInterceptor> oio::Copy for RetryWrapper<C, I> {
res.map_err(|err| err.set_persistent())
}
+ async fn close(&mut self) -> Result<Metadata> {
+ use backon::RetryableWithContext;
+
+ let inner = self.take_inner()?;
+ let mut attempt: u32 = 0;
+
+ let (inner, res) = {
+ |mut c: C| async move {
+ let res = c.close().await;
+
+ (c, res)
+ }
+ }
+ .retry(self.builder)
+ .when(|e| e.is_temporary())
+ .context(inner)
+ .notify(|err, dur| {
+ attempt += 1;
+ self.notify.intercept(RetryEvent {
+ op: Operation::Copy,
+ err,
+ retry_after: dur,
+ attempt,
+ })
+ })
+ .await;
+
+ self.inner = Some(inner);
+ res.map_err(|err| err.set_persistent())
+ }
+
async fn abort(&mut self) -> Result<()> {
use backon::RetryableWithContext;
@@ -1099,6 +1130,10 @@ mod tests {
}
}
+ async fn close(&mut self) -> Result<Metadata> {
+ Ok(Metadata::default())
+ }
+
async fn abort(&mut self) -> Result<()> {
Ok(())
}
diff --git a/core/layers/tail-cut/src/lib.rs b/core/layers/tail-cut/src/lib.rs
index 030fab382..8c12f0a0c 100644
--- a/core/layers/tail-cut/src/lib.rs
+++ b/core/layers/tail-cut/src/lib.rs
@@ -640,6 +640,19 @@ impl<C: oio::Copy> oio::Copy for TailCutWrapper<C> {
.await
}
+ async fn close(&mut self) -> Result<Metadata> {
+ let deadline = self.calculate_deadline(Operation::Copy);
+ Self::with_io_deadline(
+ deadline,
+ self.config.percentile,
+ &self.stats,
+ self.size,
+ Operation::Copy,
+ self.inner.close(),
+ )
+ .await
+ }
+
async fn abort(&mut self) -> Result<()> {
let deadline = self.calculate_deadline(Operation::Copy);
Self::with_io_deadline(
diff --git a/core/layers/timeout/src/lib.rs b/core/layers/timeout/src/lib.rs
index 004c6d8c2..b942e8245 100644
--- a/core/layers/timeout/src/lib.rs
+++ b/core/layers/timeout/src/lib.rs
@@ -376,6 +376,11 @@ impl<C: oio::Copy> oio::Copy for TimeoutWrapper<C> {
Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
}
+ async fn close(&mut self) -> Result<Metadata> {
+ let fut = self.inner.close();
+ Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
+ }
+
async fn abort(&mut self) -> Result<()> {
let fut = self.inner.abort();
Self::io_timeout(self.timeout, Operation::Copy.into_static(), fut).await
@@ -479,6 +484,10 @@ mod tests {
pending()
}
+ fn close(&mut self) -> impl Future<Output = Result<Metadata>> {
+ pending()
+ }
+
fn abort(&mut self) -> impl Future<Output = Result<()>> {
pending()
}
diff --git a/core/services/azblob/src/copier.rs b/core/services/azblob/src/copier.rs
index 5cea31d9b..f8ca7c1e1 100644
--- a/core/services/azblob/src/copier.rs
+++ b/core/services/azblob/src/copier.rs
@@ -25,6 +25,7 @@ use super::core::constants::AZBLOB_COPY_MAX_BLOCK_SIZE;
use super::core::constants::AZBLOB_COPY_MIN_BLOCK_SIZE;
use super::core::constants::X_MS_VERSION_ID;
use super::error::parse_error;
+use super::writer::AzblobWriter;
use opendal_core::raw::oio::BlockCopy;
use opendal_core::raw::*;
use opendal_core::*;
@@ -48,8 +49,7 @@ pub fn new_azblob_copier(
let Some(chunk) = opts.chunk() else {
return Ok(TwoWays::One(oio::OneShotCopier::new(async move {
- copier.copy_once().await?;
- Ok(None)
+ copier.copy_once().await
})));
};
@@ -92,14 +92,14 @@ impl oio::BlockCopy for AzblobCopier {
}
}
- async fn copy_once(&self) -> Result<()> {
+ async fn copy_once(&self) -> Result<Metadata> {
let resp = self
.core
.azblob_copy_blob(&self.from, &self.to, self.args.clone())
.await?;
match resp.status() {
- StatusCode::ACCEPTED => Ok(()),
+ StatusCode::ACCEPTED => AzblobWriter::parse_metadata(resp.headers()),
_ => Err(parse_error(resp)),
}
}
@@ -116,14 +116,15 @@ impl oio::BlockCopy for AzblobCopier {
}
}
- async fn complete_block(&self, block_ids: Vec<Uuid>) -> Result<()> {
+ async fn complete_block(&self, block_ids: Vec<Uuid>) -> Result<Metadata> {
let resp = self
.core
.azblob_complete_copy_block_list(&self.to, block_ids, &self.args)
.await?;
+ let meta = AzblobWriter::parse_metadata(resp.headers())?;
match resp.status() {
- StatusCode::CREATED | StatusCode::OK => Ok(()),
+ StatusCode::CREATED | StatusCode::OK => Ok(meta),
_ => Err(parse_error(resp)),
}
}
diff --git a/core/services/azblob/src/writer.rs b/core/services/azblob/src/writer.rs
index 6d9fc2145..d6472a53d 100644
--- a/core/services/azblob/src/writer.rs
+++ b/core/services/azblob/src/writer.rs
@@ -44,7 +44,7 @@ impl AzblobWriter {
// skip extracting `content-md5` here, as it pertains to the content of the request rather than
// the content of the block itself for the `append` and `complete put block list` operations.
- fn parse_metadata(headers: &http::HeaderMap) -> Result<Metadata> {
+ pub(crate) fn parse_metadata(headers: &http::HeaderMap) -> Result<Metadata> {
let mut metadata = Metadata::default();
if let Some(last_modified) = parse_last_modified(headers)? {
diff --git a/core/services/gcs/src/copier.rs b/core/services/gcs/src/copier.rs
index 5ddbcbff0..63d6fcf82 100644
--- a/core/services/gcs/src/copier.rs
+++ b/core/services/gcs/src/copier.rs
@@ -37,6 +37,7 @@ pub struct GcsCopier {
rewrite_token: Option<String>,
total_bytes_rewritten: u64,
completed: bool,
+ metadata: Option<Metadata>,
}
impl GcsCopier {
@@ -55,6 +56,7 @@ impl GcsCopier {
rewrite_token: None,
total_bytes_rewritten: 0,
completed: false,
+ metadata: None,
}
}
}
@@ -108,6 +110,7 @@ impl oio::Copy for GcsCopier {
if result.done {
self.completed = true;
self.rewrite_token = None;
+ self.metadata = Some(result.into_metadata(&self.to)?);
} else {
let token = result.rewrite_token.ok_or_else(|| {
Error::new(
@@ -135,9 +138,18 @@ impl oio::Copy for GcsCopier {
Ok(Some(progress))
}
+ async fn close(&mut self) -> Result<Metadata> {
+ while !self.completed {
+ self.next().await?;
+ }
+
+ Ok(self.metadata.clone().unwrap_or_default())
+ }
+
async fn abort(&mut self) -> Result<()> {
self.rewrite_token = None;
self.completed = true;
+ self.metadata = None;
Ok(())
}
}
diff --git a/core/services/gcs/src/core.rs b/core/services/gcs/src/core.rs
index 8bd7e1483..11f699f61 100644
--- a/core/services/gcs/src/core.rs
+++ b/core/services/gcs/src/core.rs
@@ -707,40 +707,48 @@ impl GcsCore {
let meta: GetObjectJsonResponse =
serde_json::from_reader(data.reader()).map_err(new_json_deserialize_error)?;
+ meta.into_metadata(path)
+ }
+}
+
+impl GetObjectJsonResponse {
+ fn into_metadata(self, path: &str) -> Result<Metadata> {
let mut m = Metadata::new(EntryMode::from_path(path));
- m.set_etag(&meta.etag);
- m.set_content_md5(&meta.md5_hash);
+ m.set_etag(&self.etag);
+ m.set_content_md5(&self.md5_hash);
- let size = meta
+ let size = self
.size
.parse::<u64>()
.map_err(|e| Error::new(ErrorKind::Unexpected, "parse u64").set_source(e))?;
m.set_content_length(size);
- if !meta.content_type.is_empty() {
- m.set_content_type(&meta.content_type);
+ if !self.content_type.is_empty() {
+ m.set_content_type(&self.content_type);
}
- if !meta.content_encoding.is_empty() {
- m.set_content_encoding(&meta.content_encoding);
+ if !self.content_encoding.is_empty() {
+ m.set_content_encoding(&self.content_encoding);
}
- if !meta.cache_control.is_empty() {
- m.set_cache_control(&meta.cache_control);
+ if !self.cache_control.is_empty() {
+ m.set_cache_control(&self.cache_control);
}
- if !meta.content_disposition.is_empty() {
- m.set_content_disposition(&meta.content_disposition);
+ if !self.content_disposition.is_empty() {
+ m.set_content_disposition(&self.content_disposition);
}
- if !meta.generation.is_empty() {
- m.set_version(&meta.generation);
+ if !self.generation.is_empty() {
+ m.set_version(&self.generation);
}
- m.set_last_modified(meta.updated.parse::<Timestamp>()?);
+ if !self.updated.is_empty() {
+ m.set_last_modified(self.updated.parse::<Timestamp>()?);
+ }
- if !meta.metadata.is_empty() {
- m = m.with_user_metadata(meta.metadata);
+ if !self.metadata.is_empty() {
+ m = m.with_user_metadata(self.metadata);
}
Ok(m)
@@ -844,6 +852,16 @@ pub struct RewriteResponse {
pub total_bytes_rewritten: String,
pub done: bool,
pub rewrite_token: Option<String>,
+ resource: Option<GetObjectJsonResponse>,
+}
+
+impl RewriteResponse {
+ pub fn into_metadata(self, path: &str) -> Result<Metadata> {
+ match self.resource {
+ Some(resource) => resource.into_metadata(path),
+ None => Ok(Metadata::default()),
+ }
+ }
}
/// The raw json response returned by [`get`](https://cloud.google.com/storage/docs/json_api/v1/objects/get)
diff --git a/core/services/s3/src/copier.rs b/core/services/s3/src/copier.rs
index 18ff64986..39dbd04f8 100644
--- a/core/services/s3/src/copier.rs
+++ b/core/services/s3/src/copier.rs
@@ -99,7 +99,7 @@ impl oio::MultipartCopy for S3Copier {
}
}
- async fn copy_once(&self) -> Result<()> {
+ async fn copy_once(&self) -> Result<Metadata> {
let resp = self
.core
.s3_copy_object(&self.from, &self.to, &self.args)
@@ -109,6 +109,8 @@ impl oio::MultipartCopy for S3Copier {
StatusCode::OK => {
let (parts, body) = resp.into_parts();
let bs = body.to_bytes();
+ let version = parse_header_to_str(&parts.headers, constants::X_AMZ_VERSION_ID)?
+ .map(str::to_string);
let result: CopyObjectResult =
quick_xml::de::from_reader(bs.as_ref()).map_err(new_xml_deserialize_error)?;
@@ -126,7 +128,16 @@ impl oio::MultipartCopy for S3Copier {
));
}
- Ok(())
+ let mut meta = Metadata::new(EntryMode::from_path(&self.to));
+ meta.set_etag(&result.etag);
+ if !result.last_modified.is_empty() {
+ meta.set_last_modified(result.last_modified.parse()?);
+ }
+ if let Some(version) = version {
+ meta.set_version(&version);
+ }
+
+ Ok(meta)
}
_ => Err(parse_error(resp)),
}
@@ -200,7 +211,11 @@ impl oio::MultipartCopy for S3Copier {
}
}
- async fn complete_copy(&self, upload_id: &str, parts: &[oio::MultipartPart]) -> Result<()> {
+ async fn complete_copy(
+ &self,
+ upload_id: &str,
+ parts: &[oio::MultipartPart],
+ ) -> Result<Metadata> {
let parts = parts
.iter()
.map(|p| CompleteMultipartUploadRequestPart {
@@ -220,6 +235,8 @@ impl oio::MultipartCopy for S3Copier {
match status {
StatusCode::OK => {
let (parts, body) = resp.into_parts();
+ let version = parse_header_to_str(&parts.headers, constants::X_AMZ_VERSION_ID)?
+ .map(str::to_string);
let ret: CompleteMultipartUploadResult =
quick_xml::de::from_reader(body.reader()).map_err(new_xml_deserialize_error)?;
@@ -236,7 +253,13 @@ impl oio::MultipartCopy for S3Copier {
));
}
- Ok(())
+ let mut meta = Metadata::new(EntryMode::from_path(&self.to));
+ meta.set_etag(&ret.etag);
+ if let Some(version) = version {
+ meta.set_version(&version);
+ }
+
+ Ok(meta)
}
_ => Err(parse_error(resp)),
}
diff --git a/core/services/tos/src/copier.rs b/core/services/tos/src/copier.rs
index 462ef0f91..608f6399b 100644
--- a/core/services/tos/src/copier.rs
+++ b/core/services/tos/src/copier.rs
@@ -98,7 +98,7 @@ impl oio::MultipartCopy for TosCopier {
}
}
- async fn copy_once(&self) -> Result<()> {
+ async fn copy_once(&self) -> Result<Metadata> {
let resp = self
.core
.tos_copy_object(&self.from, &self.to, &self.args)
@@ -116,7 +116,9 @@ impl oio::MultipartCopy for TosCopier {
.set_temporary());
}
- Ok(())
+ let mut meta = Metadata::new(EntryMode::from_path(&self.to));
+ meta.set_etag(result.etag.trim_matches('"'));
+ Ok(meta)
}
_ => Err(parse_error(resp)),
}
@@ -181,7 +183,11 @@ impl oio::MultipartCopy for TosCopier {
}
}
- async fn complete_copy(&self, upload_id: &str, parts: &[oio::MultipartPart]) -> Result<()> {
+ async fn complete_copy(
+ &self,
+ upload_id: &str,
+ parts: &[oio::MultipartPart],
+ ) -> Result<Metadata> {
let parts = parts
.iter()
.map(|p| CompleteMultipartUploadRequestPart {
@@ -204,7 +210,15 @@ impl oio::MultipartCopy for TosCopier {
return Err(Error::new(ErrorKind::Unexpected, ret.message));
}
- Ok(())
+ let mut meta = Metadata::new(EntryMode::from_path(&self.to));
+ if !ret.etag.is_empty() {
+ meta.set_etag(ret.etag.trim_matches('"'));
+ }
+ if !ret.version_id.is_empty() {
+ meta.set_version(&ret.version_id);
+ }
+
+ Ok(meta)
}
_ => Err(parse_error(resp)),
}
diff --git a/integrations/dav-server/src/fs.rs b/integrations/dav-server/src/fs.rs
index 3316d4a3f..4f259a1bf 100644
--- a/integrations/dav-server/src/fs.rs
+++ b/integrations/dav-server/src/fs.rs
@@ -200,6 +200,7 @@ impl DavFileSystem for OpendalFs {
self.op
.copy(from_path, to_path)
.await
+ .map(|_| ())
.map_err(convert_error)
}
.boxed()