Add native asyncio specialized cursors and DRY refactors (Phase 2)

Phase 2 of native asyncio cursor support:

- DRY: Add _pre_fetch flag to AthenaResultSet.__init__ to eliminate
  duplicated field initialization in AthenaAioResultSet
- DRY: Extract _prepare_query() from BaseCursor._execute so both
  sync and async _execute share non-I/O query preparation logic
- Add async metadata operations (list_databases, get_table_metadata,
  list_table_metadata) to AioBaseCursor
- Add AioPandasCursor, AioArrowCursor, AioPolarsCursor using
  asyncio.to_thread() for result set creation (no new result set
  classes needed since fetch methods are in-memory only)
- Add comprehensive tests for all new cursors and metadata ops

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: laughingman7743

pyathena/aio/common.py

@@ -5,13 +5,12 @@
 import logging
 import sys
 from datetime import datetime, timedelta, timezone
-from typing import Any, Dict, List, Optional, Tuple, Union, cast
+from typing import Any, Dict, List, Optional, Tuple, Union
 
-import pyathena
 from pyathena.aio.util import async_retry_api_call
 from pyathena.common import BaseCursor
 from pyathena.error import DatabaseError, OperationalError
-from pyathena.model import AthenaQueryExecution
+from pyathena.model import AthenaDatabase, AthenaQueryExecution, AthenaTableMetadata
 
 _logger = logging.getLogger(__name__)  # type: ignore
 
@@ -36,13 +35,7 @@ async def _execute(  # type: ignore[override]
         result_reuse_minutes: Optional[int] = None,
         paramstyle: Optional[str] = None,
     ) -> str:
-        if pyathena.paramstyle == "qmark" or paramstyle == "qmark":
-            query = operation
-            execution_parameters = cast(Optional[List[str]], parameters)
-        else:
-            query = self._formatter.format(operation, cast(Optional[Dict[str, Any]], parameters))
-            execution_parameters = None
-        _logger.debug(query)
+        query, execution_parameters = self._prepare_query(operation, parameters, paramstyle)
 
         request = self._build_start_query_execution_request(
             query=query,
@@ -216,3 +209,142 @@ async def _find_previous_query_id(  # type: ignore[override]
         except Exception:
             _logger.warning("Failed to check the cache. Moving on without cache.", exc_info=True)
         return query_id
+
+    async def _list_databases(  # type: ignore[override]
+        self,
+        catalog_name: Optional[str],
+        next_token: Optional[str] = None,
+        max_results: Optional[int] = None,
+    ) -> Tuple[Optional[str], List[AthenaDatabase]]:
+        request = self._build_list_databases_request(
+            catalog_name=catalog_name,
+            next_token=next_token,
+            max_results=max_results,
+        )
+        try:
+            response = await async_retry_api_call(
+                self.connection._client.list_databases,
+                config=self._retry_config,
+                logger=_logger,
+                **request,
+            )
+        except Exception as e:
+            _logger.exception("Failed to list databases.")
+            raise OperationalError(*e.args) from e
+        else:
+            return response.get("NextToken"), [
+                AthenaDatabase({"Database": r}) for r in response.get("DatabaseList", [])
+            ]
+
+    async def list_databases(  # type: ignore[override]
+        self,
+        catalog_name: Optional[str],
+        max_results: Optional[int] = None,
+    ) -> List[AthenaDatabase]:
+        databases: List[AthenaDatabase] = []
+        next_token = None
+        while True:
+            next_token, response = await self._list_databases(
+                catalog_name=catalog_name,
+                next_token=next_token,
+                max_results=max_results,
+            )
+            databases.extend(response)
+            if not next_token:
+                break
+        return databases
+
+    async def _get_table_metadata(  # type: ignore[override]
+        self,
+        table_name: str,
+        catalog_name: Optional[str] = None,
+        schema_name: Optional[str] = None,
+        logging_: bool = True,
+    ) -> AthenaTableMetadata:
+        request: Dict[str, Any] = {
+            "CatalogName": catalog_name if catalog_name else self._catalog_name,
+            "DatabaseName": schema_name if schema_name else self._schema_name,
+            "TableName": table_name,
+        }
+        if self._work_group:
+            request.update({"WorkGroup": self._work_group})
+        try:
+            response = await async_retry_api_call(
+                self._connection.client.get_table_metadata,
+                config=self._retry_config,
+                logger=_logger,
+                **request,
+            )
+        except Exception as e:
+            if logging_:
+                _logger.exception("Failed to get table metadata.")
+            raise OperationalError(*e.args) from e
+        else:
+            return AthenaTableMetadata(response)
+
+    async def get_table_metadata(  # type: ignore[override]
+        self,
+        table_name: str,
+        catalog_name: Optional[str] = None,
+        schema_name: Optional[str] = None,
+        logging_: bool = True,
+    ) -> AthenaTableMetadata:
+        return await self._get_table_metadata(
+            table_name=table_name,
+            catalog_name=catalog_name,
+            schema_name=schema_name,
+            logging_=logging_,
+        )
+
+    async def _list_table_metadata(  # type: ignore[override]
+        self,
+        catalog_name: Optional[str] = None,
+        schema_name: Optional[str] = None,
+        expression: Optional[str] = None,
+        next_token: Optional[str] = None,
+        max_results: Optional[int] = None,
+    ) -> Tuple[Optional[str], List[AthenaTableMetadata]]:
+        request = self._build_list_table_metadata_request(
+            catalog_name=catalog_name,
+            schema_name=schema_name,
+            expression=expression,
+            next_token=next_token,
+            max_results=max_results,
+        )
+        try:
+            response = await async_retry_api_call(
+                self.connection._client.list_table_metadata,
+                config=self._retry_config,
+                logger=_logger,
+                **request,
+            )
+        except Exception as e:
+            _logger.exception("Failed to list table metadata.")
+            raise OperationalError(*e.args) from e
+        else:
+            return response.get("NextToken"), [
+                AthenaTableMetadata({"TableMetadata": r})
+                for r in response.get("TableMetadataList", [])
+            ]
+
+    async def list_table_metadata(  # type: ignore[override]
+        self,
+        catalog_name: Optional[str] = None,
+        schema_name: Optional[str] = None,
+        expression: Optional[str] = None,
+        max_results: Optional[int] = None,
+    ) -> List[AthenaTableMetadata]:
+        metadata: List[AthenaTableMetadata] = []
+        next_token = None
+        while True:
+            next_token, response = await self._list_table_metadata(
+                catalog_name=catalog_name,
+                schema_name=schema_name,
+                expression=expression,
+                next_token=next_token,
+                max_results=max_results,
+            )
+            metadata.extend(response)
+            if not next_token:
+                break
+        return metadata

pyathena/aio/result_set.py

@@ -1,12 +1,10 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-import collections
 import logging
 from typing import (
     TYPE_CHECKING,
     Any,
-    Deque,
     Dict,
     List,
     Optional,
@@ -16,7 +14,6 @@
 )
 
 from pyathena.aio.util import async_retry_api_call
-from pyathena.common import CursorIterator
 from pyathena.converter import Converter
 from pyathena.error import OperationalError, ProgrammingError
 from pyathena.model import AthenaQueryExecution
@@ -32,9 +29,9 @@
 class AthenaAioResultSet(AthenaResultSet):
     """Async result set that provides async fetch methods.
 
-    Because ``AthenaResultSet.__init__`` calls ``_pre_fetch`` (a blocking API
-    call), this class overrides ``__init__`` to skip it and provides an
-    ``async create()`` classmethod factory instead.
+    Skips the synchronous ``_pre_fetch`` by passing ``_pre_fetch=False`` to
+    the parent ``__init__`` and provides an ``async create()`` classmethod
+    factory instead.
     """
 
     def __init__(
@@ -45,31 +42,15 @@ def __init__(
         arraysize: int,
         retry_config: RetryConfig,
     ) -> None:
-        # Replicate parent field initialization without calling _pre_fetch.
-        CursorIterator.__init__(self, arraysize=arraysize)
-        self._connection: Optional["Connection[Any]"] = connection
-        self._converter = converter
-        self._query_execution: Optional[AthenaQueryExecution] = query_execution
-        if not self._query_execution:
-            raise ProgrammingError("Required argument `query_execution` not found.")
-        self._retry_config = retry_config
-        self._client = connection.session.client(
-            "s3",
-            region_name=connection.region_name,
-            config=connection.config,
-            **connection._client_kwargs,
+        super().__init__(
+            connection=connection,
+            converter=converter,
+            query_execution=query_execution,
+            arraysize=arraysize,
+            retry_config=retry_config,
+            _pre_fetch=False,
         )
 
-        self._metadata: Optional[Tuple[Dict[str, Any], ...]] = None
-        self._rows: Deque[Union[Tuple[Optional[Any], ...], Dict[Any, Optional[Any]]]] = (
-            collections.deque()
-        )
-        self._next_token: Optional[str] = None
-
-        if self.state == AthenaQueryExecution.STATE_SUCCEEDED:
-            self._rownumber = 0
-            # NOTE: _pre_fetch is NOT called here; use create() instead.
-
     @classmethod
     async def create(
         cls,