fix(athena): skip retry of deterministic errors with configurable timeout handling

dbt-athena/.changes/unreleased/Fixes-20260330-095539.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Skip retry of deterministic errors with configurable timeout handling.
+time: 2026-03-30T09:55:39.69696+09:00
+custom:
+    Author: dtaniwaki
+    Issue: "1813 1820"

dbt-athena/src/dbt/adapters/athena/connections.py

@@ -75,6 +75,7 @@ class AthenaCredentials(Credentials):
     # Credentials in profile "athena", target "athena" invalid: Unable to create schema for 'dict'
     seed_s3_upload_args: Optional[Dict[str, Any]] = None
     lf_tags_database: Optional[Dict[str, str]] = None
+    skip_retry_on_query_timeout: bool = False
 
     @property
     def type(self) -> str:
@@ -107,6 +108,7 @@ def _connection_keys(self) -> Tuple[str, ...]:
             "seed_s3_upload_args",
             "lf_tags_database",
             "spark_work_group",
+            "skip_retry_on_query_timeout",
         )
 
 
@@ -168,12 +170,18 @@ def execute(
         @retry(
             # No need to retry if TOO_MANY_OPEN_PARTITIONS occurs.
             # Otherwise, Athena throws ICEBERG_FILESYSTEM_ERROR after retry,
-            # because not all files are removed immediately after first try to create table
+            # because not all files are removed immediately after first try to create table.
+            # Also skip retry on non-transient errors:
+            # - Query timeout: opt-in via skip_retry_on_query_timeout profile config.
+            # - Query exhausted resources: always skip (deterministic failure).
             retry=retry_if_exception(
                 lambda e: (
-                    False
-                    if catch_partitions_limit and "TOO_MANY_OPEN_PARTITIONS" in str(e)
-                    else True
+                    not (catch_partitions_limit and "TOO_MANY_OPEN_PARTITIONS" in str(e))
+                    and not (
+                        self.connection.cursor_kwargs.get("skip_retry_on_query_timeout", False)
+                        and "Query timeout" in str(e)
+                    )
+                    and "Query exhausted resources" not in str(e)
                 )
             ),
             stop=stop_after_attempt(self._retry_config.attempt),
@@ -274,6 +282,7 @@ def open(cls, connection: Connection) -> Connection:
                 cursor_kwargs={
                     "debug_query_state": creds.debug_query_state,
                     "num_iceberg_retries": creds.num_iceberg_retries,
+                    "skip_retry_on_query_timeout": creds.skip_retry_on_query_timeout,
                 },
                 formatter=AthenaParameterFormatter(),
                 poll_interval=creds.poll_interval,

dbt-athena/tests/unit/test_cursor_retry.py

@@ -0,0 +1,82 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+from pyathena.error import OperationalError
+from pyathena.util import RetryConfig
+
+from dbt.adapters.athena.connections import AthenaCursor
+
+
+@pytest.fixture()
+def athena_cursor():
+    connection = MagicMock()
+    connection.cursor_kwargs = {"num_iceberg_retries": 0}
+    retry_config = RetryConfig(attempt=3, max_delay=0, exponential_base=1)
+    cursor = AthenaCursor(
+        connection=connection,
+        converter=MagicMock(),
+        formatter=MagicMock(),
+        retry_config=retry_config,
+        s3_staging_dir="s3://test/",
+        schema_name="test_schema",
+        catalog_name="test_catalog",
+        work_group="test_wg",
+        poll_interval=0,
+        encryption_option=None,
+        kms_key=None,
+        kill_on_interrupt=False,
+        result_reuse_enable=False,
+        result_reuse_minutes=0,
+    )
+    return cursor
+
+
+class TestAthenaCursorRetry:
+    def test_retry_on_query_timeout_by_default(self, athena_cursor):
+        with patch.object(
+            athena_cursor, "_execute", side_effect=OperationalError("Query timeout")
+        ):
+            with pytest.raises(OperationalError, match="Query timeout"):
+                athena_cursor.execute("SELECT 1")
+            assert athena_cursor._execute.call_count == 3
+
+    def test_no_retry_on_query_timeout_when_skip_enabled(self, athena_cursor):
+        athena_cursor.connection.cursor_kwargs["skip_retry_on_query_timeout"] = True
+        with patch.object(
+            athena_cursor, "_execute", side_effect=OperationalError("Query timeout")
+        ):
+            with pytest.raises(OperationalError, match="Query timeout"):
+                athena_cursor.execute("SELECT 1")
+            assert athena_cursor._execute.call_count == 1
+
+    def test_no_retry_on_query_exhausted_resources(self, athena_cursor):
+        with patch.object(
+            athena_cursor, "_execute", side_effect=OperationalError("Query exhausted resources")
+        ):
+            with pytest.raises(OperationalError, match="Query exhausted resources"):
+                athena_cursor.execute("SELECT 1")
+            assert athena_cursor._execute.call_count == 1
+
+    def test_no_retry_on_too_many_open_partitions_when_catch_enabled(self, athena_cursor):
+        with patch.object(
+            athena_cursor, "_execute", side_effect=OperationalError("TOO_MANY_OPEN_PARTITIONS")
+        ):
+            with pytest.raises(OperationalError, match="TOO_MANY_OPEN_PARTITIONS"):
+                athena_cursor.execute("SELECT 1", catch_partitions_limit=True)
+            assert athena_cursor._execute.call_count == 1
+
+    def test_retry_on_too_many_open_partitions_when_catch_disabled(self, athena_cursor):
+        with patch.object(
+            athena_cursor, "_execute", side_effect=OperationalError("TOO_MANY_OPEN_PARTITIONS")
+        ):
+            with pytest.raises(OperationalError, match="TOO_MANY_OPEN_PARTITIONS"):
+                athena_cursor.execute("SELECT 1", catch_partitions_limit=False)
+            assert athena_cursor._execute.call_count == 3
+
+    def test_retry_on_transient_error(self, athena_cursor):
+        with patch.object(
+            athena_cursor, "_execute", side_effect=OperationalError("Some transient error")
+        ):
+            with pytest.raises(OperationalError, match="Some transient error"):
+                athena_cursor.execute("SELECT 1")
+            assert athena_cursor._execute.call_count == 3