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

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.