Outlast metadata throttling with jittered retries and an existence fallback

Metadata API throttling episodes observed in CI and in a controlled probe last
tens of seconds, while the default retry policy waited about 15 seconds in
total and clients retried in lockstep.

- Add uniform jitter of up to one multiplier to retry_api_call waits and raise
  the default RetryConfig attempts to seven, so the exponential waits sum to
  63 seconds plus jitter.
- Expose is_retryable_error and reuse it in the dialect: when a table-metadata
  request is still throttled after PyAthena's retries, has_table() determines
  existence with information_schema.tables and logs a warning. The fallback
  answers existence only; column, comment and table-option reflection still
  propagate the throttling error, and nothing is cached from the fallback.
- Expose retry_config on the async connection adapter.
- Cover the fallback for direct and wrapped throttling codes on existing and
  missing tables, keep the error-propagation regression for access denial and
  unrecognized messages, and add wait-bound and jitter unit tests.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: laughingman7743

docs/sqlalchemy.md

@@ -98,13 +98,18 @@ A later listing preserves metadata already fetched for a table.
 `clear_cache()` also discards this metadata; an absent entry in a listing is not cached as proof that a table does not exist.
 
 Table-metadata lookups propagate throttling and permission errors rather than reporting missing tables.
-`has_table()` propagates these failures, including access denied by Lake Formation, instead of returning or caching `False`.
+`has_table()` propagates permission failures, including access denied by Lake Formation, instead of returning or caching `False`.
 For failed metadata requests, only recognized `EntityNotFoundException` responses establish absence; unrecognized errors are propagated rather than guessed to mean a missing table.
+When a table-metadata request is still throttled after PyAthena's retries, `has_table()` determines existence with a query on `information_schema.tables` and logs a warning.
+This fallback answers only existence: it does not populate the metadata cache, and column, comment, and table-option reflection still propagate the throttling error.
+Athena applies its metadata API rate limits per account; they are not listed in Service Quotas, and throttling episodes can last tens of seconds.
+PyAthena's API retries use exponential backoff with uniform jitter; the default `RetryConfig` makes seven attempts whose waits sum to 63 seconds plus jitter.
 PyAthena recognizes Glue error codes in Athena's `MetadataException` service-error envelope and applies `RetryConfig.exceptions` to those codes.
 `RetryConfig` accepts one exception-name string or an iterable and captures the names as a tuple at construction.
 Changes to the original input list or iterator no longer change the stored policy; construct a new `RetryConfig` when changing the retry policy.
 SDK retries and PyAthena retries are separate layers, so increasing both attempt limits can multiply requests and waiting time.
 Adaptive SDK retries regulate individual clients, not the aggregate traffic from independent CI runners.
+For highly concurrent reflection or `checkfirst` DDL, bound the concurrency of metadata requests and consider `botocore.config.Config(retries={"mode": "adaptive"})` on the connection.
 
 Use SQLAlchemy's identifier quoting for reserved words or names beginning with an underscore.
 The dialect uses backticks for table DDL and double quotes for DML.

pyathena/aio/sqlalchemy/base.py

@@ -22,6 +22,7 @@
     ProgrammingError,
 )
 from pyathena.sqlalchemy.base import AthenaDialect
+from pyathena.util import RetryConfig
 
 if TYPE_CHECKING:
     from types import ModuleType
@@ -142,6 +143,10 @@ def schema_name(self) -> str | None:
     def cursor_kwargs(self) -> dict[str, Any]:
         return self._connection.cursor_kwargs  # type: ignore[no-any-return]
 
+    @property
+    def retry_config(self) -> RetryConfig:
+        return self._connection.retry_config  # type: ignore[no-any-return]
+
     def cursor(self) -> AsyncAdapt_pyathena_cursor:
         raw_cursor = self._connection.cursor()
         return AsyncAdapt_pyathena_cursor(raw_cursor)

pyathena/sqlalchemy/base.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import contextlib
+import logging
 import re
 from collections.abc import Mapping, MutableMapping
 from re import Pattern
@@ -37,7 +38,7 @@
     get_double_type,
 )
 from pyathena.sqlalchemy.util import _HashableDict
-from pyathena.util import _get_error_code, strtobool
+from pyathena.util import _get_error_code, is_retryable_error, strtobool
 
 if TYPE_CHECKING:
     from types import ModuleType
@@ -55,6 +56,8 @@
     )
     from sqlalchemy.sql.schema import SchemaItem
 
+_logger = logging.getLogger(__name__)
+
 
 ischema_names: dict[str, type[Any]] = {
     "boolean": types.BOOLEAN,
@@ -384,6 +387,40 @@ def has_table(self, connection: Connection, table_name: str, schema: str | None
             return bool(columns)
         except exc.NoSuchTableError:
             return False
+        except pyathena.error.OperationalError as e:
+            cause = e.__cause__
+            raw_connection = self._raw_connection(connection)
+            if not (
+                isinstance(cause, botocore.exceptions.ClientError)
+                and is_retryable_error(cause, raw_connection.retry_config)  # type: ignore[union-attr]
+            ):
+                raise
+            # The metadata API is still throttled after PyAthena's retries.
+            # Existence can be answered by a query; column, comment and option
+            # reflection cannot, so only this check falls back.
+            _logger.warning(
+                f"Table metadata request for {table_name} was throttled; "
+                "checking existence with information_schema."
+            )
+            return self._has_table_information_schema(connection, table_name, schema)
+
+    def _has_table_information_schema(
+        self, connection: Connection, table_name: str, schema: str | None = None
+    ) -> bool:
+        raw_connection = self._raw_connection(connection)
+        schema = schema if schema else raw_connection.schema_name  # type: ignore[union-attr]
+        catalog = raw_connection.cursor_kwargs.get("catalog_name", raw_connection.catalog_name)
+        name = (
+            str(table_name).lower()
+            if (catalog or "").lower() == "awsdatacatalog"
+            else str(table_name)
+        )
+        query = text(
+            "SELECT table_name FROM information_schema.tables "
+            "WHERE table_schema = :schema AND table_name = :table_name"
+        )
+        rows = connection.execute(query, {"schema": schema, "table_name": name}).fetchall()
+        return bool(rows)
 
     @reflection.cache
     def get_view_definition(

pyathena/util.py

@@ -7,7 +7,13 @@
 from typing import Any
 
 import tenacity
-from tenacity import after_log, retry_if_exception, stop_after_attempt, wait_exponential
+from tenacity import (
+    after_log,
+    retry_if_exception,
+    stop_after_attempt,
+    wait_exponential,
+    wait_random,
+)
 
 from pyathena import DataError
 
@@ -86,9 +92,11 @@ class RetryConfig:
 
     Attributes:
         exceptions: Tuple of AWS exception names to retry on.
-        attempt: Maximum number of retry attempts.
-        multiplier: Base multiplier for exponential backoff.
-        max_delay: Maximum delay between retries in seconds.
+        attempt: Maximum number of attempts, including the first call.
+        multiplier: Base multiplier for exponential backoff in seconds. Each
+            wait also adds uniform random jitter of up to one multiplier.
+        max_delay: Maximum exponential delay between retries in seconds,
+            before jitter is added.
         exponential_base: Base for exponential backoff calculation.
 
     Example:
@@ -116,6 +124,10 @@ class RetryConfig:
         original iterable do not change this configuration.
         Retries are applied to AWS API calls, not to SQL query execution.
         Query failures typically require manual intervention or query fixes.
+        With the default settings, the exponential waits between attempts sum
+        to 63 seconds plus jitter, which outlasts the metadata API throttling
+        episodes observed under concurrent reflection. SDK retries configured
+        on the boto3 client are a separate layer applied within each attempt.
         Recognized Glue error codes wrapped in Athena MetadataException are
         matched against exceptions in the same way as direct AWS error codes.
     """
@@ -126,7 +138,7 @@ def __init__(
             "ThrottlingException",
             "TooManyRequestsException",
         ),
-        attempt: int = 5,
+        attempt: int = 7,
         multiplier: int = 1,
         max_delay: int = 100,
         exponential_base: int = 2,
@@ -159,6 +171,23 @@ def _get_error_code(ex: BaseException, unwrap_metadata: bool = False) -> str | N
     return code if isinstance(code, str) else None
 
 
+def is_retryable_error(ex: BaseException, config: RetryConfig) -> bool:
+    """Return whether an exception matches the retry policy's AWS error codes.
+
+    Args:
+        ex: The exception raised by an AWS API call.
+        config: RetryConfig whose ``exceptions`` list the retryable error codes.
+
+    Returns:
+        True if the direct error code, or a recognized Glue error code wrapped in
+        an Athena ``MetadataException``, is listed in ``config.exceptions``.
+    """
+    return any(
+        code is not None and code in config.exceptions
+        for code in (_get_error_code(ex), _get_error_code(ex, unwrap_metadata=True))
+    )
+
+
 def retry_api_call(
     func: Callable[..., Any],
     config: RetryConfig,
@@ -169,8 +198,8 @@ def retry_api_call(
     """Execute a function with automatic retry logic for AWS API calls.
 
     This function wraps AWS API calls with retry behavior based on the provided
-    configuration. It uses exponential backoff and only retries on specific
-    AWS exceptions that indicate transient failures.
+    configuration. It uses exponential backoff with uniform jitter and only
+    retries on specific AWS exceptions that indicate transient failures.
 
     Args:
         func: The AWS API function to call.
@@ -201,20 +230,17 @@ def retry_api_call(
         Other errors are propagated without retrying.
     """
 
-    def _is_retryable(ex: BaseException) -> bool:
-        return any(
-            code is not None and code in config.exceptions
-            for code in (_get_error_code(ex), _get_error_code(ex, unwrap_metadata=True))
-        )
-
     retry = tenacity.Retrying(
-        retry=retry_if_exception(_is_retryable),
+        retry=retry_if_exception(lambda ex: is_retryable_error(ex, config)),
         stop=stop_after_attempt(config.attempt),
+        # Uniform jitter of up to one multiplier keeps concurrent clients from
+        # retrying in lockstep after a shared throttling response.
         wait=wait_exponential(
             multiplier=config.multiplier,
             max=config.max_delay,
             exp_base=config.exponential_base,
-        ),
+        )
+        + wait_random(0, config.multiplier),
         after=after_log(logger, logger.getEffectiveLevel()) if logger else None,  # type: ignore[arg-type]
         reraise=True,
     )

tests/pyathena/test_util.py

@@ -1,10 +1,17 @@
 from typing import Any
 
 import pytest
+import tenacity.nap
 from botocore.exceptions import ClientError
 
 from pyathena import DataError
-from pyathena.util import RetryConfig, parse_output_location, retry_api_call, strtobool
+from pyathena.util import (
+    RetryConfig,
+    is_retryable_error,
+    parse_output_location,
+    retry_api_call,
+    strtobool,
+)
 
 
 def test_parse_output_location():
@@ -160,3 +167,82 @@ def call():
 
         assert retry_api_call(call, config) == "success"
         assert calls == 3
+
+
+def _throttling_error() -> ClientError:
+    return ClientError(
+        {"Error": {"Code": "ThrottlingException", "Message": "Rate exceeded"}},
+        "GetTableMetadata",
+    )
+
+
+def test_retry_config_default_attempts():
+    assert RetryConfig().attempt == 7
+
+
+@pytest.mark.parametrize(
+    ("multiplier", "max_delay", "expected_bases"),
+    [
+        (1, 100, [1, 2, 4, 8, 16, 32]),
+        (2, 10, [2, 4, 8, 10, 10, 10]),
+        (0, 0, [0, 0, 0, 0, 0, 0]),
+    ],
+)
+def test_retry_api_call_waits_with_jitter(monkeypatch, multiplier, max_delay, expected_bases):
+    sleeps = []
+    monkeypatch.setattr(tenacity.nap.time, "sleep", sleeps.append)
+    error = _throttling_error()
+    calls = 0
+
+    def call():
+        nonlocal calls
+        calls += 1
+        raise error
+
+    config = RetryConfig(attempt=7, multiplier=multiplier, max_delay=max_delay)
+    with pytest.raises(ClientError):
+        retry_api_call(call, config)
+    assert calls == 7
+    assert len(sleeps) == len(expected_bases)
+    for waited, base in zip(sleeps, expected_bases, strict=True):
+        assert base <= waited <= base + multiplier
+
+
+def test_retry_api_call_jitter_varies(monkeypatch):
+    sleeps = []
+    monkeypatch.setattr(tenacity.nap.time, "sleep", sleeps.append)
+    error = _throttling_error()
+
+    def call():
+        raise error
+
+    config = RetryConfig(attempt=20, multiplier=1, max_delay=1)
+    with pytest.raises(ClientError):
+        retry_api_call(call, config)
+    assert all(1 <= waited <= 2 for waited in sleeps)
+    assert len(set(sleeps)) > 1
+
+
+@pytest.mark.parametrize(
+    ("code", "message", "expected"),
+    [
+        ("ThrottlingException", "Rate exceeded", True),
+        (
+            "MetadataException",
+            "Rate exceeded (Service: AmazonDataCatalog; Status Code: 400; "
+            "Error Code: ThrottlingException; Request ID: example; Proxy: null)",
+            True,
+        ),
+        (
+            "MetadataException",
+            "Not authorized (Service: AmazonDataCatalog; Status Code: 400; "
+            "Error Code: AccessDeniedException; Request ID: example; Proxy: null)",
+            False,
+        ),
+        ("MetadataException", "Table not found", False),
+    ],
+)
+def test_is_retryable_error(code, message, expected):
+    error = ClientError({"Error": {"Code": code, "Message": message}}, "GetTableMetadata")
+    assert is_retryable_error(error, RetryConfig()) is expected
+    assert is_retryable_error(ValueError("no response"), RetryConfig()) is False

tests/sqlalchemy/test_suite.py

@@ -264,7 +264,47 @@ def test_long_convention_name(self, type_, metadata, connection):
 
 
 class HasTableTest(_HasTableTest):
-    @sa_testing.combinations("AccessDeniedException", "ThrottlingException", None, argnames="code")
+    @sa_testing.combinations(True, False, argnames="wrapped")
+    @sa_testing.combinations(True, False, argnames="exists")
+    def test_throttled_existence_check_uses_information_schema(
+        self, connection, metadata, monkeypatch, wrapped, exists
+    ):
+        table = Table("throttled_existence", metadata, Column("id", Integer))
+        table.create(connection)
+        raw_connection = connection.connection.driver_connection
+        if connection.dialect.is_async:
+            raw_connection = raw_connection.driver_connection
+        error = ClientError(
+            {
+                "Error": {
+                    "Code": "MetadataException" if wrapped else "ThrottlingException",
+                    "Message": "Rate exceeded (Service: AmazonDataCatalog; Status Code: 400; "
+                    "Error Code: ThrottlingException; Request ID: example; Proxy: null)"
+                    if wrapped
+                    else "Rate exceeded",
+                }
+            },
+            "GetTableMetadata",
+        )
+        calls = []
+
+        def fail_metadata(**kwargs):
+            calls.append(kwargs)
+            raise error
+
+        monkeypatch.setattr(raw_connection.client, "get_table_metadata", fail_metadata)
+        monkeypatch.setattr(raw_connection.retry_config, "attempt", 1)
+        inspector = inspect(connection)
+        name = table.name if exists else "throttled_missing"
+        assert inspector.has_table(name) is exists
+        assert inspector.has_table(name.upper(), schema=raw_connection.schema_name) is exists
+        assert len(calls) == 2
+        # Only existence falls back; reflection still reports the throttled request.
+        with pytest.raises(OperationalError) as caught:
+            inspector.get_columns(table.name)
+        assert caught.value.__cause__ is error
+
+    @sa_testing.combinations("AccessDeniedException", None, argnames="code")
     def test_metadata_errors_do_not_establish_absence(self, connection, monkeypatch, code):
         raw_connection = connection.connection.driver_connection
         if connection.dialect.is_async: