setxattr() resets system metadata and rewrites the current object from a version path

[laughingman7743]

Problem

setxattr() rewrites the object with CopyObject and MetadataDirective="REPLACE", but the request carries only the user-defined metadata.

1. setxattr() resets the object's system metadata, storage class and encryption. pyathena/filesystem/s3.py:1653-1662 sends CopySource, Bucket, Key, Metadata and MetadataDirective="REPLACE", plus copy_kwargs. With REPLACE, S3 does not copy metadata that the request omits, and a CopyObject request without StorageClass or encryption parameters writes the copy as STANDARD with the bucket's default encryption.
   • For an object with ContentType: text/csv, ContentEncoding: gzip, CacheControl: max-age=60, StorageClass: STANDARD_IA and ServerSideEncryption: aws:kms with its own SSEKMSKeyId, the request sends none of these values. Adding one metadata key therefore changes how the object is served, billed and encrypted.
   • The HeadObject response that setxattr() already fetches through metadata() (s3.py:1642) contains these values.
   • S3File.setxattr() and AioS3FileSystem.setxattr() call the same method (s3.py:2468-2480, pyathena/filesystem/s3_async.py:437-449).
2. setxattr() on a version path rewrites the current object from that version. For bucket/key?versionId=OLD, s3.py:1650-1651 puts VersionId into CopySource and copies it onto bucket/key. The old version's content becomes the current object, and the content that was current becomes a noncurrent version. A call meant to add metadata silently rolls the object back. S3 cannot change the metadata of an existing version in place.

Expected:

• setxattr() keeps the object's system metadata, storage class and server-side encryption, and changes only the user-defined metadata;
• setxattr() rejects a version path with ValueError instead of copying an old version over the current object.

The stale cache entry left after setxattr() on a version path is handled separately by PR #960.

Reproduction

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

calls = []
def call(method, **request):
    name = method._extract_mock_name().split(".")[-1]
    calls.append((name, request))
    if name == "head_object":
        return {"ContentLength": 10, "ContentType": "text/csv", "ContentEncoding": "gzip",
                "CacheControl": "max-age=60", "StorageClass": "STANDARD_IA",
                "ServerSideEncryption": "aws:kms", "SSEKMSKeyId": "arn:aws:kms:us-west-2:111122223333:key/k",
                "Metadata": {"a": "1"}}
    return {}

fs = S3FileSystem(connection=MagicMock(), skip_instance_cache=True)
fs._call = call
fs.setxattr("s3://bucket/key.csv", b="2")
print(calls[-1])
# ('copy_object', {'CopySource': {'Bucket': 'bucket', 'Key': 'key.csv'}, 'Bucket': 'bucket', 'Key': 'key.csv', 'Metadata': {'a': '1', 'b': '2'}, 'MetadataDirective': 'REPLACE'})
fs.setxattr("s3://bucket/key.csv?versionId=OLD", b="2")
print(calls[-1])
# ('copy_object', {'CopySource': {'Bucket': 'bucket', 'Key': 'key.csv', 'VersionId': 'OLD'}, 'Bucket': 'bucket', 'Key': 'key.csv', 'Metadata': {'a': '1', 'b': '2'}, 'MetadataDirective': 'REPLACE'})

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 setxattr(), copy ContentType, ContentEncoding, ContentDisposition, ContentLanguage, CacheControl, Expires, StorageClass, ServerSideEncryption, SSEKMSKeyId and BucketKeyEnabled from the HeadObject response into the CopyObject request when present, and let copy_kwargs override them.
• Raise ValueError when the path has a version ID, as cp_file() already does for a versioned destination.
• Update the setxattr() docstring and docs/filesystem.md, and add offline tests that check the CopyObject request.