Fix!: sqlmesh.dbt.adapter.RuntimeAdapter.get_columns_in_relation()

returns columns in the proper Column subclass type.

fixes #1874

Author: newtonapple

pyproject.toml

@@ -272,6 +272,9 @@ filterwarnings = [
 ]
 retry_delay = 10
 
+[tool.ruff]
+line-length = 100
+
 [tool.ruff.lint]
 select = [
     "F401",

sqlmesh/core/engine_adapter/bigquery.py

@@ -266,6 +266,13 @@ def create_schema(
                 raise
             logger.warning("Failed to create schema '%s': %s", schema_name, e)
 
+    def get_bq_schema(self, table_name: TableName) -> t.List[bigquery.SchemaField]:
+        table = exp.to_table(table_name)
+        if len(table.parts) == 3 and "." in table.name:
+            self.execute(exp.select("*").from_(table).limit(0))
+            return self._query_job._query_results.schema
+        return self._get_table(table).schema
+
     def columns(
         self, table_name: TableName, include_pseudo_columns: bool = False
     ) -> t.Dict[str, exp.DataType]:

sqlmesh/dbt/adapter.py

@@ -291,14 +291,36 @@ def list_relations_without_caching(self, schema_relation: BaseRelation) -> t.Lis
         return relations
 
     def get_columns_in_relation(self, relation: BaseRelation) -> t.List[Column]:
-        from dbt.adapters.base.column import Column
-
         mapped_table = self._map_table_name(self._normalize(self._relation_to_table(relation)))
+
+        columns = self.engine_adapter.columns(table_name=mapped_table).items()
+
+        if self.project_dialect == "bigquery":
+            # dbt.adapters.bigquery.column.BigQueryColumn has a different constructor signature
+            # We need to use BigQueryColumn.create_from_field() to create the column instead
+            if (
+                hasattr(self.column_type, "create_from_field")
+                and callable(getattr(self.column_type, "create_from_field"))
+                and hasattr(self.engine_adapter, "get_bq_schema")
+                and callable(getattr(self.engine_adapter, "get_bq_schema"))
+            ):
+                return [
+                    self.column_type.create_from_field(field)  # type: ignore
+                    for field in self.engine_adapter.get_bq_schema(mapped_table)  # type: ignore
+                ]
+            from dbt.adapters.base.column import Column
+
+            return [
+                Column.from_description(
+                    name=name, raw_data_type=dtype.sql(dialect=self.project_dialect)
+                )
+                for name, dtype in columns
+            ]
         return [
-            Column.from_description(
+            self.column_type.from_description(
                 name=name, raw_data_type=dtype.sql(dialect=self.project_dialect)
             )
-            for name, dtype in self.engine_adapter.columns(table_name=mapped_table).items()
+            for name, dtype in columns
         ]
 
     def get_missing_columns(