feat(kernel): forward _pool_maxsize

Signed-off-by: Vu Anh Phung <vu.phung@databricks.com>

Author: vuanhphung

CONNECTION_PARAMETERS.md

@@ -102,7 +102,7 @@ to change without notice.
 | ------------------------------------ | ----------- | :----: | :----: | ------------- | ---------------------------------------------------------------------------------------------------------------------------------------------- |
 | `_socket_timeout`                    | `float` (s) |   ✅   |   ✅   | `900` (Thrift); `120` (kernel) | Thrift: socket send/recv/connect timeout. Kernel: total HTTP request deadline from connect through response-body completion. A positive value is forwarded; unset or `0` selects the kernel's 120s default. On the kernel path, `0` is neither unlimited nor an immediate timeout. |
 | `_pool_connections`                  | `int`       |   ✅   |   ⚠️   | `10`          | Number of urllib3 connection pools. Configures the connector's shared Python HTTP client; the kernel's query transport is its own Rust stack.  |
-| `_pool_maxsize`                      | `int`       |   ✅   |   ⚠️   | `20`          | Max connections per pool on the shared Python HTTP client. Same kernel caveat as `_pool_connections`.                                          |
+| `_pool_maxsize`                      | `int`       |   ✅   |   ✅   | `20` (Thrift); `100` (kernel when unset) | Max idle connections retained per host. An explicit value configures both the shared Python HTTP client and the kernel's Rust HTTP pool. |
 | `_proxy_auth_method`                 | `str`       |   ✅   |   ⚠️   | `None`        | `basic` or `negotiate` (Kerberos). Applies to the shared Python HTTP client; not threaded to the kernel query transport. See [`docs/proxy.md`](docs/proxy.md). |
 | `_retry_stop_after_attempts_count`   | `int`       |   ✅   |   ✅   | `30`          | Max attempts in a retry sequence. Bounded to `[1, 60]` on Thrift; forwarded to the kernel's retry policy.                                       |
 | `_retry_stop_after_attempts_duration`| `float` (s) |   ✅   |   ✅   | `900`         | Max total wall-clock seconds spent retrying. Forwarded to the kernel.                                                                           |
@@ -202,9 +202,9 @@ None — the kernel's parameter surface is currently a subset of Thrift's.
 
 ### Behavioral divergences to watch
 
-- **Connection pooling / proxy** (`_pool_connections`, `_pool_maxsize`,
-  `_proxy_auth_method`) configure the connector's shared Python HTTP client
-  (auth/telemetry); the kernel's query traffic uses its own Rust transport.
+- **Connection pooling / proxy**: `_pool_connections` and `_proxy_auth_method`
+  configure only the shared Python HTTP client. `_pool_maxsize` also configures
+  the kernel's Rust HTTP pool when explicitly set.
 - **`use_inline_params`** renders parameters inline on Thrift; the kernel uses
   native parameter binding.
 

src/databricks/sql/backend/kernel/client.py

@@ -260,6 +260,7 @@ def __init__(
         self._retry_options = kwargs.get("retry_options") or {}
         # The kernel binding owns type and range validation.
         self._request_timeout_secs = kwargs.get("request_timeout_secs")
+        self._max_connections = kwargs.get("max_connections")
         # Kernel telemetry phase 7 adds binding/runtime identity and
         # telemetry config kwargs directly to ``databricks_sql_kernel.Session``.
         self._telemetry_options = kwargs.get("telemetry_options") or {}
@@ -399,6 +400,9 @@ def open_session(
                 ]
                 if forwarded:
                     http_headers_kwargs["http_headers"] = forwarded
+            pool_kwargs: Dict[str, Any] = {}
+            if self._max_connections is not None:
+                pool_kwargs["max_connections"] = self._max_connections
             self._kernel_session = _kernel.Session(
                 host=self._server_hostname,
                 http_path=self._http_path,
@@ -417,6 +421,7 @@ def open_session(
                 # strings).
                 intervals_as_string=True,
                 request_timeout_secs=self._request_timeout_secs,
+                **pool_kwargs,
                 **auth_kwargs,
                 **tls_kwargs,
                 **retry_kwargs,

src/databricks/sql/session.py

@@ -313,6 +313,7 @@ def _create_backend(
                 auth_options=kernel_auth_options,
                 retry_options=kernel_retry_options,
                 request_timeout_secs=kwargs.get("_socket_timeout"),
+                max_connections=kwargs.get("_pool_maxsize"),
                 telemetry_options=kernel_telemetry_options,
             )
 

tests/unit/test_kernel_client.py

@@ -368,6 +368,33 @@ def fake_session(**kw):
     assert captured["request_timeout_secs"] == timeout
 
 
+@pytest.mark.parametrize("max_connections", [None, 41])
+def test_open_session_passes_max_connections_to_kernel(monkeypatch, max_connections):
+    captured = {}
+
+    def fake_session(**kw):
+        captured.update(kw)
+        sess = MagicMock()
+        sess.session_id = "sess-id"
+        return sess
+
+    monkeypatch.setattr(kernel_client._kernel, "Session", fake_session)
+    c = kernel_client.KernelDatabricksClient(
+        server_hostname="example.cloud.databricks.com",
+        http_path="/sql/1.0/warehouses/abc",
+        auth_provider=AccessTokenAuthProvider("dapi-test"),
+        ssl_options=None,
+        max_connections=max_connections,
+    )
+
+    c.open_session(session_configuration=None, catalog=None, schema=None)
+
+    if max_connections is None:
+        assert "max_connections" not in captured
+    else:
+        assert captured["max_connections"] == max_connections
+
+
 def test_open_session_passes_phase_7_telemetry_kwargs_to_kernel(monkeypatch):
     """Kernel telemetry phase 7 added binding/runtime identity and
     telemetry config kwargs to ``databricks_sql_kernel.Session``."""

tests/unit/test_session.py

@@ -466,6 +466,7 @@ def test_retry_and_socket_timeout_threaded_into_kernel_client(self):
                 _retry_stop_after_attempts_count=10,
                 _retry_stop_after_attempts_duration=600.0,
                 _socket_timeout=12.5,
+                _pool_maxsize=41,
             )
             try:
                 _, kwargs = mock_kernel_client.call_args
@@ -475,6 +476,7 @@ def test_retry_and_socket_timeout_threaded_into_kernel_client(self):
                 assert opts["retry_stop_after_attempts_count"] == 10
                 assert opts["retry_stop_after_attempts_duration"] == 600.0
                 assert kwargs["request_timeout_secs"] == 12.5
+                assert kwargs["max_connections"] == 41
             finally:
                 conn.close()