feat(observability): add wire-level sqlcommenter, zero-RTT Oracle trace piggybacking, and DuckDB statement profiling

sqlspec/adapters/adbc/config.py

@@ -1,11 +1,13 @@
 """ADBC database configuration."""
 
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, NotRequired, TypedDict, cast
 
 from sqlspec.adapters.adbc._typing import AdbcConnection, AdbcCursor, AdbcSessionContext
 from sqlspec.adapters.adbc.core import (
     apply_driver_features,
     build_connection_config,
+    configure_adbc_duckdb_profiling,
     detect_postgres_extensions,
     get_statement_config,
     is_postgres_dialect,
@@ -15,6 +17,7 @@
     resolve_driver_connect_func,
 )
 from sqlspec.adapters.adbc.driver import AdbcDriver, AdbcExceptionHandler
+from sqlspec.adapters.duckdb.core import cleanup_duckdb_profile_path, create_duckdb_profile_path
 from sqlspec.config import ExtensionConfigs, NoPoolSyncConfig
 from sqlspec.core import StatementConfig
 from sqlspec.core.capabilities import TypeCoercionCapabilities
@@ -135,6 +138,8 @@ class AdbcDriverFeatures(TypedDict):
             Only option: "poll_queue" (durable table-backed queue with lease-based retries and acknowledgements).
             ADBC does not have native pub/sub, so poll_queue is the only backend.
             Defaults to "poll_queue".
+        enable_statement_profiling: Enable DuckDB JSON statement profiling when using a DuckDB ADBC backend.
+            Defaults to False.
     """
 
     json_serializer: "NotRequired[Callable[[Any], str]]"
@@ -150,6 +155,7 @@ class AdbcDriverFeatures(TypedDict):
     enable_events: NotRequired[bool]
     on_connection_create: "NotRequired[Callable[[AdbcConnection], None]]"
     events_backend: NotRequired[Literal["poll_queue"]]
+    enable_statement_profiling: NotRequired[bool]
 
 
 class AdbcConnectionContext(SyncPoolConnectionContext):
@@ -176,22 +182,42 @@ def __exit__(
 
 
 class _AdbcSessionConnectionHandler(SyncPoolSessionFactory):
-    __slots__ = ("_connection",)
+    __slots__ = ("_connection", "_driver_features", "_profile_path")
 
     def __init__(self, config: "AdbcConfig") -> None:
         super().__init__(config)
         self._connection: AdbcConnection | None = None
+        self._driver_features: dict[str, Any] = dict(config.driver_features)
+        self._profile_path: Path | None = None
+
+    @property
+    def driver_features(self) -> "dict[str, Any]":
+        return self._driver_features
 
     def acquire_connection(self) -> "AdbcConnection":
         self._connection = self._config.create_connection()
+        if self._config._resolved_dialect == "duckdb" and bool(
+            self._driver_features.get("enable_statement_profiling", False)
+        ):
+            self._profile_path = create_duckdb_profile_path()
+            configure_adbc_duckdb_profiling(self._connection, self._profile_path)
+            self._driver_features["_duckdb_profile_path"] = str(self._profile_path)
         return cast("AdbcConnection", self._connection)
 
     def release_connection(self, _conn: "AdbcConnection", **kwargs: Any) -> None:
         if self._connection is None:
             return
-        if self._connection is not self._config.connection_instance:
-            self._connection.close()
-        self._connection = None
+        try:
+            if self._connection is not self._config.connection_instance:
+                self._connection.close()
+        finally:
+            self._connection = None
+            if self._profile_path is not None:
+                cleanup_duckdb_profile_path(self._profile_path)
+                self._profile_path = None
+            raw_path = self._driver_features.pop("_duckdb_profile_path", None)
+            if isinstance(raw_path, str):
+                cleanup_duckdb_profile_path(Path(raw_path))
 
 
 class AdbcConfig(NoPoolSyncConfig[AdbcConnection, AdbcDriver]):
@@ -369,7 +395,7 @@ def provide_session(
             acquire_connection=handler.acquire_connection,
             release_connection=handler.release_connection,
             statement_config=statement_config,
-            driver_features=self.driver_features,
+            driver_features=handler.driver_features,
             prepare_driver=self._prepare_driver,
         )
 

sqlspec/adapters/adbc/core.py

@@ -1,5 +1,6 @@
 """ADBC adapter compiled helpers."""
 
+import contextlib
 import datetime as dt
 import decimal
 import re
@@ -50,6 +51,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Mapping
+    from pathlib import Path
 
     from sqlspec.core import SQL
     from sqlspec.utils.dispatch import TypeDispatcher
@@ -59,6 +61,7 @@
     "build_connection_config",
     "build_profile",
     "collect_rows",
+    "configure_adbc_duckdb_profiling",
     "create_mapped_exception",
     "detect_dialect",
     "detect_postgres_extensions",
@@ -983,13 +986,33 @@ def apply_driver_features(
     processed_features.setdefault("enable_paradedb", True)
     processed_features.setdefault("enable_pg_textsearch", True)
     processed_features.setdefault("enable_lowercase_column_names", resolved_dialect in {"oracle", "db2"})
+    processed_features.setdefault("enable_statement_profiling", False)
 
     if json_serializer is not None:
         statement_config = _apply_adbc_json_serializer(statement_config, json_serializer)
 
     return statement_config, processed_features
 
 
+def configure_adbc_duckdb_profiling(connection: Any, profile_path: "Path") -> None:
+    """Enable DuckDB JSON statement profiling on an ADBC DuckDB connection.
+
+    Args:
+        connection: Active ADBC DuckDB connection.
+        profile_path: Temporary file path where DuckDB writes statement profiles.
+    """
+    escaped_path = str(profile_path).replace("'", "''")
+    cursor = connection.cursor()
+    try:
+        cursor.execute("PRAGMA enable_profiling='json'")
+        with contextlib.suppress(Exception):
+            cursor.execute("SET profiling_coverage='ALL'")
+        cursor.execute(f"SET profiling_output='{escaped_path}'")
+    finally:
+        with contextlib.suppress(Exception):
+            cursor.close()
+
+
 def normalize_column_names(
     column_names: "list[str]", driver_features: "Mapping[str, Any] | None" = None
 ) -> "list[str]":

sqlspec/adapters/adbc/driver.py

@@ -5,12 +5,15 @@
 """
 
 import contextlib
+from pathlib import Path
+from time import perf_counter
 from typing import TYPE_CHECKING, Any, Literal, cast, final
 
 from sqlspec.adapters.adbc._typing import AdbcCursor, AdbcNativeError, AdbcSessionContext
 from sqlspec.adapters.adbc.core import (
     apply_driver_features,
     collect_rows,
+    configure_adbc_duckdb_profiling,
     create_mapped_exception,
     detect_dialect,
     driver_profile,
@@ -31,6 +34,11 @@
     resolve_rowcount,
 )
 from sqlspec.adapters.adbc.data_dictionary import AdbcDataDictionary
+from sqlspec.adapters.duckdb.core import (
+    cleanup_duckdb_profile_path,
+    create_duckdb_profile_path,
+    read_duckdb_profile_json,
+)
 from sqlspec.core import (
     SQL,
     StatementConfig,
@@ -239,6 +247,7 @@ def dispatch_execute(self, cursor: "AdbcRawCursor", statement: SQL) -> "Executio
             Execution result with data or row count
         """
         sql, prepared_parameters = self._compiled_sql(statement, self.statement_config)
+        self._ensure_statement_profiling()
 
         try:
             execute_parameters = normalize_postgres_empty_parameters(self._dialect_name, prepared_parameters)
@@ -252,6 +261,7 @@ def dispatch_execute(self, cursor: "AdbcRawCursor", statement: SQL) -> "Executio
 
         if is_select_like:
             arrow_table = cursor.fetch_arrow_table()
+            self._capture_statement_profile(sql)
             raw_column_names = list(arrow_table.column_names)
             column_names = normalize_column_names(raw_column_names, self.driver_features)
             if column_names != raw_column_names and hasattr(arrow_table, "rename_columns"):
@@ -270,6 +280,7 @@ def dispatch_execute(self, cursor: "AdbcRawCursor", statement: SQL) -> "Executio
             )
 
         row_count = self._resolve_count_result_rowcount(cursor, fallback=resolve_rowcount(cursor))
+        self._capture_statement_profile(sql)
         return self.create_execution_result(cursor, rowcount_override=row_count)
 
     def dispatch_execute_many(self, cursor: "AdbcRawCursor", statement: SQL) -> "ExecutionResult":
@@ -338,28 +349,57 @@ def dispatch_execute_script(self, cursor: "AdbcRawCursor", statement: "SQL") ->
 
         successful_count = 0
         last_rowcount = 0
+        total_statements = len(statements)
+        runtime = self._observability
+        has_observers = runtime is not None and runtime.has_statement_observers
+        profiling_enabled = self._dialect_name == "duckdb" and bool(
+            self.driver_features.get("enable_statement_profiling", False)
+        )
+        if profiling_enabled:
+            self._ensure_statement_profiling()
         try:
-            for stmt in statements:
+            for idx, stmt in enumerate(statements):
+                stmt_started = perf_counter() if has_observers else 0.0
+                stmt_params: Any = ()
                 if prepared_parameters and (
-                    len(statements) == 1 or any(marker in stmt for marker in ("?", "$", "%", "@", ":"))
+                    total_statements == 1 or any(marker in stmt for marker in ("?", "$", "%", "@", ":"))
                 ):
-                    postgres_compatible_params = normalize_postgres_empty_parameters(
-                        self._dialect_name, prepared_parameters
-                    )
-                    cursor.execute(stmt, parameters=postgres_compatible_params)
+                    stmt_params = normalize_postgres_empty_parameters(self._dialect_name, prepared_parameters)
+                    cursor.execute(stmt, parameters=stmt_params)
                 else:
                     cursor.execute(stmt)
                 successful_count += 1
-                last_rowcount = self._resolve_count_result_rowcount(
+                stmt_rowcount = self._resolve_count_result_rowcount(
                     cursor, fallback=normalize_script_rowcount(last_rowcount, cursor)
                 )
+                last_rowcount = stmt_rowcount
+                stmt_profile = self._capture_statement_profile(stmt) if profiling_enabled else None
+                if has_observers:
+                    stmt_duration = perf_counter() - stmt_started
+                    single_stmt = self._sub_statement(stmt, stmt_params)
+                    try:
+                        stmt_operation = single_stmt.operation_type
+                    finally:
+                        self._release_pooled_statement(single_stmt)
+                    self._emit_script_statement_event(
+                        sql=stmt,
+                        parameters=stmt_params,
+                        operation=stmt_operation,
+                        rows_affected=stmt_rowcount,
+                        duration_s=stmt_duration,
+                        started_at=stmt_started,
+                        profile=stmt_profile,
+                        script_ordinal=idx,
+                        script_statement_count=total_statements,
+                    )
         except Exception:
             handle_postgres_rollback(self._dialect_name, cursor, logger)
             raise
 
+        self._pending_statement_profile = None
         return self.create_execution_result(
             cursor,
-            statement_count=len(statements),
+            statement_count=total_statements,
             successful_statements=successful_count,
             rowcount_override=last_rowcount,
             is_script_result=True,
@@ -826,6 +866,33 @@ def _detect_flightsql_connection(connection: "AdbcConnection") -> bool:
     def _resolve_column_names(self, description: Any) -> list[str]:
         return resolve_column_names(description, self._column_name_cache, self.driver_features)
 
+    def _ensure_statement_profiling(self) -> "Path | None":
+        if self._dialect_name != "duckdb" or not bool(self.driver_features.get("enable_statement_profiling", False)):
+            return None
+        raw_path = self.driver_features.get("_duckdb_profile_path")
+        if isinstance(raw_path, Path):
+            return raw_path
+        if isinstance(raw_path, str) and raw_path:
+            return Path(raw_path)
+        profile_path = create_duckdb_profile_path()
+        try:
+            configure_adbc_duckdb_profiling(self.connection, profile_path)
+        except Exception:
+            cleanup_duckdb_profile_path(profile_path)
+            return None
+        self.driver_features["_duckdb_profile_path"] = str(profile_path)
+        return profile_path
+
+    def _capture_statement_profile(self, executed_sql: str) -> "dict[str, Any] | None":
+        if self._dialect_name != "duckdb" or not bool(self.driver_features.get("enable_statement_profiling", False)):
+            return None
+        profile_path = self._ensure_statement_profiling()
+        if profile_path is None:
+            return None
+        profile = read_duckdb_profile_json(profile_path, executed_sql)
+        self._pending_statement_profile = profile
+        return profile
+
     @staticmethod
     def _resolve_count_result_rowcount(cursor: "AdbcRawCursor", *, fallback: int) -> int:
         """Consume ADBC count result streams and return their row count."""

sqlspec/adapters/duckdb/config.py

@@ -176,6 +176,8 @@ class DuckDBDriverFeatures(TypedDict):
             extension_config["events"] is set; requires it for migrations.
         events_backend: Only "poll_queue" is supported, with lease-based retries
             and acknowledgements. DuckDB has no native pub/sub.
+        enable_statement_profiling: Enable per-statement JSON operator tree
+            profiling attached to ``StatementEvent.profile`` (default False).
     """
 
     extensions: NotRequired[Sequence[DuckDBExtensionConfig]]
@@ -186,6 +188,7 @@ class DuckDBDriverFeatures(TypedDict):
     extension_flags: NotRequired["dict[str, Any]"]
     enable_events: NotRequired[bool]
     events_backend: NotRequired[Literal["poll_queue"]]
+    enable_statement_profiling: NotRequired[bool]
 
 
 class DuckDBConnectionContext(SyncPoolConnectionContext):
@@ -369,6 +372,7 @@ def _create_pool(self) -> DuckDBConnectionPool:
             extensions=extensions_dicts,
             secrets=secrets_dicts,
             on_connection_create=self._user_connection_hook,
+            enable_statement_profiling=bool(self.driver_features.get("enable_statement_profiling", False)),
             **pool_kwargs,
         )