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,69 @@ 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 open (returns ``True``) only when the signature
+    can't be introspected, so a future non-introspectable binding still gets
+    the kwargs.
+    """
+    try:
+        params = inspect.signature(_kernel.Session).parameters
+    except (TypeError, ValueError):
+        return True
+    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 +291,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 +407,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 +451,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

tests/unit/test_kernel_client.py

@@ -368,6 +368,157 @@ def fake_session(**kw):
     assert captured["request_timeout_secs"] == timeout
 
 
+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``."""
+    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)
+    monkeypatch.setattr(
+        kernel_client.TelemetryHelper,
+        "get_driver_system_configuration",
+        lambda: types.SimpleNamespace(
+            driver_name="Databricks SQL Python Connector",
+            driver_version="1.2.3",
+            runtime_name="Python 3.12.0",
+            runtime_version="3.12.0",
+            runtime_vendor="CPython",
+            os_name="Linux",
+            os_version="6.1",
+            os_arch="x86_64",
+            client_app_name=None,
+            locale_name="en_US",
+            char_set_encoding="utf-8",
+        ),
+    )
+
+    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,
+        telemetry_options={
+            "enable_telemetry": True,
+            "telemetry_batch_size": 17,
+            "telemetry_circuit_breaker_enabled": False,
+        },
+    )
+    c.open_session(session_configuration=None, catalog=None, schema=None)
+
+    assert captured["driver_name"] == "Databricks SQL Python Connector"
+    assert captured["driver_version"] == "1.2.3"
+    assert captured["runtime_name"] == "Python 3.12.0"
+    assert captured["runtime_version"] == "3.12.0"
+    assert captured["runtime_vendor"] == "CPython"
+    assert captured["os_name"] == "Linux"
+    assert captured["os_version"] == "6.1"
+    assert captured["os_arch"] == "x86_64"
+    assert captured["client_app_name"] is None
+    assert captured["locale_name"] == "en_US"
+    assert captured["char_set_encoding"] == "utf-8"
+    assert captured["process_name"] is None
+    assert captured["telemetry_enabled"] is True
+    assert captured["telemetry_batch_size"] == 17
+    assert captured["telemetry_circuit_breaker_enabled"] is False
+
+
+def test_open_session_omits_phase_7_kwargs_kernel_does_not_accept(monkeypatch):
+    """Phase-7 identity/telemetry kwargs must NOT be forwarded to a kernel
+    ``Session`` whose (fixed, no-``**kwargs``) constructor doesn't declare
+    them.
+
+    The real ``databricks_sql_kernel.Session`` is a PyO3 class with a fixed
+    signature; the pinned ``^0.2.0`` wheel predates phase 7 and accepts none
+    of these kwargs, so forwarding them unconditionally raises ``TypeError``
+    and breaks every ``use_kernel=True`` connection. The other tests here use
+    a ``**kwargs`` MagicMock that silently swallows the kwargs and hides the
+    break; this one uses a fixed-signature fake mirroring the real 0.2.0
+    surface to prove the client gates on what the installed Session supports.
+    """
+    captured = {}
+
+    # Fixed signature mirroring the pinned 0.2.0 kernel Session: it accepts
+    # the base connection/tls/retry kwargs but NONE of the phase-7 identity
+    # or telemetry kwargs, and has no **kwargs catch-all.
+    def fake_session_v0_2_0(
+        host,
+        http_path,
+        *,
+        auth_type=None,
+        access_token=None,
+        client_id=None,
+        client_secret=None,
+        oauth_scopes=None,
+        token_url=None,
+        redirect_port=None,
+        oauth_callback_timeout_secs=None,
+        tls_ca_cert=None,
+        tls_skip_verify=False,
+        tls_skip_hostname_verify=False,
+        tls_client_cert=None,
+        tls_client_key=None,
+        retry_min_wait_secs=None,
+        retry_max_wait_secs=None,
+        retry_max_attempts=None,
+        retry_overall_timeout_secs=None,
+        http_headers=None,
+        catalog=None,
+        schema=None,
+        session_conf=None,
+        complex_types_as_json=False,
+        intervals_as_string=False,
+        request_timeout_secs=None,
+    ):
+        captured["host"] = host
+        sess = MagicMock()
+        sess.session_id = "sess-id"
+        return sess
+
+    monkeypatch.setattr(kernel_client._kernel, "Session", fake_session_v0_2_0)
+    monkeypatch.setattr(
+        kernel_client.TelemetryHelper,
+        "get_driver_system_configuration",
+        lambda: types.SimpleNamespace(
+            driver_name="Databricks SQL Python Connector",
+            driver_version="1.2.3",
+            runtime_name="Python 3.12.0",
+            runtime_version="3.12.0",
+            runtime_vendor="CPython",
+            os_name="Linux",
+            os_version="6.1",
+            os_arch="x86_64",
+            client_app_name=None,
+            locale_name="en_US",
+            char_set_encoding="utf-8",
+        ),
+    )
+
+    # The kwargs the client builds must be filtered to what fake_session
+    # accepts, so open_session succeeds instead of raising TypeError.
+    kwargs = kernel_client._kernel_telemetry_kwargs(
+        {"enable_telemetry": True, "telemetry_batch_size": 17}
+    )
+    assert kwargs == {}, f"expected no phase-7 kwargs on 0.2.0 Session, got {kwargs}"
+
+    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,
+        telemetry_options={"enable_telemetry": True, "telemetry_batch_size": 17},
+    )
+    # Would raise TypeError: unexpected keyword argument if the client
+    # forwarded phase-7 kwargs the fixed-signature Session doesn't declare.
+    c.open_session(session_configuration=None, catalog=None, schema=None)
+    assert captured["host"] == "example.cloud.databricks.com"
+
+
 def test_execute_command_forwards_parameters_to_bind_param():
     """``execute_command(parameters=[...])`` routes each parameter
     through ``bind_tspark_params`` onto the kernel statement before