[laughingman7743]

WHAT

Implements the #1323 design (option A, Q4 (i)).

• get_file() reads waves of parallel ranges. The new S3File._download() splits the object into ranges of the block size, also for max_workers=1. It reads one wave at a time with S3Executor.map() on the file's scope, and writes the bodies of the wave in order, dropping each once written, before it reads the next wave.
  • A wave has max_workers ranges, capped at the size of the request pool, since no more run at once.
  • The download holds at most one wave of bodies: window × block_size, 50 MiB with the defaults. A thread of the request pool that has just read a range keeps its body until its call returns (_S3WorkItem.run()), which no reordering can avoid in CPython; the bound is stated for what the download holds.
  • A block size that is not positive reads the object in one range, as read() reads the rest.
  • Each range is read by S3File._get_range(), which checks the stop scopes when it starts (Run the batch operations of S3FileSystem concurrently (GH-1322) #1344); a body read when a stop came is not written.
• No data cache. get_file() drops the caller's cache_options and cache_type, and opens the file with cache_type="none". open() constructed the cache before, and cache_type="all" read the whole object. The file is still opened through open(), so the argument handling, the version and IfMatch of S3FileTarget.resolve(), and the request parameters are those of open(), unchanged. This takes the place of the shared helper that the design proposed: no argument handling is duplicated, and no data cache is constructed.
• Failures while writing. A failure while a wave is written, including a failure of the write or of the progress callback, and an interrupt, drops the bodies of the wave. The new S3MapResult.release(error) clears the outcomes and the completed frames of the traceback of the error and of its chained errors (__cause__, __context__). A file-like that wraps its own failure therefore keeps no body either (F5). The try block also covers the assignment of each wave, for an interrupt at the handoff.
• compression. The codec is resolved as fsspec's open() does, with "infer" from the key without the version. With a codec, the object is decompressed into the destination, read in order through a "bytes" block cache (the caller's cache type does not apply here either), and the callback gets an unknown size. A block size that is not positive takes the filesystem's, or DEFAULT_BLOCK_SIZE if that is not positive either. Before, the decompressor, which has no size or blocksize, made get_file(..., compression=...) raise AttributeError after truncating the destination. "infer" without a codec reads in parallel ranges.
• AioS3FileSystem._get() pairs its paths with S3FileSystem._copy_pairs(), as S3FileSystem.get() does, instead of only for a version path, and checks the destinations named after the sources (check_contained). The str and path-like entries of a destination list are made absolute, as fsspec's _get() does for one source, so that fsspec can create their parent directories. It passes each _get_file() max_workers = max(1, pool // n), or the caller's smaller value, through _batch_kwargs(), where n is the number of files or a smaller positive batch_size. Differences from the design:
  • The threads of the filesystem are not counted. Its thread pool and request pool have the same size, so pool // min(n, pool) and pool // n give the same window, and the mutation check confirmed that the term changed nothing.
  • fsspec's default batch size, when none is given, is not counted either. Counting it could only make the window larger.
• Docs. docs/filesystem.md describes the waves, the memory bound, the consistency, the partial file of a failed download (Q4 (i)), and compression. It drops the note on cache_type="background", since no cache reads a download now. docs/aio.md describes the window of _get().

Changes that callers observe (release notes):

• get()/get_file() download each object with up to max_workers parallel range requests of the block size (at most the request pool), instead of one at a time.
• cache_type and cache_options passed to get_file() are not used.
• get_file(..., compression=...) decompresses the object, which raised AttributeError.
• AioS3FileSystem._get() pairs its paths as S3FileSystem.get() does, and each of its downloads gets max(1, max_workers // n) range reads instead of max_workers.

Not changed: B (a sliding window) and C, for the step 5 measurement; _fetch_range() and AioS3StreamedFile reads; the default block size; a failed download still leaves the partly written file (Q4 (i)).

WHY

Closes #1323. Part of #1315 and of step 5 of #1332.

TEST

Tested at 05eb56b, rebased onto 758adee (#1394, #1378).

• just lint, just docs lint.
• Local AWS: uv run --with 'botocore[crt]' --env-file .env pytest -n 4 tests/pyathena/filesystem/ at 9ff12b2 (after the rebase) → 1747 passed, 8 failed, 6 skipped: the two new aio tests still called a helper that Accept os.PathLike paths in every S3 filesystem method (GH-1380) #1394 moved; fixed in 05eb56b, after which tests/pyathena/filesystem/test_s3_async.py → 223 passed, 3 skipped. just benchmark test → 340 passed, 1 skipped. Before the rebase, full runs followed each repair (25d0994: 1656 passed; 3a5ea80: 1661; 7c9bc98: 1663).
• New tests in test_s3.py:
  • test_get_file_reads_waves_of_ranges: the waves, each written before the next is read; one range at a time with one worker; the cap at the pool.
  • test_get_file_contents: an empty object, less than a block, an exact multiple, and a remainder, with the progress values.
  • test_get_file_constructs_no_data_cache, test_get_file_compression_ignores_cache_type (one worker, so that an eager whole-object read is one request).
  • test_get_file_non_positive_block_size: 0 and -1, with and without gzip, with a negative filesystem default.
  • test_get_file_failed_range_reads_no_further_wave (the failing range waits until the other has been requested).
  • test_get_file_holds_one_wave: the bodies of the first wave die, within a bound, before the second wave is read.
  • test_get_file_failure_while_writing_keeps_no_bodies: write, wrapped write, callback, and interrupt, with the cyclic GC disabled.
  • test_get_file_request_parameters: version, IfMatch, and call parameters on every range.
  • test_get_file_compression: gzip, "infer" from a versioned key, and "infer" without a codec.
• New tests in test_s3_async.py: test_get_window (one file, two, more than max_workers, batch_size 1 and -1, the caller's smaller max_workers) and test_get_relative_destination_list.
• test_get_file_stopped_writes_no_further_block is rewritten for the waves, with one range per wave, so that the stop always comes after the read.
• Mutation check: 17 mutants, one per change. They cover the pool cap, sequential waves, the release, the chained frames, the drop of written bodies, the cache on both paths, the compression path and its block size and fallback, the non-positive block size, the window of _get() and its batch_size, the thread term, the normalization of destination lists, and the write-time stop check. 16 are killed. The thread term survived and was removed. The stop, failed-range, and one-wave tests first depended on timing and were made deterministic; five repeated runs passed.
• Reviews: two self-review rounds; Codex independent review (5 findings), follow-ups (4, 1, CLEAN), and a review of the rebase (CLEAN; static, so it did not see the moved test helper), recorded inline.
• AWS CI: runs when the PR is marked Ready.

🤖 Generated with Claude Code

[laughingman7743]

Self-review round one (implementation behavior): base d03e034f947b9ab6ad5696ec2bd942f8a63e7b5d, head 0f5f21e72ea385e8d995c773ceebb42a8056946f (initial pass at 25d09949).

Covered: S3File._download() (wave partition for every max_workers including 1, the pool cap, the range read through _get_range() and its stop check, the write-time stop check, the per-body drop, the try scope over the assignment of a wave, S3MapResult.release() with chained errors and cycles); get_file() (the directory and missing-object paths, unchanged; cache_type="none" and the dropped cache_options; the codec resolution and the decompressing path; the order of opening the remote and the local file); callers: synchronous get() (task threads, _batch_kwargs() window, outfile in the calling thread), AioS3FileSystem._get_file() (a thread of the filesystem waits on the request pool, T1), and AioS3FileSystem._get() (pairing through _copy_pairs(), check_contained, the window, fsspec popping batch_size); the new and rewritten tests and their mutants.

FINDINGS, all repaired:

1. The codec was resolved from remote.path, which carries a ?versionId= suffix, so compression="infer" found no codec for a versioned .gz key. It is now resolved from the key, before the file is opened, so that "infer" without a codec takes the parallel path (25d0994, before the PR); test_get_file_compression covers the three cases.
2. The docstring of get_file() said that a replacement always fails the download; a download of a version other than null is unaffected. Corrected (25d0994).
3. A bucket path with compression="infer" reached infer_compression(None) and raised AttributeError; the key defaults to "" (0f5f21e).

Not changed: an empty pairing in AioS3FileSystem._get() returns without callback.set_size(0), which fsspec's _get() called; a lpath that is a single file-like object fails in the pairing, as it failed in fsspec's _get() before.

[laughingman7743]

Self-review round two (claims, compatibility, operations): base d03e034f947b9ab6ad5696ec2bd942f8a63e7b5d, head 0f5f21e72ea385e8d995c773ceebb42a8056946f.

Claims checked:

• "At most one wave of bodies is held": bodies[i] = b"" after each write, and the list is replaced only by the next map(); test_get_file_holds_one_wave sees no body of the first wave alive when the second is read (killed without the drop).
• "No data cache": cache_type="none" makes fsspec construct BaseCache, which reads nothing; test_get_file_constructs_no_data_cache with cache_type="all" sees each range once.
• "The version, IfMatch, and parameters of open()": get_file() still opens through open(), so S3FileTarget.resolve() and the precedence of _open()/S3File.__init__() apply unchanged; test_get_file_request_parameters.
• "compression raised AttributeError": remote.size on fsspec's decompressor (GzipFile) at d03e034.
• Docs: the removed cache_type="background" note concerned the block prefetched by the cache of a download, which no longer exists; the other statements match the code. The AioS3FileSystem._get() window omits the thread term and fsspec's default batch size, as the PR body explains.

Existing callers: get_file() keeps its signature; cache_type/cache_options are now ignored instead of applied; AioS3FileSystem._get() now pairs non-version paths with S3PathPairing, the pairing of S3FileSystem.get(), which the fsspec conformance tests of get cover for both filesystems (AWS, passed).

Operator: the number of GetObject requests per object stays one per block (the old first read requested two blocks); up to window of them now run at once per download, within the request pool. Memory per download is window × block_size (50 MiB with the defaults); a synchronous or aio batch gives each download max(1, pool // n). Throughput is not measured here; that is step 5.

Evidence: local AWS filesystem suite and just benchmark test (rerun at the reviewed head pending); mutants per change. FINDINGS: the PR body stated a tested commit whose full run predated the last codec change; corrected in the body.

[laughingman7743]

Independent review (relayed): Codex CLI codex exec -s read-only (model gpt-6-astra, session 01a1d07a-27e5-7b92-a922-57e29bec2157), static review only, base d03e034f947b9ab6ad5696ec2bd942f8a63e7b5d, head 0f5f21e72ea385e8d995c773ceebb42a8056946f, detached snapshot unchanged. Prompt: the diff, the agreed design's contract, repository conventions and docs/testing.md; no PR text, commits, or prior findings.

Covered: the exact diff; range partitioning, waves, worker caps, ordered writes, stop scopes, map draining, F5 retention; open() arguments, versions/IfMatch, compression inference, closing, callbacks, partial files; aio pairing, allocation and cancellation against fsspec 2026.9.0; callers in pyathena/ and tests/, docs, docstrings, and the ten added or rewritten tests (each judged sensitive to its change; determinism issues in findings 4 and 5). Result: FINDINGS (5).

1. P2 — Negative block size silently produces an empty download (reviewed at s3.py:4574). For a nonempty object, get_file(..., block_size=-1, max_workers=1, cache_type="none") opens/truncates the destination, then range(0, size, -1) performs no iterations and returns successfully. The previous read loop downloaded the object for these arguments.

[laughingman7743]

Verified and fixed in 3a5ea80: a block size that is not positive reads the object in one range, as read() reads the rest; with 0, range() raised ValueError. test_get_file_non_positive_block_size[0, -1] checks the contents and the single range; killed without the fix.

[laughingman7743]

Independent review 2. P2 — Aio pairing loses destination-list normalization (reviewed at s3_async.py:821). _get("s3://bucket/key", ["download.txt"]) now passes two lists to fsspec, which bypasses its make_path_posix() step, so its os.makedirs(os.path.dirname("download.txt")) raises FileNotFoundError. Previously this input used an absolute destination and succeeded.

[laughingman7743]

Verified and fixed in 3a5ea80: _get() makes the str and path-like entries of a destination list absolute with make_path_posix, as fsspec's _get() does for one source. File-like entries are left as given. test_get_relative_destination_list covers a scalar source and a list source, each with a relative destination, and is killed without the normalization. The list-to-list case also failed before this PR, in fsspec's makedirs("").

[laughingman7743]

Independent review 3. P2 — Compression bypasses the promised no-cache behavior (reviewed at s3.py:2908). compression="gzip", cache_type="all" constructs AllBytes, fetching and retaining the entire compressed object before writing; cache_options also remains active, and "background" can issue reads outside the enclosing stop context. This contradicts the docs.

[laughingman7743]

Verified and fixed in 3a5ea80: get_file() drops the caller's cache_options and sets cache_type on both paths: "none" for the ranged download, and "bytes" for the decompressing path, whose small reads need a block cache. So neither "all" nor "background" applies. test_get_file_compression_ignores_cache_type checks that no request reads the whole compressed object when cache_type="all" is passed, and is killed when the override is limited to the uncompressed path. The docs state it.

[laughingman7743]

Independent review 4. P2 — Completed workers can retain bodies across waves (reviewed at s3.py:4592). Clearing bodies[i] does not clear _S3WorkItem.run()'s local result; its future becomes done before that worker frame returns. A worker paused after publishing its result can still hold the body while the download writes that wave and reads the next, exceeding the strict one-wave bound and making test_get_file_holds_one_wave scheduling-dependent.

[laughingman7743]

Verified; documented rather than changed. A pool thread holds the value it returned until _S3WorkItem.run() returns, after the future is set. In CPython the reference is on the worker's frame or stack while set_result() wakes the waiter, so no reordering inside run() can close the window. The bound is therefore stated for what the download holds: S3File._download() says that a pool thread that has just returned a body drops it when its call returns. test_get_file_holds_one_wave now waits, with a 2 s bound, for the first wave's bodies to die when the second wave is read, instead of requiring it at once. It still fails when the download keeps its bodies (killed by the no-drop mutant) and no longer depends on the scheduling. Five repeated runs passed.

[laughingman7743]

Independent review 5. P2 — The failed-range test assumes an unsynchronized request order (reviewed at test_s3.py:4694). If bytes=4-7 fails before bytes=0-3 reaches its stop check, cancellation correctly prevents the latter request, but the test requires both requests and fails.

[laughingman7743]

Verified and fixed in 3a5ea80: the failing bytes=4-7 request waits, with a bound, until bytes=0-3 has been requested before raising, so the batch stop cannot cancel the first range before it starts. Five repeated runs passed.

[laughingman7743]

Independent follow-up review (relayed): Codex CLI codex exec -s read-only (model gpt-6-astra, session 01a1d08b-191d-7b41-acec-50512b9d6e66), static review only, repair 0f5f21e72ea385e8d995c773ceebb42a8056946f..3a5ea80a86ea4aba5383934790a632f809da3768 (same base d03e034f947b9ab6ad5696ec2bd942f8a63e7b5d), detached snapshot unchanged.

Covered: the repair diff, all five changed files, sync get() forwarding, aio destination handling, fsspec 2026.9.0 caches, gzip reads, worker ownership, interrupt/cancellation cleanup, the changed tests. The ranged non-positive block size and relative-destination repairs are correct, and their tests detect the prior failures; both synchronization repairs are sound; the documented ownership choice for the pool thread is acceptable; aio file-like list entries still hit fsspec's existing dirname() TypeError, not introduced here. Result: FINDINGS (4).

1. P2 — Forced bytes caching breaks compressed downloads with negative block sizes (reviewed at s3.py:2914). A gzip download with block_size=-1, max_workers=1 reaches BytesCache._fetch(), whose trimming divides by blocksize + 1, raising ZeroDivisionError; with zero, reader.read(0) produces an empty destination.

[laughingman7743]

Verified and fixed in 7065102: on the decompressing path, a block size that is not a positive integer takes the filesystem's default_block_size before the file and its bytes cache are constructed. The ranged path keeps one range for the whole object. test_get_file_non_positive_block_size now covers 0 and -1 with and without gzip, and is killed when the normalization is removed.

[laughingman7743]

Independent follow-up 1, finding 2. P3 — The compression cache is incorrectly documented as one block (reviewed at filesystem.md:146, s3.py:2912). BytesCache reads ahead beyond the requested bytes, and gzip requests 128 KiB of compressed input even with block_size=1024.

[laughingman7743]

Verified and fixed in 7065102: the docs and the docstring say the object is read in order through a block cache, which reads ahead of the decompressor. They no longer say one block at a time.

[laughingman7743]

Independent follow-up 1, finding 3. P3 — The weaker guarantee of finding 4 is not stated consistently (reviewed at s3.py:2860, filesystem.md:139). get_file() still says at most one wave "is held", and the guide omits the body that a returning pool thread keeps.

[laughingman7743]

Verified and fixed in 7065102: the get_file() docstring and docs/filesystem.md now say that the download holds at most one wave, apart from a range that a thread of the request pool has just read, until its call returns. This matches S3File._download().

[laughingman7743]

Independent follow-up 1, finding 4. P3 — The compression test does not detect retaining the caller's cache type (reviewed at test_s3.py:4703). "all" fetches the whole object through _fetch_range(), which splits it into smaller requests, so the forbidden whole-object Range never appears; only the rejected "unused" option made a full revert fail.

[laughingman7743]

Verified and fixed in 7065102: the test reads with max_workers=1, so that the eager whole-object read of "all" is one request, and it no longer passes cache_options. It is killed by the mutant that keeps the caller's cache_type on the decompressing path while still dropping cache_options.

[laughingman7743]

Independent follow-up review 2 (relayed): Codex CLI codex exec -s read-only (model gpt-6-astra, session 01a1d094-f808-7e70-9256-f60cfbeac7b1), static review only, repair 3a5ea80a86ea4aba5383934790a632f809da3768..706510292721e2312d15716e521b0586ce605f33 (same base d03e034f947b9ab6ad5696ec2bd942f8a63e7b5d), detached snapshot unchanged.

Covered: the repair diff, block-size propagation through sync/batch/async downloads, fsspec caches, wave-body lifetime, interrupt/cancellation cleanup, documentation, the changed tests. The read-ahead and returning-thread documentation repairs are correct; the new compressed cases fail without the normalization, and the one-worker cache test detects keeping "all"; no assertion depends on thread scheduling. Result: FINDINGS (1).

1. P2 — The fallback block size can still be negative (reviewed at s3.py:2921). S3FileSystem(default_block_size=-1) is accepted, so get_file(..., compression="gzip", block_size=0, max_workers=1) replaces 0 with -1 and still raises ZeroDivisionError in BytesCache; with several workers, _get_ranges() keeps generating decreasing offsets.

[laughingman7743]

Verified and fixed in 7c9bc98: the fallback is the filesystem's default_block_size when it is positive, and DEFAULT_BLOCK_SIZE otherwise. test_get_file_non_positive_block_size sets default_block_size = -1 and is killed when the fallback is the raw default. Validating the constructor argument itself is outside this PR.

[laughingman7743]

Independent follow-up review 3 (relayed): Codex CLI codex exec -s read-only (model gpt-6-astra, session 01a1d09a-2ff1-7550-92eb-abdffdb87087), static review only, repair 706510292721e2312d15716e521b0586ce605f33..7c9bc988867daa687c91850d3c3244ea0d8694cf (same base d03e034f947b9ab6ad5696ec2bd942f8a63e7b5d), detached snapshot unchanged.

Covered: the repair diff; constructor defaults; compressed get_file() → open() → fsspec BytesCache; uncompressed downloads; the changed test and its fixture isolation.

CLEAN. The repair falls back to the positive 5 MiB constant; positive explicit sizes, positive filesystem defaults, and uncompressed behavior are unchanged. Both compressed parameterizations would fail without the repair (propagating -1 divides by zero in fsspec's cache trimming), and the test is deterministic: a fresh mocked filesystem, fixed content, one worker, no timing-sensitive assertion.

[laughingman7743]

Rebase onto 758adee9eb1b12d5bd939611f56e0ff5e433a743 (#1394, #1378): the patch series is unchanged except for one conflict resolution in AioS3FileSystem._get(), which keeps upstream's rpath = self._sync_fs._stringify_paths(rpath) before the new pairing (git range-diff d03e034f..7c9bc988 758adee9..9ff12b21). Force-pushed with --force-with-lease against 7c9bc988.

Independent rebase review (relayed): Codex CLI codex exec -s read-only (model gpt-6-astra, session 01a1d0aa-2b6f-7012-a9bc-3e5b32221a75), static review only, on the range-diff and the upstream diff of s3.py, s3_async.py, and s3_executor.py (d03e034f..758adee9), snapshot at 9ff12b211ee498fa05d8f241f5f47800fede4fcc, unchanged.

Covered: the conflict resolution; path-like sources and destinations through get_file(), get(), aio _get_file() and _get() (scalar/list conversion, versioned pairing, absolute destination-list entries); client=, retry_config=, deprecated connection=, and s3_filesystem(), with the request pool sized by max_workers or the injected client's max_pool_connections, which the waves and the batch windows use; the waves, writes, body release, cache exclusion, and the compression path; the upstream tests of path-like get() and of the client and pool size, and the conformance fixtures switched to s3_filesystem().

CLEAN. The conflict resolution is correct, and no rebase regression was found.

[laughingman7743]

The local AWS run at 9ff12b2 (1747 passed, 8 failed, 6 skipped) found what the static rebase review missed. #1394 moved the _serve_keys helper of TestAioS3FileSystem to tests.pyathena.util.serve_s3_keys(). Its existing callers were converted upstream, but the two tests added here (test_get_window, test_get_relative_destination_list) still called the removed method. They are fixed in 05eb56b (test-only), and tests/pyathena/filesystem/test_s3_async.py passes in full (223 passed, 3 skipped). The other 1747 tests passed at 9ff12b2, with just benchmark test (340 passed).