Multipart copies drop the source metadata, and the async multipart copy does not abort on failure

[laughingman7743]

Problem

cp_file() copies objects larger than 5 GiB with a multipart upload (CreateMultipartUpload, UploadPartCopy, CompleteMultipartUpload). That path differs from the single CopyObject path in two ways.

1. The destination loses the source's content type and user metadata. CopyObject copies the source's system metadata (such as Content-Type) and user metadata (x-amz-meta-*) by default. A multipart upload starts with neither, and _copy_object_with_multipart_upload() creates it with only Bucket, Key and the caller's keyword arguments (pyathena/filesystem/s3.py:1256-1260; async pyathena/filesystem/s3_async.py:299-304). For a source with ContentType="text/csv" and Metadata={"owner": "etl"}, the request is {'Bucket': 'bucket', 'Key': 'dst.csv'}, so the copy gets S3's default content type and no metadata. Whether a copy keeps its metadata therefore depends on the object size.
2. The async multipart copy does not abort the upload when a part fails. AioS3FileSystem._copy_object_with_multipart_upload() runs the parts with asyncio.gather() and calls CompleteMultipartUpload directly (pyathena/filesystem/s3_async.py:321-330). When a part or the completion fails, the error propagates and AbortMultipartUpload is never sent, so the incomplete upload and its copied parts stay in the destination bucket until a lifecycle rule removes them. The sync version goes through _finish_multipart_upload(), which cancels the remaining parts and aborts the upload (pyathena/filesystem/s3.py:1338-1384).

Expected:

• a multipart copy keeps the same metadata as CopyObject: the source's content type, other system metadata and user metadata, unless the caller passes them;
• a failed async multipart copy aborts the upload, as the sync one does.

Related: #951 (fixed by PR #955) changed the part ranges of both _copy_object_with_multipart_upload() functions, and #967 (block_size/max_workers sent to S3) changes how their keyword arguments are routed. A fix here touches the same code.

Reproduction

from unittest.mock import MagicMock
from pyathena.filesystem.s3 import S3FileSystem
from pyathena.filesystem.s3_async import AioS3FileSystem

calls = []
def call(method, **request):
    name = method._extract_mock_name().split(".")[-1]
    calls.append((name, request))
    if name == "head_object":  # a 6 GiB source object
        return {"ContentLength": 6 * 2**30, "ETag": '"e"', "ContentType": "text/csv", "Metadata": {"owner": "etl"}}
    if name == "create_multipart_upload":
        return {"Bucket": request["Bucket"], "Key": request["Key"], "UploadId": "u"}
    if name == "upload_part_copy":
        if request["PartNumber"] == 2:
            raise OSError("part 2 failed")
        return {"CopyPartResult": {"ETag": '"p"'}}
    return {}

fs = S3FileSystem(connection=MagicMock(), skip_instance_cache=True)
fs._call = call
try:
    fs.cp_file("s3://bucket/src.csv", "s3://bucket/dst.csv")
except OSError as e:
    print(e)
# part 2 failed
print(next(c for c in calls if c[0] == "create_multipart_upload"))
# ('create_multipart_upload', {'Bucket': 'bucket', 'Key': 'dst.csv'}): no ContentType, no Metadata
print([c[0] for c in calls if c[0] != "head_object"])
# ['create_multipart_upload', 'upload_part_copy', 'upload_part_copy', 'abort_multipart_upload']

afs = AioS3FileSystem(connection=MagicMock(), skip_instance_cache=True)
afs._sync_fs._call = call
calls.clear()
try:
    afs.cp_file("s3://bucket/src.csv", "s3://bucket/dst.csv")
except OSError as e:
    print(e)
# part 2 failed
print([c[0] for c in calls if c[0] != "head_object"])
# ['create_multipart_upload', 'upload_part_copy', 'upload_part_copy']: no abort_multipart_upload

Environment

• PyAthena master e0e85da, fsspec 2026.9.0, botocore from uv.lock, Python 3.13.1.
• Found in a full audit of pyathena/filesystem/. Unless noted, checked offline with mocked S3 responses (fs._call replaced) or botocore's Stubber with dummy credentials.

Proposed fix (optional)

• In the multipart copy, read the source with HeadObject (with its VersionId, if any; a cached listing entry has no content type or metadata) and pass its content type, other system metadata and user metadata to CreateMultipartUpload, with the caller's keyword arguments taking precedence.
• Run the async part copies and the completion through the sync _finish_multipart_upload() (or an equivalent try/except) so a failure aborts the upload.
• Add tests that check the CreateMultipartUpload parameters and the abort on a failed part for both classes.

[laughingman7743]

A related gap in the same code, found during PR #1000 (#969), to cover with this fix: how the multipart copy handles the CopyObject parameters given to cp_file().

For an object larger than 5 GiB, _copy_object_with_multipart_upload() (sync and async) passes all of cp_file()'s keyword arguments to CreateMultipartUpload. Those arguments are CopyObject's parameters, and the botocore S3 model in uv.lock splits them across the multipart requests as follows:

• Source conditions (CopySourceIfMatch, CopySourceIfNoneMatch, CopySourceIfModifiedSince, CopySourceIfUnmodifiedSince), as well as CopySourceSSECustomer* and ExpectedSourceBucketOwner, are members of UploadPartCopy, not of CreateMultipartUpload. Today they fail CreateMultipartUpload's parameter validation; with this fix they should go to the part copies.
• MetadataDirective, TaggingDirective, AnnotationDirective are members of CopyObject only; no multipart request accepts them. Today they fail validation as well. They have to be implemented, not routed: COPY (CopyObject's default) means passing the source's metadata/tags to CreateMultipartUpload (Metadata, ContentType, …, Tagging), which is this issue's first item, and REPLACE means using the caller's values instead.

PR #1000 tried filtering CreateMultipartUpload's parameters by its input shape, so that the source conditions reach the part copies. It reverted that: with the filter, MetadataDirective="COPY" or TaggingDirective="COPY" was silently dropped, so the copy succeeded without the metadata or tags that were explicitly requested, where it used to fail validation. After PR #1000, the part copies, the completion and the abort already receive the parameters of the copy that they accept; only CreateMultipartUpload still receives all of them.

Suggested scope for the fix here: implement the directives (default COPY, as CopyObject does) when building CreateMultipartUpload's parameters, then pass it the subset it accepts; reject a directive value that cannot be honored rather than dropping it; and cover a source condition reaching the part copies, in sync and async.