[laughingman7743]

WHAT

Step 3 of #1332: one fan-out primitive for the parallel leaf requests of the S3 filesystems, as designed in the step 3 comment.

• S3Executor.map(fn, items, *, stop_on_failure=True, counts=None) and map_async() return an S3MapResult (outcomes in input order, and failure, the first counted failure in completion order). They are concrete methods of S3Executor, built on submit_to(), so any executor works, including one given to S3File(executor=) with only submit() and shutdown(). Every item is submitted at once; the executor applies its own limit.

  • A call checks the batch's stop before it starts, and the first failure that counts is recorded in the thread of the failed call, so that the call that takes its place makes no request. A call that the executor fails or cancels without running it also counts as a failure, recorded by a done callback when its future is set.
  • The waiting thread waits for one future. The done callbacks set it when every call has settled or when a failure stops the batch, and the states are rechecked every second.
  • An error of counts in a done callback becomes the failure instead of escaping the callback.
  • With stop_on_failure, the first counted failure cancels the queued calls, stops the retries of the running ones, and is raised once they have ended (B2).
  • An interrupt (sync) or a cancellation (asyncio) of the wait does the same whatever stop_on_failure is, setting the stop first, and is re-raised (B3). A second one stops the wait; the running calls finish in the executor (amended I1).
  • map_async drains in its own task, so the shutdown of asyncio.run(), which cancels the task, also waits for the running calls (B6). A loop closed without cancelling its tasks is a documented limit.
  • The futures and outcomes are dropped on every path. On an error, the frames below the map are cleared, including those of a further interrupt that stopped the drain (F5). The asyncio outcomes are retrieved, so asyncio logs no "never retrieved".

• submit_to() cancels the submission when the caller's future is cancelled, so that a scope drops a waiting call at once (S8); _S3WorkItem.link() owns both links.

• S3Executor.waiter() wraps a future for asyncio and retrieves its outcome (shared by wait_shielded(), the map, and the streamed drain). S3ClientError.code_of() returns the code of the ClientError an error was raised from.

• Sites moved to the primitive:

  • S3File._fetch_range and AioS3StreamedFile._fetch_range;
  • recursive S3FileSystem.chmod;
  • S3FileSystem._delete_objects (rm) and AioS3FileSystem._delete_objects (_rm), with stop_on_failure=False;
  • S3FileSystem.clear_multipart_uploads, with stop_on_failure=False and counts excluding NoSuchUpload;
  • the annotation copies of the asyncio multipart copy.

  The operations run through S3FileSystem._map() and AioS3FileSystem._map_async(), which own the one-shot scope and shut it down with wait=False, so that a second interrupt is not followed by another wait in __exit__.

Changes that callers observe:

Site	Change
S3File and AioS3StreamedFile parallel reads	a failed, interrupted, or cancelled read raises once its running reads have ended (they used to run on after it raised)
rm, clear_multipart_uploads	an interrupt cancels the requests that have not started (they used to be sent)
AioS3FileSystem._rm()	a cancellation cancels the queued batches and waits for the running ones (it used to return while they ran)
asyncio multipart copy	a cancellation during the annotation copies waits for the running ones

Docs: docs/filesystem.md (interrupts of parallel reads, rm, clear_multipart_uploads, recursive chmod) and docs/aio.md (cancellation of the fan-outs, S3Executor.map(), and the comparison table, whose asyncio.gather and per-call thread pool entries were obsolete).

WHY

#1332 step 3: the track-before-schedule, cancel-before-stop, and frame-clearing rules were written by hand at each site, with different failure semantics. rm and clear_multipart_uploads waited for every request on an interrupt, and the asyncio fan-outs returned on a cancellation while their requests ran (B6 gap). Step 4 (#1322 batch operations, #1344) builds on this primitive and its counts hook (B5).

Out of scope, per the design: the finalizer of an unclosed multipart file (#1364); the cooperative stop of writers in B2 (step 4); rm(batch_size=) and per-call task threads (step 4); _S3WriteSession and ReadAheadStream, which own their tracking.

TEST

Tested commit: 6b8f5f1.

• just lint: passed.

• Full local suite (uv run --with 'botocore[crt]' --env-file .env pytest -n 8 tests/pyathena, real AWS): 4089 passed, 15 skipped.

• Filesystem suite at the same head: 1479 passed and 6 skipped, with 4 fixture setup errors in TestS3FileSystemGet that pass on rerun (45 passed). The full run above passed them.

• New tests:

  • TestS3Executor:
    • test_map
    • test_map_executor_without_limit
    • test_map_executor_rejection[failed|cancelled]
    • test_map_failure_cancels_queued_calls
    • test_map_deferred_cancellation_stops_batch (and the asyncio variant)
    • test_map_records_failure_before_callbacks_run
    • test_map_interrupted_in_counts
    • test_map_interrupted_while_ending_wait
    • test_map_result_frees_outcomes
    • test_map_async_counts_error_in_callback
    • test_map_failure_stops_batch (with a delayed waiter, so the freed place is taken first)
    • test_map_raises_first_failure_to_end
    • test_map_counts_failures
    • test_map_interrupted_without_stop_on_failure
    • test_map_interrupted_twice
    • test_map_failure_frees_outcomes
    • test_map_interrupted_at_each_opcode
    • test_map_async_cancelled
    • test_map_async_failures_are_retrieved
    • test_map_async_at_loop_shutdown
    • test_submit_to_cancel_frees_waiting_call
  • test_s3.py:
    • test_interrupted_operation_cancels_queued_requests[rm|clear_multipart_uploads] (real SIGINT)
    • test_clear_multipart_uploads_raises_first_failure
    • test_failed_read_waits_for_running_reads
  • test_s3_async.py:
    • test_copy_object_with_multipart_upload_cancelled_annotation_copy
    • test_close_after_interrupted_cancel: the file-level test of the drain's one-second recheck, deferred from Share one request pool per filesystem among its files and operations (GH-1332) #1359.

• Updated tests:

  • test_rm_cancel_invalidates_cache_after_each_request became test_rm_cancel_waits_for_running_requests.
  • test_close_waits_for_cancelled_read now cancels the read twice, since one cancellation now drains.
  • test_cancelled_read_frees_ranges no longer depends on the old per-range tasks.
  • test_submit_to_cancelled_executor_failure starts the submission before cancelling.
  • The injection tests of the read cleanup target S3Executor.map.
  • Patches of pyathena.filesystem.s3.wait moved to s3_executor.wait. The interrupted-read tests make their reads wait until the wait is entered, since the wait is skipped when every call has ended.
  • test_interrupted_read_frees_ranges patches with a plain function, since a MagicMock keeps its arguments.

• Mutation check: each new test fails under its mutation. The mutations were:

  • no gate;
  • the last failure recorded instead of the first;
  • counts ignored;
  • no stop on interrupt;
  • no sync or asyncio drain;
  • a drain that swallows a second interrupt;
  • futures kept;
  • no retrieval of asyncio outcomes;
  • no cancellation chained from submit_to();
  • no one-second recheck in the streamed drain;
  • no recording of failures that the executor reports;
  • rm with stop_on_failure.

  The site tests fail on master's source. The exception is test_clear_multipart_uploads_raises_first_failure, which guards existing precedence and fails under the counts mutation.

Reviews:

• Self-review rounds 1 and 2.
• Codex independent review and three follow-ups (6, 5, 4, and 3 findings).
• One finding is deferred, outside the amended I1: an interrupt inside a done callback that a caller-supplied executor runs on the main thread while the map waits in another thread can keep that callback's future.

Not run locally: just test sqla and just test sqla-async. This change does not touch the dialect, and CI runs them after Ready.

🤖 Generated with Claude Code

[laughingman7743]

Self-review round 1: implementation behavior (Claude Code; /simplify with four read-only agents, then /code-review at high effort)

Base 2a60ecc, head 4f90bd4 (round start) → 5ff74f6 (repairs).

Covered: S3Executor.map/map_async, _S3Batch (gate, latch, wait, drain, release), submit_to and _S3WorkItem, the seven sites in s3.py and s3_async.py, the docs, and the new and updated tests against B1–B6, S8, F2, F5, T1 and the amended I1.

FINDINGS, repaired:

1. The executor's own failures were not counted (s3_executor.py, _S3Batch.wait; /code-review). A call that the executor failed or cancelled without running it settled its future through settle(), not _call. It was neither latched nor did it stop the batch, so a parallel read failed in b"".join with a TypeError, and chmod or clear_multipart_uploads ignored it. A cancelled outcome made result() raise CancelledError. Repair (5ff74f6): the waiting thread records the outcomes of the futures it sees end, and a cancelled outcome is a CancelledError value. Test: test_map_executor_rejection[failed|cancelled], which fails without the repair.
2. A second interrupt before _stop() left the batch running (/code-review). release() now sets the stop in its finally.
3. Simplification and reuse (6590afd):
   • S3FileSystem._map() and AioS3FileSystem._map_async() own the one-shot scope and its shutdown(wait=False), replacing five copies;
   • S3Executor.waiter() replaces three copies of wrap-and-retrieve, and the asyncio drain reuses the map's wrappers;
   • _S3WorkItem.link() owns both future links and drops the submission once done;
   • a single all-completed wait without stop_on_failure;
   • S3ClientError.code_of() for the InvalidRange and NoSuchUpload checks;
   • _map_async moved to AioS3FileSystem;
   • an empty asyncio delete returns early.

Deferred or rejected, with reasons:

• Removing counts in favor of a wrapping fn (raised by two agents): rejected. It is in the posted design and is the hook for B5 in step 4.
• outcomes typed list[T | BaseException], which needs the two casts: deferred to step 4, where the result shape of the batch operations is decided.
• A shared helper for the four "wait, recheck every second" loops: deferred. They differ in cancellation handling.
• Data that a second interrupt keeps in the frames of the drain (/code-review Retry processing with API call #3): within the amended I1 and documented in docs/filesystem.md as best effort.
• O(n·k) re-waits with counts and stop_on_failure=True: no caller yet; revisit in step 4.
• A single completion signal per batch, and a shared test stub for copy plans: optional; skipped.

Validation at 5ff74f6:

• just lint passed.
• The filesystem suite: 1473 passed, 6 skipped.
• Mutation check: 17 mutant/test pairs over 12 mutations, plus the new rejection test, all killed.

[laughingman7743]

Self-review round 2: claims, compatibility, operations (Claude Code)

Base 2a60ecc, head 5ff74f6 → 58ef05d (repairs to docs and a docstring).

Claims checked against the code and tests:

• PR body, S3MapResult.failure "first counted failure in completion order": true for failures of calls; the executor's own failures are recorded when the waiting thread sees them (added to the body).
• "The call that takes its place makes no request": test_map_failure_stops_batch holds the waiter until the freed thread has taken the next call; the gate mutant fails it.
• "asyncio.run() shutdown waits for the running calls": test_map_async_at_loop_shutdown. The limit for a loop closed without cancelling is documented in map_async's docstring.
• docs/filesystem.md, failed or interrupted reads "raised once they have ended", and rm/clear_multipart_uploads interrupts: test_failed_read_waits_for_running_reads and test_interrupted_operation_cancels_queued_requests (real SIGINT; it fails on master with both requests sent).
• docs/aio.md, cancellation waits for "the range reads of a file": corrected. It holds for open_async() files only. An AioS3File read runs in asyncio.to_thread, whose cancellation does not wait for the thread.
• docs/aio.md comparison table: corrected. The bulk delete no longer uses asyncio.gather, and since Share one request pool per filesystem among its files and operations (GH-1332) #1359 the sync filesystem runs its parallel requests in its request pool, not per-call thread pools.
• AioS3FileSystem docstring, "run parallel requests with asyncio.gather": corrected to per-object calls such as _mv() copies.

Existing callers:

• S3File(executor=...) with a custom S3Executor keeps working: map uses only submit() through submit_to() (test_map_executor_without_limit).
• The new public names are S3Executor.map, map_async, waiter, and S3MapResult; nothing is removed.
• submit_to() now cancels a cancelled caller's submission. Its callers (write sessions, the map, the asyncio copy) only gain the earlier release of queued arguments (S8).

AWS operator:

• An interrupt now stops the retries of a running DeleteObjects request; aborts keep unstoppable_retries() (B4).
• _delete_batch still invalidates the cache in its finally.
• No request count changes on success; a failed or interrupted read now holds the caller until the running GETs end, which is bounded by their stopped retries.

Evidence: the local filesystem suite only, on a WSL2 host with real S3. CI and the full just test pyathena/sqla/sqla-async runs are pending before Ready.

[laughingman7743]

Independent review (relayed): Codex CLI (codex exec -s read-only), session 01a1b220-30fe-7313-a250-e7670c88b3a8. Static review only: no edits, execution, or network access.

Snapshot: detached worktree at 58ef05d. Base 2a60ecc. The package contained the literal diff, the specification and design comments, CLAUDE.md, docs/contributing.md, and docs/testing.md "Choose what to test". It contained no PR text or prior findings.

Covered: map/map_async and _S3Batch (the gate and latch, executor failures, wait, drain, release, F5), submit_to and link, every sync and asyncio site, pool and scope compatibility, the docs, and the added and rewritten tests.

FINDINGS (6), all verified and repaired in 81a0c9c:

1. [P2] Executor-reported failures reached the latch too late (_S3Batch.wait). Later calls could start, and a later failure could win the latch. Repair: a done callback on each future (_settled) records the failure when the future is set, before the setting thread takes another call. test_map_executor_rejection now also asserts that the next item does not run; it fails when the callback is removed.
   • Follow-on found while validating: the callbacks formed a cycle between the batch and its futures, which kept outcomes until the cyclic GC ran (test_read_interrupted_at_each_opcode_frees_ranges failed deterministically). Repair: batch.clear() in a finally on every path. New test: test_map_result_frees_outcomes (GC disabled). It fails without the clear.
2. [P2] The completion-order tests raced, because events were set before the failure was latched. Repair: test_map_raises_first_failure_to_end, test_map_counts_failures, and test_clear_multipart_uploads_raises_first_failure now wait for the recording of the preceding failure (a pass-through patch of _S3Batch._record).
3. [P2] The rewritten asyncio rm test lost the per-request invalidation check. Repair: separate releases. The first object's cache is gone while the second request still runs and the task is not done.
4. [P2] The retrieval test could pass without the callback. Repair: the second call fails only after the first stopped the batch, that is, during the drain. The waiting loop no longer retrieves anything, and the GC and handler checks span three loop passes.
5. [P2] The sync map tests had no external bound. Repair: they run under the new call_bounded() helper (a daemon thread with a join timeout). The opcode-injection test still runs inline, since its tracer is per thread.
6. [P3] The bare-pool self-map limit was undocumented. Repair: stated in S3Executor.map() and docs/aio.md.

Validation at 81a0c9c:

• just lint passed.
• The filesystem suite: 1474 passed, 6 skipped.
• Mutation check: 18 mutant/test pairs, all killed.

The full tests/pyathena run at 58ef05d had 22 failures and 36 errors after the exported session credentials expired mid-run (overlapping runs). The erroring conformance classes pass on rerun (90 passed), and the full run will be repeated at the final head.

[laughingman7743]

Independent follow-up 1 (relayed): Codex CLI, session 01a1b253-771a-7380-9c91-5013f7848c6c. Static review of the repair range 58ef05d..81a0c9c at snapshot 81a0c9c.

FINDINGS (5), all verified and repaired in e0f0bfa:

1. [P1] An error of counts in a done callback escaped it and skipped the later callbacks. For a rejection reported by another thread, that included asyncio's, so map_async waited forever. Repair: _record catches the predicate's error and records it as the failure. Test: test_map_async_counts_error_in_callback, which raises a non-Exception BaseException, since asyncio re-raises SystemExit out of the loop. It fails when the guard is narrowed.
2. [P2] A future wakes its waiters before it runs its callbacks, so result() could see a failure that was not yet recorded. Repair: the waiting thread also records the futures it sees end, and result() records the finished ones before deciding.
   • Found while repairing: recording every future in result() called exception() on unfinished ones, which waits. A stopped batch would then have waited for its queued calls before cancelling them. Repair: only finished futures are recorded. New test: test_map_failure_cancels_queued_calls, whose executor never starts the held call; it fails without the done() check.
3. [P2] An interrupt entering the finally that cleared the batch kept the bodies in the traceback. Repair: result() clears the futures itself before returning, so the except handler (release) covers every point. test_map_result_frees_outcomes fails without the clear.
4. [P2] The rejection test could pass with waiter-side recording. Repair: its executor runs the other calls in the submitting thread, so the call after the rejection starts before the map waits. It fails when the callback is removed.
5. [P3] The opcode-injection test was unbounded. Repair: the whole interrupt_each_opcode runs under call_bounded (120 s).

Also: test_map_counts_failures now orders its failures so that the first to end is not the last item. The latch mutant used to survive because result() re-records in input order.

Validation at e0f0bfa:

• just lint passed.
• The filesystem suite: 1476 passed, 6 skipped.
• Mutation check: 20 mutant/test pairs plus the two new tests' mutants, all killed.

[laughingman7743]

Independent follow-up 2 (relayed): Codex CLI, session 01a1b27e-7f21-7722-946f-5b8018cd7d47. Static review of the repair range 81a0c9c..e0f0bfa at snapshot e0f0bfa.

FINDINGS (4), verified and repaired in 13860df:

1. [P1] A cancellation that the executor reported after the wait began latched the failure, but did not wake a FIRST_EXCEPTION wait. A call held until cancelled then kept the map waiting forever. Repair: the waiting thread waits for one future (_ended). The done callbacks set it when every call has settled (a count under the batch lock), or _latch sets it when a failure stops the batch. The states are also rechecked every second. This also removes the repeated waits on every pending future that the efficiency review noted. Tests: test_map_deferred_cancellation_stops_batch and test_map_async_deferred_cancellation_stops_batch. Both fail (bounded) when a stop does not end the wait.
2. [P2] A Ctrl-C in counts, evaluated in the waiting thread, was swallowed. Repair: only the done callback (_settled) catches and latches the predicate's error; _record lets it propagate. Test: test_map_interrupted_in_counts, which fails when _record catches.
3. [P2] The callback test depended on timing. Repair: the rejecting thread waits until the map has wrapped the future for asyncio. The test also asserts that the predicate's error does not escape into that thread, which is what the guard now protects, since the wait no longer depends on the skipped callbacks.
4. [P2] Coverage gaps, both added:
   • test_map_records_failure_before_callbacks_run. A Future whose callbacks are deferred is patched into the module, so result()'s own recording is the only path. It fails without it.
   • The map opcode-injection test now also asserts that outcomes are freed while the interrupt is held.

Found while validating: with the single wait, the reads of test_read_interrupted_twice_frees_ranges could end before the patched wait was entered. Once that was fixed, the test showed that a second interrupt in the drain's own wait keeps the futures in that wait's frame. Repair: release() also clears the frames of the propagating exception (sys.exception()). It fails without this. The interrupted-read tests now make their reads wait until the wait is entered.

Validation at 13860df:

• just lint passed.
• The filesystem suite: 1480 passed, 6 skipped.
• Mutation check: 24 mutant/test pairs plus the release-frames mutant, all killed.

[laughingman7743]

Independent follow-up 3 (relayed): Codex CLI, session 01a1b2b6-1d16-7df3-bd0e-3ded3c232e04. Static review of the repair range e0f0bfa..13860df at snapshot 13860df.

FINDINGS (3):

1. [P1] An interrupt inside _end(), in a callback that records a failure, left _ended pending, and the second _latch returned early. A call held until cancelled then kept the map waiting. Repaired in 6b8f5f1:

   • _latch sets the stop and ends the wait for every counted failure under stop_on_failure, not only the first;
   • the waits also end once a failure has stopped the batch.

   Test: test_map_interrupted_while_ending_wait, which injects at each opcode of _end under call_bounded. Without the repair it hangs until the bound.

2. [P2] An interrupt caught by _settled while a custom executor completes a call on the map's other thread. If the waiting thread releases before that callback returns, the callback's own executing frame, which clear_frames skips, keeps the future and its body. Deferred, outside the amended I1. It needs a caller-supplied executor that runs callbacks in the interrupted main thread while the map waits in another thread. PyAthena's executors run the callbacks of calls in pool threads, and signals are delivered only to the main thread. Under the amendment, data retention under such interleavings is best effort.

3. [P2] The deferred-cancellation tests used timers, and the callback test could check escaped before the rejecting thread had returned. Repaired:

   • the cancellations happen once the map waits (a patched wait and a watched S3Executor.waiter);
   • the callback test waits for the rejecting thread to finish before asserting.

Validation at 6b8f5f1:

• just lint passed.
• The filesystem suite: 1479 passed, 6 skipped. 4 fixture setup errors in TestS3FileSystemGet pass on rerun (45 passed).
• The new mutant was killed.

The independent review loop ends here. The initial review and three follow-ups found 6, 5, 4, and 3 findings. The last open item is the deferred interleaving above, which needs a caller-supplied executor and an interrupt inside its callback. The remaining validation is the full local AWS run at the final head and CI after Ready.