feat: Add instrumentation for recording add_query, get_relation_config, is_uniform, has_dbr_capability, and is_cluster methods (#1508)

<!-- Please review our pull request review process in CONTRIBUTING.md
before your proceed. -->

Resolves #N/A.

<!---
Include the number of the issue addressed by this PR above if
applicable.

  Example:
    resolves #1234

Please review our pull request review process in CONTRIBUTING.md before
your proceed.
-->

### Description

Instruments the following adapter functions for recording:
- `has_dbr_capability` and `is_cluster` (via the auto record decorator)
- `add_query`
- `get_relation_config`
- `is_uniform`

### Checklist

- [X] I have run this code in development and it appears to resolve the
stated issue
- [X] This PR includes tests, or tests are not required/relevant for
this PR
- [ ] I have updated the `CHANGELOG.md` and added information about my
change to the "dbt-databricks next" section.

---------

Signed-off-by: Anna Lee <anna.lee@dbtlabs.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ajhlee-dbt

dbt/adapters/databricks/impl.py

@@ -32,6 +32,7 @@
 from dbt_common.behavior_flags import BehaviorFlag
 from dbt_common.contracts.config.base import BaseConfig, MergeBehavior
 from dbt_common.exceptions import DbtConfigError, DbtInternalError, DbtRuntimeError
+from dbt_common.record import auto_record_function, record_function
 from dbt_common.utils import executor
 from dbt_common.utils.dict import AttrDict
 from packaging import version
@@ -62,6 +63,11 @@
     ServerlessClusterPythonJobHelper,
     WorkflowPythonJobHelper,
 )
+from dbt.adapters.databricks.record.record_types import (
+    DatabricksAdapterAddQueryRecord,
+    DatabricksAdapterGetRelationConfigRecord,
+    DatabricksAdapterIsUniformRecord,
+)
 from dbt.adapters.databricks.relation import (
     KEY_TABLE_PROVIDER,
     DatabricksRelation,
@@ -332,6 +338,12 @@ def quote(self, identifier):  # type: ignore[override,no-untyped-def]
         return quote(identifier)
 
     @available.parse(lambda *a, **k: 0)
+    @record_function(
+        DatabricksAdapterIsUniformRecord,
+        method=True,
+        index_on_thread_id=True,
+        id_field_name="thread_id",
+    )
     def is_uniform(self, config: BaseConfig) -> bool:
         catalog_relation: DatabricksCatalogRelation = self.build_catalog_relation(config.model)  # type:ignore
         if catalog_relation.table_format == constants.ICEBERG_TABLE_FORMAT:
@@ -425,6 +437,7 @@ def has_capability(self, capability: DBRCapability) -> bool:
         return conn.has_capability(capability)
 
     @available.parse(lambda *a, **k: False)
+    @auto_record_function("AdapterHasDbrCapability", group="Available")
     def has_dbr_capability(self, capability_name: str) -> bool:
         """
         Check if a DBR capability is available for current compute.
@@ -889,6 +902,13 @@ def get_behavior_flag_no_warn(self, behavior_flag_name: str) -> bool:
         behavior_flag = getattr(self.behavior, behavior_flag_name)
         return behavior_flag.no_warn
 
+    @available.parse(lambda *a, **k: (None, None))
+    @record_function(
+        DatabricksAdapterAddQueryRecord,
+        method=True,
+        index_on_thread_id=True,
+        id_field_name="thread_id",
+    )
     def add_query(
         self,
         sql: str,
@@ -1059,6 +1079,12 @@ def parse_columns_and_constraints(
         return enriched_columns, parsed_constraints
 
     @available.parse(lambda *a, **k: {})
+    @record_function(
+        DatabricksAdapterGetRelationConfigRecord,
+        method=True,
+        index_on_thread_id=True,
+        id_field_name="thread_id",
+    )
     def get_relation_config(
         self,
         relation: DatabricksRelation,
@@ -1097,6 +1123,7 @@ def get_config_from_model(self, model: RelationConfig) -> DatabricksRelationConf
             )
 
     @available
+    @auto_record_function("AdapterIsCluster", group="Available")
     def is_cluster(self) -> bool:
         """Check if the current connection is a cluster."""
         return self.connections.is_cluster()

dbt/adapters/databricks/record/record_types.py

@@ -0,0 +1,149 @@
+import dataclasses
+import json
+from typing import Any, Optional
+
+from dbt.adapters.record.serialization import serialize_bindings
+from dbt_common.record import Record, Recorder
+
+from dbt.adapters.databricks.relation import DatabricksRelation
+from dbt.adapters.databricks.relation_configs.base import DatabricksRelationConfigBase
+
+
+def _serialize_component(component: Any) -> dict[str, Any]:
+    """Serialize a relation config component to a JSON-native dict.
+
+    pydantic v2 exposes ``model_dump(mode="json")``; v1 only has ``json()``,
+    which is round-tripped to coerce tuples, sets, and enums to JSON types.
+    Under v1 the ``model_config`` ConfigDict is treated as a regular field and
+    leaks into the output, so it is dropped to match the v2 shape.
+    """
+    if hasattr(component, "model_dump"):
+        return component.model_dump(mode="json")
+    serialized = json.loads(component.json())
+    serialized.pop("model_config", None)
+    return serialized
+
+
+@dataclasses.dataclass
+class DatabricksAdapterAddQueryParams:
+    thread_id: str
+    sql: str
+    auto_begin: bool = True
+    bindings: Optional[Any] = None
+    abridge_sql_log: bool = False
+    close_cursor: bool = False
+
+    def _to_dict(self) -> dict[str, Any]:
+        return {
+            "thread_id": self.thread_id,
+            "sql": self.sql,
+            "auto_begin": self.auto_begin,
+            "bindings": serialize_bindings(self.bindings),
+            "abridge_sql_log": self.abridge_sql_log,
+            "close_cursor": self.close_cursor,
+        }
+
+
+@dataclasses.dataclass
+class DatabricksAdapterAddQueryResult:
+    return_val: tuple
+
+    def _to_dict(self) -> dict[str, Any]:
+        return {
+            "return_val": {
+                "conn": "conn",
+                "cursor": "cursor",
+            }
+        }
+
+
+@Recorder.register_record_type
+class DatabricksAdapterAddQueryRecord(Record):
+    """Implements record/replay support for the DatabricksAdapter.add_query() method."""
+
+    params_cls = DatabricksAdapterAddQueryParams
+    result_cls = DatabricksAdapterAddQueryResult
+    group = "Available"
+
+
+@dataclasses.dataclass
+class DatabricksAdapterIsUniformParams:
+    thread_id: str
+    config: Any
+
+    def _to_dict(self) -> dict[str, Any]:
+        return {
+            "thread_id": self.thread_id,
+            "config_model": self.config.model.to_dict(),
+            "materialized": self.config.get("materialized"),
+        }
+
+
+@dataclasses.dataclass
+class DatabricksAdapterIsUniformResult:
+    return_val: bool
+
+
+@Recorder.register_record_type
+class DatabricksAdapterIsUniformRecord(Record):
+    """Implements record/replay support for the DatabricksAdapter.is_uniform() method."""
+
+    params_cls = DatabricksAdapterIsUniformParams
+    result_cls = DatabricksAdapterIsUniformResult
+    group = "Available"
+
+
+@dataclasses.dataclass
+class DatabricksAdapterGetRelationConfigParams:
+    thread_id: str
+    relation: DatabricksRelation
+    model_config: Optional[DatabricksRelationConfigBase]
+
+    def _to_dict(self) -> dict[str, Any]:
+        from dbt.adapters.record.serialization import serialize_base_relation
+
+        return {
+            "thread_id": self.thread_id,
+            "relation": serialize_base_relation(self.relation),
+            "model_config": (
+                {
+                    "config": {
+                        k: _serialize_component(v) for k, v in self.model_config.config.items()
+                    }
+                }
+                if self.model_config
+                else None
+            ),
+        }
+
+
+@dataclasses.dataclass
+class DatabricksAdapterGetRelationConfigResult:
+    return_val: Optional[dict]  # DatabricksRelationConfigBase serialized as dict, or None
+
+    def _to_dict(self) -> dict[str, Any]:
+        # return_val is already converted to dict by the constructor
+        return {
+            "return_val": self.return_val,
+        }
+
+    def __init__(self, return_val: Optional[Any]) -> None:
+        if return_val is not None and not isinstance(return_val, dict):
+            self.return_val = {
+                "config": {k: _serialize_component(v) for k, v in return_val.config.items()}
+            }
+        else:
+            self.return_val = return_val
+
+    @classmethod
+    def _from_dict(cls, data: dict[str, Any]) -> "DatabricksAdapterGetRelationConfigResult":
+        return cls(return_val=data.get("return_val"))
+
+
+@Recorder.register_record_type
+class DatabricksAdapterGetRelationConfigRecord(Record):
+    """Implements record/replay support for the DatabricksAdapter.get_relation_config() method."""
+
+    params_cls = DatabricksAdapterGetRelationConfigParams
+    result_cls = DatabricksAdapterGetRelationConfigResult
+    group = "Available"

tests/unit/record/__init__.py

@@ -0,0 +1,122 @@
+{
+  "database": "test_schema",
+  "schema": "test_schema",
+  "name": "m_table",
+  "resource_type": "model",
+  "package_name": "test_project",
+  "path": "m_table.sql",
+  "original_file_path": "models/m_table.sql",
+  "unique_id": "model.test_project.m_table",
+  "fqn": [
+    "test_project",
+    "m_table"
+  ],
+  "alias": "m_table",
+  "checksum": {
+    "name": "sha256",
+    "checksum": ""
+  },
+  "config": {
+    "enabled": true,
+    "alias": null,
+    "schema": null,
+    "database": null,
+    "tags": [],
+    "meta": {},
+    "group": null,
+    "materialized": "table",
+    "incremental_strategy": null,
+    "batch_size": null,
+    "lookback": 1,
+    "begin": null,
+    "persist_docs": {},
+    "post-hook": [],
+    "pre-hook": [],
+    "quoting": {},
+    "column_types": {},
+    "full_refresh": null,
+    "unique_key": null,
+    "on_schema_change": "ignore",
+    "on_configuration_change": "apply",
+    "grants": {},
+    "packages": [],
+    "docs": {
+      "show": true,
+      "node_color": null
+    },
+    "contract": {
+      "enforced": false,
+      "alias_types": true
+    },
+    "event_time": null,
+    "concurrent_batches": null,
+    "access": "protected",
+    "freshness": null
+  },
+  "tags": [],
+  "description": "",
+  "columns": {
+    "id": {
+      "name": "id",
+      "description": "",
+      "meta": {},
+      "data_type": null,
+      "constraints": [],
+      "quote": null,
+      "config": {
+        "meta": {},
+        "tags": []
+      },
+      "tags": [],
+      "granularity": null,
+      "doc_blocks": []
+    }
+  },
+  "meta": {},
+  "group": null,
+  "docs": {
+    "show": true,
+    "node_color": null
+  },
+  "patch_path": "test_project://models/schema.yml",
+  "build_path": null,
+  "unrendered_config": {
+    "materialized": "table"
+  },
+  "created_at": 0,
+  "config_call_dict": {
+    "materialized": "table"
+  },
+  "unrendered_config_call_dict": {},
+  "relation_name": "`test_schema`.`test_schema`.`m_table`",
+  "raw_code": "{{ config(materialized='table') }}\nselect 1 as id, 'a' as val",
+  "doc_blocks": [],
+  "language": "sql",
+  "refs": [],
+  "sources": [],
+  "metrics": [],
+  "functions": [],
+  "depends_on": {
+    "macros": [],
+    "nodes": []
+  },
+  "compiled_path": "target/compiled/test_project/models/m_table.sql",
+  "compiled": true,
+  "compiled_code": "\nselect 1 as id, 'a' as val",
+  "extra_ctes_injected": true,
+  "extra_ctes": [],
+  "contract": {
+    "enforced": false,
+    "alias_types": true,
+    "checksum": null
+  },
+  "access": "protected",
+  "constraints": [],
+  "version": null,
+  "latest_version": null,
+  "deprecation_date": null,
+  "defer_relation": null,
+  "primary_key": [],
+  "time_spine": null,
+  "batch": null
+}