BREAKING: Reject unknown cursor keyword arguments

docs/usage.md

@@ -40,6 +40,12 @@ cursor.execute("SELECT %(value)s", {"value": 1}, work_group="YOUR_WORK_GROUP")
 Existing keyword arguments retain their defaults and behavior.
 The cursor class can still be selected positionally with `connection.cursor(DictCursor)`.
 
+Unknown keyword arguments raise `TypeError` in cursor constructors and in `execute()` instead of being ignored.
+The connection's `cursor_kwargs` are passed to every cursor that it creates, so the requested cursor class must accept them.
+A cursor subclass that receives settings of its own must not pass them on to the base class constructor.
+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.
+
 ## 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,

pyathena/aio/arrow/cursor.py

@@ -144,13 +144,15 @@ async def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Result-set overrides: ``block_size``, ``connect_timeout``,
+                and ``request_timeout``. Unknown names raise TypeError.
                 ``block_size`` sets the read block size for this query, and
                 ``connect_timeout`` and ``request_timeout`` override the cursor's values.
 
         Returns:
             Self reference for method chaining.
         """
+        self._validate_execute_kwargs(kwargs, ("block_size", "connect_timeout", "request_timeout"))
         self._reset_state()
         options = ExecuteOptions.resolve(
             options,

pyathena/aio/cursor.py

@@ -139,11 +139,12 @@ async def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Unknown keyword arguments raise TypeError.
 
         Returns:
             Self reference for method chaining.
         """
+        self._validate_execute_kwargs(kwargs)
         self._reset_state()
         options = ExecuteOptions.resolve(
             options,

pyathena/aio/s3fs/cursor.py

@@ -147,13 +147,14 @@ async def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Supported S3FS result-set overrides. Unknown names raise TypeError.
                 ``block_size`` sets the read block size for this query, and
                 ``csv_reader`` overrides the cursor's value.
 
         Returns:
             Self reference for method chaining.
         """
+        self._validate_execute_kwargs(kwargs, ("block_size", "csv_reader"))
         self._reset_state()
         options = ExecuteOptions.resolve(
             options,

pyathena/aio/spark/cursor.py

@@ -353,11 +353,12 @@ async def execute(
             description: Calculation description.
             client_request_token: Idempotency token.
             work_group: Unused, kept for API compatibility.
-            **kwargs: Additional parameters.
+            **kwargs: Unknown keyword arguments raise TypeError.
 
         Returns:
             Self reference for method chaining.
         """
+        self._validate_execute_kwargs(kwargs)
         # A failure below must not leave the previous calculation on the cursor.
         self._calculation_id = None
         self._calculation_execution = None

pyathena/aio/sqlalchemy/base.py

@@ -276,6 +276,20 @@ def cursor(self, cursor: Any = None, **kwargs: Any) -> AsyncAdapt_pyathena_curso
         raw_cursor = self._connection.cursor(_ASYNC_CURSOR_CLASSES.get(cursor, cursor), **kwargs)
         return AsyncAdapt_pyathena_cursor(raw_cursor)
 
+    def _internal_cursor(self, cursor: Any) -> AsyncAdapt_pyathena_cursor:
+        """Create a cursor with the wrapped connection's ``_internal_cursor()`` and adapt it.
+
+        A synchronous cursor class is replaced as in ``cursor()``.
+
+        Args:
+            cursor: The cursor class to create.
+
+        Returns:
+            The created cursor wrapped in ``AsyncAdapt_pyathena_cursor``.
+        """
+        raw_cursor = self._connection._internal_cursor(_ASYNC_CURSOR_CLASSES.get(cursor, cursor))
+        return AsyncAdapt_pyathena_cursor(raw_cursor)
+
     def close(self) -> None:
         """Close the wrapped connection."""
         self._connection.close()

pyathena/arrow/async_cursor.py

@@ -212,13 +212,15 @@ def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Result-set overrides: ``block_size``, ``connect_timeout``,
+                and ``request_timeout``. Unknown names raise TypeError.
                 ``block_size`` sets the read block size for this query, and
                 ``connect_timeout`` and ``request_timeout`` override the cursor's values.
 
         Returns:
             Tuple of (query_id, future) where future resolves to AthenaArrowResultSet.
         """
+        self._validate_execute_kwargs(kwargs, ("block_size", "connect_timeout", "request_timeout"))
         options = ExecuteOptions.resolve(
             options,
             work_group=work_group,

pyathena/arrow/cursor.py

@@ -170,7 +170,8 @@ def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Result-set overrides: ``block_size``, ``connect_timeout``,
+                and ``request_timeout``. Unknown names raise TypeError.
                 ``block_size`` sets the read block size for this query, and
                 ``connect_timeout`` and ``request_timeout`` override the cursor's values.
 
@@ -181,6 +182,7 @@ def execute(
             >>> cursor.execute("SELECT * FROM sales WHERE year = 2023")
             >>> table = cursor.as_arrow()  # Returns Apache Arrow Table
         """
+        self._validate_execute_kwargs(kwargs, ("block_size", "connect_timeout", "request_timeout"))
         self._reset_state()
         options = ExecuteOptions.resolve(
             options,

pyathena/async_cursor.py

@@ -241,7 +241,7 @@ def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Unknown keyword arguments raise TypeError.
 
         Returns:
             Tuple of (query_id, future) where:
@@ -254,6 +254,7 @@ def execute(
             >>> # Do other work while query runs...
             >>> result_set = future.result()  # Wait for completion
         """
+        self._validate_execute_kwargs(kwargs)
         options = ExecuteOptions.resolve(
             options,
             work_group=work_group,

pyathena/common.py

@@ -7,9 +7,10 @@
 import threading
 import time
 from abc import ABCMeta, abstractmethod
-from collections.abc import Callable
+from collections.abc import Callable, Collection
 from concurrent.futures import Future, wait
 from datetime import UTC, datetime, timedelta
+from inspect import Parameter, signature
 from typing import TYPE_CHECKING, Any, TypeVar, cast
 
 from botocore.exceptions import BotoCoreError, ClientError
@@ -212,7 +213,6 @@ def __init__(
         result_reuse_minutes: int,
         on_start_query_execution: Callable[[str], None] | None = None,
         on_poll: OnPollCallback | None = None,
-        **kwargs,
     ) -> None:
         """Initialize the cursor with the settings it uses to run queries.
 
@@ -236,7 +236,6 @@ def __init__(
                 waits for the query, by cursors whose ``execute()`` supports it.
             on_poll: Callback invoked once per poll iteration with the current
                 execution object.
-            **kwargs: Ignored.
         """
         super().__init__()
         self._connection = connection
@@ -272,11 +271,48 @@ def get_default_converter(unload: bool = False) -> DefaultTypeConverter | Any:
         """
         return DefaultTypeConverter()
 
+    @classmethod
+    def _constructor_keyword_names(cls) -> set[str]:
+        """Get the keyword names accepted by the constructors in this class's MRO.
+
+        Returns:
+            The named parameters of each ``__init__`` defined in the MRO,
+            excluding ``self``.
+        """
+        names: set[str] = set()
+        for base in cls.__mro__:
+            if "__init__" in vars(base):
+                names.update(
+                    name
+                    for name, parameter in signature(vars(base)["__init__"]).parameters.items()
+                    if parameter.kind in (Parameter.POSITIONAL_OR_KEYWORD, Parameter.KEYWORD_ONLY)
+                )
+        names.discard("self")
+        return names
+
     @property
     def connection(self) -> Connection[Any]:
         """The connection that created this cursor."""
         return self._connection
 
+    def _validate_execute_kwargs(
+        self, kwargs: dict[str, Any], allowed: Collection[str] = ()
+    ) -> None:
+        """Reject the keyword arguments that ``execute()`` does not support.
+
+        Args:
+            kwargs: The extra keyword arguments given to ``execute()``.
+            allowed: The extra keyword names that this cursor supports.
+
+        Raises:
+            TypeError: If a keyword name is not in ``allowed``.
+        """
+        for name in kwargs:
+            if name not in allowed:
+                raise TypeError(
+                    f"{type(self).__name__}.execute() got an unexpected keyword argument '{name}'"
+                )
+
     def _build_start_query_execution_request(
         self,
         query: str,

pyathena/connection.py

@@ -612,7 +612,21 @@ def cursor(
             >>> pandas_cursor = connection.cursor(PandasCursor)
             >>> df = pandas_cursor.execute("SELECT * FROM my_table").fetchall()
         """
-        kwargs = {**self.cursor_kwargs, **kwargs}
+        return self._cursor(cursor, **{**self.cursor_kwargs, **kwargs})
+
+    def _cursor(
+        self, cursor: type[FunctionalCursor] | None, /, **kwargs: Any
+    ) -> FunctionalCursor | ConnectionCursor:
+        """Create a cursor without applying the connection's ``cursor_kwargs``.
+
+        Args:
+            cursor: The cursor class. If None, the connection's default cursor class.
+            **kwargs: The cursor constructor arguments. Omitted connection settings
+                default to the connection's values.
+
+        Returns:
+            The created cursor.
+        """
         _cursor = cursor or self.cursor_class
         converter = kwargs.pop("converter", self._converter)
         if not converter:
@@ -639,6 +653,26 @@ def cursor(
             **kwargs,
         )
 
+    def _internal_cursor(self, cursor: type[FunctionalCursor]) -> FunctionalCursor:
+        """Create a cursor for queries that PyAthena runs on its own behalf.
+
+        Only the connection's ``cursor_kwargs`` that the cursor class accepts are
+        applied, leaving out options of the default cursor class such as ``unload``
+        of a DataFrame cursor. ``arraysize`` is left out too, since a DataFrame
+        cursor allows values above this cursor's page size limit. The cursor
+        class's default converter is used.
+
+        Args:
+            cursor: The cursor class.
+
+        Returns:
+            The created cursor.
+        """
+        accepted = cursor._constructor_keyword_names() - {"arraysize"}
+        kwargs = {k: v for k, v in self.cursor_kwargs.items() if k in accepted}
+        kwargs["converter"] = cursor.get_default_converter()
+        return cast(FunctionalCursor, self._cursor(cursor, **kwargs))
+
     def close(self) -> None:
         """Close the connection.
 

pyathena/cursor.py

@@ -149,7 +149,7 @@ def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Unknown keyword arguments raise TypeError.
 
         Returns:
             Self reference for method chaining.
@@ -163,6 +163,7 @@ def execute(
             ...     }
             ... )
         """
+        self._validate_execute_kwargs(kwargs)
         self._reset_state()
         options = ExecuteOptions.resolve(
             options,

pyathena/s3fs/async_cursor.py

@@ -215,9 +215,10 @@ def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Supported S3FS result-set overrides. Unknown names raise TypeError.
                 ``block_size`` sets the read block size for this query, and
                 ``csv_reader`` overrides the cursor's value.
+                ``filesystem_class`` overrides the filesystem used to read results.
 
         Returns:
             Tuple of (query_id, Future[AthenaS3FSResultSet]).
@@ -227,6 +228,7 @@ def execute(
             >>> result_set = future.result()
             >>> rows = result_set.fetchall()
         """
+        self._validate_execute_kwargs(kwargs, ("block_size", "csv_reader", "filesystem_class"))
         options = ExecuteOptions.resolve(
             options,
             work_group=work_group,

pyathena/s3fs/cursor.py

@@ -165,9 +165,10 @@ def execute(
             options: Shared execution options as an
                 :class:`~pyathena.options.ExecuteOptions` instance. Individual
                 keyword arguments take precedence over ``options`` fields.
-            **kwargs: Additional execution parameters.
+            **kwargs: Supported S3FS result-set overrides. Unknown names raise TypeError.
                 ``block_size`` sets the read block size for this query, and
                 ``csv_reader`` overrides the cursor's value.
+                ``filesystem_class`` overrides the filesystem used to read results.
 
         Returns:
             Self reference for method chaining.
@@ -176,6 +177,7 @@ def execute(
             >>> cursor.execute("SELECT * FROM my_table WHERE id = %(id)s", {"id": 123})
             >>> rows = cursor.fetchall()
         """
+        self._validate_execute_kwargs(kwargs, ("block_size", "csv_reader", "filesystem_class"))
         self._reset_state()
         options = ExecuteOptions.resolve(
             options,

pyathena/spark/async_cursor.py

@@ -212,6 +212,7 @@ def execute(
         work_group: str | None = None,
         **kwargs,
     ) -> tuple[str, "Future[AthenaQueryExecution | AthenaCalculationExecution]"]:
+        self._validate_execute_kwargs(kwargs)
         calculation_id = self._calculate(
             session_id=session_id if session_id else self._session_id,
             code_block=operation,

pyathena/spark/cursor.py

@@ -111,6 +111,7 @@ def execute(
         work_group: str | None = None,
         **kwargs,
     ) -> SparkCursor:
+        self._validate_execute_kwargs(kwargs)
         # A failure below must not leave the previous calculation on the cursor.
         self._calculation_id = None
         self._calculation_execution = None

pyathena/sqlalchemy/base.py

@@ -536,9 +536,7 @@ def _internal_cursor(raw_connection: PoolProxiedConnection) -> Any:
         returns its wrapper, so this is typed by the interface used here rather
         than by the class requested.
         """
-        return raw_connection.driver_connection.cursor(  # type: ignore[union-attr]
-            Cursor, converter=Cursor.get_default_converter()
-        )
+        return raw_connection.driver_connection._internal_cursor(Cursor)  # type: ignore[union-attr]
 
     def _column(self, name: str | None, type_: str, comment: str | None, partition: bool | None):
         return {