Skip to content

Separate query and S3 read worker settings #1103

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
laughingman7743 merged 7 commits into from
Oct 10, 2026
Merged

Separate query and S3 read worker settings #1103

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
dda6aae
Separate cursor query and S3 read worker settings
laughingman7743 Oct 5, 2026
d30dff2
Migrate benchmark worker options and serialize Arrow S3 client creation
laughingman7743 Oct 5, 2026
8f1d1af
Use standalone functions for S3 worker option tests
laughingman7743 Oct 9, 2026
c120e30
Limit S3 worker settings to the pandas and Polars cursors
laughingman7743 Oct 9, 2026
f316843
Note that chunked Polars reads do not use s3_max_workers
laughingman7743 Oct 9, 2026
0233108
Merge remote-tracking branch 'origin/master' into feat/1094-separate-…
laughingman7743 Oct 10, 2026
9df509e
Describe where the former max_workers keyword fails
laughingman7743 Oct 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 4 additions & 8 deletions benchmarks/pyathena_bench/adapters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,8 +179,10 @@ def connection(self) -> Any:
def cursor(self, connection: Any) -> Any:
case = self.case
kwargs: dict[str, Any] = {"arraysize": case.arraysize}
if case.api == "thread" or case.family in {"pandas", "polars"}:
if case.api == "thread":
kwargs["max_workers"] = self.settings.executor_workers
if case.family in {"pandas", "polars"}:
kwargs["s3_max_workers"] = self.settings.executor_workers
if case.family in {"pandas", "polars", "arrow"}:
kwargs["unload"] = case.transport == "unload"
if case.family in {"pandas", "polars"}:
Expand Down Expand Up @@ -251,14 +253,8 @@ async def measure_query(adapter: Adapter, cursor: Any, sql: str, expected: int)
count = consume(result, case)
query_id = None # The botocore observer captures IDs even for iterators.
elif case.api == "thread":
# AsyncPandasCursor forwards result-reader options through execute().
options = (
{"max_workers": adapter.settings.executor_workers}
if case.family == "pandas"
else {}
)
query_id, future = await asyncio.to_thread(
cursor.execute, sql, cache_size=0, result_reuse_enable=False, **options
cursor.execute, sql, cache_size=0, result_reuse_enable=False
)
result = await asyncio.wrap_future(future)
ready = time.perf_counter()
Expand Down
9 changes: 7 additions & 2 deletions benchmarks/tests/test_adapters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,13 +129,18 @@ def unexpected_request(*args, **kwargs):
session = boto3.Session(
aws_access_key_id="testing", aws_secret_access_key="testing", region_name="us-west-2"
)
for case in matrix(Settings(), "single", "flat"):
settings = Settings(executor_workers=3)
for case in matrix(settings, "single", "flat"):
if case.family == "wrangler" or case.unsupported:
continue
adapter = Adapter(Settings(), case, session, "scratch", "s3://out/", "temp")
adapter = Adapter(settings, case, session, "scratch", "s3://out/", "temp")
with adapter.connection() as connection:
cursor = adapter.cursor(connection)
assert cursor.arraysize == case.arraysize
if case.api == "thread":
assert cursor._executor._max_workers == 3
if case.family in {"pandas", "polars"}:
assert cursor._s3_max_workers == 3
cursor.close()

@pytest.mark.parametrize("api", ["sync", "thread", "aio"])
Expand Down
23 changes: 23 additions & 0 deletions docs/usage.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,29 @@ A cursor subclass that receives settings of its own must not pass them on to the
The pandas and Polars cursors pass the `execute()` keyword arguments that they do not use themselves to the reader, such as `pandas.read_csv()`.
The reader rejects an unknown name when it reads the query results, which PyAthena raises as `OperationalError`; the name is not checked when there are no results to read.

## Worker settings in PyAthena 4.0

The pandas and Polars cursors take the number of threads that read each result file from S3 as `s3_max_workers`.
It can be set in the cursor constructor and overridden for one query in `execute()`.
The default is `(cpu_count() or 1) * 5`.
Polars uses it only for CSV results read without `chunksize`.

@laughingman7743 laughingman7743 Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Relayed independent static review — FINDINGS (one Low documentation finding), repaired; follow-up CLEAN.

Reviewer: Codex CLI 0.161.0, model gpt-6.1-sol (a different model from the Claude author of this revision), codex exec -s read-only in a detached clean snapshot. No edits, builds, tests, network, or GitHub access; the prompt omitted the PR number, body, commits, and prior findings.

Initial review — session 01a1216d-29c9-7751-a8c5-e1630b9bb34f, base f3532e9fb57fd769441e533f0b8375fd854ece69, head c120e3011a9d20dbf8f12e27765df0794da39973. Covered all six pandas/Polars cursors (constructor settings, defaults, routing, per-query overrides), thread-pool independence, unchanged result-set/filesystem names and Arrow docstring, all repository callers (benchmarks, tests, docs, README, connection factories, sync/async SQLAlchemy dialects), regression coverage, docstrings, and simplicity. No implementation defect; it noted the tests use mocked result sets and do not test overlapping overrides.
Finding (P3): docs/usage.md said only UNLOAD bypasses s3_max_workers for Polars, but chunked CSV reads also do — pyathena/polars/result_set.py:778 passes _parquet_storage_options (native object_store, no worker value) to pl.scan_csv. Verified by the author.

An earlier delegated Codex run (thread 01a1216c-1a5d-7d02-8b74-810ca7993e7f) ended with a placeholder message and no review; it is not counted.

Repair f316843f84e1dc839e1dd243e1ca3fa76b188c18: the docs now say Polars uses the setting only for CSV results read without chunksize, and chunked CSV and UNLOAD reads use Polars' native readers.

Follow-up — session 01a12170-fac9-7bb0-88a7-61d533338321, diff c120e3011a9d20dbf8f12e27765df0794da39973..f316843f84e1dc839e1dd243e1ca3fa76b188c18: CLEAN. Verified non-chunked CSV, chunked CSV, non-chunked and chunked UNLOAD Parquet, and results without an output file; user storage_options replace the defaults, so "only for" is not overstated. The review snapshot remained clean. Static review only.

Chunked CSV reads and UNLOAD results use Polars' native readers, which do not use this setting.

The thread-pool cursors, such as `AsyncCursor`, `AsyncPandasCursor`, and `AsyncPolarsCursor`, keep `max_workers` for the size of the thread pool that waits for queries and collects their results.

```python
from pyathena.pandas.async_cursor import AsyncPandasCursor

cursor = connection.cursor(AsyncPandasCursor, max_workers=4, s3_max_workers=2)
query_id, future = cursor.execute("SELECT 1", s3_max_workers=3)
```

Before PyAthena 4.0, `max_workers` set the S3 read workers of `PandasCursor`, `PolarsCursor`, `AioPandasCursor`, and `AioPolarsCursor`, and of `execute()` in the pandas and Polars cursors.

@laughingman7743 laughingman7743 Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Self-review round two after scope reduction — CLEAN after the PR-body correction recorded in round one.

Base: f3532e9fb57fd769441e533f0b8375fd854ece69
Head: c120e3011a9d20dbf8f12e27765df0794da39973

Claims checked: (1) s3_max_workers default equals the former default — (cpu_count() or 1) * 5 in the pandas/Polars cursors, the result sets, and AsyncCursor.max_workers (the value AsyncPolarsCursor used to forward). (2) AsyncPolarsCursor used max_workers for both pools — base pyathena/polars/async_cursor.py passed self._max_workers to the result set. (3) Polars UNLOAD does not use the setting — _read_parquet/scan_parquet use _parquet_storage_options (native object_store); only _csv_storage_options passes max_workers. (4) The usage example — Connection.cursor(cursor, **kwargs) forwards both arguments. (5) Arrow's native PyArrow S3FileSystem has no per-file worker option. (6) No other doc uses the renamed pandas/Polars argument; remaining max_workers examples are thread-pool cursors or to_sql.

Correction left in history: commit c120e301's message says reader options are rejected by the readers; the former execute() keyword actually fails as a duplicate result-set argument. The PR body is accurate; the commit was not rewritten.

Limits: docs example and AWS paths are not run locally; just docs build shows only the existing warnings. AWS suites run after Ready.

@laughingman7743 laughingman7743 Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Round-two follow-up for the documentation repair — CLEAN. Old head c120e3011a9d20dbf8f12e27765df0794da39973, new head f316843f84e1dc839e1dd243e1ca3fa76b188c18. Claim checked: "Polars uses it only for CSV results read without chunksize; chunked CSV reads and UNLOAD results use Polars' native readers". Supported by pyathena/polars/result_set.py:577 (_csv_storage_options) versus :637, :654, :779, and :818 (_parquet_storage_options). Results without an output file are read from GetQueryResults bytes and use no S3 reader, which the wording does not contradict. No other prose changed.

@laughingman7743 laughingman7743 Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Final runtime validation: head f316843f84e1dc839e1dd243e1ca3fa76b188c18. Ready-triggered Test run 37959031361 succeeded on Python 3.14: PyAthena 3648 passed / 12 skipped; SQLAlchemy sync compliance 589 passed / 759 skipped; async compliance 589 passed / 759 skipped; lint and changes passed. The skipped test/test-sqla/test-sqla-async entries belong to the earlier Draft run 37957549120. It started after the #1102 run 37957475564 completed, with no other Test run active. PR is Ready, MERGEABLE, CLEAN. Not merged.

@laughingman7743 laughingman7743 Oct 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Round-two follow-up after merging master — CLEAN.
Merge-base dafee9a4264ae2bb765cc7c5e1e3ef842bd6cfb8, head 9df509eaa4a8432db13e62837d4613948352be76.

Claims: (1) "Passing max_workers to those four constructors raises TypeError" — the probe shows BaseCursor.__init__() got an unexpected keyword argument 'max_workers' for PandasCursor, PolarsCursor, AioPandasCursor, AioPolarsCursor. (2) "Passing it to execute() raises TypeError after the query has run; with AsyncPandasCursor and AsyncPolarsCursor, the returned future raises it" — the probe shows _execute called before the duplicate-keyword error in all six; the thread-pool cursors collect results in _collect_result_set (pandas/async_cursor.py:190, polars/async_cursor.py:192). (3) #1102's general statement that pandas/Polars reader options fail as OperationalError still holds for other names; max_workers is the documented exception. The PR body states the post-#1102 behavior. AWS CI for the merge commit was triggered on push; results to follow.

@laughingman7743 laughingman7743 Oct 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Final runtime validation after the master merge: head 9df509eaa4a8432db13e62837d4613948352be76. Test run 38020038709 succeeded on Python 3.14: PyAthena 3740 passed / 12 skipped; SQLAlchemy sync compliance 589 passed / 759 skipped; async compliance 589 passed / 759 skipped; lint and changes passed. Run 38019766760 for merge commit 02331086 was cancelled when the wording repair was pushed. PR is Ready, MERGEABLE, CLEAN. Not merged.

Replace it with `s3_max_workers` in those places.
Passing `max_workers` to those four constructors raises `TypeError`.

@laughingman7743 laughingman7743 Oct 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Relayed independent static follow-up for the master merge — FINDINGS (one Low doc wording), repaired; follow-up CLEAN.

Reviewer: Codex CLI 0.161.0, model gpt-6.1-sol, codex exec -s read-only, detached clean snapshot; no edits, builds, tests, network, or GitHub.
Merge review — session 01a123cd-4816-73d2-ac90-455f4d88b4a8, merge-base dafee9a4264ae2bb765cc7c5e1e3ef842bd6cfb8, head 02331086a17e04a60895e792941f2aa58927b512. Covered the conflict resolution, the full branch diff against master, all six pandas/Polars cursors, base keyword validation, signature inspection, public/internal connection option routing, sync/async SQLAlchemy dialects and reflection adapters, fixtures, tests, benchmarks, and worker docs. No code-level merge regression or stale S3 max_workers caller found.
Finding (P3): docs/usage.md:68 said max_workers raises in execute(); for AsyncPandasCursor/AsyncPolarsCursor execute() returns normally and future.result() raises (pandas/async_cursor.py:190, polars/async_cursor.py:192). Verified.
Repair 9df509eaa4a8432db13e62837d4613948352be76: the docs say the returned future raises it for those two cursors.
Follow-up — session 01a123d0-e10e-76d2-ad9c-92e1790f58db, diff 02331086a17e04a60895e792941f2aa58927b512..9df509eaa4a8432db13e62837d4613948352be76: CLEAN for all six cursors' constructors and execute()/future paths. Snapshot remained clean. Static review only.

Passing it to `execute()` raises `TypeError` after the query has run; with `AsyncPandasCursor` and `AsyncPolarsCursor`, the returned future raises it.
`AsyncPolarsCursor` used its `max_workers` for both the thread pool and the S3 read workers; pass both arguments to keep the former values.

## Managed query result storage

When using a workgroup with [managed query result storage](https://docs.aws.amazon.com/athena/latest/ug/managed-results.html) enabled,
Expand Down
10 changes: 5 additions & 5 deletions pyathena/aio/pandas/cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ def __init__(
chunksize: int | None = None,
block_size: int | None = None,
cache_type: str | None = None,
max_workers: int = (cpu_count() or 1) * 5,
s3_max_workers: int = (cpu_count() or 1) * 5,
result_reuse_enable: bool = False,
result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
auto_optimize_chunksize: bool = False,
Expand All @@ -84,7 +84,7 @@ def __init__(
it takes precedence over ``auto_optimize_chunksize``.
block_size: Default block size of the S3 filesystem that reads the results.
cache_type: Default cache type of the S3 filesystem that reads the results.
max_workers: Maximum number of workers of the S3 filesystem.
s3_max_workers: Maximum number of workers of the S3 filesystem.
result_reuse_enable: Whether to enable Athena query result reuse.
result_reuse_minutes: Maximum age of a reused query result in minutes.
auto_optimize_chunksize: Whether to choose a chunk size from the size of the
Expand All @@ -110,7 +110,7 @@ def __init__(
self._chunksize = chunksize
self._block_size = block_size
self._cache_type = cache_type
self._max_workers = max_workers
self._s3_max_workers = s3_max_workers
self._auto_optimize_chunksize = auto_optimize_chunksize
self._result_set: AthenaPandasResultSet | None = None

Expand Down Expand Up @@ -168,7 +168,7 @@ async def execute(
:class:`~pyathena.options.ExecuteOptions` instance. Individual
keyword arguments take precedence over ``options`` fields.
**kwargs: Additional pandas read_csv/read_parquet parameters.
``engine``, ``chunksize``, ``block_size``, ``cache_type``, ``max_workers``,
``engine``, ``chunksize``, ``block_size``, ``cache_type``, ``s3_max_workers``,
and ``auto_optimize_chunksize`` override the cursor's values for this query.
``storage_options`` and, for UNLOAD results, ``filesystem`` replace
PyAthena's S3 filesystem (see
Expand Down Expand Up @@ -218,7 +218,7 @@ async def execute(
chunksize=kwargs.pop("chunksize", self._chunksize),
block_size=kwargs.pop("block_size", self._block_size),
cache_type=kwargs.pop("cache_type", self._cache_type),
max_workers=kwargs.pop("max_workers", self._max_workers),
max_workers=kwargs.pop("s3_max_workers", self._s3_max_workers),
auto_optimize_chunksize=kwargs.pop(
"auto_optimize_chunksize", self._auto_optimize_chunksize
),
Expand Down
10 changes: 5 additions & 5 deletions pyathena/aio/polars/cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ def __init__(
result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
block_size: int | None = None,
cache_type: str | None = None,
max_workers: int = (cpu_count() or 1) * 5,
s3_max_workers: int = (cpu_count() or 1) * 5,
chunksize: int | None = None,
**kwargs,
) -> None:
Expand All @@ -78,7 +78,7 @@ def __init__(
result_reuse_minutes: Maximum age of a reused query result in minutes.
block_size: Default block size of the S3 filesystem that reads the results.
cache_type: Default cache type of the S3 filesystem that reads the results.
max_workers: Maximum number of workers of the S3 filesystem.
s3_max_workers: Maximum number of workers of the S3 filesystem.
chunksize: Number of rows per chunk. If set, result files in S3 are read
lazily in chunks of this size.
**kwargs: Other cursor arguments, such as ``connection`` and ``arraysize``,
Expand All @@ -100,7 +100,7 @@ def __init__(
self._unload = unload
self._block_size = block_size
self._cache_type = cache_type
self._max_workers = max_workers
self._s3_max_workers = s3_max_workers
self._chunksize = chunksize
self._result_set: AthenaPolarsResultSet | None = None

Expand Down Expand Up @@ -152,7 +152,7 @@ async def execute(
:class:`~pyathena.options.ExecuteOptions` instance. Individual
keyword arguments take precedence over ``options`` fields.
**kwargs: Additional execution parameters passed to Polars read functions.
``block_size``, ``cache_type``, ``max_workers``, and ``chunksize``
``block_size``, ``cache_type``, ``s3_max_workers``, and ``chunksize``
override the cursor's values for this query.
Read function arguments replace the ones the result set chooses, such as
``separator``, ``has_header``, ``schema_overrides``, and ``storage_options``
Expand Down Expand Up @@ -197,7 +197,7 @@ async def execute(
unload_location=unload_location,
block_size=kwargs.pop("block_size", self._block_size),
cache_type=kwargs.pop("cache_type", self._cache_type),
max_workers=kwargs.pop("max_workers", self._max_workers),
max_workers=kwargs.pop("s3_max_workers", self._s3_max_workers),
chunksize=kwargs.pop("chunksize", self._chunksize),
result_set_type_hints=options.result_set_type_hints,
**kwargs,
Expand Down
2 changes: 1 addition & 1 deletion pyathena/arrow/async_cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ def __init__(
encryption_option: S3 encryption option (SSE_S3, SSE_KMS, CSE_KMS).
kms_key: KMS key ARN for encryption.
kill_on_interrupt: Cancel running query on keyboard interrupt.
max_workers: Maximum number of workers for concurrent execution.
max_workers: Size of the cursor thread pool for waiting and collecting results.
arraysize: Number of rows to fetch per batch.
unload: Enable UNLOAD for high-performance Parquet output.
result_reuse_enable: Enable Athena query result reuse.
Expand Down
2 changes: 1 addition & 1 deletion pyathena/async_cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ def __init__(
kill_on_interrupt: Cancel a query whose start in ``execute()`` is interrupted by
``KeyboardInterrupt``. Waiting runs on worker threads, which do not
receive the interrupt.
max_workers: Maximum number of threads in the cursor's thread pool.
max_workers: Size of the cursor thread pool for waiting and collecting results.
arraysize: Default number of rows per ``fetchmany()`` call of the result
sets the cursor creates.
result_reuse_enable: Enable Athena query result reuse.
Expand Down
8 changes: 6 additions & 2 deletions pyathena/pandas/async_cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ def __init__(
block_size: int | None = None,
cache_type: str | None = None,
auto_optimize_chunksize: bool = False,
s3_max_workers: int = (cpu_count() or 1) * 5,
**kwargs,
) -> None:
"""Initialize an AsyncPandasCursor.
Expand All @@ -98,7 +99,7 @@ def __init__(
kill_on_interrupt: Cancel a query whose start in ``execute()`` is interrupted by
``KeyboardInterrupt``. Waiting runs on worker threads, which do not
receive the interrupt.
max_workers: Maximum number of threads that run queries concurrently.
max_workers: Size of the cursor thread pool for waiting and collecting results.
arraysize: Number of rows to fetch per batch. Must be a positive integer.
unload: Whether to wrap queries in ``UNLOAD`` and read the Parquet output.
engine: Parsing engine (``auto``, ``c``, ``python``, or ``pyarrow``).
Expand All @@ -110,6 +111,7 @@ def __init__(
cache_type: Default cache type of the S3 filesystem that reads the results.
auto_optimize_chunksize: Whether to choose a chunk size from the size of the
CSV result file when ``chunksize`` is None.
s3_max_workers: Maximum worker threads per S3 file reader.
**kwargs: Other cursor arguments, such as ``connection`` and ``converter``,
passed to ``AsyncCursor.__init__``.
"""
Expand All @@ -128,6 +130,7 @@ def __init__(
result_reuse_minutes=result_reuse_minutes,
**kwargs,
)
self._s3_max_workers = s3_max_workers
self._unload = unload
self._engine = engine
self._chunksize = chunksize
Expand Down Expand Up @@ -184,6 +187,7 @@ def _collect_result_set(
chunksize=kwargs.pop("chunksize", self._chunksize),
block_size=kwargs.pop("block_size", self._block_size),
cache_type=kwargs.pop("cache_type", self._cache_type),
max_workers=kwargs.pop("s3_max_workers", self._s3_max_workers),
auto_optimize_chunksize=kwargs.pop(
"auto_optimize_chunksize", self._auto_optimize_chunksize
),
Expand Down Expand Up @@ -237,7 +241,7 @@ def execute(
``storage_options`` and, for UNLOAD results, ``filesystem`` replace
PyAthena's S3 filesystem (see
:class:`~pyathena.pandas.result_set.AthenaPandasResultSet`).
``max_workers`` sets the number of S3 read workers for this query.
``s3_max_workers`` sets the number of S3 read workers for this query.

Returns:
Tuple of (query_id, future) where future resolves to AthenaPandasResultSet.
Expand Down
10 changes: 5 additions & 5 deletions pyathena/pandas/cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ def __init__(
chunksize: int | None = None,
block_size: int | None = None,
cache_type: str | None = None,
max_workers: int = (cpu_count() or 1) * 5,
s3_max_workers: int = (cpu_count() or 1) * 5,
result_reuse_enable: bool = False,
result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
auto_optimize_chunksize: bool = False,
Expand All @@ -104,7 +104,7 @@ def __init__(
If specified, takes precedence over auto_optimize_chunksize.
block_size: S3 read block size.
cache_type: S3 caching strategy.
max_workers: Maximum worker threads for parallel processing.
s3_max_workers: Maximum worker threads per S3 file reader.
result_reuse_enable: Enable query result reuse.
result_reuse_minutes: Result reuse duration in minutes.
auto_optimize_chunksize: Enable automatic chunksize determination for
Expand Down Expand Up @@ -133,7 +133,7 @@ def __init__(
self._chunksize = chunksize
self._block_size = block_size
self._cache_type = cache_type
self._max_workers = max_workers
self._s3_max_workers = s3_max_workers
self._auto_optimize_chunksize = auto_optimize_chunksize

@staticmethod
Expand Down Expand Up @@ -194,7 +194,7 @@ def execute(
:class:`~pyathena.options.ExecuteOptions` instance. Individual
keyword arguments take precedence over ``options`` fields.
**kwargs: Additional pandas read_csv/read_parquet parameters.
``engine``, ``chunksize``, ``block_size``, ``cache_type``, ``max_workers``,
``engine``, ``chunksize``, ``block_size``, ``cache_type``, ``s3_max_workers``,
and ``auto_optimize_chunksize`` override the cursor's values for this query.
``storage_options`` and, for UNLOAD results, ``filesystem`` replace
PyAthena's S3 filesystem (see
Expand Down Expand Up @@ -247,7 +247,7 @@ def execute(
chunksize=kwargs.pop("chunksize", self._chunksize),
block_size=kwargs.pop("block_size", self._block_size),
cache_type=kwargs.pop("cache_type", self._cache_type),
max_workers=kwargs.pop("max_workers", self._max_workers),
max_workers=kwargs.pop("s3_max_workers", self._s3_max_workers),
auto_optimize_chunksize=kwargs.pop(
"auto_optimize_chunksize", self._auto_optimize_chunksize
),
Expand Down
9 changes: 6 additions & 3 deletions pyathena/polars/async_cursor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ def __init__(
block_size: int | None = None,
cache_type: str | None = None,
chunksize: int | None = None,
s3_max_workers: int = (cpu_count() or 1) * 5,
**kwargs,
) -> None:
"""Initialize an AsyncPolarsCursor.
Expand All @@ -91,7 +92,7 @@ def __init__(
encryption_option: S3 encryption option (SSE_S3, SSE_KMS, CSE_KMS).
kms_key: KMS key ARN for encryption.
kill_on_interrupt: Cancel running query on keyboard interrupt.
max_workers: Maximum number of workers for concurrent execution.
max_workers: Size of the cursor thread pool for waiting and collecting results.
arraysize: Number of rows to fetch per batch.
unload: Enable UNLOAD for high-performance Parquet output.
result_reuse_enable: Enable Athena query result reuse.
Expand All @@ -101,6 +102,7 @@ def __init__(
chunksize: Number of rows per chunk for memory-efficient processing.
If specified, data is loaded lazily in chunks for all data
access methods including fetchone(), fetchmany(), and iter_chunks().
s3_max_workers: Maximum worker threads per S3 file reader.
**kwargs: Additional connection parameters.

Example:
Expand All @@ -123,6 +125,7 @@ def __init__(
result_reuse_minutes=result_reuse_minutes,
**kwargs,
)
self._s3_max_workers = s3_max_workers
self._unload = unload
self._block_size = block_size
self._cache_type = cache_type
Expand Down Expand Up @@ -186,7 +189,7 @@ def _collect_result_set(
unload_location=unload_location,
block_size=kwargs.pop("block_size", self._block_size),
cache_type=kwargs.pop("cache_type", self._cache_type),
max_workers=kwargs.pop("max_workers", self._max_workers),
max_workers=kwargs.pop("s3_max_workers", self._s3_max_workers),

@laughingman7743 laughingman7743 Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Self-review round one after scope reduction — FINDINGS (one claim correction, no code repair).

Base: f3532e9fb57fd769441e533f0b8375fd854ece69
Head: c120e3011a9d20dbf8f12e27765df0794da39973
Prior reviewed head: 8f1d1afc2db345bb56ac31b6c248001cd8afa973. The new commit removes the Arrow s3_max_workers path (dedicated S3 client, S3FileSystem s3_client argument, PyArrow fsspec adapter) and _validate_s3_max_workers, so this is a full pass over the 17-file diff rather than a range-diff follow-up.

Covered: constructor and execute() routing in all six pandas/Polars cursors (sync, thread-pool, native asyncio); AsyncPolarsCursor no longer forwards its pool size; AsyncPandasCursor's new constructor default equals the result-set default it used before; Arrow cursors identical to base except the AsyncArrowCursor docstring; benchmark adapter/tests; the six test_read_options changes. The thread-pool cases construct with max_workers=1, s3_max_workers=2 and assert both values, which fails against base AsyncPolarsCursor.

Finding (fixed in the PR body): the body said a former max_workers in execute() is rejected by the reader. An offline probe with mocked _execute/_poll showed all six cursors raise TypeError: ... got multiple values for keyword argument 'max_workers' when the result set (or asyncio.to_thread) is called, i.e. after the query has run. It never silently changes behavior, so no pre-check is added; the body now states this. Constructors ignore the former name until #1102 rejects unknown keywords.

Evidence: just format, just lint, just docs lint passed; 18 test_read_options cases across the nine cursors passed; just benchmark test 108 passed / 1 packaging skip. No AWS run for this head yet.

@laughingman7743 laughingman7743 Oct 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Round-one follow-up for the independent documentation finding — CLEAN. Old head c120e3011a9d20dbf8f12e27765df0794da39973, new head f316843f84e1dc839e1dd243e1ca3fa76b188c18, same base. The repair changes two lines of docs/usage.md only. Traced each Polars read path in pyathena/polars/result_set.py: _read_csv (non-chunked) passes _csv_storage_options with max_workers; _iter_csv_chunks (scan_csv) and the Parquet paths (read_parquet/scan_parquet) use _parquet_storage_options without it. No code changed; just docs lint passed.

@laughingman7743 laughingman7743 Oct 10, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Round-one follow-up after merging master — CLEAN after one documentation repair.
Old base/head f3532e9fb57fd769441e533f0b8375fd854ece69/f316843f84e1dc839e1dd243e1ca3fa76b188c18; new merge-base dafee9a4264ae2bb765cc7c5e1e3ef842bd6cfb8 (includes #1102 and #1107); merge commit 02331086a17e04a60895e792941f2aa58927b512; head 9df509eaa4a8432db13e62837d4613948352be76.

The merge conflicted only in docs/usage.md (#1102's unknown-keyword paragraph and this PR's worker section), resolved by keeping both. Code merged without conflicts. Checked upstream contracts touching the rename: BaseCursor without **kwargs, constructor signature caching, Connection._internal_cursor cursor_kwargs filtering, pandas/Polars execute() reader forwarding, SQLAlchemy dialect cursor options, and fixtures. None names the S3 max_workers or mishandles s3_max_workers.
Probe at the merge commit (mocked _execute/_poll): the four renamed constructors raise TypeError for max_workers; execute(max_workers=2) raises duplicate-keyword TypeError after _execute in all six cursors. I added this to docs/usage.md; independent review found the thread-pool wording imprecise, and 9df509eaa4a8432db13e62837d4613948352be76 fixes it.
Validation: just lint, just docs lint; 18 test_read_options; 218 connection/options tests; just benchmark test 108 passed / 1 skip.

chunksize=kwargs.pop("chunksize", self._chunksize),
result_set_type_hints=result_set_type_hints,
**kwargs,
Expand Down Expand Up @@ -231,7 +234,7 @@ def execute(
:class:`~pyathena.options.ExecuteOptions` instance. Individual
keyword arguments take precedence over ``options`` fields.
**kwargs: Additional execution parameters passed to Polars read functions.
``block_size``, ``cache_type``, ``max_workers``, and ``chunksize``
``block_size``, ``cache_type``, ``s3_max_workers``, and ``chunksize``
override the cursor's values for this query.
Read function arguments replace the ones the result set chooses, such as
``separator``, ``has_header``, ``schema_overrides``, and ``storage_options``
Expand Down
Loading
Loading