Repository navigation
Give multipart writes a single owner of their parts and an explicit state #1211
Description
Use case
Multipart writes have no single owner of their parts. S3File (also used by AioS3File) and AioS3StreamedFile each track the part futures, the parts and copies in flight, the failure checks, the cancellation and waiting before an abort, and the abort itself, with S3MultipartWriter owning only the upload identity and the requests. The write lifecycle has no explicit state either: it is inferred from buffer, closed, forced, multipart_upload, multipart_upload_parts, and _committing.
Consequences found during #1186 (PR #1192):
- Every fix to part handling is written twice, and the two files already differ:
S3File.discard()keeps the parts when the abort fails so that a laterdiscard()retries it, whileAioS3StreamedFile._discard()logs the failure. - Several defects come from implicit states. A retried
commit()after a failed multipart commit uploaded only the last part with PutObject (fixed in Free the data of failed S3 requests and clean up async copies at loop shutdown #1192 by closing the file), anddiscard()followed bycommit()still uploads the buffered data of the last part alone; after adiscard()whose abort failed, acommit()waits for the kept parts and may complete the discarded upload. - The futures of failed parts stay referenced from the files and from the frames that raise their errors, forming reference cycles that only the garbage collector frees. Free the data of failed S3 requests and clean up async copies at loop shutdown #1192 releases the request data (part bodies, read results) at the executor and request boundaries, but a per-call-site attempt to remove the cycles themselves needed an invariant that every caller had to keep by hand, and kept introducing new bugs.
Proposed change
A write session in pyathena/filesystem/s3_writer.py, composed with S3MultipartWriter, that exclusively owns one multipart write:
- Ownership: the creation request, the part uploads and part copies, bounded concurrency (
max_workersparts in flight), ordered part metadata, the first failure, and completion and abort. Request futures are consumed as they finish: a finished part becomes its part metadata or a recorded failure, and the future is dropped, so neither the files nor the frames that raise a failure hold part futures. - Explicit state: for example
BUFFERING,CREATING,ACTIVE,COMPLETING,COMPLETED,ABORTING,ABORTED, andABORT_FAILED(keeps the upload identity for a retried abort). After a failure or a discard, completing is impossible; later operations raise a clear state error instead of re-raising an earlier error or uploading the buffer. - API sketch:
start(),submit_part(body),submit_copy(range), a readiness handle that resolves when another part may be submitted (ready()),check()to raise a recorded failure,complete(), andabort(). The synchronous file blocks on the readiness handle; the asyncio file awaits it (asyncio.wrap_future), so the event loop keeps running and a one-thread pool never waits for itself. - Adapters:
S3FileandAioS3StreamedFilekeep only the buffering and the fsspec protocol (flush,commit,discard, transactions), and delegate part handling to the session.
Risks: callback re-entrancy and locking; Future.cancel() runs callbacks on the cancelling thread; the "tracked before scheduled" invariant that lets an interrupt still cancel or wait for a part; completion racing an abort; pool starvation; and the public S3File.multipart_upload_parts list, which would become derived from the session. Append (part copies of the existing object), exclusive create (IfNoneMatch), and transactions must keep their behavior.
Migration: the session and its unit tests first; then S3File/AioS3File; then AioS3StreamedFile, removing the duplicated tracking; each in its own PR with the behavior changes listed for the release notes.
Validation plan (if implementing)
Unit tests of the session's state transitions and of interrupt, cancellation, failure, and abort-failure paths without AWS; the existing S3File and AioS3StreamedFile tests, which must keep passing; tests that a failed part's body and the parts of a failed write are released while the error is held; and the filesystem integration tests (tests/pyathena/filesystem/) against the CI bucket.
Activity
laughingman7743 commentedon Oct 17, 2026
Design decisions (2026-10-17)
An audit found four owners of the multipart part lifecycle, not two, and they diverge in abort-failure handling, in-flight bounding, stopping after a failure, recovery of an interrupted creation, and the completion-versus-abort race:
S3File(alsoAioS3File):_initiate_upload,_upload_chunk,commit,discard,_close_without_commit.AioS3StreamedFile:_initiate_upload,_upload_chunk,_complete_upload,_discard,_close_without_commit.S3FileSystem._copy_object_with_multipart_upload, with_finish_multipart_uploadand_abort_multipart_upload.AioS3FileSystem._copy_object_with_multipart_upload: a semaphore, afailedflag, and_abort()in a filesystem thread.The draft design was reviewed with Codex before these decisions.
Scope and delivery
All four owners move to one session, one PR per sub-issue, in order:
S3FileandAioS3Fileas its first consumer, so that the hardest contracts (deferred commit after the executor shutdown, empty writes, fsspec rollback) shape the session before it is reused.AioS3StreamedFile.S3FileSystemmultipart copy.AioS3FileSystemmultipart copy.This issue is closed by the PR of the last sub-issue.
Contract
_S3WriteSessioninpyathena/filesystem/s3_writer.py, composed withS3MultipartWriter, whose public contract is unchanged. The session exclusively owns its writer. Making the session public is reconsidered after the migration.IDLE,CREATING,ACTIVE,COMPLETING,COMPLETED,FAILED,ABORTING,ABORTED,ABORT_FAILED.FAILEDmeans that completing is prohibited, not that an upload exists. Late request callbacks never move the session out ofABORTINGor back toACTIVE.IDLE→ABORTEDwithout a request;CREATINGcancels a queued creation or recovers a running one, then aborts the resulting upload;ACTIVE/FAILEDseal submissions, cancel queued requests, wait for running ones, and abort;COMPLETINGfollows the completion policy and never sends a concurrent abort;COMPLETEDandABORTEDdo nothing;ABORTINGjoins the running cleanup;ABORT_FAILEDretries against the retained upload and never reopens writing.begin_complete()entersCOMPLETINGbefore the blocking finalizer is queued (what_committingdoes today), and is serialized with part submission, so that a cancellation cannot abort a queued commit and a finalizer cannot overtake a part that is registered but not yet scheduled. The completion policy distinguishes a queued finalizer from a running CompleteMultipartUpload request: the streamed file protects a queued commit, while the asynchronous copy may still cancel a completion that has not started.S3File.close()has shut the executor down before a deferred transaction commit. Done callbacks never block.submit_to()(tracked before scheduled); a scheduling failure cancels the future if it can, and otherwise keeps it until it settles. Uploads and copies both count againstmax_workers;submit_*()checks the state and reserves capacity atomically, and readiness is only a hint. Each waiter gets its own readiness future, resolved outside the lock, and a cancelled waiter is dropped. A creation in progress keeps readiness pending. Cancelled futures that may never have been scheduled are never passed towait().iter_parts,iter_copy_parts) stays inS3MultipartWriter. An empty final flush at the part limit stays valid.exception()aftercancelled(),result()only on success) and drops it. The first failure is handed to the caller once: it is removed from the session under the lock and raised outside it. The session keeps a permanent failure marker and a summary without a traceback, and later operations raiseS3MultipartStateError(RuntimeError).S3AioThreadPool.wait_shielded()waits on still holds a failed completion.NoSuchUploadis treated narrowly as an upload that no longer exists. A failed or lost completion does not promise that the destination is unchanged.User-visible changes (4.2.0 release notes)
S3File.multipart_uploadbecomes read-only (the setter is removed).S3File.multipart_upload_parts(a list of futures) is removed; a read-only snapshot of the completed parts in part-number order (for examplecompleted_parts: tuple[S3MultipartUploadPart, ...]) replaces it.commit()afterdiscard(), or after a failed write or commit, raisesS3MultipartStateErrorinstead of doing nothing or uploading the buffered last part. Inside an fsspec transaction, a caller that catches a write error and continues makes the transaction completion fail, and fsspec then discards the remaining deferred files.discard()stays repeatable after a failed commit or abort.