Record additional parameters from BaseColumn subclasses in adapter.get_columns_for_relation (#1383)

Author: MichelleArk

dbt-adapters/.changes/unreleased/Under the Hood-20251010-153203.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Record all fields in subclasses of BaseColumns when recording adapter.get_columns_in_relation
+time: 2025-10-10T15:32:03.863594-04:00
+custom:
+    Author: michelleark
+    Issue: "1383"

dbt-adapters/src/dbt/adapters/base/impl.py

@@ -31,6 +31,7 @@
     AdapterConvertTypeRecord,
     AdapterStandardizeGrantsDictRecord,
     AdapterListRelationsWithoutCachingRecord,
+    AdapterGetColumnsInRelationRecord,
 )
 from dbt_common.behavior_flags import Behavior, BehaviorFlag
 from dbt_common.clients.jinja import CallableMacroGenerator
@@ -745,7 +746,12 @@ def rename_relation(self, from_relation: BaseRelation, to_relation: BaseRelation
         """
         raise NotImplementedError("`rename_relation` is not implemented for this adapter!")
 
-    @auto_record_function("AdapterGetColumnsInRelation", group="Available")
+    @record_function(
+        AdapterGetColumnsInRelationRecord,
+        method=True,
+        index_on_thread_id=True,
+        id_field_name="thread_id",
+    )
     @abc.abstractmethod
     @available.parse_list
     def get_columns_in_relation(self, relation: BaseRelation) -> List[BaseColumn]:

dbt-adapters/src/dbt/adapters/record/base.py

@@ -11,6 +11,7 @@
 if TYPE_CHECKING:
     from agate import Table
     from dbt.adapters.base.relation import BaseRelation
+    from dbt.adapters.base.column import Column as BaseColumn
 
 
 @dataclasses.dataclass
@@ -244,3 +245,47 @@ class AdapterListRelationsWithoutCachingRecord(Record):
     params_cls = AdapterListRelationsWithoutCachingParams
     result_cls = AdapterListRelationsWithoutCachingResult
     group = "Available"
+
+
+@dataclasses.dataclass
+class AdapterGetColumnsInRelationParams:
+    thread_id: str
+    relation: "BaseRelation"
+
+    def _to_dict(self):
+        from dbt.adapters.record.serialization import serialize_base_relation
+
+        return {
+            "thread_id": self.thread_id,
+            "relation": serialize_base_relation(self.relation),
+        }
+
+    def _from_dict(self, data: Dict[str, Any]):
+        from dbt.adapters.record.serialization import deserialize_base_relation
+
+        self.thread_id = data["thread_id"]
+        self.relation = deserialize_base_relation(data["relation"])
+
+
+@dataclasses.dataclass
+class AdapterGetColumnsInRelationResult:
+    return_val: List["BaseColumn"]
+
+    def _to_dict(self):
+        from dbt.adapters.record.serialization import serialize_base_column_list
+
+        return {"return_val": serialize_base_column_list(self.return_val)}
+
+    def _from_dict(self, data: Dict[str, Any]):
+        from dbt.adapters.record.serialization import deserialize_base_column_list
+
+        self.return_val = deserialize_base_column_list(data["return_val"])
+
+
+@Recorder.register_record_type
+class AdapterGetColumnsInRelationRecord(Record):
+    """Implements record/replay support for the BaseAdapter.get_columns_in_relation() method."""
+
+    params_cls = AdapterGetColumnsInRelationParams
+    result_cls = AdapterGetColumnsInRelationResult
+    group = "Available"

dbt-adapters/src/dbt/adapters/record/serialization.py

@@ -1,10 +1,12 @@
+import dataclasses
 from datetime import datetime, date
 from decimal import Decimal
 from typing import Any, Dict, TYPE_CHECKING, List, Union
 
 if TYPE_CHECKING:
     from agate import Table
     from dbt.adapters.base.relation import BaseRelation
+    from dbt.adapters.base.column import Column as BaseColumn
 
 
 def _column_filter(val: Any) -> Any:
@@ -61,3 +63,27 @@ def deserialize_base_relation(relation_dict: Dict[str, Any]) -> "BaseRelation":
 def deserialize_base_relation_list(relations_data: List[Dict[str, Any]]) -> List["BaseRelation"]:
     """Deserialize a list of BaseRelation objects from dictionaries."""
     return [deserialize_base_relation(relation_dict) for relation_dict in relations_data]
+
+
+def serialize_base_column_list(columns: List["BaseColumn"]) -> List[Dict[str, Any]]:
+    return [serialize_base_column(column) for column in columns]
+
+
+def serialize_base_column(column: "BaseColumn") -> Dict[str, Any]:
+    column_dict = dataclasses.asdict(column)
+    return column_dict
+
+
+def deserialize_base_column_list(columns_data: List[Dict[str, Any]]) -> List["BaseColumn"]:
+    return [deserialize_base_column(column_dict) for column_dict in columns_data]
+
+
+def deserialize_base_column(column_dict: Dict[str, Any]) -> "BaseColumn":
+    # Only include fields that are present in the base column class
+    params_dict = {
+        field.name: column_dict[field.name]
+        for field in dataclasses.fields(BaseColumn)
+        if field.name in column_dict
+    }
+
+    return BaseColumn(**params_dict)