Fix issue with seeds using data_type from columns too

Author: georgesittas

sqlmesh/dbt/seed.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import copy
 import typing as t
 
 import agate
@@ -50,15 +49,11 @@ def to_sqlmesh(
         """Converts the dbt seed into a SQLMesh model."""
         seed_path = self.path.absolute().as_posix()
 
-        if column_types := self.column_types:
-            column_types_override = copy.deepcopy(self.columns)
-            for name, data_type in column_types.items():
-                column = column_types_override.setdefault(name, ColumnConfig(name=name))
-                column.data_type = data_type
-                column.quote = self.quote_columns or column.quote
-                kwargs = self.sqlmesh_model_kwargs(context, column_types_override)
-        else:
-            kwargs = self.sqlmesh_model_kwargs(context)
+        column_types_override = {
+            name: ColumnConfig(name=name, data_type=data_type, quote=self.quote_columns)
+            for name, data_type in (self.column_types or {}).items()
+        }
+        kwargs = self.sqlmesh_model_kwargs(context, column_types_override)
 
         columns = kwargs.get("columns") or {}
 

tests/dbt/test_transformation.py

@@ -677,6 +677,27 @@ def test_seed_column_types():
     assert sqlmesh_seed.columns_to_types == expected_column_types
     assert sqlmesh_seed.column_descriptions == expected_column_descriptions
 
+    seed = SeedConfig(
+        name="foo",
+        package="package",
+        path=Path("examples/sushi_dbt/seeds/waiter_names.csv"),
+        column_types={
+            "name": "text",
+        },
+        columns={
+            # The `data_type` field does not affect the materialized seed's column type
+            "id": ColumnConfig(name="name", data_type="text"),
+        },
+        quote_columns=True,
+    )
+
+    expected_column_types = {
+        "id": exp.DataType.build("int"),
+        "name": exp.DataType.build("text"),
+    }
+    sqlmesh_seed = seed.to_sqlmesh(context)
+    assert sqlmesh_seed.columns_to_types == expected_column_types
+
 
 def test_seed_column_inference(tmp_path):
     seed_csv = tmp_path / "seed.csv"