Forward telemetry parameters to the kernel binding

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

@@ -22,6 +22,7 @@
 
 from __future__ import annotations
 
+import inspect
 import logging
 import threading
 import uuid
@@ -47,6 +48,7 @@
     NotSupportedError,
     ProgrammingError,
 )
+from databricks.sql.telemetry.telemetry_client import TelemetryHelper
 
 if TYPE_CHECKING:
     from databricks.sql.client import Cursor
@@ -172,6 +174,75 @@ def _is_staging_statement(operation: str) -> bool:
     return verb in _STAGING_VERBS
 
 
+def _kernel_session_accepts_kwarg(name: str) -> bool:
+    """True iff the installed ``databricks_sql_kernel.Session`` constructor
+    declares keyword ``name``.
+
+    The kernel ``Session`` is a PyO3 class with a **fixed** signature (no
+    ``**kwargs`` catch-all), so forwarding a kwarg it doesn't declare raises
+    ``TypeError`` at construction. The phase-7 identity/telemetry kwargs
+    (``driver_name`` etc.) only exist on wheels newer than the pinned
+    ``^0.2.0`` (whose ``Session`` accepts none of them), so we must gate them
+    on what the actually-installed wheel supports rather than pass them
+    unconditionally. Falls **closed** (returns ``False``) when the signature
+    can't be introspected: a PyO3 class only exposes ``__text_signature__``
+    (and thus an introspectable signature) when built with
+    ``#[pyo3(signature=...)]``; otherwise ``inspect.signature`` raises
+    ``ValueError``. Since the pinned ``^0.2.0`` ``Session`` accepts none of
+    these kwargs, forwarding one it doesn't declare is a hard ``TypeError`` at
+    construction that breaks every ``use_kernel=True`` connection, whereas
+    omitting one the wheel *would* have accepted only loses telemetry
+    richness — so we omit the kwarg on introspection failure.
+    """
+    try:
+        params = inspect.signature(_kernel.Session).parameters
+    except (TypeError, ValueError):
+        return False
+    if any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()):
+        return True
+    return name in params
+
+
+def _kernel_telemetry_kwargs(options: Dict[str, Any]) -> Dict[str, Any]:
+    """Build phase-7 telemetry/system kwargs for ``databricks_sql_kernel.Session``.
+
+    Only kwargs the installed ``Session`` constructor actually accepts are
+    returned; on the pinned ``^0.2.0`` wheel (which predates phase 7) this is
+    empty, so ``open_session`` doesn't break with ``TypeError`` on a wheel
+    that doesn't yet know these kwargs.
+    """
+    system = TelemetryHelper.get_driver_system_configuration()
+    candidates: Dict[str, Any] = {
+        "driver_name": system.driver_name,
+        "driver_version": system.driver_version,
+        "runtime_name": system.runtime_name,
+        "runtime_version": system.runtime_version,
+        "runtime_vendor": system.runtime_vendor,
+        "os_name": system.os_name,
+        "os_version": system.os_version,
+        "os_arch": system.os_arch,
+        "client_app_name": system.client_app_name,
+        "locale_name": system.locale_name,
+        "char_set_encoding": system.char_set_encoding,
+        # The Python telemetry model does not currently track process
+        # name; omit it and let the kernel fill what it can derive.
+        "process_name": None,
+    }
+    if options.get("enable_telemetry") is not None:
+        candidates["telemetry_enabled"] = bool(options["enable_telemetry"])
+    if options.get("telemetry_batch_size") is not None:
+        candidates["telemetry_batch_size"] = options["telemetry_batch_size"]
+    if options.get("telemetry_circuit_breaker_enabled") is not None:
+        candidates["telemetry_circuit_breaker_enabled"] = options[
+            "telemetry_circuit_breaker_enabled"
+        ]
+    return {
+        name: value
+        for name, value in candidates.items()
+        if _kernel_session_accepts_kwarg(name)
+    }
+
+
 # ─── Client ─────────────────────────────────────────────────────────────────
 
 
@@ -226,6 +297,9 @@ 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")
+        # 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 {}
         self._catalog = catalog
         self._schema = schema
         # ``_use_arrow_native_complex_types`` is the connector-side
@@ -339,6 +413,7 @@ def open_session(
             # Translate the connector's ``_retry_*`` kwargs into the
             # kernel's ``retry_*`` kwargs. Empty when at defaults.
             retry_kwargs = _kernel_retry_kwargs(self._retry_options)
+            telemetry_kwargs = _kernel_telemetry_kwargs(self._telemetry_options)
             # Forward caller / connector HTTP headers. The kernel applies
             # them on every request; a caller ``User-Agent`` is appended
             # to the kernel's base UA. Only pass the kwarg when there's
@@ -382,6 +457,7 @@ def open_session(
                 **auth_kwargs,
                 **tls_kwargs,
                 **retry_kwargs,
+                **telemetry_kwargs,
                 **http_headers_kwargs,
             )
         except Exception as exc:

src/databricks/sql/client.py

@@ -178,6 +178,16 @@ def __init__(
                 ``ImportError``. Supports PAT, OAuth M2M, and OAuth
                 U2M auth, and native (positional and named) parameter
                 binding. Mutually exclusive with ``use_sea``.
+
+                Telemetry note: on the kernel path telemetry is
+                strictly opt-in. An explicit ``enable_telemetry=True``
+                turns telemetry on unconditionally; it is NOT gated by
+                the server-side ``enableTelemetryForPythonDriver``
+                feature flag that applies on the Thrift/SEA paths. When
+                ``enable_telemetry`` is unset the kernel owns the enable
+                decision. This is an intentional divergence from the
+                Thrift/SEA paths, where an explicit ``True`` can still
+                be suppressed by the feature flag.
             :param use_hybrid_disposition: `bool`, optional (default is False)
                 Use the hybrid disposition instead of the inline disposition.
             :param server_hostname: Databricks instance host name.
@@ -402,8 +412,12 @@ def read(self) -> Optional[OAuthToken]:
             )
             self.session.open()
         except Exception as e:
-            # Respect user's telemetry preference even during connection failure
-            enable_telemetry = kwargs.get("enable_telemetry", True)
+            # 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
+            )
             TelemetryClientFactory.connection_failure_log(
                 error_name="Exception",
                 error_message=str(e),

src/databricks/sql/session.py

@@ -13,6 +13,7 @@
 from databricks.sql.backend.types import SessionId, BackendType
 from databricks.sql.common.unified_http_client import UnifiedHttpClient
 from databricks.sql.common.agent import detect as detect_agent
+from databricks.sql.telemetry.telemetry_client import TelemetryClientFactory
 
 if TYPE_CHECKING:
     from databricks.sql.backend.thrift_backend import ThriftDatabricksClient
@@ -250,6 +251,55 @@ def _create_backend(
                     "_retry_stop_after_attempts_duration"
                 ),
             }
+            # Forward the binding/runtime identity and telemetry knobs
+            # added by kernel telemetry phase 7. Python-side telemetry
+            # still owns feature-flag evaluation and event export for the
+            # Thrift/SEA paths; the kernel path needs the same driver
+            # identity at Session construction time so kernel-owned
+            # telemetry can populate its system configuration.
+            kernel_telemetry_options = {
+                # Preserve the caller's explicit telemetry choice. When unset,
+                # leave it as None so the kernel owns the enable decision --
+                # this is an INTENTIONAL divergence from the Thrift/SEA path,
+                # not an oversight. There, an unset enable_telemetry defaults to
+                # True (client.py) but only actually emits when the
+                # `enableTelemetryForPythonDriver` server feature flag is on
+                # (telemetry_client.py). That feature-flag gate is Python-side
+                # and is bypassed on the kernel path (is_telemetry_enabled
+                # short-circuits to False for use_kernel), so forwarding the
+                # connector's True default here would force telemetry on without
+                # an equivalent gate. Passing None instead defers to the
+                # kernel's own default/gating, which is expected to mirror the
+                # feature-flag-gated wrapper behaviour; only an explicit caller
+                # opt-in/opt-out overrides it.
+                "enable_telemetry": kwargs.get("enable_telemetry"),
+                # Match the connector's default batch size (client.py forwards
+                # the same TelemetryClientFactory.DEFAULT_BATCH_SIZE fallback)
+                # so an unset telemetry_batch_size resolves to the same value
+                # on the kernel path as on the Thrift/SEA path, rather than
+                # letting the kernel silently pick its own internal default.
+                "telemetry_batch_size": kwargs.get(
+                    "telemetry_batch_size", TelemetryClientFactory.DEFAULT_BATCH_SIZE
+                ),
+                # Preserve the caller's explicit circuit-breaker choice. When
+                # unset, leave it as None so the kernel owns the decision --
+                # mirroring the enable_telemetry handling above rather than
+                # forcing a default. This is deliberately NOT defaulted to True:
+                # although ClientContext's signature default is True
+                # (auth/common.py:55), the Thrift/SEA path never reaches it --
+                # build_client_context (utils.py:1018) always passes
+                # _telemetry_circuit_breaker_enabled explicitly (None when the
+                # caller leaves it unset), and ClientContext coerces it with
+                # bool(None) -> False (auth/common.py:89). So the *effective*
+                # Thrift/SEA default when unset is False, not True; forwarding
+                # True here would turn the circuit breaker on for an
+                # unconfigured connection while Thrift/SEA leaves it off.
+                # Passing None instead defers to the kernel's own default;
+                # only an explicit caller value overrides it.
+                "telemetry_circuit_breaker_enabled": kwargs.get(
+                    "_telemetry_circuit_breaker_enabled"
+                ),
+            }
             return KernelDatabricksClient(
                 server_hostname=server_hostname,
                 http_path=http_path,
@@ -263,6 +313,7 @@ def _create_backend(
                 auth_options=kernel_auth_options,
                 retry_options=kernel_retry_options,
                 request_timeout_secs=kwargs.get("_socket_timeout"),
+                telemetry_options=kernel_telemetry_options,
             )
 
         # These reference the lazily-resolved module attributes defined via

src/databricks/sql/telemetry/telemetry_client.py

@@ -122,6 +122,9 @@ def get_auth_flow(auth_provider):
 
     @staticmethod
     def is_telemetry_enabled(connection: "Connection") -> bool:
+        if getattr(connection.session, "use_kernel", False):
+            return False
+
         # Fast path: force enabled - skip feature flag fetch entirely
         if connection.force_enable_telemetry:
             return True