perf(thrift): buffer reply body, use accelerated protocol, log payloads lazily

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 # Unreleased
 - Fix inline parameters (`use_inline_params=True`) so a sequence directly after `IN` renders as a list again: `x IN %(ids)s` renders as `x IN (1,2,3)` instead of `x IN ARRAY(1,2,3)`, which Spark rejects. Inside the list, tuples render as rows for a multi-column `IN`, and a `set` or `frozenset` is accepted again. Sequences anywhere else still render as `ARRAY(...)` ([#629](https://github.com/databricks/databricks-sql-python/issues/629))
+- Speed up Thrift response handling: read replies in 1 MiB chunks and decode them with Thrift's C-accelerated protocol, and stop stringifying every request/response for the default no-op debug logger. Wide result schemas, metadata calls with many rows, and inline Arrow results benefit most.
 
 # 4.9.0 (2026-11-04)
 - Upgrade Databricks SQL Kernel to 1.4.0, which keeps Lakehouse Real-Time session affinity and flushes shared telemetry only after the final session closes.

src/databricks/sql/auth/thrift_http_client.py

@@ -11,6 +11,7 @@
 from http.client import HTTPResponse
 from io import BytesIO
 
+from thrift.transport.TTransport import CReadableTransport
 from urllib3 import HTTPConnectionPool, HTTPSConnectionPool, ProxyManager
 from urllib3.util import make_headers
 from databricks.sql.auth.retry import CommandType, DatabricksRetryPolicy
@@ -22,7 +23,7 @@
 logger = logging.getLogger(__name__)
 
 
-class THttpClient(thrift.transport.THttpClient.THttpClient):
+class THttpClient(thrift.transport.THttpClient.THttpClient, CReadableTransport):
     realhost: Optional[str]
     realport: Optional[int]
     proxy_uri: Optional[str]
@@ -94,6 +95,7 @@ def __init__(
         self.retry_policy = retry_policy
 
         self.__wbuf = BytesIO()
+        self.__rbuf = BytesIO()
         self.__resp: Union[None, HTTPResponse] = None
         self.__timeout = None
         self.__custom_headers = None
@@ -153,9 +155,38 @@ def close(self):
         self.__resp and self.__resp.drain_conn()
         self.__resp and self.__resp.release_conn()
         self.__resp = None
+        self.__rbuf = BytesIO()
+
+    # Thrift decodes the reply one field at a time, and one urllib3 read per field
+    # costs far more than the decode, so the body is read in chunks of this size.
+    _READ_CHUNK_SIZE = 1024 * 1024
 
     def read(self, sz):
-        return self.__resp.read(sz)
+        ret = self.__rbuf.read(sz)
+        if ret:
+            return ret
+        self.__rbuf = BytesIO(self.__resp.read(max(sz, self._READ_CHUNK_SIZE)))
+        return self.__rbuf.read(sz)
+
+    # CReadableTransport: lets TBinaryProtocolAccelerated decode the chunks in C.
+    @property
+    def cstringio_buf(self):
+        return self.__rbuf
+
+    def cstringio_refill(self, partialread, reqlen):
+        # As in TBufferedTransport: read one more chunk, or all of a larger field.
+        parts, have = [partialread], len(partialread)
+        if reqlen < self._READ_CHUNK_SIZE:
+            parts.append(self.__resp.read(self._READ_CHUNK_SIZE))
+            have += len(parts[-1])
+        while have < reqlen:
+            chunk = self.__resp.read(reqlen - have)
+            if not chunk:
+                raise EOFError()
+            parts.append(chunk)
+            have += len(chunk)
+        self.__rbuf = BytesIO(b"".join(parts))
+        return self.__rbuf
 
     def isOpen(self):
         return self.__resp is not None
@@ -206,6 +237,7 @@ def flush(self):
         self.code = self.__resp.status
         self.message = self.__resp.reason
         self.headers = self.__resp.headers
+        self.__rbuf = BytesIO()
 
         logger.info(
             "HTTP Response with status code {}, message: {}".format(

src/databricks/sql/backend/thrift_backend.py

@@ -253,12 +253,18 @@ def __init__(
         self._transport.setTimeout(timeout and (float(timeout) * 1000.0))
 
         self._transport.setCustomHeaders(dict(http_headers))
-        protocol = thrift.protocol.TBinaryProtocol.TBinaryProtocol(
+        # Replies decode with thrift's C extension when available, else in pure Python.
+        input_protocol = thrift.protocol.TBinaryProtocol.TBinaryProtocolAccelerated(
             self._transport,
             string_length_limit=THRIFT_BINARY_PROTOCOL_STRING_LENGTH_LIMIT,
             container_length_limit=THRIFT_BINARY_PROTOCOL_CONTAINER_LENGTH_LIMIT,
         )
-        self._client = TCLIService.Client(protocol)
+        # Requests use the pure-Python encoder: the C one rejects structs nested
+        # deeper than ~64 levels, e.g. native array parameters nested 62+ deep.
+        output_protocol = thrift.protocol.TBinaryProtocol.TBinaryProtocol(
+            self._transport
+        )
+        self._client = TCLIService.Client(input_protocol, output_protocol)
 
         try:
             self._transport.open()
@@ -443,7 +449,9 @@ def attempt_request(attempt):
                 this_method_name = getattr(method, "__name__")
 
                 logger.debug("Sending request: {}(<REDACTED>)".format(this_method_name))
-                unsafe_logger.debug("Sending request: {}".format(request))
+                # Lazy %s args: unsafe_logger is DEBUG-level with a NullHandler by
+                # default, so eager formatting would stringify every payload.
+                unsafe_logger.debug("Sending request: %s", request)
 
                 # These three lines are no-ops if the v3 retry policy is not in use
                 if self.enable_v3_retries:
@@ -457,7 +465,7 @@ def attempt_request(attempt):
                 logger.debug(
                     "Received response: {}(<REDACTED>)".format(type(response).__name__)
                 )
-                unsafe_logger.debug("Received response: {}".format(response))
+                unsafe_logger.debug("Received response: %s", response)
                 return response
 
             except urllib3.exceptions.HTTPError as err: