将S3对象回退到历史版本是否真的需要使用multipart copy？

首先明确核心问题结论：同存储桶内将历史版本对象复制到自身的回退操作，受普通PUT复制5GB的上限限制，但无需采用下载到本地再分片上传的低效率方案，可通过S3原生的服务端分片复制能力完成操作，全程无本地数据中转。

适配不同大小对象的回退方案 #

• 当待回退对象的目标历史版本大小 ≤ 5GB时，直接调用CopyObject接口即可实现回退，指定复制源时传入目标版本ID，目标路径填写原对象路径，S3会自动生成对应内容的最新版本，符合官方推荐的回退逻辑。
• 当待回退对象的目标历史版本大小 > 5GB时，使用S3服务端分片复制能力完成回退，全流程操作均在S3服务端执行，无本地带宽消耗，最大支持5TB大小的对象回退，操作步骤如下：
  1. 调用CreateMultipartUpload接口，初始化针对原对象路径的分片上传任务
  2. 按指定分片大小（推荐100MB~500MB，最大支持5GB）拆分目标历史版本的范围，调用UploadPartCopy接口逐个上传分片
  3. 所有分片上传完成后，调用CompleteMultipartUpload接口提交任务，S3会自动拼接分片生成新的最新版本

代码示例（Python + Boto3） #

import boto3

s3_client = boto3.client("s3")
BUCKET_NAME = "替换为你的存储桶名称"
OBJECT_KEY = "替换为待回退对象的路径"
TARGET_VERSION_ID = "替换为需要回退到的历史版本ID"
PART_SIZE = 100 * 1024 * 1024  # 分片大小设置为100MB，可按需调整

# 查询目标历史版本的对象大小
object_meta = s3_client.head_object(
    Bucket=BUCKET_NAME,
    Key=OBJECT_KEY,
    VersionId=TARGET_VERSION_ID
)
object_size = object_meta["ContentLength"]

if object_size <= 5 * 1024 * 1024 * 1024:
    # 小于等于5GB直接使用普通复制
    s3_client.copy_object(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY,
        CopySource={
            "Bucket": BUCKET_NAME,
            "Key": OBJECT_KEY,
            "VersionId": TARGET_VERSION_ID
        }
    )
else:
    # 大于5GB使用服务端分片复制
    multipart_task = s3_client.create_multipart_upload(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY
    )
    upload_id = multipart_task["UploadId"]
    parts_list = []
    total_parts = (object_size + PART_SIZE - 1) // PART_SIZE

    for part_num in range(total_parts):
        start_byte = part_num * PART_SIZE
        end_byte = min(start_byte + PART_SIZE - 1, object_size - 1)
        part_resp = s3_client.upload_part_copy(
            Bucket=BUCKET_NAME,
            Key=OBJECT_KEY,
            PartNumber=part_num + 1,
            UploadId=upload_id,
            CopySource={
                "Bucket": BUCKET_NAME,
                "Key": OBJECT_KEY,
                "VersionId": TARGET_VERSION_ID
            },
            CopySourceRange=f"bytes={start_byte}-{end_byte}"
        )
        parts_list.append({
            "ETag": part_resp["CopyPartResult"]["ETag"],
            "PartNumber": part_num + 1
        })
    
    s3_client.complete_multipart_upload(
        Bucket=BUCKET_NAME,
        Key=OBJECT_KEY,
        UploadId=upload_id,
        MultipartUpload={"Parts": parts_list}
    )

内容的提问来源于stack exchange，提问作者landau