perf(kernel): convert small results to rows without pandas

CHANGELOG.md

@@ -4,6 +4,7 @@
 - Fix the kernel backend ignoring `use_cloud_fetch=False`, so complete results can be fetched without cloud-storage downloads.
 - 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.
+- Speed up fetching small results on the kernel backend (`use_kernel=True`): they are converted to rows without building a pandas DataFrame when that is cheaper and gives the same values, which cut client CPU per query by 47% in a local benchmark of 200 concurrent sessions running one-row queries against a mock server.
 
 # 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/backend/kernel/result_set.py

@@ -32,7 +32,7 @@
 
 import logging
 from collections import deque
-from typing import Any, Deque, List, Optional, TYPE_CHECKING, cast
+from typing import Any, Deque, List, Optional, Tuple, TYPE_CHECKING, cast
 
 import pyarrow
 
@@ -48,6 +48,109 @@
 
 logger = logging.getLogger(__name__)
 
+# Small results are converted with to_pylist() instead of through a pandas
+# DataFrame, whose fixed cost (100-300 µs) dominates them. to_pylist() costs
+# more per value, so a result skips pandas only while its row count times the
+# sum of its column weights (per-value cost relative to an integer) is within
+# _MAX_COST_WITHOUT_PANDAS, which kept it at least 25% cheaper than pandas for
+# every type and width measured (pyarrow 23, pandas 1.3-3.1).
+_MAX_ROWS_WITHOUT_PANDAS = 128
+_MAX_COST_WITHOUT_PANDAS = 1024
+
+# Types for which to_pylist() returns the same values as the pandas path in
+# ResultSet._convert_arrow_table, with their weights. Floats only qualify
+# without NaN, which pandas turns into None or keeps depending on its version;
+# timestamps only when not in nanoseconds.
+_FLOAT_TYPE_IDS = frozenset(t.id for t in (pyarrow.float32(), pyarrow.float64()))
+_TIMESTAMP_TYPE_ID = pyarrow.timestamp("us").id
+_VALUE_WEIGHTS = {
+    **{
+        t.id: 1
+        for t in (
+            pyarrow.int8(),
+            pyarrow.int16(),
+            pyarrow.int32(),
+            pyarrow.int64(),
+            pyarrow.uint8(),
+            pyarrow.uint16(),
+            pyarrow.uint32(),
+            pyarrow.uint64(),
+            pyarrow.bool_(),
+            pyarrow.float32(),
+            pyarrow.float64(),
+            pyarrow.null(),
+        )
+    },
+    **{t.id: 3 for t in (pyarrow.string(), pyarrow.large_string())},
+    **{
+        t.id: 6
+        for t in (
+            pyarrow.binary(),
+            pyarrow.large_binary(),
+            pyarrow.binary(1),
+            pyarrow.decimal128(1),
+            pyarrow.decimal256(1),
+        )
+    },
+    pyarrow.date32().id: 12,
+    _TIMESTAMP_TYPE_ID: 12,
+}
+# pyarrow looks the time zone up again for every tz-aware value, which costs
+# tens of microseconds when pytz isn't installed.
+_TZ_TIMESTAMP_WEIGHT = 400
+
+
+def _max_rows_without_pandas(types: List[pyarrow.DataType]) -> int:
+    """The most rows of a result with these column types that are converted
+    with to_pylist(); -1 if such results always go through pandas."""
+    total = 0
+    for arrow_type in types:
+        weight = _VALUE_WEIGHTS.get(arrow_type.id)
+        if weight is None:
+            return -1
+        if arrow_type.id == _TIMESTAMP_TYPE_ID:
+            if arrow_type.unit == "ns":
+                return -1
+            if arrow_type.tz is not None:
+                weight = _TZ_TIMESTAMP_WEIGHT
+        total += weight
+    if total == 0:
+        return -1
+    return min(_MAX_ROWS_WITHOUT_PANDAS, _MAX_COST_WITHOUT_PANDAS // total)
+
+
+def _rows_without_pandas(
+    table: pyarrow.Table, description: List[Tuple]
+) -> Optional[List[Row]]:
+    """The rows of ``table`` built from ``to_pylist()``, or None if the table
+    must go through pandas (too costly, or a column whose values could
+    differ)."""
+    num_rows = table.num_rows
+    # No weight is below 1, so this rules out larger results without
+    # looking at every column type.
+    if num_rows * table.num_columns > _MAX_COST_WITHOUT_PANDAS:
+        return None
+    if num_rows > _max_rows_without_pandas(table.schema.types):
+        return None
+    columns = []
+    for column in table.itercolumns():
+        try:
+            values = column.to_pylist()
+        except Exception:
+            # E.g. a date past 9999-12-31: pandas then raises its own error,
+            # as it does without this fast path.
+            logger.debug(
+                "to_pylist() failed for a %s column; using pandas",
+                column.type,
+                exc_info=True,
+            )
+            return None
+        if column.type.id in _FLOAT_TYPE_IDS and any(v != v for v in values):
+            return None
+        columns.append(values)
+    row_factory = Row(*[c[0] for c in description])
+    return [row_factory(*values) for values in zip(*columns)]
+
 
 class KernelResultSet(ResultSet):
     """Streaming ``ResultSet`` over a kernel handle.
@@ -202,6 +305,13 @@ def fetchmany_arrow(self, size: int) -> pyarrow.Table:
 
     # ----- Row fetches -----
 
+    def _convert_arrow_table(self, table):
+        if self.connection.disable_pandas is not True:
+            rows = _rows_without_pandas(table, self.description)
+            if rows is not None:
+                return rows
+        return super()._convert_arrow_table(table)
+
     def fetchone(self) -> Optional[Row]:
         self._ensure_buffered(1)
         if self._buffered_rows() == 0: