Add AioS3FileSystem using asyncio for parallel S3 operations

[laughingman7743]

Summary

The current S3FileSystem uses ThreadPoolExecutor for parallel S3 operations (range reads, batch deletes, multipart uploads). Add an AioS3FileSystem variant that uses asyncio.gather + asyncio.to_thread instead, providing better integration with the asyncio event loop for aio cursors.

Motivation

When aio cursors use the current S3FileSystem, operations are double-wrapped:

1. The cursor wraps the result set creation in asyncio.to_thread()
2. Inside that thread, S3FileSystem spawns more threads via ThreadPoolExecutor

An AioS3FileSystem would allow aio cursors to use async S3 operations directly, eliminating the thread-in-thread pattern and integrating naturally with the event loop.

Design

• Keep S3FileSystem as-is — no breaking changes to the existing synchronous implementation
• Add AioS3FileSystem — uses asyncio.gather + asyncio.to_thread for individual boto3 calls instead of ThreadPoolExecutor
• Follow fsspec's _async_impl pattern — implement async methods (_cat_file, _ls, etc.) so fsspec provides sync wrappers automatically
• No new dependencies — uses asyncio.to_thread to wrap synchronous boto3 calls (no aiobotocore needed)

ThreadPoolExecutor usage to replace

Location	Current usage	Async replacement
S3File._fetch_range()	Parallel range GETs	asyncio.gather(*[asyncio.to_thread(...)])
S3FileSystem._delete_objects()	Batch deletes	asyncio.gather(*[asyncio.to_thread(...)])
S3File._upload_chunk()	Multipart uploads	asyncio.gather(*[asyncio.to_thread(...)])

User-facing API

Users can choose which filesystem to use. Aio cursors could default to AioS3FileSystem when available:

from pyathena.filesystem.s3 import AioS3FileSystem

# Direct usage
fs = AioS3FileSystem(connection=connection)
async with fs.open("s3://bucket/key", "rb") as f:
    data = await f.read()

Future consideration

If benchmarks show the aio version performs better, it could become the default implementation for aio cursors or even replace the ThreadPoolExecutor-based implementation entirely.

Related

• AioPandasCursor/AioPolarsCursor: chunksize fetch blocks the event loop #672 — chunksize fetch blocking the event loop (fixed in Wrap fetch methods in asyncio.to_thread for aio pandas/arrow/polars cursors #674 with asyncio.to_thread wrapping)
• Wrap fetch methods in asyncio.to_thread for aio pandas/arrow/polars cursors #674 — wraps fetch methods in asyncio.to_thread for aio pandas/arrow/polars cursors

[laughingman7743]

Implementation Plan

Architecture: Composition with asyncio.to_thread

AioS3FileSystem extends fsspec.asyn.AsyncFileSystem and internally holds an S3FileSystem instance for delegation.

Why composition (not inheritance)?

• S3FileSystem extends fsspec.AbstractFileSystem (sync) and AioS3FileSystem must extend AsyncFileSystem — diamond inheritance would cause MRO conflicts
• Avoids refactoring existing S3FileSystem (no breaking changes)
• Maximizes code reuse (all boto3 logic stays in S3FileSystem)

Key properties:

• async_impl = True, mirror_sync_methods = True — fsspec auto-generates sync wrappers
• asyncio.to_thread() for individual boto3 calls
• asyncio.gather(*[asyncio.to_thread(...)]) replaces ThreadPoolExecutor for parallel ops
• Shared dircache between async and sync instances

Files

File	Action
pyathena/filesystem/s3_async.py	New — AioS3FileSystem class
tests/pyathena/filesystem/test_s3_async.py	New — async test suite
pyathena/filesystem/s3.py	No changes

Method Mapping

Core async methods (fsspec required)

Async Method	Implementation
_info(path)	asyncio.to_thread(self._sync_fs.info, ...)
_ls(path, detail)	asyncio.to_thread(self._sync_fs.ls, ...)
_cat_file(path, start, end)	asyncio.to_thread(self._sync_fs.cat_file, ...)
_exists(path)	asyncio.to_thread(self._sync_fs.exists, ...)
_rm_file(path)	asyncio.to_thread(self._sync_fs.rm_file, ...)
_put_file(lpath, rpath)	asyncio.to_thread(self._sync_fs.put_file, ...)
_get_file(rpath, lpath)	asyncio.to_thread(self._sync_fs.get_file, ...)
_pipe_file(path, value)	asyncio.to_thread(self._sync_fs.pipe_file, ...)
_mkdir / _makedirs	asyncio.to_thread(self._sync_fs.mkdir/makedirs, ...)

Parallel async methods (replacing ThreadPoolExecutor)

Method	What it replaces
_rm(path, recursive)	S3FileSystem._delete_objects() — chunks into batches of 1000, asyncio.gather(*[asyncio.to_thread(delete_objects, ...)])
_cp_file(path1, path2)	S3FileSystem._copy_object_with_multipart_upload() — parallel part uploads via asyncio.gather(*[asyncio.to_thread(upload_part_copy, ...)])

Other methods

Method	Strategy
parse_path(path)	Delegate to S3FileSystem.parse_path (static)
_open(path, mode, ...)	Return S3File via self._sync_fs._open(...)
invalidate_cache(path)	Delegate to self._sync_fs.invalidate_cache(path)
sign, checksum, created, modified	Delegate to sync fs

Design Decisions

1. No AioS3File class — fsspec's _open is always synchronous (even in AsyncFileSystem); S3File works as-is via _open delegation. S3File's internal ThreadPoolExecutor for range reads is acceptable since file-level operations are inherently sync in fsspec.

2. Shared dircache — self.dircache = self._sync_fs.dircache prevents stale data between sync and async paths.

3. No new dependencies — asyncio.to_thread wraps synchronous boto3 calls (no aiobotocore required).

4. S3FileSystem unchanged — zero breaking changes to existing sync implementation.

Future Considerations

• If benchmarks show benefit, aio cursors could default to AioS3FileSystem
• An AioS3File variant using asyncio tasks instead of ThreadPoolExecutor for range reads could be a follow-up enhancement
• If aiobotocore becomes a viable dependency, native async boto3 calls could replace asyncio.to_thread