[laughingman7743]

WHAT

Step 4a of the #1332 design: the batch operations of the synchronous S3FileSystem, built on the fan-out primitive of #1367. This branch was rebuilt on master; the earlier _run_batch with its own refill loop is gone.

• cat, cat_ranges, get, put, pipe with a dict, copy, and mv run their per-object calls (cat_file, get_file, put_file, pipe_file, cp_file, _copy_file) in threads of a pool created for the call, with S3Executor.map(), never in the request pool (T1). A single item runs in the calling thread.
• batch_size is a new argument of these methods, of rm, and of S3FileSystem (keyword-only). None (default) and -1 mean max_workers, the size of the request pool (O1). Other non-positive values raise ValueError.
• Inside a batch of n concurrent calls, each object gets max_workers // n parallel requests (at least 1) through the existing max_workers keyword; a caller's max_workers is resolved (None/0 take the window) and capped at it (O2). A single object keeps the filesystem's window.
• on_error keeps fsspec's meaning (B5): with "omit"/"return" of cat, "return" of cat_ranges, and "ignore" of copy for a missing source, an Exception does not stop the batch; a BaseException raised in a call always does.
• A failure or an interrupt starts no further call, stops the retries of the running ones, waits for them, and raises (B2/B3, via map). mv deletes only after every copy succeeded (O3).
• Cooperative stop (B2): pyathena.util.raise_if_stopped() raises concurrent.futures.CancelledError when an enclosing stop_retries_on() scope is set. It is checked before the creation of a multipart upload, before a part is tracked and again when it starts (a part queued behind a full pool), after the wait for the parts right before CompleteMultipartUpload (inside the commit, so the write fails and is aborted), before single-request commits (PutObject of S3File and pipe_file, both CopyObject paths of cp_file), before each range read of S3File, and before get_file writes a block locally. The annotation copies of a committed multipart copy run under unstoppable_retries().
• Conflicting pairs raise ValueError before any transfer: two pairs of get/put/copy/pipe with the same destination after normalization, the same existing local file (e.g. hard links) or the same file-like object, and a copy destination that is another pair's source. S3PathPairing.check_copy_pairs() shares the checks of move_pairs(); a self-copy and a skipped prefix-only directory do not conflict. get(..., outfile=) runs one object at a time, in order.
• Progress callbacks of get/put and their branches are used through a lock (_S3LockedCallback), since fsspec callbacks need not be thread-safe nor their branches independent.
• rm(batch_size=) counts DeleteObjects requests; rm(max_workers=) is a deprecated alias (DeprecationWarning); both together raise TypeError (O4).
• docs/filesystem.md documents the batch operations.

Release notes (behavior changes):

• The listed operations transfer objects concurrently. After a failure, at most batch_size - 1 other calls are still running and stop before their next request; objects completed by other threads before the failure stay written, and their number is not bounded by batch_size.
• Pairs that the sequential loop resolved by order now raise ValueError (nonexistent local paths are compared as strings after os.path.normcase), e.g. copy("b/d/", "b/d/sub/", recursive=True) when b/d/sub/ already holds objects.
• Within a stopped stop_retries_on() scope, reads and writes of the S3 filesystem now raise CancelledError before their next request instead of continuing. Inside PyAthena these scopes are the step 3 primitive, the aio task pool, pyathena/aio/util.py's thread helpers, the Arrow reader, and the session's own abort.
• get accepts an os.PathLike destination for any source; copy no longer passes its copy parameters to expand_path(); get and put still pass theirs, as fsspec does. get(outfile=) runs in the calling thread.
• rm(max_workers=) warns.

WHY

Closes #1322 (part of #1315). Step 4a of #1332; 4b (AioS3FileSystem, #1337) follows separately.

In the s3fs comparison (#1318), S3FileSystem ran these operations 7 to 59 times slower than s3fs, because fsspec's loops call the per-object methods one after another. The design was reviewed by Codex (3 rounds) and Claude Fable 5.1 before implementation, and the maintainer decided -1 = pool size, CancelledError for the stop, and ValueError for conflicting pairs.

TEST

Tested commit: eaa1441, rebased onto master edfafcf (#1362, #1369, #1370, #1377). The suite runs below are on ed4f431, the rebased series before two test-only fixes (a stub signature and batch_size=1 in an order-dependent Stubber test), which were rerun alone.

• just lint: passed. rumdl on docs/filesystem.md: passed.
• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/ (local AWS) on ed4f431: 1587 passed, 1 failed (the stub fixed in fb082b3). Before the rebase, on 8e5beda: 1554 passed.
• Filesystem unit tests at fc1a168 (-m "not integration"): 417 passed.
• just benchmark test: 340 passed, 1 skipped.
• New tests: 4 in test_s3_writer.py and 55 in test_s3.py (concurrency per operation, batch_size resolution and validation, O2 windows, failure stop, real-SIGINT drain, on_error of cat/cat_ranges/copy, F5 of the cat assembly, O3 with a late success, conflicts, skipped prefix and in-place copies, expansion kwargs, outfile, callback serialization and contexts, rm alias, range-read and download stops, annotation retries after the stop).
• Mutation check: 32 mutants, all killed.
• Reviews: self-review rounds 1 and 2, Codex independent review (8 findings) and three follow-ups (3, 1, CLEAN), and a Codex review of the rebase (1 test finding, fixed; follow-up CLEAN); records inline.

Not run yet: AWS CI (starts when marked Ready); the in-region measurement of #1322 (step 5 of #1332).

🤖 Generated with Claude Code

[laughingman7743]

Paused. The design was decided in #1332: #1332 (comment)

Concurrent requests here multiply to batch_size * max_workers, because each object keeps its own per-file pool, and batch_size=-1 starts one thread per object. This PR will be rebuilt as step 4, after the filesystem-owned request pool (step 2) and the fan-out primitive (step 3). _run_batch and its tests are the starting point for step 3.

Review findings from the first /code-review round that carry over:

• get() no longer forwards **kwargs to expand_path().
• The one-at-a-time path counts failed calls as progress.
• S3FileSystem(batch_size=...) is ignored.
• chmod(recursive=True) repeats the fan-out logic.
• put() copies fsspec's path pairing and makes an extra lookup of the destination.

[laughingman7743]

Self-review round 1 (implementation behavior) — base 7fcbcc64b02f9a5efe16ec04815b4143d1a2c615 (merge-base with master), head 24d9872ff4026dc67e6d9bd63fed6080fe11b120.

Covered: _run_batch / _batch_limit / _batch_kwargs, each batch operation (cat, cat_ranges, get, put, pipe, copy, mv, rm), the stop checks in _S3WriteSession (start, _submit, _send_part, _put_object, _complete_parts), S3File._get_range, get_file, the annotation loop, S3PathPairing.check_copy_pairs, _S3LockedCallback, and the tests.

Checked: T1 (the task pool is a separate S3ThreadPool; scopes on the request pool see a non-pool thread), the stop check in _submit sits before the request is tracked (a raise after tracking would leave abort() waiting), the completion check is inside _commit after the wait (the write becomes FAILED and abort aborts), the session's own _stop is set only by _stop_requests() (abort), and every caller of stop_retries_on() in pyathena/ (aio thread helpers, Arrow reader, primitive, writer) — none encloses an S3 write except the aio task pool (#1337).

FINDINGS (2, tests only), found by the mutation check and fixed in 24d9872f:

1. test_get_file_stopped_writes_no_further_block read the same block forever, so removing the write check hung instead of failing; the stub now returns a finite sequence.
2. Making the single-item path of _run_batch swallow a BaseException survived: the interrupt test ran two items. It is now parametrized over 1 and 2 items.

After the fixes, all 23 mutants are killed.

[laughingman7743]

Self-review round 2 (claims, compatibility, operations) — base 7fcbcc64b02f9a5efe16ec04815b4143d1a2c615, head 24d9872ff4026dc67e6d9bd63fed6080fe11b120.

Claims checked against the code: cat_file and the small pipe_file send their request from the task thread (so batch_size > max_workers raises total concurrency — documented); mv deletes only after map returns; a committed multipart copy still copies its annotations under unstoppable_retries(); the conflict rules match S3PathPairing._check (self-copy kept, prefix-only skips excluded, same existing file by st_dev/st_ino, file-like by identity); rm(max_workers=) warns with stacklevel=2.

Existing callers: positional arguments of S3FileSystem.__init__ unchanged (batch_size is keyword-only after *args); AioS3FileSystem keeps its own batch_size and does not pass it to the internal filesystem; AioS3FileSystem._mv still uses _move_pairs, whose return is unchanged; no rm(max_workers=) call in pyathena/ or benchmarks/.

Operational: a stopped stop_retries_on() scope now also makes S3 reads and writes raise CancelledError before their next request (release note in the PR body). The design comment said retry_api_call and wait_with_retry_stop would reuse the new test; they keep their own snapshot of the events, which is equivalent, so that sentence is not implemented and is not claimed in the PR.

Evidence: local AWS filesystem suite 1549 passed (implementation of 24d9872f); AWS CI not run yet (Draft). CLEAN.

[laughingman7743]

Independent review (relayed): Codex CLI (codex exec, read-only sandbox), static review — base 7fcbcc64b02f9a5efe16ec04815b4143d1a2c615, head 24d9872ff4026dc67e6d9bd63fed6080fe11b120, session 01a1b4e0-b3e5-78d1-bcf7-3b3be249ff7f. Package: detached worktree at the head, the literal diff, the #1332 specification and step 4 design, repository conventions; no PR text or prior findings. No edits, builds, tests, or GitHub access.

Coverage reported: all eight batch operations, executor/pool/scope interactions, writer and read stop placement, aio consumers, conflicts, callbacks, window arithmetic, fsspec 2026.9.0 compatibility, docs, and the 20 added test functions. No T1/T2 deadlock; stop placement, on_error, O2, O3, O4 found correct.

FINDINGS (8), all verified and fixed in 8e5bedad:

1. Nonexistent local destinations differing only by case conflict on case-insensitive filesystems → compared with os.path.normcase(os.path.realpath(...)) (s3.py _check_local_destinations).
2. get/put no longer passed **kwargs to the expansion as fsspec does (e.g. prefix=) → _copy_pairs(..., **kwargs) and LocalFileSystem.expand_path(..., **kwargs); test test_get_put_pass_kwargs_to_expansion.
3. An interrupt between _run_batch() and the cleanup try of cat kept the outcomes → the batch runs inside the try.
4. copy(["b/dir", "b/x"], ["b/x", "b/y"]) with b/dir a bare prefix raised a conflict → conflict_candidates(copy=True) looks up every conflicting unversioned source for a copy; test test_copy_skipped_directory_does_not_conflict. mv keeps its rule (same input raises there too, before this PR; its contract tests assume it) — to be filed separately.
5. The callback proxy did not enter/exit the wrapped callback → __enter__/__exit__ delegate under the lock; the callback test now uses a branch that requires its context.
6. get(outfile=) still checked the ignored local paths and used a task thread → it skips the check and runs in the calling thread, in order; test asserts stream contents and thread.
7. Docs window formula → max(1, max_workers // n) with n = min(batch_size, objects); copy docstring too.
8. Docs stop guarantees → a request already sent (small PutObject) runs to its end; cache_type="background" prefetches may still read.

Test remarks (P3) taken: the interrupt test waits for the batch's stop instead of sleeping; the mv test synchronizes both copies so one succeeds after the failure; cat_ranges on_error and the range-read stop got tests. Deferred, with reasons: the completion-timer ordering in the writer test (the mutation that removes the post-wait check is killed); a bounded driver around the writer test; an end-to-end multipart put batch (covered by the writer and batch tests separately); rm request grouping (unchanged _delete_objects, covered by its existing tests).

Validation after the repair: just lint; new mutants (copy flag, callback enter, expansion kwargs ×2, outfile thread/check, cat cleanup, stop no-op) all killed; filesystem suite 1554 passed, 1 failed (test_copy_glob_to_new_directory, during a logged credential refresh failure), which passes on rerun with its class (45 passed).

[laughingman7743]

Independent follow-up 1 (relayed): Codex CLI, static, session 01a1b506-c887-7f62-9b1b-1374c12d9cf2, range 24d9872ff4026dc67e6d9bd63fed6080fe11b120..8e5bedad2f518e07c64e13d21e6f2ca905e81060 with the earlier findings. Repairs 2, 3, 4, 6, 7, 8 confirmed. FINDINGS (3), handled in fc1a1689:

1. os.path.normcase does not fold case on macOS, so nonexistent out/A and out/a still pass. Not fixed in code: folding case on POSIX would reject distinct files on case-sensitive filesystems. Documented in docs/filesystem.md (nonexistent local paths are compared as strings after normcase; give destinations that differ by more than case on such filesystems).
2. The proxy discarded the callback that a branch returns from __enter__() → it now wraps that callback; the test uses a branch whose __enter__() returns another callback and checks that each branch is exited. Mutants (returning self, dropping __exit__) killed.
3. The mv test could let the second copy finish before the failure was recorded → it now waits for the batch stop (raise_if_stopped) before succeeding; a mutant that deletes after a failure is killed.

Validation: just lint; filesystem unit tests (-m "not integration") 417 passed.

[laughingman7743]

Independent follow-up 2 (relayed): Codex CLI, static, session 01a1b50e-fe7b-7dc0-af19-b75f405cd238, range 8e5bedad2f518e07c64e13d21e6f2ca905e81060..fc1a16893f. The documented case limit and the __enter__() proxy were confirmed. FINDINGS (1, test only): if the stop stopped propagating, the late copy in the mv test would fail inside its worker and the batch would keep the first PermissionError, so the test could pass without the late success. Fixed in 8d4c1b78: the worker records its success after the wait and the test asserts it. just lint; the test passes.

[laughingman7743]

Independent follow-up 3 (relayed): Codex CLI, static, session 01a1b512-8bb4-7272-a5de-9a0d5ca01734, range fc1a16893f..8d4c1b78: the mv test now requires the late success marker. CLEAN.

[laughingman7743]

Rebase onto master — from base 7fcbcc64b02f9a5efe16ec04815b4143d1a2c615 (head 8d4c1b7898dd7cf145125f54cfaf6ae84746088a) onto edfafcf54a, which brought #1362 (source parameters for the lookups of copies and moves), #1369, #1370, and #1377 (authorization parameters with every request).

Conflict resolution: mv keeps the parallel copies and passes source_params to _move_pairs and the deletion; _conflict_lookups takes both source_params (upstream) and copy (this PR); copy now also sends the source's parameters to its conflict lookups, as mv does upstream. The stub of test_copy_skipped_directory_does_not_conflict accepts the new lookup_kwargs of _head_object (fb082b33).

Independent review of the rebase (relayed): Codex CLI, static, session 01a1b523-03cf-7632-a21b-d725a6a79bf1, input the git range-diff of the two series. Production resolution confirmed for both contracts. FINDINGS (1, test): upstream test_mv_source_lookup_and_delete_params answers requests in a fixed Stubber order, which parallel copies do not keep, so it could fail by scheduling. Fixed in eaa14413 with batch_size=1 in that test (the aio counterpart uses _mv, unchanged here).

Validation: filesystem suite on ed4f4311 (rebased): 1587 passed, 1 failed (the stub above, fixed); after the fixes the affected tests pass (skipped_directory, conflicts, batch: 43 passed; the mv source-parameter test three times).

[laughingman7743]

Independent follow-up of the rebase fix (relayed): Codex CLI, static, session 01a1b528-caa3-7cc0-9eef-adb65cdd526a, range fb082b33..eaa14413: batch_size=1 matches the Stubber order, and the stubbed expectations still check the source parameters, the filtered deletion, and the moved object. CLEAN.