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/18 06:29:29 UTC
(opendal) branch main updated: feat(services/gcs): add rewrite copier (#7545)
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 5e8670e43 feat(services/gcs): add rewrite copier (#7545)
5e8670e43 is described below
commit 5e8670e43683036764f9f5c3230e0225fed9d93e
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon May 18 14:29:24 2026 +0800
feat(services/gcs): add rewrite copier (#7545)
---
core/services/gcs/src/backend.rs | 26 ++++---
core/services/gcs/src/copier.rs | 143 +++++++++++++++++++++++++++++++++++++++
core/services/gcs/src/core.rs | 45 ++++++++++--
core/services/gcs/src/lib.rs | 1 +
4 files changed, 200 insertions(+), 15 deletions(-)
diff --git a/core/services/gcs/src/backend.rs b/core/services/gcs/src/backend.rs
index d066a5e9a..8df101d6e 100644
--- a/core/services/gcs/src/backend.rs
+++ b/core/services/gcs/src/backend.rs
@@ -39,6 +39,9 @@ use reqsign_google::VmMetadataCredentialProvider;
use super::GCS_SCHEME;
use super::config::GcsConfig;
+use super::copier::GcsCopier;
+use super::core::constants::GCS_REWRITE_MAX_CHUNK_SIZE;
+use super::core::constants::GCS_REWRITE_MIN_CHUNK_SIZE;
use super::core::*;
use super::deleter::GcsDeleter;
use super::error::parse_error;
@@ -366,7 +369,15 @@ impl Builder for GcsBuilder {
delete: true,
delete_max_size: Some(100),
+
copy: true,
+ copy_can_multi: true,
+ // GCS rewrite requires maxBytesRewrittenPerCall to be an
+ // integral multiple of 1 MiB if specified.
+ //
+ // ref: <https://cloud.google.com/storage/docs/json_api/v1/objects/rewrite>
+ copy_multi_min_size: Some(GCS_REWRITE_MIN_CHUNK_SIZE),
+ copy_multi_max_size: Some(GCS_REWRITE_MAX_CHUNK_SIZE),
list: true,
list_with_limit: true,
@@ -410,7 +421,7 @@ impl Access for GcsBackend {
type Writer = GcsWriters;
type Lister = oio::PageLister<GcsLister>;
type Deleter = oio::BatchDeleter<GcsDeleter>;
- type Copier = ();
+ type Copier = GcsCopier;
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -480,16 +491,11 @@ impl Access for GcsBackend {
&self,
from: &str,
to: &str,
- _: OpCopy,
- _opts: OpCopier,
+ args: OpCopy,
+ opts: OpCopier,
) -> Result<(RpCopy, Self::Copier)> {
- let resp = self.core.gcs_copy_object(from, to).await?;
-
- if resp.status().is_success() {
- Ok((RpCopy::default(), ()))
- } else {
- Err(parse_error(resp))
- }
+ let copier = GcsCopier::new(self.core.clone(), from, to, args, opts);
+ Ok((RpCopy::default(), copier))
}
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
diff --git a/core/services/gcs/src/copier.rs b/core/services/gcs/src/copier.rs
new file mode 100644
index 000000000..5ddbcbff0
--- /dev/null
+++ b/core/services/gcs/src/copier.rs
@@ -0,0 +1,143 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use bytes::Buf;
+
+use super::core::GcsCore;
+use super::core::RewriteResponse;
+use super::core::constants::GCS_REWRITE_MAX_CHUNK_SIZE;
+use super::core::constants::GCS_REWRITE_MIN_CHUNK_SIZE;
+use super::error::parse_error;
+use opendal_core::raw::*;
+use opendal_core::*;
+
+pub struct GcsCopier {
+ core: Arc<GcsCore>,
+ from: String,
+ to: String,
+ args: OpCopy,
+
+ chunk: Option<usize>,
+ rewrite_token: Option<String>,
+ total_bytes_rewritten: u64,
+ completed: bool,
+}
+
+impl GcsCopier {
+ pub fn new(core: Arc<GcsCore>, from: &str, to: &str, args: OpCopy, opts: OpCopier) -> Self {
+ let chunk = opts.chunk().map(|v| {
+ let v = v.clamp(GCS_REWRITE_MIN_CHUNK_SIZE, GCS_REWRITE_MAX_CHUNK_SIZE);
+ v / GCS_REWRITE_MIN_CHUNK_SIZE * GCS_REWRITE_MIN_CHUNK_SIZE
+ });
+
+ Self {
+ core,
+ from: from.to_string(),
+ to: to.to_string(),
+ args,
+ chunk,
+ rewrite_token: None,
+ total_bytes_rewritten: 0,
+ completed: false,
+ }
+ }
+}
+
+impl oio::Copy for GcsCopier {
+ async fn next(&mut self) -> Result<Option<usize>> {
+ if self.completed {
+ return Ok(None);
+ }
+
+ let resp = self
+ .core
+ .gcs_rewrite_object(
+ &self.from,
+ &self.to,
+ &self.args,
+ self.chunk,
+ self.rewrite_token.as_deref(),
+ )
+ .await?;
+
+ if !resp.status().is_success() {
+ return Err(parse_error(resp));
+ }
+
+ let result: RewriteResponse = serde_json::from_reader(resp.into_body().reader())
+ .map_err(new_json_deserialize_error)?;
+
+ let total = result.total_bytes_rewritten.parse::<u64>().map_err(|err| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "invalid totalBytesRewritten in rewrite response",
+ )
+ .with_context("totalBytesRewritten", &result.total_bytes_rewritten)
+ .set_source(err)
+ })?;
+
+ let progress = total
+ .checked_sub(self.total_bytes_rewritten)
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "rewrite response totalBytesRewritten moved backwards",
+ )
+ .with_context("previous", self.total_bytes_rewritten)
+ .with_context("current", total)
+ })?;
+
+ self.total_bytes_rewritten = total;
+
+ if result.done {
+ self.completed = true;
+ self.rewrite_token = None;
+ } else {
+ let token = result.rewrite_token.ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "rewrite response is not done but rewriteToken is missing",
+ )
+ })?;
+ self.rewrite_token = Some(token);
+ }
+
+ if progress == 0 {
+ return if self.completed {
+ Ok(None)
+ } else {
+ Err(
+ Error::new(ErrorKind::Unexpected, "rewrite response made no progress")
+ .set_temporary(),
+ )
+ };
+ }
+
+ let progress = progress
+ .try_into()
+ .map_err(|_| Error::new(ErrorKind::Unexpected, "rewrite progress exceeds usize"))?;
+ Ok(Some(progress))
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.rewrite_token = None;
+ self.completed = true;
+ Ok(())
+ }
+}
diff --git a/core/services/gcs/src/core.rs b/core/services/gcs/src/core.rs
index 2d12cde86..a05fea9ad 100644
--- a/core/services/gcs/src/core.rs
+++ b/core/services/gcs/src/core.rs
@@ -46,6 +46,14 @@ use opendal_core::raw::*;
use opendal_core::*;
pub mod constants {
+ pub const GCS_REWRITE_MIN_CHUNK_SIZE: usize = 1024 * 1024;
+ #[cfg(target_pointer_width = "64")]
+ pub const GCS_REWRITE_MAX_CHUNK_SIZE: usize =
+ i64::MAX as usize / GCS_REWRITE_MIN_CHUNK_SIZE * GCS_REWRITE_MIN_CHUNK_SIZE;
+ #[cfg(not(target_pointer_width = "64"))]
+ pub const GCS_REWRITE_MAX_CHUNK_SIZE: usize =
+ usize::MAX / GCS_REWRITE_MIN_CHUNK_SIZE * GCS_REWRITE_MIN_CHUNK_SIZE;
+
pub const X_GOOG_ACL: &str = "x-goog-acl";
pub const X_GOOG_STORAGE_CLASS: &str = "x-goog-storage-class";
pub const X_GOOG_META_PREFIX: &str = "x-goog-meta-";
@@ -466,12 +474,19 @@ impl GcsCore {
self.send(req).await
}
- pub async fn gcs_copy_object(&self, from: &str, to: &str) -> Result<Response<Buffer>> {
+ pub async fn gcs_rewrite_object(
+ &self,
+ from: &str,
+ to: &str,
+ args: &OpCopy,
+ max_bytes_rewritten_per_call: Option<usize>,
+ rewrite_token: Option<&str>,
+ ) -> Result<Response<Buffer>> {
let source = build_abs_path(&self.root, from);
let dest = build_abs_path(&self.root, to);
- let req_uri = format!(
- "{}/storage/v1/b/{}/o/{}/copyTo/b/{}/o/{}",
+ let url = format!(
+ "{}/storage/v1/b/{}/o/{}/rewriteTo/b/{}/o/{}",
self.endpoint,
self.bucket,
percent_encode_path(&source),
@@ -479,10 +494,22 @@ impl GcsCore {
percent_encode_path(&dest)
);
- let req = Request::post(req_uri)
+ let mut url = QueryPairsWriter::new(&url);
+
+ if args.if_not_exists() {
+ url = url.push("ifGenerationMatch", "0");
+ }
+ if let Some(max_bytes) = max_bytes_rewritten_per_call {
+ url = url.push("maxBytesRewrittenPerCall", &max_bytes.to_string());
+ }
+ if let Some(token) = rewrite_token {
+ url = url.push("rewriteToken", &percent_encode_path(token));
+ }
+
+ let req = Request::post(url.finish())
.header(CONTENT_LENGTH, 0)
.extension(Operation::Copy)
- .extension(ServiceOperation("CopyObject"))
+ .extension(ServiceOperation("RewriteObject"))
.body(Buffer::new())
.map_err(new_request_build_error)?;
@@ -818,6 +845,14 @@ pub struct CompleteMultipartUploadRequestPart {
pub etag: String,
}
+#[derive(Debug, Default, Deserialize)]
+#[serde(default, rename_all = "camelCase")]
+pub struct RewriteResponse {
+ pub total_bytes_rewritten: String,
+ pub done: bool,
+ pub rewrite_token: Option<String>,
+}
+
/// The raw json response returned by [`get`](https://cloud.google.com/storage/docs/json_api/v1/objects/get)
#[derive(Debug, Default, Deserialize)]
#[serde(default, rename_all = "camelCase")]
diff --git a/core/services/gcs/src/lib.rs b/core/services/gcs/src/lib.rs
index 551eceeee..3e0f9c42d 100644
--- a/core/services/gcs/src/lib.rs
+++ b/core/services/gcs/src/lib.rs
@@ -21,6 +21,7 @@
mod backend;
mod config;
+mod copier;
mod core;
mod deleter;
mod error;