Fix client_prefetch_threads=0 normalization

DESCRIPTION.md

@@ -9,6 +9,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
 # Release Notes
 - Upcoming Release
   - Fixed sdist to only install the minicore binary matching the current platform (SNOW-3526469). Previous 4.x releases copied every platform's minicore `.so`/`.dylib`/`.dll` into the install prefix, breaking downstream packagers (e.g. Homebrew) whose audits reject foreign-arch binaries.
+  - Fixed a bug where `client_prefetch_threads=0` could fall back to the default thread count instead of following the existing lower-bound correction, and aligned async `client_prefetch_threads` validation with the sync path.
 
 - v4.5.0(May 12,2026)
   - Fixed `write_pandas` temp stage name collisions (SNOW-3481510). The old PRNG could produce identical name sequences in forked processes (e.g. Notebook kernels), causing `CREATE TEMPORARY STAGE` to fail with "Object already exists".

src/snowflake/connector/aio/_connection.py

@@ -266,7 +266,7 @@ async def __open_connection(self):
                 PARAMETER_CLIENT_SESSION_KEEP_ALIVE_HEARTBEAT_FREQUENCY
             ] = self._validate_client_session_keep_alive_heartbeat_frequency()
 
-        if self.client_prefetch_threads:
+        if self.client_prefetch_threads is not None:
             self._session_parameters[PARAMETER_CLIENT_PREFETCH_THREADS] = (
                 self._validate_client_prefetch_threads()
             )
@@ -842,6 +842,7 @@ def client_prefetch_threads(self) -> int:
     @client_prefetch_threads.setter
     def client_prefetch_threads(self, value) -> None:
         self._client_prefetch_threads = value
+        self._validate_client_prefetch_threads()
 
     @property
     def errorhandler(self) -> None:

src/snowflake/connector/connection.py

@@ -2312,12 +2312,19 @@ def _validate_client_session_keep_alive_heartbeat_frequency(self) -> int:
         return value
 
     def _validate_client_prefetch_threads(self) -> int:
-        if self.client_prefetch_threads <= 0:
-            self._client_prefetch_threads = 1
-        elif self.client_prefetch_threads > MAX_CLIENT_PREFETCH_THREADS:
-            self._client_prefetch_threads = MAX_CLIENT_PREFETCH_THREADS
-        self._client_prefetch_threads = int(self.client_prefetch_threads)
-        return self.client_prefetch_threads
+        value = self._client_prefetch_threads
+
+        if value is None:
+            value = DEFAULT_CLIENT_PREFETCH_THREADS
+        else:
+            value = int(value)
+            if value <= 0:
+                value = 1
+            elif value > MAX_CLIENT_PREFETCH_THREADS:
+                value = MAX_CLIENT_PREFETCH_THREADS
+
+        self._client_prefetch_threads = value
+        return value
 
     def _update_parameters(
         self,

test/unit/aio/test_connection_async_unit.py

@@ -110,6 +110,20 @@ async def mock_post_request(request, url, headers, json_body, **kwargs):
 
     return request_body
 
+async def test_connect_client_prefetch_threads_zero_clamps_to_one(mock_post_requests):
+    async with fake_db_conn(client_prefetch_threads=0) as conn:
+        assert conn.client_prefetch_threads == 1
+        assert (
+            mock_post_requests["data"]["SESSION_PARAMETERS"]["CLIENT_PREFETCH_THREADS"]
+            == 1
+        )
+
+
+async def test_client_prefetch_threads_setter_zero_clamps_to_one(mock_post_requests):
+    async with fake_db_conn() as conn:
+        conn.client_prefetch_threads = 0
+        assert conn.client_prefetch_threads == 1
+
 
 async def test_connect_with_service_name(mock_post_requests):
     async with fake_db_conn() as conn:

test/unit/test_connection.py

@@ -113,6 +113,27 @@ def test_connect_with_service_name(mock_post_requests):
     assert fake_connector().service_name == "FAKE_SERVICE_NAME"
 
 
+def test_connect_client_prefetch_threads_zero_clamps_to_one(mock_post_requests):
+    conn = fake_connector(client_prefetch_threads=0)
+    try:
+        assert conn.client_prefetch_threads == 1
+        assert (
+            mock_post_requests["data"]["SESSION_PARAMETERS"]["CLIENT_PREFETCH_THREADS"]
+            == 1
+        )
+    finally:
+        conn.close()
+
+
+def test_client_prefetch_threads_setter_zero_clamps_to_one(mock_post_requests):
+    conn = fake_connector()
+    try:
+        conn.client_prefetch_threads = 0
+        assert conn.client_prefetch_threads == 1
+    finally:
+        conn.close()
+
+
 @pytest.mark.skip(reason="Mock doesn't work as expected.")
 @patch("snowflake.connector.network.SnowflakeRestful._post_request")
 def test_connection_ignore_exception(mockSnowflakeRestfulPostRequest):