Merge main and resolve release history conflict

Signed-off-by: TangoEnSkai <21152231+TangoEnSkai@users.noreply.github.com>

Author: TangoEnSkai

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 # Unreleased
 - Fix: a rejected token-federation exchange now reports the reason the endpoint gave. `_exchange_token` raised `KeyError: 'access_token'` on an OAuth error body, so the connection logged `Token exchange failed, using external token: 'access_token'` and the endpoint's `error` / `error_description` were discarded. It now raises a `ValueError` naming the endpoint, the HTTP status, and the returned error, and a non-JSON body reports the endpoint and status instead of surfacing a `JSONDecodeError`. The graceful fallback to the external token is unchanged ([#904](https://github.com/databricks/databricks-sql-python/issues/904))
+- Transparently auto-recover Thrift connections to Reyden / Real-Time warehouses: when a warehouse rejects the default Thrift protocol (SQLSTATE `KP001`), the session is re-opened on the kernel backend and the warehouse is remembered so later connections skip Thrift. Applies only when no backend was chosen explicitly.
+- Reject an mTLS private key without a client certificate, and identify missing or empty client certificate/key files in connection errors.
 
 # 4.5.0 (2026-09-01)
 - Upgrade Databricks SQL Kernel to 1.0.0.

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

@@ -53,6 +53,7 @@
 if TYPE_CHECKING:
     from databricks.sql.client import Cursor
     from databricks.sql.result_set import ResultSet
+    from databricks.sql.types import SSLOptions
 
     # Type-annotation-only import (deferred by ``from __future__ import
     # annotations``). ``execute_command`` accepts the Thrift-shaped
@@ -1078,7 +1079,7 @@ def max_download_threads(self) -> int:
 }
 
 
-def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
+def _kernel_tls_kwargs(ssl_options: Optional[SSLOptions]) -> Dict[str, Any]:
     """Translate the connector's ``SSLOptions`` into the kernel
     ``Session``'s ``tls_*`` kwargs.
 
@@ -1102,6 +1103,11 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
     if ssl_options is None:
         return {}
 
+    # The kernel rejects an in-memory key without a certificate, but a lone key file
+    # used to be dropped here before it could reach that validation. Reject the
+    # incomplete connector configuration directly instead.
+    ssl_options.validate_client_identity()
+
     kwargs: Dict[str, Any] = {}
 
     # Inverted booleans. Emit only the insecure (skip) direction so the
@@ -1111,18 +1117,18 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
     # own semantics (``create_ssl_context`` sets ``check_hostname=False``
     # whenever ``tls_verify`` is False). Without this the kernel could
     # still attempt a hostname check the connector considers disabled.
-    if getattr(ssl_options, "tls_verify", True) is False:
+    if ssl_options.tls_verify is False:
         kwargs["tls_skip_verify"] = True
         kwargs["tls_skip_hostname_verify"] = True
-    elif getattr(ssl_options, "tls_verify_hostname", True) is False:
+    elif ssl_options.tls_verify_hostname is False:
         kwargs["tls_skip_hostname_verify"] = True
 
-    ca_file = getattr(ssl_options, "tls_trusted_ca_file", None)
+    ca_file = ssl_options.tls_trusted_ca_file
     if ca_file:
         kwargs["tls_ca_cert"] = _read_pem_bytes(ca_file, "tls_trusted_ca_file")
 
-    cert_file = getattr(ssl_options, "tls_client_cert_file", None)
-    key_file = getattr(ssl_options, "tls_client_cert_key_file", None)
+    cert_file = ssl_options.tls_client_cert_file
+    key_file = ssl_options.tls_client_cert_key_file
     if cert_file:
         # The kernel pairs cert + key for mutual TLS; a cert without a
         # key (or vice versa) is rejected kernel-side. The connector's
@@ -1135,7 +1141,7 @@ def _kernel_tls_kwargs(ssl_options) -> Dict[str, Any]:
         # The kernel has no surface for an encrypted client key today.
         # Reject loudly rather than hand the kernel a key it can't
         # decrypt (which would fail with an opaque TLS parse error).
-        if getattr(ssl_options, "tls_client_cert_key_password", None):
+        if ssl_options.tls_client_cert_key_password:
             raise NotSupportedError(
                 "use_kernel=True does not support a password-protected mTLS "
                 "client key (tls_client_cert_key_password). Provide an "

src/databricks/sql/backend/reyden_warehouse_cache.py

@@ -0,0 +1,101 @@
+"""Process-wide cache of warehouses known to reject the legacy Thrift protocol.
+
+A Reyden / Real-Time SQL warehouse rejects a Thrift ``OpenSession`` — the SQL
+Gateway proxy stamps SQLSTATE ``KP001`` on the rejection. When the driver
+auto-recovers by re-opening on the kernel backend, it records the warehouse
+here so later connections to the same warehouse skip the doomed Thrift attempt
+and open on the kernel directly.
+
+Keyed by ``(host, warehouse_id)``. Warehouse ids are globally unique, so the
+warehouse id alone identifies the warehouse — even on a SPOG host shared by many
+workspaces (where only the ``?o=<workspace-id>`` path param distinguishes them),
+there is no cross-workspace collision. The host is kept in the key only as a
+cheap optimization (scoping lookups) and defense-in-depth, not for correctness.
+Entries expire after ``_TTL_SECONDS`` so a warehouse later reconfigured to accept
+Thrift is eventually retried.
+"""
+
+import re
+import threading
+import time
+from typing import Dict, Optional, Tuple
+
+# A warehouse's Reyden membership can change (an id may be recreated on a
+# Thrift-capable endpoint), so cached entries are re-validated after this long.
+# Matches the ADBC driver's 6-hour horizon.
+_TTL_SECONDS = 6 * 60 * 60
+
+# Warehouse paths look like ``/sql/1.0/warehouses/<id>`` or
+# ``.../endpoints/<id>``; the id stops at the next ``/``, ``?`` or ``&`` (e.g. a
+# ``?o=`` SPOG routing param). All-purpose-compute cluster paths carry no
+# warehouse id and never match — they are never Reyden warehouses.
+_WAREHOUSE_PATH_RE = re.compile(r".*/(?:warehouses|endpoints)/([^?&/]+)")
+
+
+def extract_warehouse_id(http_path: Optional[str]) -> Optional[str]:
+    """Return the warehouse/endpoint id embedded in ``http_path``, or ``None``."""
+    if not http_path:
+        return None
+    match = _WAREHOUSE_PATH_RE.match(http_path)
+    return match.group(1) if match else None
+
+
+class _ReydenWarehouseCache:
+    def __init__(self, ttl_seconds: float = _TTL_SECONDS) -> None:
+        self._ttl_seconds = ttl_seconds
+        self._lock = threading.Lock()
+        # (host_lowercased, warehouse_id) -> monotonic expiry deadline
+        self._expiry: Dict[Tuple[str, str], float] = {}
+
+    @staticmethod
+    def _key(host: str, warehouse_id: str) -> Tuple[str, str]:
+        return (host.lower(), warehouse_id)
+
+    def mark_reyden(self, host: str, warehouse_id: str) -> None:
+        now = time.monotonic()
+        with self._lock:
+            # Opportunistic sweep: mark_reyden only runs on an actual Thrift
+            # rejection (rare), so purging every expired entry here is near-free
+            # and bounds the cache to warehouses seen within the TTL window
+            # rather than every warehouse ever seen (the per-key lazy eviction
+            # in is_known_reyden never reclaims a warehouse that is not looked
+            # up again).
+            for key in [k for k, deadline in self._expiry.items() if deadline <= now]:
+                del self._expiry[key]
+            self._expiry[self._key(host, warehouse_id)] = now + self._ttl_seconds
+
+    def is_known_reyden(self, host: str, warehouse_id: str) -> bool:
+        key = self._key(host, warehouse_id)
+        now = time.monotonic()
+        with self._lock:
+            deadline = self._expiry.get(key)
+            if deadline is None:
+                return False
+            if deadline <= now:
+                # Lazily evict so a reconfigured warehouse is retried over Thrift.
+                del self._expiry[key]
+                return False
+            return True
+
+    def clear(self) -> None:
+        with self._lock:
+            self._expiry.clear()
+
+
+# Process-wide singleton; multi-tenant safe via the host component of the key.
+_CACHE = _ReydenWarehouseCache()
+
+
+def mark_reyden(host: str, warehouse_id: str) -> None:
+    """Record that ``warehouse_id`` on ``host`` rejects the Thrift protocol."""
+    _CACHE.mark_reyden(host, warehouse_id)
+
+
+def is_known_reyden(host: str, warehouse_id: str) -> bool:
+    """Whether ``warehouse_id`` on ``host`` is known (unexpired) to reject Thrift."""
+    return _CACHE.is_known_reyden(host, warehouse_id)
+
+
+def clear_cache() -> None:
+    """Reset the cache. Intended for tests."""
+    _CACHE.clear()

src/databricks/sql/backend/thrift_backend.py

@@ -284,11 +284,23 @@ def _initialize_retry_args(self, kwargs):
             )
 
     @staticmethod
-    def _check_response_for_error(response, host_url=None):
+    def _check_response_for_error(response, host_url=None, detect_reyden=False):
         if response.status and response.status.statusCode in [
             ttypes.TStatusCode.ERROR_STATUS,
             ttypes.TStatusCode.INVALID_HANDLE_STATUS,
         ]:
+            # A Reyden / Real-Time warehouse rejects the legacy Thrift protocol
+            # with SQLSTATE KP001, but only at OpenSession. `detect_reyden` gates
+            # the marker to that call so a stray KP001 on any other RPC surfaces
+            # as a normal DatabaseError (the connection-layer recovery only wraps
+            # session open). host_url is deliberately omitted on the marker: it is
+            # a recoverable signal, not a terminal failure, so it must not emit a
+            # failure-telemetry event here.
+            if (
+                detect_reyden
+                and response.status.sqlState == ReydenThriftUnsupportedError.SQL_STATE
+            ):
+                raise ReydenThriftUnsupportedError(response.status.errorMessage)
             raise DatabaseError(
                 response.status.errorMessage,
                 host_url=host_url,
@@ -520,7 +532,14 @@ def attempt_request(attempt):
             if not isinstance(response_or_error_info, RequestErrorInfo):
                 # log nothing here, presume that main request logging covers
                 response = response_or_error_info
-                ThriftDatabricksClient._check_response_for_error(response, self._host)
+                # Only OpenSession opts into KP001→Reyden-marker detection (the
+                # rejection is stamped only there). Mirrors the method.__name__
+                # discrimination already used above for GetOperationStatus.
+                ThriftDatabricksClient._check_response_for_error(
+                    response,
+                    self._host,
+                    detect_reyden=getattr(method, "__name__", None) == "OpenSession",
+                )
                 return response
 
             error_info = response_or_error_info

src/databricks/sql/client.py

@@ -35,6 +35,12 @@
     ProgrammingError,
     TransactionError,
     DatabaseError,
+    ReydenThriftUnsupportedError,
+)
+from databricks.sql.backend.reyden_warehouse_cache import (
+    extract_warehouse_id,
+    is_known_reyden,
+    mark_reyden,
 )
 
 from databricks.sql.backend.databricks_client import DatabricksClient
@@ -66,7 +72,7 @@
 from databricks.sql.session import Session
 from databricks.sql.backend.types import CommandId, BackendType, CommandState, SessionId
 
-from databricks.sql.auth.common import ClientContext
+from databricks.sql.auth.common import AuthType, ClientContext
 from databricks.sql.common.unified_http_client import UnifiedHttpClient
 from databricks.sql.common.http import HttpMethod
 
@@ -399,24 +405,30 @@ def read(self) -> Optional[OAuthToken]:
         self.http_client = UnifiedHttpClient(client_context)
 
         try:
-            self.session = Session(
+            self.session = self._open_session_with_reyden_fallback(
                 server_hostname,
                 http_path,
-                self.http_client,
                 http_headers,
                 session_configuration,
                 catalog,
                 schema,
                 _use_arrow_native_complex_types,
-                **kwargs,
+                kwargs,
             )
-            self.session.open()
         except Exception as e:
             # Respect user's telemetry preference even during connection failure.
-            # For use_kernel connections the kernel owns telemetry, so suppress
-            # the wrapper-side failure log to avoid wrapper-vs-kernel duplication.
-            enable_telemetry = kwargs.get("enable_telemetry", True) and not kwargs.get(
-                "use_kernel", False
+            # For a kernel connection the kernel owns telemetry, so suppress the
+            # wrapper-side failure log to avoid wrapper-vs-kernel duplication.
+            # Read the backend from the session that actually failed rather than
+            # the caller's kwargs: on the Reyden auto-recovery path we retry on
+            # the kernel via a kwargs copy, so the original kwargs still says
+            # Thrift. If the kernel never got constructed (e.g. its wheel is
+            # missing), self.session is the Thrift session and we still log.
+            attempted_kernel = getattr(
+                getattr(self, "session", None), "use_kernel", False
+            )
+            enable_telemetry = (
+                kwargs.get("enable_telemetry", True) and not attempted_kernel
             )
             TelemetryClientFactory.connection_failure_log(
                 error_name="Exception",
@@ -512,6 +524,106 @@ def read(self) -> Optional[OAuthToken]:
             session_id=self.get_session_id_hex(),
         )
 
+    def _open_session_with_reyden_fallback(
+        self,
+        server_hostname: str,
+        http_path: str,
+        http_headers,
+        session_configuration,
+        catalog,
+        schema,
+        _use_arrow_native_complex_types,
+        kwargs: dict,
+    ) -> Session:
+        """Open a ``Session``, transparently recovering onto the kernel backend
+        when a Reyden / Real-Time warehouse rejects the default Thrift protocol.
+
+        Reyden warehouses reject a Thrift ``OpenSession`` (SQLSTATE ``KP001``);
+        the kernel (SEA) backend is the supported path. Auto-recovery applies
+        only when the caller did not pick a backend explicitly (neither
+        ``use_kernel`` nor ``use_sea``). On a rejection the warehouse is
+        remembered so later connections skip the doomed Thrift attempt.
+        """
+
+        def build_session(session_kwargs: dict) -> Session:
+            # Assign self.session before open() so a failed open still leaves the
+            # attempted session on the connection — __del__ and the failure
+            # telemetry log both rely on self.session being present.
+            self.session = Session(
+                server_hostname,
+                http_path,
+                self.http_client,
+                http_headers,
+                session_configuration,
+                catalog,
+                schema,
+                _use_arrow_native_complex_types,
+                **session_kwargs,
+            )
+            self.session.open()
+            return self.session
+
+        def kernel_recovery_kwargs() -> dict:
+            # Kwargs for re-opening on the kernel. The Thrift path treats an
+            # unset auth_type as databricks-oauth (see get_auth_provider); the
+            # kernel path has no such fallback and rejects auth_type=None unless
+            # a credential shape (PAT / OAuth M2M) is present. Mirror the Thrift
+            # default so a bare OAuth-U2M connection recovers instead of failing
+            # with NotSupportedError. Skip the injection when a credential shape
+            # is already present — the kernel routes on it regardless of
+            # auth_type, and forcing databricks-oauth alongside an M2M secret or
+            # a credentials_provider would change that routing.
+            recovery_kwargs = {**kwargs, "use_kernel": True}
+            has_credential_shape = (
+                recovery_kwargs.get("access_token")
+                or recovery_kwargs.get("oauth_client_secret")
+                or recovery_kwargs.get("oauth_jwt_key_file")
+                or recovery_kwargs.get("credentials_provider")
+            )
+            if recovery_kwargs.get("auth_type") is None and not has_credential_shape:
+                recovery_kwargs["auth_type"] = AuthType.DATABRICKS_OAUTH.value
+            return recovery_kwargs
+
+        # An explicit backend choice is always honored — auto-recovery engages
+        # only on the default (Thrift) path.
+        explicit_backend = kwargs.get("use_kernel", False) or kwargs.get(
+            "use_sea", False
+        )
+        if explicit_backend:
+            return build_session(kwargs)
+
+        warehouse_id = extract_warehouse_id(http_path)
+
+        # Pre-check: a warehouse already seen to reject Thrift opens straight on
+        # the kernel, skipping the doomed Thrift OpenSession round-trip.
+        if warehouse_id and is_known_reyden(server_hostname, warehouse_id):
+            logger.info(
+                "Warehouse %s on %s is known to require the kernel backend; "
+                "opening on the kernel and skipping Thrift.",
+                warehouse_id,
+                server_hostname,
+            )
+            return build_session(kernel_recovery_kwargs())
+
+        try:
+            return build_session(kwargs)
+        except ReydenThriftUnsupportedError as thrift_ex:
+            logger.info(
+                "Thrift is not supported for this Reyden/Real-Time warehouse; "
+                "transparently re-opening the session on the kernel backend."
+            )
+            # Remember the rejection regardless of the retry's outcome — the
+            # warehouse is Reyden either way, so future connects should skip
+            # Thrift; a kernel failure below is a separate, orthogonal problem.
+            if warehouse_id:
+                mark_reyden(server_hostname, warehouse_id)
+            try:
+                return build_session(kernel_recovery_kwargs())
+            except Exception as kernel_ex:
+                # Surface the kernel failure (the actionable one) while keeping
+                # the original Thrift rejection in the chain for diagnosis.
+                raise kernel_ex from thrift_ex
+
     def _set_use_inline_params_with_warning(self, value: Union[bool, str]):
         """Valid values are True, False, and "silent"