Finish the public S3Core API extraction from filesystem adapters

[laughingman7743]

Use case

Complete the remaining S3 API extraction in #1053.
Steps 1–4 are merged: paths (#1052), typed listing/HEAD operations (#1060), delete/multipart/copy/pairing (#1064, #1069, #1072, #1074, #1081), and the multipart writer (#1085).
The native sub-issues #1049, #1059, #1063 and #1077 are closed.

On master 855d4a7b652d6dad106c27ee7112264ca45b923e, several operations still build S3 requests and apply S3 rules inside S3FileSystem.
They use S3Core.call() for transport, retries and error translation, but a direct core caller has no typed operation taking S3Path for these requests.
Moving only GET/PUT, tags and ACLs would leave other request construction in the adapter and would not finish the separation proposed in #1053.

The remaining inventory in pyathena/filesystem/s3.py is:

Area	Current owner on the recorded master
Object GET/PUT	_get_object() (2843), _put_object() (2892): request construction, ranges, body handling and PUT result conversion
Tags and ACLs	get_tags() (2313), put_tags() (2336), chmod() (2377): request construction, version selection, tag merge and ACL validation
Metadata updates	setxattr() (2227): metadata changes, retained system/storage/encryption fields, and the self-copy request; reads already use core.head_object()
Multipart upload listing	list_multipart_uploads() (2429): requests, pagination and conversion to upload objects
Presigned URLs	sign() (2147): S3 request parameters and SDK signing
Bucket lifecycle	mkdir() (1240), rmdir() (1336): create/delete requests and region/ACL rules
Multipart size planning	_check_multipart_upload_size() (1744): S3 part-count validation also reached by the aio transaction helpers

Proposed change

Add the remaining public, synchronous operations and pure S3 planning rules to S3Core, S3MultipartWriter, or an appropriate fsspec-independent type.
Object operations accept S3Path; bucket operations use a clearly documented bucket-only argument.
Results expose documented types, including the existing S3PutObject, S3Metadata and S3MultipartUpload where appropriate.
Settle the GET result and ownership contract before implementation: whether a result exposes a response body or bytes determines who reads and closes it, and how read failures are handled.

Move request construction, operation-specific parameter rules, version handling, response conversion, and S3 validation out of the adapters for the inventory above.
Extract page-level multipart upload listing separately from iteration, following the existing core listing API.
The core uses S3 prefix semantics; the adapter retains its documented key-or-descendant filter, which excludes sibling keys sharing a prefix.
Make metadata replacement and tag merging use core operations or pure planners while preserving their existing requests and results.
Centralize the multipart size rule for sync and aio callers without changing the existing error conditions.

The adapters keep path parsing and fsspec behavior: directory synthesis/expansion, recursive traversal, chmod(recursive=True), callbacks, cache invalidation, buffering, transactions, executor scheduling and async bridging.
Bucket lifecycle opt-ins (allow_bucket_creation and allow_bucket_deletion) remain enforced by the filesystem before it invokes a core primitive.
The public core bucket methods document that a direct call creates or deletes a bucket; an adapter option is not a core permission mechanism.

Preserve the existing public filesystem signatures, return shapes, errors, request order/count, per-call parameter precedence, checksum behavior and cleanup.
Preserve open-ended and suffix reads, empty-range handling, empty PUT bodies, tag overwrite/merge modes, version-qualified paths, metadata retention/override rules, ACL validation before recursive mutation, bucket region handling and multipart listing markers.
The core stays synchronous and imports no fsspec or aiobotocore; scheduling stays with the adapters as decided on #1053.
New public names and result contracts should be agreed in this issue before implementation.
The implementation can use several independently reviewable PRs, each keeping the adapters functional.

Completion criteria

• Cover every inventory item with a public core operation or pure S3 planner, document its arguments/results/errors, and make the sync/aio adapters delegate to it.
• Audit the remaining SDK calls and S3 request builders in both adapters and S3File. Record each remaining site's reason; SDK client compatibility, raw-call compatibility shims, fsspec translation and orchestration remain adapter responsibilities.
• Keep the existing DirCache storage, options, invalidation and sync/aio sharing. Record the disposition of step 5: centralize cache keys from S3Path only if this extraction needs it; otherwise explicitly record that the conditional step was not needed, as decided in Separate an S3 core from the fsspec adapter in pyathena.filesystem #1053.
• Preserve the public adapter API and the internal users in pyathena/s3fs/, pyathena/pandas/, pyathena/aio/s3fs/ and Polars/fsspec registration.
• Complete core regression tests, affected adapter/runtime validation, public API documentation, both self-review rounds, independent review and applicable current CI for each implementation PR.
• After all implementation PRs merge, re-audit the inventory and the live sub-issue hierarchy of Separate an S3 core from the fsspec adapter in pyathena.filesystem #1053. Record the completed scope and cache decision, then close this issue and Separate an S3 core from the fsspec adapter in pyathena.filesystem #1053 if no work in the agreed scope remains.

The separate mv() behavior defect #1083 keeps its own issue and implementation.
This extraction preserves existing behavior; new S3 features and behavior changes require separate agreement, as specified by #1053.

Validation plan (if implementing)

Use botocore Stubber and self-contained tests for the new operations, typed results and planners.
Cover inherited/per-call parameter precedence, version IDs including null, request bodies/ranges, body ownership and read failures, metadata/encryption retention, tag modes, ACL validation, bucket regions, multipart pagination and size limits, plus translated failures.
Pin expected requests independently of the moved implementation.

Run just format and just lint, then the affected tests in tests/pyathena/filesystem/test_s3_core.py, test_s3_writer.py, test_s3.py and test_s3_async.py.
Check the existing filesystem request sequences, cache invalidation, transactions, conditional creation, multipart checksums and interruption/cancellation cleanup.
Exercise internal consumers that use these read/write operations and fsspec registration.
Run documentation lint/build for API changes and the applicable full PyAthena AWS CI before declaring an implementation PR ready.

Use the repository's existing AWS test environment and fixtures for live S3 validation.
Serialize live runs with other Test workflows/local tests; do not provision new persistent infrastructure for this extraction.
Bucket creation/deletion coverage must use the explicit opt-ins and disposable test resources.
Record exact tested commits, commands, results and skipped coverage, separating static, offline and live AWS evidence.
This proposal is based on source and GitHub-state inspection; no new tests or AWS operations were run to prepare it.

[laughingman7743]

Agreed design

The maintainer agreed to these names and contracts on 2026-10-04, after reviews of a draft by claude-fable-5-1 and Codex.
All operations are public and synchronous, are sent through S3Core.call(), and preserve the adapters' current requests, unless a point below says otherwise.
Object operations take an S3Path, and bucket operations take a bucket name.

1. Object GET and PUT

• S3Core.get_object(path, range_=None, **params) -> bytes
  • range_ is (start, end) with an exclusive end, or (start, None) to read to the end.
    A negative start with no end reads the last -start bytes.
    None sends no Range of its own, so a Range in params still applies.
  • An empty range raises ValueError, because S3 would return the whole object.
    A negative start with an end also raises ValueError; no adapter path reaches it.
  • The core reads the whole body and closes it, also when the read fails.
    A read failure of the body propagates as the untranslated botocore exception, and the GET is not retried, as today.
    A start past the end raises the translated OSError, whose __cause__ is the ClientError with code InvalidRange.
  • The adapter keeps the InvalidRange to b"" mapping of cat_file(), the range resolution based on info(), and the parallel range splitting in S3File._fetch_range().
    Failures are still observed in completion order, and the parts are joined in range order.
    S3FileSystem._get_object(), S3File._format_ranges() and S3File._merge_objects() are removed.
• S3Core.put_object(path, body=None, **params) -> S3PutObject
  • An empty or None body sends no Body of its own, so a Body in params still applies.
    The request fields take precedence over params.
  • ValueError if the path has no key or has a version ID.
    The adapters keep their own checks and messages first.
  • S3FileSystem._put_object() is removed.

2. Tags and ACLs

• S3Core.get_object_tagging(path, **params) -> dict[str, str] and S3Core.put_object_tagging(path, tags, **params) -> None.
  Both accept a version ID, including null.
  put_tags(mode="m") stays in the adapter: two requests, the core get, then the core put of the merged tags.
  plan_multipart_copy() reuses get_object_tagging().
• S3Core.put_object_acl(path, acl, **params) -> None and S3Core.put_bucket_acl(bucket, acl, **params) -> None take a canned ACL.
  They raise ValueError if the ACL is not in S3Core.OBJECT_ACLS or S3Core.BUCKET_ACLS, respectively.
  chmod() validates against the same sets before any recursive change.
• Breaking (4.0.0 release note): S3FileSystem.OBJECT_ACLS and S3FileSystem.BUCKET_ACLS are removed, as the MULTIPART_UPLOAD_* limits were.

3. Metadata replacement

• S3Core.replace_object_metadata(path, head, metadata, **params) -> None sends one CopyObject request.
  The request copies the object onto itself with MetadataDirective="REPLACE" and Metadata=metadata.
  It retains from head the content headers, Expires, WebsiteRedirectLocation and StorageClass.
  Unless params set an encryption parameter, it also retains ServerSideEncryption, SSEKMSKeyId and BucketKeyEnabled.
  params take precedence over the retained fields.
  A field that the operation itself sets, such as Metadata, still raises TypeError when it is given in params.
  head must be the HeadObject result of path.
• setxattr() keeps its validation before the HEAD, the HEAD through metadata(), the None deletions, and the cache invalidation.

4. Multipart upload listing

• S3ListMultipartUploadsPage(bucket, uploads, is_truncated, next_key_marker, next_upload_id_marker) is a frozen dataclass with from_response().
  Its uploads field is a tuple[S3MultipartUpload, ...].
• S3Core.list_multipart_uploads_page(bucket, prefix=None, key_marker=None, upload_id_marker=None, **params) lists one page.
  The S3Core.list_multipart_uploads(...) iterator yields the pages.
  • A None prefix sends no Prefix, as today.
  • The iterator stops when a page is not truncated or lacks either next marker, as today.
    It raises TypeError for KeyMarker and UploadIdMarker in params.
• S3FileSystem.list_multipart_uploads() keeps the filter that selects only the key and the keys under it.

5. Presigned URLs

• S3Core.generate_presigned_url(path, client_method="get_object", expires_in=3600, **params) -> str.
  params take precedence over the bucket, key and version ID of the path, as today.
  request_kwargs are not added, because the method is not an API operation.
  A path without a key is not rejected; botocore validates the parameters.

6. Bucket lifecycle

• S3Core.create_bucket(bucket, acl=None, region_name=None, **params) -> None.
  • The ACL is validated against BUCKET_ACLS.
  • The region defaults to the client's region.
  • No LocationConstraint is sent for us-east-1.
• S3Core.delete_bucket(bucket, **params) -> None.
• A direct call to either method creates or deletes the bucket.
  allow_bucket_creation and allow_bucket_deletion are filesystem options, not core permissions.
• mkdir() and rmdir() keep the path checks, the exists() checks, the opt-ins, the ParamValidationError to ValueError translation, and the cache eviction.

7. Multipart size check

• S3Core.check_multipart_upload_size(size, block_size) -> None raises ValueError when size > block_size * MULTIPART_UPLOAD_MAX_PARTS.
  The error condition is unchanged.
  The message is generic: it gives the minimum block size, but not the path or the filesystem option.
  The sync adapter and the aio transaction helpers call it directly.
  S3FileSystem._check_multipart_upload_size() is removed.

Kept as is

• S3FileSystem._call() stays a compatibility delegate.
• The cache stays in DirCache.
• The adapters keep scheduling, callbacks, transactions, invalidation and fsspec behavior.

Pull requests

1. GET and PUT (section 1).
2. Tags, ACLs, metadata replacement and presigned URLs (sections 2, 3 and 5).
3. Bucket lifecycle, multipart upload listing and the size check (sections 4, 6 and 7).

After the merges, I will audit the remaining SDK call sites and record the disposition of step 5.
I will then close this issue and #1053 if no work in the agreed scope remains.
#1083 stays separate.