Add native asyncio S3FS and Spark cursors with boilerplate deduplication (Phase 3)

[laughingman7743]

Summary

Phase 3 of native asyncio cursor implementation (#662), building on Phase 2 (PR #667).

Part 1: Boilerplate Deduplication — WithAsyncFetch mixin

Extract shared boilerplate from 4 existing aio SQL cursors into a WithAsyncFetch mixin in pyathena/aio/common.py:

• Properties: arraysize (getter/setter), result_set (getter/setter), query_id (getter/setter), rownumber, rowcount
• Lifecycle: close(), executemany(), cancel()
• Default sync fetch: fetchone(), fetchmany(), fetchall() — for cursors that load data eagerly in __init__
• Async protocol: __aiter__, __anext__, __aenter__, __aexit__

Net reduction of ~520 lines of duplicated code across AioCursor, AioPandasCursor, AioArrowCursor, AioPolarsCursor.

Part 2: AioS3FSCursor

New pyathena/aio/s3fs/cursor.py — async CSV cursor using S3FileSystem.

Unlike other aio cursors (Arrow/Pandas/Polars) that load data eagerly, AthenaS3FSResultSet lazily streams rows from S3 via a CSV reader. Therefore:

• Result set creation: asyncio.to_thread(AthenaS3FSResultSet, ...)
• Fetch methods: async via asyncio.to_thread(result_set.fetch*, ...) — reads from S3 on each call
• No UNLOAD support (consistent with sync S3FSCursor)

Part 3: AioSparkCursor

New pyathena/aio/spark/cursor.py — async PySpark code execution.

• Inherits from SparkBaseCursor and WithCalculationExecution directly (no intermediate base class — aio only has one Spark cursor variant)
• Post-init I/O methods (_poll, _cancel, _terminate_session, _read_s3_file_as_text, _calculate) overridden with async equivalents
• Session management stays synchronous in __init__ (runs inside asyncio.to_thread at cursor creation)
• _read_s3_file_as_text uses async_retry_api_call for retry consistency with sync version

Additional changes

• Removed on_start_query_execution callback from all aio cursors — unnecessary in async context where await provides direct control flow. Consistent with AsyncCursor which also omits this callback.

Class hierarchy

BaseCursor
  └─ AioBaseCursor (async I/O: _execute, _poll, _cancel, metadata)
      └─ WithAsyncFetch (properties, close, cancel, sync fetch, async protocol)
          ├─ AioCursor / AioDictCursor     — async fetch (paginated API)
          ├─ AioPandasCursor               — eager load, as_pandas()
          ├─ AioArrowCursor                — eager load, as_arrow(), as_polars()
          ├─ AioPolarsCursor               — eager load, as_polars(), as_arrow()
          └─ AioS3FSCursor                 — lazy streaming, async fetch

SparkBaseCursor + WithCalculationExecution
  └─ AioSparkCursor (async calculation I/O, session, stdout/stderr)

Files changed

Action	File	Description
Modify	pyathena/aio/common.py	Add WithAsyncFetch mixin
Modify	pyathena/aio/cursor.py	Extend WithAsyncFetch, remove boilerplate
Modify	pyathena/aio/pandas/cursor.py	Extend WithAsyncFetch, remove boilerplate
Modify	pyathena/aio/arrow/cursor.py	Extend WithAsyncFetch, remove boilerplate
Modify	pyathena/aio/polars/cursor.py	Extend WithAsyncFetch, remove boilerplate
Create	pyathena/aio/s3fs/__init__.py	Package marker
Create	pyathena/aio/s3fs/cursor.py	AioS3FSCursor
Create	pyathena/aio/spark/__init__.py	Package marker
Create	pyathena/aio/spark/cursor.py	AioSparkCursor
Modify	tests/pyathena/aio/conftest.py	Add s3fs + spark fixtures
Create	tests/pyathena/aio/s3fs/	S3FS cursor tests
Create	tests/pyathena/aio/spark/	Spark cursor tests
Modify	tests/pyathena/aio/test_cursor.py	Remove callback test

Test plan

• make fmt — formatting clean
• make chk — lint + mypy clean
• All existing aio tests pass (no regressions from Phase 2)
• New AioS3FSCursor tests: fetchone/many/all, async iterator, description, cancel, executemany, arraysize, context manager
• New AioSparkCursor tests: spark_dataframe, spark_sql, failed, cancel, executemany, context manager

Related issues

• Closes Add native asyncio cursor support (AioCursor) #662 (native asyncio cursor support)
• Related: Extract shared boilerplate from sync cursors into WithFetch mixin #669 (sync cursor deduplication — same pattern, separate PR)
• Related: Extract shared UNLOAD and execute() control flow into helper methods #670 (UNLOAD/execute() control flow deduplication)

🤖 Generated with Claude Code