Fix AWS CI failures exposed by debug matrix

.github/workflows/build_test.yml

@@ -38,6 +38,7 @@ jobs:
     outputs:
       test_matrix: ${{ steps.test_matrix.outputs.test_matrix }}
       build_matrix: ${{ steps.test_matrix.outputs.build_matrix }}
+      lambda_matrix: ${{ steps.test_matrix.outputs.lambda_matrix }}
     steps:
       - name: Checkout code
         uses: actions/checkout@v5
@@ -55,6 +56,7 @@ jobs:
           fi
           echo "test_matrix=$MATRIX" >> "$GITHUB_OUTPUT"
           echo "build_matrix=$(echo "$MATRIX" | jq -c '[group_by(.os_download_name, .["python-version"]) | .[] | first]')" >> "$GITHUB_OUTPUT"
+          echo "lambda_matrix=$(echo "$MATRIX" | jq -c '[.[] | select(.cloud-provider == "aws" and .os_download_name == "manylinux_x86_64")]')" >> "$GITHUB_OUTPUT"
   lint:
     name: Check linting
     runs-on: ubuntu-latest
@@ -380,15 +382,12 @@ jobs:
 
   test-lambda:
     name: Test Lambda linux-${{ matrix.python-version }}-${{ matrix.cloud-provider }}
-    needs: build
+    needs: [build, define-matrix]
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
-        include: ${{ fromJSON(needs.define-matrix.outputs.test_matrix) }}
-        exclude:
-          - cloud-provider: gcp
-          - cloud-provider: azure
+        include: ${{ fromJSON(needs.define-matrix.outputs.lambda_matrix) }}
     steps:
       - name: Set shortver
         run: echo "shortver=${longver//./}" >> $GITHUB_ENV

.github/workflows/generated_pr_matrix.json

@@ -1,4 +1,22 @@
 [
+  {
+    "os_image_name": "macos-latest",
+    "os_download_name": "macosx_x86_64",
+    "python-version": "3.12",
+    "cloud-provider": "aws"
+  },
+  {
+    "os_image_name": "macos-latest",
+    "os_download_name": "macosx_x86_64",
+    "python-version": "3.13",
+    "cloud-provider": "aws"
+  },
+  {
+    "os_image_name": "macos-latest",
+    "os_download_name": "macosx_x86_64",
+    "python-version": "3.14",
+    "cloud-provider": "aws"
+  },
   {
     "os_image_name": "ubuntu-latest",
     "os_download_name": "manylinux_x86_64",
@@ -8,31 +26,37 @@
   {
     "os_image_name": "ubuntu-latest",
     "os_download_name": "manylinux_x86_64",
-    "python-version": "3.14",
+    "python-version": "3.10",
     "cloud-provider": "aws"
   },
   {
-    "os_image_name": "macos-latest",
-    "os_download_name": "macosx_x86_64",
-    "python-version": "3.9",
-    "cloud-provider": "azure"
+    "os_image_name": "ubuntu-latest",
+    "os_download_name": "manylinux_x86_64",
+    "python-version": "3.11",
+    "cloud-provider": "aws"
   },
   {
-    "os_image_name": "macos-latest",
-    "os_download_name": "macosx_x86_64",
-    "python-version": "3.14",
-    "cloud-provider": "azure"
+    "os_image_name": "ubuntu-latest",
+    "os_download_name": "manylinux_x86_64",
+    "python-version": "3.12",
+    "cloud-provider": "aws"
   },
   {
     "os_image_name": "windows-latest",
     "os_download_name": "win_amd64",
     "python-version": "3.9",
-    "cloud-provider": "gcp"
+    "cloud-provider": "aws"
   },
   {
     "os_image_name": "windows-latest",
     "os_download_name": "win_amd64",
-    "python-version": "3.14",
-    "cloud-provider": "gcp"
+    "python-version": "3.12",
+    "cloud-provider": "aws"
+  },
+  {
+    "os_image_name": "windows-latest",
+    "os_download_name": "win_amd64",
+    "python-version": "3.13",
+    "cloud-provider": "aws"
   }
 ]

ci/generate_full_matrix.py

@@ -91,6 +91,21 @@ class CSP(Enum):
 # Example: {"with_snowpark": "true"}
 ADDITIONAL_FIELDS = {}
 
+# Temporary PR debug matrix for the 10 AWS jobs that were intermittently failing
+# in CI and need to be exercised together.
+PR_AWS_DEBUG_MATRIX: List[Tuple[OperatingSystem, str]] = [
+    (OperatingSystem.MACOS, "3.12"),
+    (OperatingSystem.MACOS, "3.13"),
+    (OperatingSystem.MACOS, "3.14"),
+    (OperatingSystem.UBUNTU, "3.9"),
+    (OperatingSystem.UBUNTU, "3.10"),
+    (OperatingSystem.UBUNTU, "3.11"),
+    (OperatingSystem.UBUNTU, "3.12"),
+    (OperatingSystem.WINDOWS, "3.9"),
+    (OperatingSystem.WINDOWS, "3.12"),
+    (OperatingSystem.WINDOWS, "3.13"),
+]
+
 # Output file paths (relative to repository root)
 _WORKFLOWS_DIR = Path(__file__).parent.parent / ".github" / "workflows"
 FULL_MATRIX_FILE = _WORKFLOWS_DIR / "generated_full_matrix.json"
@@ -128,13 +143,14 @@ def generate_matrix(pr_only: bool = False):
     matrix = []
 
     if pr_only:
-        csp_to_test = list(CSP)
-        for system in OperatingSystem:
-            os_config = system.value
-            csp_name = csp_to_test.pop(0).value if csp_to_test else CSP.AWS.value
-            for py_version in Python:
-                if py_version.value.test_on_pr:
-                    _add_to_matrix(matrix, os_config, csp_name, py_version.value)
+        python_versions = {py_enum.value.version: py_enum.value for py_enum in Python}
+        for system, python_version in PR_AWS_DEBUG_MATRIX:
+            _add_to_matrix(
+                matrix,
+                system.value,
+                CSP.AWS.value,
+                python_versions[python_version],
+            )
     else:
         operating_systems = [os_enum.value for os_enum in OperatingSystem]
         python_versions = [py_enum.value for py_enum in Python]

test/integ/aio_it/conftest.py

@@ -9,10 +9,11 @@
     get_db_parameters,
     is_public_testaccount,
 )
-from typing import Any, AsyncContextManager, AsyncGenerator, Callable
+from typing import Any, AsyncContextManager, AsyncGenerator, Callable, Generator
 
 import pytest
 
+import snowflake.connector
 from snowflake.connector.aio import SnowflakeConnection
 from snowflake.connector.aio import connect as async_connect
 from snowflake.connector.aio._telemetry import TelemetryClient
@@ -143,6 +144,16 @@ def conn_cnx():
     return db
 
 
+@pytest.fixture(autouse=True)
+def reset_default_paramstyle() -> Generator[None, None, None]:
+    """Keep async integration tests isolated from global paramstyle changes."""
+    snowflake.connector.paramstyle = "pyformat"
+    try:
+        yield
+    finally:
+        snowflake.connector.paramstyle = "pyformat"
+
+
 @pytest.fixture()
 async def conn_testaccount() -> AsyncGenerator[SnowflakeConnection, None]:
     connection = await create_connection("default")

test/integ/aio_it/test_multi_statement_async.py

@@ -73,7 +73,7 @@ async def _check_multi_statement_results(
 
 async def test_multi_statement_basic(conn_cnx, skip_to_last_set: bool):
     """Selects fixed integer data using statement level parameters."""
-    async with conn_cnx() as con:
+    async with conn_cnx(paramstyle="pyformat") as con:
         async with con.cursor() as cur:
             statement_params = dict()
             await cur.execute(
@@ -144,7 +144,7 @@ async def test_binding_multi(conn_cnx, style: str, skip_to_last_set: bool):
 @pytest.mark.parametrize("cursor_class", [SnowflakeCursor, DictCursor])
 async def test_async_exec_multi(conn_cnx, cursor_class, skip_to_last_set: bool):
     """Tests whether async execution query works within a multi-statement"""
-    async with conn_cnx() as con:
+    async with conn_cnx(paramstyle="pyformat") as con:
         async with con.cursor(cursor_class) as cur:
             await cur.execute_async(
                 "select 1; select 2; select count(*) from table(generator(timeLimit => 1)); select 'b';",
@@ -153,7 +153,7 @@ async def test_async_exec_multi(conn_cnx, cursor_class, skip_to_last_set: bool):
             q_id = cur.sfqid
             assert con.is_still_running(await con.get_query_status(q_id))
         await _wait_while_query_running_async(con, q_id, sleep_time=1)
-    async with conn_cnx() as con:
+    async with conn_cnx(paramstyle="pyformat") as con:
         async with con.cursor(cursor_class) as cur:
             await _wait_until_query_success_async(
                 con, q_id, num_checks=3, sleep_per_check=1
@@ -318,7 +318,7 @@ async def test_executemany_multi(conn_cnx, skip_to_last_set: bool):
     """Tests executemany with multi-statement optimizations enabled through the num_statements parameter."""
     table1 = random_string(5, "test_executemany_multi_")
     table2 = random_string(5, "test_executemany_multi_")
-    async with conn_cnx() as con:
+    async with conn_cnx(paramstyle="pyformat") as con:
         async with con.cursor() as cur:
             await cur.execute(
                 f"create temp table {table1} (aa number); create temp table {table2} (bb number);",

test/integ/aio_it/test_numpy_binding_async.py

@@ -62,7 +62,7 @@ async def test_numpy_datatype_binding(conn_cnx, db_parameters):
         },
     ]
     try:
-        async with conn_cnx(numpy=True) as cnx:
+        async with conn_cnx(numpy=True, paramstyle="pyformat") as cnx:
             await cnx.cursor().execute(
                 """
 CREATE OR REPLACE TABLE {name} (

test/integ/conftest.py

@@ -406,6 +406,16 @@ def fin():
     return connection
 
 
+@pytest.fixture(autouse=True)
+def reset_default_paramstyle() -> Generator[None]:
+    """Keep tests isolated from process-global paramstyle changes."""
+    snowflake.connector.paramstyle = "pyformat"
+    try:
+        yield
+    finally:
+        snowflake.connector.paramstyle = "pyformat"
+
+
 @pytest.fixture()
 def conn_cnx() -> Callable[..., ContextManager[SnowflakeConnection]]:
     return db

test/integ/test_large_result_set.py

@@ -4,6 +4,7 @@
 import logging
 import threading
 from unittest.mock import Mock
+from urllib.parse import urlparse
 
 import pytest
 
@@ -274,7 +275,8 @@ def send_strip_encoding(self, request, *args, **kwargs):
         HTTP round-trip but before requests reads the body."""
         nonlocal chunks_intercepted
         response = original_send(self, request, *args, **kwargs)
-        if response.status_code == 200:
+        hostname = urlparse(request.url).hostname or ""
+        if response.status_code == 200 and "snowflakecomputing." not in hostname:
             _strip_content_encoding(response)
             with chunks_intercepted_lock:
                 chunks_intercepted += 1

test/integ/test_multi_statement.py

@@ -86,7 +86,7 @@ def _check_multi_statement_results(
 
 def test_multi_statement_basic(conn_cnx, skip_to_last_set: bool):
     """Selects fixed integer data using statement level parameters."""
-    with conn_cnx() as con:
+    with conn_cnx(paramstyle="pyformat") as con:
         with con.cursor() as cur:
             statement_params = dict()
             cur.execute(
@@ -157,7 +157,7 @@ def test_binding_multi(conn_cnx, style: str, skip_to_last_set: bool):
 @pytest.mark.parametrize("cursor_class", [SnowflakeCursor, DictCursor])
 def test_async_exec_multi(conn_cnx, cursor_class, skip_to_last_set: bool):
     """Tests whether async execution query works within a multi-statement"""
-    with conn_cnx() as con:
+    with conn_cnx(paramstyle="pyformat") as con:
         with con.cursor(cursor_class) as cur:
             cur.execute_async(
                 "select 1; select 2; select count(*) from table(generator(timeLimit => 1)); select 'b';",
@@ -166,7 +166,7 @@ def test_async_exec_multi(conn_cnx, cursor_class, skip_to_last_set: bool):
             q_id = cur.sfqid
             assert con.is_still_running(con.get_query_status(q_id))
         _wait_while_query_running(con, q_id, sleep_time=1)
-    with conn_cnx() as con:
+    with conn_cnx(paramstyle="pyformat") as con:
         with con.cursor(cursor_class) as cur:
             _wait_until_query_success(con, q_id, num_checks=3, sleep_per_check=1)
             assert con.get_query_status_throw_if_error(q_id) == QueryStatus.SUCCESS
@@ -322,7 +322,7 @@ def test_executemany_multi(conn_cnx, skip_to_last_set: bool):
     """Tests executemany with multi-statement optimizations enabled through the num_statements parameter."""
     table1 = random_string(5, "test_executemany_multi_")
     table2 = random_string(5, "test_executemany_multi_")
-    with conn_cnx() as con:
+    with conn_cnx(paramstyle="pyformat") as con:
         with con.cursor() as cur:
             cur.execute(
                 f"create temp table {table1} (aa number); create temp table {table2} (bb number);",

test/integ/test_numpy_binding.py

@@ -58,7 +58,7 @@ def test_numpy_datatype_binding(conn_cnx, db_parameters):
         },
     ]
     try:
-        with conn_cnx(numpy=True) as cnx:
+        with conn_cnx(numpy=True, paramstyle="pyformat") as cnx:
             cnx.cursor().execute(
                 """
 CREATE OR REPLACE TABLE {name} (

tox.ini

@@ -40,7 +40,7 @@ setenv =
     unit: SNOWFLAKE_TEST_TYPE = unit
     integ: SNOWFLAKE_TEST_TYPE = integ
     single: SNOWFLAKE_TEST_TYPE = single
-    parallel: SNOWFLAKE_PYTEST_OPTS = {env:SNOWFLAKE_PYTEST_OPTS:} -n auto
+    parallel: SNOWFLAKE_PYTEST_OPTS = {env:SNOWFLAKE_PYTEST_OPTS:} -n auto --dist worksteal
     # Add common parts into pytest command
     SNOWFLAKE_PYTEST_COV_LOCATION = {env:JUNIT_REPORT_DIR:{toxworkdir}}/junit.{envname}-{env:cloud_provider:dev}.xml
     SNOWFLAKE_PYTEST_COV_CMD = --cov snowflake.connector --junitxml {env:SNOWFLAKE_PYTEST_COV_LOCATION} --cov-report=
@@ -185,8 +185,9 @@ depends = py39, py310, py311, py312, py313, py314
 
 [pytest]
 log_level = info
-addopts = -ra --strict-markers
+addopts = -ra --strict-markers -vvv
 junit_family = xunit2
+timeout = 1200
 filterwarnings =
     error::UserWarning:cryptography.*
     error::cryptography.utils.CryptographyDeprecationWarning