Implement exclusive create for open("xb"), put_file() and pipe_file()

[laughingman7743]

WHAT

Implement fsspec's exclusive create for S3FileSystem and AioS3FileSystem: open(path, "xb"), put_file(..., mode="create") and pipe_file(..., mode="create") raise FileExistsError for an existing object and never replace it.

• open(path, "xb") (S3File, also AioS3File): checks exists() when the file is opened, before any data is uploaded, and raises FileExistsError(path). The upload is then committed with IfNoneMatch="*", so an object created after the file was opened is not replaced either. The parameter is added to the file's parameters, so the per-operation filtering from Route S3 request parameters to the operations that accept them #1000 sends it only to the operations that accept it: PutObject (including the empty-object PutObject from touch() in commit()) and CompleteMultipartUpload, not CreateMultipartUpload or UploadPart. A failed completion aborts the multipart upload through the existing path.
• put_file(lpath, rpath, callback, mode="overwrite", **kwargs) (sync; async _put_file() and its in-transaction path): mode is now a parameter. "create" opens the remote file with "xb"; mode never reaches S3.
• pipe_file(mode="create"): on the single-request path, the existing exists() check is kept and the PutObject also sends IfNoneMatch="*". The buffered path (large data, or inside a transaction) and the async in-transaction path open the file with "xb" instead of checking exists() and opening with "wb".
• Errors: S3ClientError maps PreconditionFailed with Condition: If-None-Match to FileExistsError. Other PreconditionFailed errors (e.g. the If-Match of a read) keep the OSError(EINVAL) mapping.
• Docs: the error translation table in docs/filesystem.md lists the new mapping.

Behavior notes:

• When an object is created after the file is opened, FileExistsError is raised from close()/commit(), after the data is uploaded. Inside an fsspec transaction, it is raised when the transaction commits.
• The up-front check uses exists(), as pipe_file(mode="create") already did. It may be answered from the listing cache; otherwise, for a missing key, it costs a HeadObject and a ListObjectsV2 per open. A prefix of the same name (a "directory") also counts as existing.
• S3 documents a 409 ConditionalRequestConflict for conflicting concurrent conditional writes to the same key. It is not translated (it stays OSError(EBUSY) through the 409 fallback); this case was not measured.
• A principal that HeadObject denies (403), e.g. one that may only call PutObject, gets the PermissionError of the check and cannot create the object exclusively. pipe_file(mode="create") already behaved this way; kept by the maintainer's choice so that permission errors are not hidden.

Release-note items:

• open(path, "xb") raises FileExistsError for an existing object; it used to replace it.
• put_file()/put() accept fsspec's mode="create" and raise FileExistsError for an existing object.
• pipe_file(mode="create") also does not replace an object created after its existence check.

WHY

Closes #972.

The maintainer chose to check up front and also write conditionally: the check fails before a large upload, and IfNoneMatch makes the write atomic. Both are applied to open("xb"), put_file() and pipe_file().

On current master, put_file(mode="overwrite") no longer raises ParamValidationError: since #1000, the file's parameters are filtered by each operation's input shape, so mode was already dropped. The issue was observed at e0e85da, before #1000. mode="create" still did not check for an existing object; that is fixed here.

Measured on S3 with IfNoneMatch="*" on an existing key:

• PutObject and CompleteMultipartUpload both fail with HTTP 412, {"Code": "PreconditionFailed", "Condition": "If-None-Match"}. A failed IfMatch read returns "Condition": "If-Match", so the two can be told apart.
• After the failed completion, the multipart upload can still be aborted, and the existing object is unchanged.

Overlap: #1003 changes the same two pipe_file buffered-path lines (super().pipe_file(...) and the async with self.open(path, "wb")). Whichever merges second needs a small rebase: open with "xb" if mode == "create" else "wb" through _write_and_close(), and drop the exists() check that "xb" now does.

TEST

Tested commit: bf5743b (on 28826be). 08bcff5 adds only the docs table row (just docs lint).

• just format, just lint, just docs lint: pass.
• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/ against AWS: 453 passed.
• New and updated tests:
  • TestS3ClientError::test_os_error_precondition_failed: If-None-Match → FileExistsError; If-Match and no condition → OSError(EINVAL).
  • TestS3FileSystem::test_open_exclusive_create (empty, single PutObject, multipart): IfNoneMatch="*" only on PutObject/CompleteMultipartUpload.
  • test_open_exclusive_create_existing, test_put_file_create_existing: FileExistsError with no S3 request.
  • test_open_exclusive_create_created_since (single, multipart): a 412 at commit raises FileExistsError, and the multipart upload is aborted.
  • test_put_file_mode (sync and async): mode is not sent; "create" opens "xb" and sends IfNoneMatch.
  • test_pipe_file_create_created_since (single-request and buffered paths).
  • TestAioS3FileSystem::test_transaction_pipe_put_file_create_existing, test_put_file_in_transaction_open_parameters (parametrized by mode).
  • Live test_exclusive_create: "xb" and put_file(mode="create") on an existing object, and an object created between open("xb") and close() is kept.
• With the source changes reverted, 15 of the new and updated offline tests fail. The overwrite and non-If-None-Match cases pass both ways, as expected.
• The offline tests run with --noconftest and dummy credentials, using a real botocore service model for the parameter filtering.
• test_put_file_mode in test_s3_async.py builds its filesystem with a mock connection. AioS3FileSystem does not pass skip_instance_cache to its internal S3FileSystem, so instances built with the same arguments share it, and test_transaction_pipe_put_file replaces _put_object on the shared instance.

🤖 Generated with Claude Code

[laughingman7743]

Self-review round 1 (behavior and implementation) — FINDINGS (1, resolved without code change)

Base 28826be7d37983cb54050e849b32a4fe2331ea7d, head bf5743b6d3fca17a1d37ce716cad6312a858044c.

Covered: S3File.__init__ x-mode check and IfNoneMatch routing through _get_request_kwargs to PutObject / CompleteMultipartUpload / touch() (not CreateMultipartUpload / UploadPart; checked against the botocore input shapes); commit() failure paths (412 on PutObject; 412 on completion → _finish_multipart_upload abort, multipart_upload reset); pipe_file() single-request and buffered paths; put_file() sync, async, and async in-transaction; S3ClientError translation (If-None-Match vs If-Match vs no condition); fsspec Transaction.complete() on a failed commit (re-queues and discards the rest, clears _intrans); fsspec put()/_put() forwarding mode through **kwargs; executor on a FileExistsError at open (created lazily, as on the read path's FileNotFoundError); botocore floor 1.41.2 has IfNoneMatch on both operations.

Finding: the up-front exists() propagates PermissionError for a key that HeadObject denies (pyathena/filesystem/s3.py:917-923 catches only FileNotFoundError), so a PutObject-only principal cannot use xb/create. Pre-existing for pipe_file(mode=\"create\"). Maintainer chose to keep it (permission errors are not hidden); recorded in the PR body.

Tests: the regression tests fail on the original code (15 cases), and the live test exercises the real 412 path. The async test_put_file_mode avoids the shared internal S3FileSystem instance (pre-existing test pollution, noted in the PR body).

[laughingman7743]

Self-review round 2 (claims, callers, operations) — FINDINGS (2, repaired in 08bcff5 and the PR body)

Base 28826be7d37983cb54050e849b32a4fe2331ea7d, head bf5743b6d3fca17a1d37ce716cad6312a858044c (repair head 08bcff53).

Claims checked:

• 412 shapes for PutObject/CompleteMultipartUpload/If-Match and abort-after-failed-completion: measured on S3 before implementation; live test_exclusive_create reproduces the commit-time 412.
• mode already dropped on master since Route S3 request parameters to the operations that accept them #1000: test_put_file_mode[overwrite] (sync and async) passes with the source changes reverted.
• 15 failing tests without the fix, 453 passing live: match the recorded runs on bf5743b's tree.
• Write pipe_file() data without committing a failed write #1003 overlap: its diff changes super().pipe_file(...) and the async with self.open(path, \"wb\"); the described resolution matches.
• fsspec 2026.9.0 (uv.lock): _Cached.__call__ pops skip_instance_cache (spec.py:108), confirming the test-isolation note; Transaction.complete() re-raises a failed commit and discards the rest.
• Callers: put_file's new mode sits where fsspec's base has it (after callback); no positional argument beyond callback was possible before. fsspec put()/_put() pass mode by keyword.
• Retries: RetryConfig retries throttling codes only, so a 412 is not retried.

Findings and repairs:

1. docs/filesystem.md error-translation table omitted the new FileExistsError mapping — row added (08bcff5), just docs lint passes.
2. The PR body did not state the operational cost or edge cases of the up-front check — added: HeadObject + ListObjectsV2 per open of a missing key (info() at pyathena/filesystem/s3.py:704-725), a same-name prefix counts as existing, and S3's documented 409 ConditionalRequestConflict stays OSError(EBUSY) (unmeasured).

[laughingman7743]

Independent review (relayed) — CLEAN

Reviewer: Codex CLI 0.160.0 (codex exec, model gpt-6-astra, reasoning effort high, ChatGPT login), session 01a10176-8a61-79b2-86a3-bca5b26d7380. Read-only sandbox in a detached snapshot; no PR number, description, commit messages or prior findings given. Static review only: no tests, builds or network.

Base 28826be7d37983cb54050e849b32a4fe2331ea7d, head 08bcff533b171628a138e7d1ac86ec3216bc36f6. Snapshot and PR worktree verified unchanged afterwards.

Covered (as reported): S3File/AioS3File exclusive create, existing objects and creation races; sync/async put_file/pipe_file inside and outside transactions; empty-object touch(), single PutObject and multipart completion; multipart abort, pending-part cleanup, executor shutdown and transaction failure handling; mode consumption and per-operation IfNoneMatch routing against botocore 1.43.102; error translation, caller compatibility, docs, and tests.

Result: "no actionable defect introduced by the exact diff was found."

Reported coverage limits (not defects): no test for a conflict on the empty-file touch() commit, on a deferred (transaction) commit, or for executor shutdown right after a conditional-write failure. Deferred: these reuse the paths that are tested (touch() → _put_object, commit() from fsspec's Transaction.complete(), S3File.close()'s finally), so no change in this PR.