@@ -7,6 +7,8 @@ https://docs.snowflake.com/
 Source code is also available at: https://github.com/snowflakedb/snowflake-connector-python
 
 # Release Notes
+- v4.8.1 (Unreleased)
+  - Fixed large result set batch chunk download logging every transient retry at ERROR level with a traceback. Intermediate retries now log at WARNING; only the final failure after all retries are exhausted logs at ERROR with a traceback. These retry and failure logs now also include the S3 request IDs (`x-amz-request-id` / `x-amz-id-2`) when present, to aid troubleshooting and correlation with AWS-side records (SNOW-4218352).
 - v4.8.0(Sep 30,2026)
   - Fixed external-browser (SSO) authentication to validate the `Origin` header on the local callback server, rejecting tokens delivered from unexpected origins. A trailing slash in the origin (e.g. `https://account.snowflakecomputing.com/`) is now accepted on par with the bare origin, matching JDBC and other driver behaviour. Preconnect probe connections (empty recv) no longer count against the retry budget and no longer abort the login flow.
   - Added the `SNOWFLAKE_TLS_CIPHERS` environment variable to restrict which TLS ciphers the connector offers. It takes a colon-separated list; names beginning with `TLS_` are applied as TLS 1.3 cipher suites and the remainder as the cipher list for TLS 1.2 and below, so a single variable covers both. Leaving it unset keeps OpenSSL's defaults unchanged, and an unrecognized cipher name is rejected rather than silently ignored. The restriction covers Snowflake API traffic, cloud-storage (stage) transfers, OCSP/CRL fetches and IdP requests. Requests issued by the AWS and Azure SDKs, and asynchronous connections, are not covered — for TLS 1.3 suites specifically they cannot be, because the Python standard library exposes no API for restricting them.
 
@@ -34,6 +34,7 @@
 from snowflake.connector.result_batch import ResultBatch as ResultBatchSync
 from snowflake.connector.result_batch import (
     _create_nanoarrow_iterator,
+    _s3_request_id_log_suffix,
     inline_first_chunk_rowcount,
 )
 from snowflake.connector.secret_detector import SecretDetector
@@ -232,6 +233,7 @@ async def download_chunk(http_session):
 
         content, encoding = None, None
         for retry in range(max(MAX_DOWNLOAD_RETRY, 1)):
+            amz_request_id_suffix = ""
             try:
 
                 async with TimerContextManager() as download_metric:
@@ -284,6 +286,7 @@ async def download_chunk(http_session):
                         ) as session:
                             response, content, encoding = await download_chunk(session)
 
+                    amz_request_id_suffix = _s3_request_id_log_suffix(response)
                     if response.status == OK:
                         break
                     # Raise error here to correctly go in to exception clause
@@ -301,12 +304,18 @@ async def download_chunk(http_session):
                 if retry == MAX_DOWNLOAD_RETRY - 1:
                     # Re-throw if we failed on the last retry
                     e = e.args[0] if isinstance(e, RetryRequest) else e
+                    logger.error(
+                        f"Failed to fetch the large result set batch "
+                        f"{self.id} after {MAX_DOWNLOAD_RETRY} attempts, "
+                        f"giving up for the reason: '{e}'{amz_request_id_suffix}",
+                        exc_info=True,
+                    )
                     raise e
                 sleep_timer = next(backoff)
-                logger.exception(
+                logger.warning(
                     f"Failed to fetch the large result set batch "
                     f"{self.id} for the {retry + 1} th time, "
-                    f"backing off for {sleep_timer}s for the reason: '{e}'"
+                    f"backing off for {sleep_timer}s for the reason: '{e}'{amz_request_id_suffix}"
                 )
                 await asyncio.sleep(sleep_timer)
 
 
@@ -55,6 +55,29 @@
 MAX_INCOMPLETE_CHUNK_RETRY = 1
 
 
+def _s3_request_id_log_suffix(response: Response | None) -> str:
+    """Return a ``', S3 request id: ..., S3 extended request id: ...'`` log suffix, or ``''``.
+
+    S3 returns these trace headers (``x-amz-request-id`` / ``x-amz-id-2``) on every
+    response, including error responses, so surfacing them in the retry/failure logs
+    lets support correlate a failed chunk download with AWS-side records. Empty string
+    when no response arrived (pure network failure) or the headers are absent (non-S3
+    stage).
+    """
+    headers = getattr(response, "headers", None)
+    if headers is None or not hasattr(headers, "get"):
+        return ""
+    parts = []
+    for name, label in (
+        ("x-amz-request-id", "S3 request id"),
+        ("x-amz-id-2", "S3 extended request id"),
+    ):
+        value = headers.get(name)
+        if isinstance(value, str) and value:
+            parts.append(f"{label}: {value}")
+    return f", {', '.join(parts)}" if parts else ""
+
+
 def inline_first_chunk_rowcount(data: dict[str, Any]) -> int:
     """Rows attributed to the inline first chunk: ``total`` minus remote chunk rowCounts.
 
@@ -520,6 +543,7 @@ def _download(
             else exponential_backoff()()
         )
         for retry in range(MAX_DOWNLOAD_RETRY):
+            amz_request_id_suffix = ""
             try:
                 with TimerContextManager() as download_metric:
                     logger.debug(f"started downloading result batch id: {self.id}")
@@ -560,6 +584,7 @@ def _download(
                         )
                         response = local_session_manager.get(**request_data)
 
+                    amz_request_id_suffix = _s3_request_id_log_suffix(response)
                     if response.status_code == OK:
                         logger.debug(
                             f"successfully downloaded result batch id: {self.id}"
@@ -581,12 +606,18 @@ def _download(
                 if retry == MAX_DOWNLOAD_RETRY - 1:
                     # Re-throw if we failed on the last retry
                     e = e.args[0] if isinstance(e, RetryRequest) else e
+                    logger.error(
+                        f"Failed to fetch the large result set batch "
+                        f"{self.id} after {MAX_DOWNLOAD_RETRY} attempts, "
+                        f"giving up for the reason: '{e}'{amz_request_id_suffix}",
+                        exc_info=True,
+                    )
                     raise e
                 sleep_timer = next(backoff)
-                logger.exception(
+                logger.warning(
                     f"Failed to fetch the large result set batch "
                     f"{self.id} for the {retry + 1} th time, "
-                    f"backing off for {sleep_timer}s for the reason: '{e}'"
+                    f"backing off for {sleep_timer}s for the reason: '{e}'{amz_request_id_suffix}"
                 )
                 time.sleep(sleep_timer)
 
 
@@ -48,27 +48,37 @@
     pass
 
 
-def create_mock_response(status_code: int) -> Mock:
+def create_mock_response(status_code: int, headers: dict | None = None) -> Mock:
     """Create a Mock "Response" with a given status code. See `test_result_batch.py` for examples.
     Args:
         status_code: the status code of the response.
+        headers: optional response headers (e.g. S3 ``x-amz-*`` trace headers).
     Returns:
         A Mock object that can be used as a Mock Response in tests.
     """
     mock_resp = Mock()
     mock_resp.status_code = status_code
     mock_resp.raw = "success" if status_code == OK else "fail"
+    mock_resp.headers = headers or {}
     return mock_resp
 
 
-def create_async_mock_response(status: int) -> AsyncMock:
-    async def _create_async_mock_response(url, *, status, **kwargs):
+def create_async_mock_response(status: int, headers: dict | None = None) -> AsyncMock:
+    # ``response_headers`` is named distinctly from the request ``headers`` kwarg the
+    # real ``http_session.get(**request_data)`` passes, so the call-time request
+    # headers don't override the bound response headers.
+    async def _create_async_mock_response(
+        url, *, status, response_headers=None, **kwargs
+    ):
         resp = AsyncMock(status=status)
         resp.read.return_value = "success" if status == OK else "fail"
         resp.status = status
+        resp.headers = response_headers or {}
         return resp
 
-    return functools.partial(_create_async_mock_response, status=status)
+    return functools.partial(
+        _create_async_mock_response, status=status, response_headers=headers
+    )
 
 
 def verify_log_tuple(
 
@@ -99,23 +99,44 @@ async def test_ok_response_download(mock_get):
         (555, OtherHTTPRetryableError),  # random 5xx error
     ],
 )
-async def test_retryable_response_download(errcode, error_class):
+async def test_retryable_response_download(errcode, error_class, caplog):
     """This test checks that responses which are deemed 'retryable' are handled correctly."""
     # retryable exceptions
+    amz_headers = {"x-amz-request-id": "test-req-id", "x-amz-id-2": "test-ext-id"}
     with mock.patch(
-        REQUEST_MODULE_PATH + ".get", side_effect=create_async_mock_response(errcode)
+        REQUEST_MODULE_PATH + ".get",
+        side_effect=create_async_mock_response(errcode, headers=amz_headers),
     ) as mock_get:
         # mock_get.return_value = create_async_mock_response(errcode)
 
         with mock.patch("asyncio.sleep", return_value=None):
-            with pytest.raises(error_class) as ex:
-                _ = await result_batch._download()
+            with caplog.at_level(
+                logging.WARNING, logger="snowflake.connector.aio._result_batch"
+            ):
+                with pytest.raises(error_class) as ex:
+                    _ = await result_batch._download()
             err_msg = ex.value.msg
             if isinstance(errcode, HTTPStatus):
                 assert str(errcode.value) in err_msg
             else:
                 assert str(errcode) in err_msg
         assert mock_get.call_count == MAX_DOWNLOAD_RETRY
+        batch_logs = [
+            r
+            for r in caplog.records
+            if r.name == "snowflake.connector.aio._result_batch"
+        ]
+        assert (
+            sum(r.levelno == logging.WARNING for r in batch_logs)
+            == MAX_DOWNLOAD_RETRY - 1
+        )
+        assert sum(r.levelno == logging.ERROR for r in batch_logs) == 1
+        # S3 trace headers are surfaced in every retry/failure log line (SNOW-4218352)
+        assert all(
+            "S3 request id: test-req-id" in r.getMessage()
+            and "S3 extended request id: test-ext-id" in r.getMessage()
+            for r in batch_logs
+        )
 
 
 async def test_unauthorized_response_download():
@@ -150,21 +171,36 @@ async def test_non_200_response_download(status_code):
         assert mock_get.call_count == MAX_DOWNLOAD_RETRY
 
 
-async def test_retries_until_success():
+async def test_retries_until_success(caplog):
     with mock.patch(REQUEST_MODULE_PATH + ".get") as mock_get:
         error_codes = [BAD_REQUEST, UNAUTHORIZED, 201]
         # There is an OK added to the list of responses so that there is a success
         # and the retry loop ends.
+        amz_headers = {"x-amz-request-id": "test-req-id", "x-amz-id-2": "test-ext-id"}
         mock_responses = [
-            create_async_mock_response(code)("") for code in error_codes + [OK]
+            create_async_mock_response(code, headers=amz_headers)("")
+            for code in error_codes + [OK]
         ]
         mock_get.side_effect = mock_responses
 
         with mock.patch("asyncio.sleep", return_value=None):
-            res, _ = await result_batch._download()
+            with caplog.at_level(
+                logging.WARNING, logger="snowflake.connector.aio._result_batch"
+            ):
+                res, _ = await result_batch._download()
             assert res == "success"
         # call `get` once for each error and one last time when it succeeds
         assert mock_get.call_count == len(error_codes) + 1
+        batch_logs = [
+            r
+            for r in caplog.records
+            if r.name == "snowflake.connector.aio._result_batch"
+        ]
+        assert sum(r.levelno == logging.WARNING for r in batch_logs) == len(error_codes)
+        assert sum(r.levelno == logging.ERROR for r in batch_logs) == 0
+        warning_logs = [r for r in batch_logs if r.levelno == logging.WARNING]
+        assert warning_logs
+        assert all("S3 request id: test-req-id" in r.getMessage() for r in warning_logs)
 
 
 @pytest.mark.skipolddriver
 
@@ -123,21 +123,39 @@ def test_ok_response_download(mock_get):
         (555, OtherHTTPRetryableError),  # random 5xx error
     ],
 )
-def test_retryable_response_download(errcode, error_class):
+def test_retryable_response_download(errcode, error_class, caplog):
     """This test checks that responses which are deemed 'retryable' are handled correctly."""
     # retryable exceptions
+    amz_headers = {"x-amz-request-id": "test-req-id", "x-amz-id-2": "test-ext-id"}
     with mock.patch(SESSION_FROM_REQUEST_MODULE_PATH + ".get") as mock_get:
-        mock_get.return_value = create_mock_response(errcode)
+        mock_get.return_value = create_mock_response(errcode, headers=amz_headers)
 
         with mock.patch("time.sleep", return_value=None):
-            with pytest.raises(error_class) as ex:
-                _ = result_batch._download()
+            with caplog.at_level(
+                logging.WARNING, logger="snowflake.connector.result_batch"
+            ):
+                with pytest.raises(error_class) as ex:
+                    _ = result_batch._download()
             err_msg = ex.value.msg
             if isinstance(errcode, HTTPStatus):
                 assert str(errcode.value) in err_msg
             else:
                 assert str(errcode) in err_msg
         assert mock_get.call_count == MAX_DOWNLOAD_RETRY
+        batch_logs = [
+            r for r in caplog.records if r.name == "snowflake.connector.result_batch"
+        ]
+        assert (
+            sum(r.levelno == logging.WARNING for r in batch_logs)
+            == MAX_DOWNLOAD_RETRY - 1
+        )
+        assert sum(r.levelno == logging.ERROR for r in batch_logs) == 1
+        # S3 trace headers are surfaced in every retry/failure log line (SNOW-4218352)
+        assert all(
+            "S3 request id: test-req-id" in r.getMessage()
+            and "S3 extended request id: test-ext-id" in r.getMessage()
+            for r in batch_logs
+        )
 
 
 def test_unauthorized_response_download():
@@ -170,19 +188,34 @@ def test_non_200_response_download(status_code):
         assert mock_get.call_count == MAX_DOWNLOAD_RETRY
 
 
-def test_retries_until_success():
+def test_retries_until_success(caplog):
     with mock.patch(SESSION_FROM_REQUEST_MODULE_PATH + ".get") as mock_get:
         error_codes = [BAD_REQUEST, UNAUTHORIZED, 201]
+        amz_headers = {"x-amz-request-id": "test-req-id", "x-amz-id-2": "test-ext-id"}
         # There is an OK added to the list of responses so that there is a success
         # and the retry loop ends.
-        mock_responses = [create_mock_response(code) for code in error_codes + [OK]]
+        mock_responses = [
+            create_mock_response(code, headers=amz_headers)
+            for code in error_codes + [OK]
+        ]
         mock_get.side_effect = mock_responses
 
         with mock.patch("time.sleep", return_value=None):
-            res = result_batch._download()
+            with caplog.at_level(
+                logging.WARNING, logger="snowflake.connector.result_batch"
+            ):
+                res = result_batch._download()
             assert res.raw == "success"
         # call `get` once for each error and one last time when it succeeds
         assert mock_get.call_count == len(error_codes) + 1
+        batch_logs = [
+            r for r in caplog.records if r.name == "snowflake.connector.result_batch"
+        ]
+        assert sum(r.levelno == logging.WARNING for r in batch_logs) == len(error_codes)
+        assert sum(r.levelno == logging.ERROR for r in batch_logs) == 0
+        warning_logs = [r for r in batch_logs if r.levelno == logging.WARNING]
+        assert warning_logs
+        assert all("S3 request id: test-req-id" in r.getMessage() for r in warning_logs)
 
 
 # ---------------------------------------------------------------------------