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 05:20:14 UTC
(opendal) branch xuanwo/copier-gcs-rewrite created (now 5fa9f753a)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/copier-gcs-rewrite
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 5fa9f753a feat(services/gcs): add rewrite copier
This branch includes the following new commits:
new 5fa9f753a feat(services/gcs): add rewrite copier
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/01: feat(services/gcs): add rewrite copier
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/copier-gcs-rewrite
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 5fa9f753aec425335d24b6bcb56b59a148cde15d
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon May 18 13:19:57 2026 +0800
feat(services/gcs): add rewrite copier
---
core/services/gcs/src/backend.rs | 24 +++---
core/services/gcs/src/copier.rs | 182 +++++++++++++++++++++++++++++++++++++++
core/services/gcs/src/core.rs | 37 ++++++--
core/services/gcs/src/lib.rs | 1 +
4 files changed, 229 insertions(+), 15 deletions(-)
diff --git a/core/services/gcs/src/backend.rs b/core/services/gcs/src/backend.rs
index d066a5e9a..28ce41fc7 100644
--- a/core/services/gcs/src/backend.rs
+++ b/core/services/gcs/src/backend.rs
@@ -39,6 +39,7 @@ use reqsign_google::VmMetadataCredentialProvider;
use super::GCS_SCHEME;
use super::config::GcsConfig;
+use super::copier::GcsCopier;
use super::core::*;
use super::deleter::GcsDeleter;
use super::error::parse_error;
@@ -366,7 +367,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(GcsCopier::min_chunk_size()),
+ copy_multi_max_size: Some(GcsCopier::max_chunk_size()),
list: true,
list_with_limit: true,
@@ -410,7 +419,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 +489,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..1524484c9
--- /dev/null
+++ b/core/services/gcs/src/copier.rs
@@ -0,0 +1,182 @@
+// 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::error::parse_error;
+use opendal_core::raw::*;
+use opendal_core::*;
+
+const GCS_REWRITE_MIN_CHUNK_SIZE: usize = 1024 * 1024;
+
+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| v.clamp(GCS_REWRITE_MIN_CHUNK_SIZE, gcs_rewrite_max_chunk_size()))
+ .map(align_gcs_rewrite_chunk_size);
+
+ Self {
+ core,
+ from: from.to_string(),
+ to: to.to_string(),
+ args,
+ chunk,
+ rewrite_token: None,
+ total_bytes_rewritten: 0,
+ completed: false,
+ }
+ }
+
+ pub fn min_chunk_size() -> usize {
+ GCS_REWRITE_MIN_CHUNK_SIZE
+ }
+
+ pub fn max_chunk_size() -> usize {
+ gcs_rewrite_max_chunk_size()
+ }
+}
+
+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(())
+ }
+}
+
+fn gcs_rewrite_max_chunk_size() -> usize {
+ const I64_MAX_AS_USIZE: usize = i64::MAX as usize;
+
+ I64_MAX_AS_USIZE / GCS_REWRITE_MIN_CHUNK_SIZE * GCS_REWRITE_MIN_CHUNK_SIZE
+}
+
+fn align_gcs_rewrite_chunk_size(size: usize) -> usize {
+ size / GCS_REWRITE_MIN_CHUNK_SIZE * GCS_REWRITE_MIN_CHUNK_SIZE
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_align_gcs_rewrite_chunk_size() {
+ assert_eq!(
+ align_gcs_rewrite_chunk_size(GCS_REWRITE_MIN_CHUNK_SIZE),
+ GCS_REWRITE_MIN_CHUNK_SIZE
+ );
+ assert_eq!(
+ align_gcs_rewrite_chunk_size(GCS_REWRITE_MIN_CHUNK_SIZE + 1),
+ GCS_REWRITE_MIN_CHUNK_SIZE
+ );
+ assert_eq!(
+ align_gcs_rewrite_chunk_size(GCS_REWRITE_MIN_CHUNK_SIZE * 2),
+ GCS_REWRITE_MIN_CHUNK_SIZE * 2
+ );
+ }
+}
diff --git a/core/services/gcs/src/core.rs b/core/services/gcs/src/core.rs
index 2d12cde86..bd1444db3 100644
--- a/core/services/gcs/src/core.rs
+++ b/core/services/gcs/src/core.rs
@@ -466,12 +466,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 +486,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 +837,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;