Repository: TobikoData/sqlmesh
Pull request #2458 (2024-04-17 00:53:52)

fix!: normalize scd type 2 columns
Prior to this change we would treat the column properties in SCD Type 2 model kind as strings. This would cause issues in dialects like Snowflake where we need to properly normalize (uppercase) in order for the column to be used properly. 

This change makes it so that we now consider these fields as columns and normalize/quote them. As shown by the tests, this means casing is preserved properly. 

* Breaking: Once someone starts using this version they will start storing column values in state in a different format than someone on a different version.
* Empty migration script: The validator is designed to handle columns, identifiers, or strings and so strings that are currently stored in state will be processed correctly in the migration that happens from the empty migration script. 


Patch:
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index feb1d09a..0eb6c633 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -1150,10 +1150,10 @@ class EngineAdapter:
         target_table: TableName,
         source_table: QueryOrDF,
         unique_key: t.Sequence[exp.Expression],
-        valid_from_name: str,
-        valid_to_name: str,
+        valid_from_col: exp.Column,
+        valid_to_col: exp.Column,
         execution_time: TimeLike,
-        updated_at_name: str,
+        updated_at_col: exp.Column,
         invalidate_hard_deletes: bool = True,
         updated_at_as_valid_from: bool = False,
         columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
@@ -1166,10 +1166,10 @@ class EngineAdapter:
             target_table=target_table,
             source_table=source_table,
             unique_key=unique_key,
-            valid_from_name=valid_from_name,
-            valid_to_name=valid_to_name,
+            valid_from_col=valid_from_col,
+            valid_to_col=valid_to_col,
             execution_time=execution_time,
-            updated_at_name=updated_at_name,
+            updated_at_col=updated_at_col,
             invalidate_hard_deletes=invalidate_hard_deletes,
             updated_at_as_valid_from=updated_at_as_valid_from,
             columns_to_types=columns_to_types,
@@ -1183,8 +1183,8 @@ class EngineAdapter:
         target_table: TableName,
         source_table: QueryOrDF,
         unique_key: t.Sequence[exp.Expression],
-        valid_from_name: str,
-        valid_to_name: str,
+        valid_from_col: exp.Column,
+        valid_to_col: exp.Column,
         execution_time: TimeLike,
         check_columns: t.Union[exp.Star, t.Sequence[exp.Column]],
         invalidate_hard_deletes: bool = True,
@@ -1199,8 +1199,8 @@ class EngineAdapter:
             target_table=target_table,
             source_table=source_table,
             unique_key=unique_key,
-            valid_from_name=valid_from_name,
-            valid_to_name=valid_to_name,
+            valid_from_col=valid_from_col,
+            valid_to_col=valid_to_col,
             execution_time=execution_time,
             check_columns=check_columns,
             columns_to_types=columns_to_types,
@@ -1216,11 +1216,11 @@ class EngineAdapter:
         target_table: TableName,
         source_table: QueryOrDF,
         unique_key: t.Sequence[exp.Expression],
-        valid_from_name: str,
-        valid_to_name: str,
+        valid_from_col: exp.Column,
+        valid_to_col: exp.Column,
         execution_time: TimeLike,
         invalidate_hard_deletes: bool = True,
-        updated_at_name: t.Optional[str] = None,
+        updated_at_col: t.Optional[exp.Column] = None,
         check_columns: t.Optional[t.Union[exp.Star, t.Sequence[exp.Column]]] = None,
         updated_at_as_valid_from: bool = False,
         execution_time_as_valid_from: bool = False,
@@ -1233,6 +1233,9 @@ class EngineAdapter:
             source_table, columns_to_types, target_table=target_table, batch_size=0
         )
         columns_to_types = columns_to_types or self.columns(target_table)
+        valid_from_name = valid_from_col.name
+        valid_to_name = valid_to_col.name
+        updated_at_name = updated_at_col.name if updated_at_col else None
         if (
             valid_from_name not in columns_to_types
             or valid_to_name not in columns_to_types
@@ -1243,7 +1246,7 @@ class EngineAdapter:
             raise SQLMeshError(f"Could not get columns_to_types. Does {target_table} exist?")
         if not unique_key:
             raise SQLMeshError("unique_key must be provided for SCD Type 2")
-        if check_columns and updated_at_name:
+        if check_columns and updated_at_col:
             raise SQLMeshError(
                 "Cannot use both `check_columns` and `updated_at_name` for SCD Type 2"
             )
@@ -1270,7 +1273,7 @@ class EngineAdapter:
         table_columns = [exp.column(c, quoted=True) for c in columns_to_types]
         if updated_at_name:
             select_source_columns.append(
-                exp.cast(updated_at_name, time_data_type).as_(updated_at_name)
+                exp.cast(updated_at_col, time_data_type).as_(updated_at_col.this)  # type: ignore
             )
 
         # If a star is provided, we include all unmanaged columns in the check.
@@ -1282,25 +1285,29 @@ class EngineAdapter:
             check_columns = [exp.column(col) for col in unmanaged_columns]
         execution_ts = to_time_column(execution_time, time_data_type)
         if updated_at_as_valid_from:
-            if not updated_at_name:
+            if not updated_at_col:
                 raise SQLMeshError(
                     "Cannot use `updated_at_as_valid_from` without `updated_at_name` for SCD Type 2"
                 )
-            update_valid_from_start: t.Union[str, exp.Expression] = updated_at_name
+            update_valid_from_start: t.Union[str, exp.Expression] = updated_at_col
         elif execution_time_as_valid_from:
             update_valid_from_start = execution_ts
         else:
             update_valid_from_start = to_time_column("1970-01-01 00:00:00+00:00", time_data_type)
-        insert_valid_from_start = execution_ts if check_columns else exp.column(updated_at_name)  # type: ignore
+        insert_valid_from_start = execution_ts if check_columns else updated_at_col  # type: ignore
         # joined._exists IS NULL is saying "if the row is deleted"
         delete_check = (
             exp.column("_exists", "joined").is_(exp.Null()) if invalidate_hard_deletes else None
         )
+        prefixed_valid_to_col = valid_to_col.copy()
+        prefixed_valid_to_col.this.set("this", f"t_{prefixed_valid_to_col.name}")
+        prefixed_valid_from_col = valid_from_col.copy()
+        prefixed_valid_from_col.this.set("this", f"t_{valid_from_col.name}")
         if check_columns:
             row_check_conditions = []
             for col in check_columns:
                 t_col = col.copy()
-                t_col.set("this", exp.to_identifier(f"t_{col.name}"))
+                t_col.this.set("this", f"t_{col.name}")
                 row_check_conditions.extend(
                     [
                         col.neq(t_col),
@@ -1312,7 +1319,7 @@ class EngineAdapter:
             unique_key_conditions = []
             for col in unique_key:
                 t_col = col.copy()
-                t_col.set("this", exp.to_identifier(f"t_{col.name}"))
+                t_col.this.set("this", f"t_{col.name}")
                 unique_key_conditions.extend(
                     [t_col.is_(exp.Null()).not_(), col.is_(exp.Null()).not_()]
                 )
@@ -1331,52 +1338,62 @@ class EngineAdapter:
                     ),
                     execution_ts,
                 )
-                .else_(exp.column(f"t_{valid_to_name}"))
-                .as_(valid_to_name)
+                .else_(prefixed_valid_to_col)
+                .as_(valid_to_col.this)
             )
             valid_from_case_stmt = exp.func(
                 "COALESCE",
-                exp.column(f"t_{valid_from_name}"),
+                prefixed_valid_from_col,
                 update_valid_from_start,
-            ).as_(valid_from_name)
+            ).as_(valid_from_col.this)
         else:
-            assert updated_at_name is not None
-            updated_row_filter = exp.column(updated_at_name) > exp.column(f"t_{updated_at_name}")
+            assert updated_at_col is not None
+            prefixed_updated_at_col = updated_at_col.copy()
+            prefixed_updated_at_col.this.set("this", f"t_{updated_at_col.name}")
+            updated_row_filter = updated_at_col > prefixed_updated_at_col
 
-            valid_to_case_stmt_builder = exp.Case().when(
-                updated_row_filter, exp.column(updated_at_name)
-            )
+            valid_to_case_stmt_builder = exp.Case().when(updated_row_filter, updated_at_col)
             if delete_check:
                 valid_to_case_stmt_builder = valid_to_case_stmt_builder.when(
                     delete_check, execution_ts
                 )
-            valid_to_case_stmt = valid_to_case_stmt_builder.else_(
-                exp.column(f"t_{valid_to_name}")
-            ).as_(valid_to_name)
+            valid_to_case_stmt = valid_to_case_stmt_builder.else_(prefixed_valid_to_col).as_(
+                valid_to_col.this
+            )
 
             valid_from_case_stmt = (
                 exp.Case()
                 .when(
                     exp.and_(
-                        exp.column(f"t_{valid_from_name}").is_(exp.Null()),
+                        prefixed_valid_from_col.is_(exp.Null()),
                         exp.column("_exists", "latest_deleted").is_(exp.Null()).not_(),
                     ),
                     exp.Case()
                     .when(
-                        exp.column(valid_to_name, "latest_deleted") > exp.column(updated_at_name),
-                        exp.column(valid_to_name, "latest_deleted"),
+                        exp.column(valid_to_col.this, "latest_deleted") > updated_at_col,
+                        exp.column(valid_to_col.this, "latest_deleted"),
                     )
-                    .else_(exp.column(updated_at_name)),
+                    .else_(updated_at_col),
                 )
-                .when(exp.column(f"t_{valid_from_name}").is_(exp.Null()), update_valid_from_start)
-                .else_(exp.column(f"t_{valid_from_name}"))
-            ).as_(valid_from_name)
+                .when(prefixed_valid_from_col.is_(exp.Null()), update_valid_from_start)
+                .else_(prefixed_valid_from_col)
+            ).as_(valid_from_col.this)
 
         existing_rows_query = exp.select(*table_columns).from_(target_table)
         if truncate:
             existing_rows_query = existing_rows_query.limit(0)
 
         with source_queries[0] as source_query:
+            prefixed_columns_to_types = []
+            for column in columns_to_types:
+                prefixed_col = exp.column(column).copy()
+                prefixed_col.this.set("this", f"t_{prefixed_col.name}")
+                prefixed_columns_to_types.append(prefixed_col)
+            prefixed_unmanaged_columns = []
+            for column in unmanaged_columns:
+                prefixed_col = exp.column(column).copy()
+                prefixed_col.this.set("this", f"t_{prefixed_col.name}")
+                prefixed_unmanaged_columns.append(prefixed_col)
             query = (
                 exp.Select()  # type: ignore
                 .with_(
@@ -1388,17 +1405,17 @@ class EngineAdapter:
                 # Historical Records that Do Not Change
                 .with_(
                     "static",
-                    existing_rows_query.where(f"{valid_to_name} IS NOT NULL"),
+                    existing_rows_query.where(valid_to_col.is_(exp.Null()).not_()),
                 )
                 # Latest Records that can be updated
                 .with_(
                     "latest",
-                    existing_rows_query.where(f"{valid_to_name} IS NULL"),
+                    existing_rows_query.where(valid_to_col.is_(exp.Null())),
                 )
                 # Deleted records which can be used to determine `valid_from` for undeleted source records
                 .with_(
                     "deleted",
-                    exp.select(*[f"static.{col}" for col in columns_to_types])
+                    exp.select(*[exp.column(col, "static") for col in columns_to_types])
                     .from_("static")
                     .join(
                         "latest",
@@ -1410,7 +1427,7 @@ class EngineAdapter:
                         ),
                         join_type="left",
                     )
-                    .where(f"latest.{valid_to_name} IS NULL"),
+                    .where(exp.column(valid_to_col.this, "latest").is_(exp.Null())),
                 )
                 # Get the latest `valid_to` deleted record for each unique key
                 .with_(
@@ -1418,7 +1435,7 @@ class EngineAdapter:
                     exp.select(
                         exp.true().as_("_exists"),
                         *(part.as_(f"_key{i}") for i, part in enumerate(unique_key)),
-                        f"MAX({valid_to_name}) AS {valid_to_name}",
+                        exp.Max(this=valid_to_col).as_(valid_to_col.this),
                     )
                     .from_("deleted")
                     .group_by(*unique_key),
@@ -1430,8 +1447,8 @@ class EngineAdapter:
                     exp.select(
                         exp.column("_exists", table="source"),
                         *(
-                            exp.column(col, table="latest").as_(f"t_{col}")
-                            for col in columns_to_types
+                            exp.column(col, table="latest").as_(prefixed_columns_to_types[i].this)
+                            for i, col in enumerate(columns_to_types)
                         ),
                         *(exp.column(col, table="source").as_(col) for col in unmanaged_columns),
                     )
@@ -1450,8 +1467,10 @@ class EngineAdapter:
                         exp.select(
                             exp.column("_exists", table="source"),
                             *(
-                                exp.column(col, table="latest").as_(f"t_{col}")
-                                for col in columns_to_types
+                                exp.column(col, table="latest").as_(
+                                    prefixed_columns_to_types[i].this
+                                )
+                                for i, col in enumerate(columns_to_types)
                             ),
                             *(
                                 exp.column(col, table="source").as_(col)
@@ -1478,10 +1497,10 @@ class EngineAdapter:
                         *(
                             exp.func(
                                 "COALESCE",
-                                exp.column(f"t_{col}", table="joined"),
+                                exp.column(prefixed_unmanaged_columns[i].this, table="joined"),
                                 exp.column(col, table="joined"),
                             ).as_(col)
-                            for col in unmanaged_columns
+                            for i, col in enumerate(unmanaged_columns)
                         ),
                         valid_from_case_stmt,
                         valid_to_case_stmt,
@@ -1505,8 +1524,8 @@ class EngineAdapter:
                     "inserted_rows",
                     exp.select(
                         *unmanaged_columns,
-                        insert_valid_from_start.as_(valid_from_name),
-                        to_time_column(exp.null(), time_data_type).as_(valid_to_name),
+                        insert_valid_from_start.as_(valid_from_col.this),  # type: ignore
+                        to_time_column(exp.null(), time_data_type).as_(valid_to_col.this),
                     )
                     .from_("joined")
                     .where(updated_row_filter),
diff --git a/sqlmesh/core/engine_adapter/trino.py b/sqlmesh/core/engine_adapter/trino.py
index 21190193..2a8e6fd4 100644
--- a/sqlmesh/core/engine_adapter/trino.py
+++ b/sqlmesh/core/engine_adapter/trino.py
@@ -205,11 +205,11 @@ class TrinoEngineAdapter(
         target_table: TableName,
         source_table: QueryOrDF,
         unique_key: t.Sequence[exp.Expression],
-        valid_from_name: str,
-        valid_to_name: str,
+        valid_from_col: exp.Column,
+        valid_to_col: exp.Column,
         execution_time: TimeLike,
         invalidate_hard_deletes: bool = True,
-        updated_at_name: t.Optional[str] = None,
+        updated_at_col: t.Optional[exp.Column] = None,
         check_columns: t.Optional[t.Union[exp.Star, t.Sequence[exp.Column]]] = None,
         updated_at_as_valid_from: bool = False,
         execution_time_as_valid_from: bool = False,
@@ -225,11 +225,11 @@ class TrinoEngineAdapter(
             target_table,
             source_table,
             unique_key,
-            valid_from_name,
-            valid_to_name,
+            valid_from_col,
+            valid_to_col,
             execution_time,
             invalidate_hard_deletes,
-            updated_at_name,
+            updated_at_col,
             check_columns,
             updated_at_as_valid_from,
             execution_time_as_valid_from,
diff --git a/sqlmesh/core/model/kind.py b/sqlmesh/core/model/kind.py
index fc167788..31be0c32 100644
--- a/sqlmesh/core/model/kind.py
+++ b/sqlmesh/core/model/kind.py
@@ -18,10 +18,12 @@ from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import (
     PydanticModel,
     SQLGlotBool,
+    SQLGlotColumn,
     SQLGlotListOfColumnsOrStar,
     SQLGlotListOfFields,
     SQLGlotPositiveInt,
     SQLGlotString,
+    column_validator,
     field_validator,
     field_validator_v1_args,
     model_validator,
@@ -422,14 +424,19 @@ class FullKind(_ModelKind):
 class _SCDType2Kind(_ModelKind):
     dialect: str = ""
     unique_key: SQLGlotListOfFields
-    valid_from_name: SQLGlotString = "valid_from"
-    valid_to_name: SQLGlotString = "valid_to"
+    valid_from_name: SQLGlotColumn = Field(exp.column("valid_from"), validate_default=True)
+    valid_to_name: SQLGlotColumn = Field(exp.column("valid_to"), validate_default=True)
     invalidate_hard_deletes: SQLGlotBool = False
     time_data_type: exp.DataType = Field(exp.DataType.build("TIMESTAMP"), validate_default=True)
 
     forward_only: SQLGlotBool = True
     disable_restatement: SQLGlotBool = True
 
+    # Remove once Pydantic 1 is deprecated
+    _always_validate_column = field_validator(
+        "valid_from_name", "valid_to_name", mode="before", always=True
+    )(column_validator)
+
     # always=True can be removed once Pydantic 1 is deprecated
     @field_validator("time_data_type", mode="before", always=True)
     @classmethod
@@ -447,8 +454,8 @@ class _SCDType2Kind(_ModelKind):
     @property
     def managed_columns(self) -> t.Dict[str, exp.DataType]:
         return {
-            self.valid_from_name: self.time_data_type,
-            self.valid_to_name: self.time_data_type,
+            self.valid_from_name.name: self.time_data_type,
+            self.valid_to_name.name: self.time_data_type,
         }
 
     @property
@@ -457,8 +464,8 @@ class _SCDType2Kind(_ModelKind):
             *super().data_hash_values,
             self.dialect,
             *(gen(k) for k in self.unique_key),
-            self.valid_from_name,
-            self.valid_to_name,
+            gen(self.valid_from_name),
+            gen(self.valid_to_name),
             str(self.invalidate_hard_deletes),
             gen(self.time_data_type),
         ]
@@ -476,12 +483,21 @@ class SCDType2ByTimeKind(_SCDType2Kind):
     name: Literal[ModelKindName.SCD_TYPE_2, ModelKindName.SCD_TYPE_2_BY_TIME] = (
         ModelKindName.SCD_TYPE_2_BY_TIME
     )
-    updated_at_name: SQLGlotString = "updated_at"
+    updated_at_name: SQLGlotColumn = Field(exp.column("updated_at"), validate_default=True)
     updated_at_as_valid_from: SQLGlotBool = False
 
+    # Remove once Pydantic 1 is deprecated
+    _always_validate_updated_at = field_validator("updated_at_name", mode="before", always=True)(
+        column_validator
+    )
+
     @property
     def data_hash_values(self) -> t.List[t.Optional[str]]:
-        return [*super().data_hash_values, self.updated_at_name, str(self.updated_at_as_valid_from)]
+        return [
+            *super().data_hash_values,
+            gen(self.updated_at_name),
+            str(self.updated_at_as_valid_from),
+        ]
 
 
 class SCDType2ByColumnKind(_SCDType2Kind):
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 0e68ed05..785dcfec 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -1276,7 +1276,7 @@ class SCDType2Strategy(MaterializableStrategy):
             logger.info("Creating table '%s'", name)
             columns_to_types = model.columns_to_types_or_raise
             if isinstance(model.kind, SCDType2ByTimeKind):
-                columns_to_types[model.kind.updated_at_name] = model.kind.time_data_type
+                columns_to_types[model.kind.updated_at_name.name] = model.kind.time_data_type
             self.adapter.create_table(
                 name,
                 columns_to_types=columns_to_types,
@@ -1316,9 +1316,9 @@ class SCDType2Strategy(MaterializableStrategy):
                 target_table=name,
                 source_table=query_or_df,
                 unique_key=model.unique_key,
-                valid_from_name=model.kind.valid_from_name,
-                valid_to_name=model.kind.valid_to_name,
-                updated_at_name=model.kind.updated_at_name,
+                valid_from_col=model.kind.valid_from_name,
+                valid_to_col=model.kind.valid_to_name,
+                updated_at_col=model.kind.updated_at_name,
                 invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
                 updated_at_as_valid_from=model.kind.updated_at_as_valid_from,
                 columns_to_types=model.columns_to_types,
@@ -1332,8 +1332,8 @@ class SCDType2Strategy(MaterializableStrategy):
                 target_table=name,
                 source_table=query_or_df,
                 unique_key=model.unique_key,
-                valid_from_name=model.kind.valid_from_name,
-                valid_to_name=model.kind.valid_to_name,
+                valid_from_col=model.kind.valid_from_name,
+                valid_to_col=model.kind.valid_to_name,
                 check_columns=model.kind.columns,
                 columns_to_types=model.columns_to_types,
                 invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
@@ -1363,9 +1363,9 @@ class SCDType2Strategy(MaterializableStrategy):
                 target_table=table_name,
                 source_table=query_or_df,
                 unique_key=model.unique_key,
-                valid_from_name=model.kind.valid_from_name,
-                valid_to_name=model.kind.valid_to_name,
-                updated_at_name=model.kind.updated_at_name,
+                valid_from_col=model.kind.valid_from_name,
+                valid_to_col=model.kind.valid_to_name,
+                updated_at_col=model.kind.updated_at_name,
                 invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
                 updated_at_as_valid_from=model.kind.updated_at_as_valid_from,
                 columns_to_types=model.columns_to_types,
@@ -1378,8 +1378,8 @@ class SCDType2Strategy(MaterializableStrategy):
                 target_table=table_name,
                 source_table=query_or_df,
                 unique_key=model.unique_key,
-                valid_from_name=model.kind.valid_from_name,
-                valid_to_name=model.kind.valid_to_name,
+                valid_from_col=model.kind.valid_from_name,
+                valid_to_col=model.kind.valid_to_name,
                 check_columns=model.kind.columns,
                 columns_to_types=model.columns_to_types,
                 invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
diff --git a/sqlmesh/migrations/v0047_change_scd_string_to_column.py b/sqlmesh/migrations/v0047_change_scd_string_to_column.py
new file mode 100644
index 00000000..72ebbf26
--- /dev/null
+++ b/sqlmesh/migrations/v0047_change_scd_string_to_column.py
@@ -0,0 +1,5 @@
+"""Changes the SCD Type 2 columns from strings to columns."""
+
+
+def migrate(state_sync, **kwargs):  # type: ignore
+    pass
diff --git a/sqlmesh/utils/pydantic.py b/sqlmesh/utils/pydantic.py
index 4ada7c93..f8465ff7 100644
--- a/sqlmesh/utils/pydantic.py
+++ b/sqlmesh/utils/pydantic.py
@@ -290,6 +290,27 @@ def positive_int_validator(v: t.Any) -> int:
     return v
 
 
+def _get_field(
+    v: t.Any,
+    values: t.Any,
+) -> exp.Expression:
+    values = values if isinstance(values, dict) else values.data
+    dialect = values.get("dialect")
+
+    if isinstance(v, exp.Expression):
+        expression = v
+    else:
+        expression = parse_one(v, dialect=dialect)
+
+    expression = exp.column(expression) if isinstance(expression, exp.Identifier) else expression
+    expression = quote_identifiers(
+        normalize_identifiers(expression, dialect=dialect), dialect=dialect
+    )
+    expression.meta["dialect"] = dialect
+
+    return expression
+
+
 def _get_fields(
     v: t.Any,
     values: t.Any,
@@ -310,10 +331,7 @@ def _get_fields(
     results = []
 
     for expr in expressions:
-        expr = exp.column(expr) if isinstance(expr, exp.Identifier) else expr
-        expr = quote_identifiers(normalize_identifiers(expr, dialect=dialect), dialect=dialect)
-        expr.meta["dialect"] = dialect
-        results.append(expr)
+        results.append(_get_field(expr, values))
 
     return results
 
@@ -322,6 +340,13 @@ def list_of_fields_validator(v: t.Any, values: t.Any) -> t.List[exp.Expression]:
     return _get_fields(v, values)
 
 
+def column_validator(v: t.Any, values: t.Any) -> exp.Column:
+    expression = _get_field(v, values)
+    if not isinstance(expression, exp.Column):
+        raise ValueError(f"Invalid column {expression}. Value must be a column")
+    return expression
+
+
 def list_of_columns_or_star_validator(
     v: t.Any, values: t.Any
 ) -> t.Union[exp.Star, t.List[exp.Column]]:
@@ -336,6 +361,7 @@ if t.TYPE_CHECKING:
     SQLGlotString = str
     SQLGlotBool = bool
     SQLGlotPositiveInt = int
+    SQLGlotColumn = exp.Column
     SQLGlotListOfFields = t.List[exp.Expression]
     SQLGlotListOfColumnsOrStar = t.Union[t.List[exp.Column], exp.Star]
 elif PYDANTIC_MAJOR_VERSION >= 2:
@@ -345,6 +371,7 @@ elif PYDANTIC_MAJOR_VERSION >= 2:
     SQLGlotString = Annotated[str, BeforeValidator(validate_string)]
     SQLGlotBool = Annotated[bool, BeforeValidator(bool_validator)]
     SQLGlotPositiveInt = Annotated[int, BeforeValidator(positive_int_validator)]
+    SQLGlotColumn = Annotated[exp.Expression, BeforeValidator(column_validator)]
     SQLGlotListOfFields = Annotated[
         t.List[exp.Expression], BeforeValidator(list_of_fields_validator)
     ]
@@ -372,6 +399,9 @@ else:
     class SQLGlotPositiveInt(PydanticTypeProxy[int]):
         validate = positive_int_validator
 
+    class SQLGlotColumn(PydanticTypeProxy[exp.Column]):
+        validate = column_validator
+
     class SQLGlotListOfFields(PydanticTypeProxy[t.List[exp.Expression]]):
         validate = list_of_fields_validator
 


Test patch:
diff --git a/tests/core/engine_adapter/test_base.py b/tests/core/engine_adapter/test_base.py
index efe5de4f..2d67bf48 100644
--- a/tests/core/engine_adapter/test_base.py
+++ b/tests/core/engine_adapter/test_base.py
@@ -977,20 +977,20 @@ def test_scd_type_2_by_time(make_mocked_engine_adapter: t.Callable):
     adapter.scd_type_2_by_time(
         target_table="target",
         source_table=t.cast(
-            exp.Select, parse_one("SELECT id, name, price, test_updated_at FROM source")
+            exp.Select, parse_one("SELECT id, name, price, test_UPDATED_at FROM source")
         ),
         unique_key=[
             parse_one("""COALESCE("id", '') || '|' || COALESCE("name", '')"""),
             parse_one("""COALESCE("name", '')"""),
         ],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
-        updated_at_name="test_updated_at",
+        valid_from_col=exp.column("test_valid_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
+        updated_at_col=exp.column("test_UPDATED_at", quoted=True),
         columns_to_types={
             "id": exp.DataType.build("INT"),
             "name": exp.DataType.build("VARCHAR"),
             "price": exp.DataType.build("DOUBLE"),
-            "test_updated_at": exp.DataType.build("TIMESTAMP"),
+            "test_UPDATED_at": exp.DataType.build("TIMESTAMP"),
             "test_valid_from": exp.DataType.build("TIMESTAMP"),
             "test_valid_to": exp.DataType.build("TIMESTAMP"),
         },
@@ -1008,13 +1008,13 @@ WITH "source" AS (
     "id",
     "name",
     "price",
-    CAST("test_updated_at" AS TIMESTAMP) AS "test_updated_at"
+    CAST("test_UPDATED_at" AS TIMESTAMP) AS "test_UPDATED_at"
   FROM (
     SELECT
       "id",
       "name",
       "price",
-      "test_updated_at"
+      "test_UPDATED_at"
     FROM "source"
   ) AS "raw_source"
 ), "static" AS (
@@ -1022,7 +1022,7 @@ WITH "source" AS (
     "id",
     "name",
     "price",
-    "test_updated_at",
+    "test_UPDATED_at",
     "test_valid_from",
     "test_valid_to"
   FROM "target"
@@ -1033,7 +1033,7 @@ WITH "source" AS (
     "id",
     "name",
     "price",
-    "test_updated_at",
+    "test_UPDATED_at",
     "test_valid_from",
     "test_valid_to"
   FROM "target"
@@ -1044,7 +1044,7 @@ WITH "source" AS (
     "static"."id",
     "static"."name",
     "static"."price",
-    "static"."test_updated_at",
+    "static"."test_UPDATED_at",
     "static"."test_valid_from",
     "static"."test_valid_to"
   FROM "static"
@@ -1073,13 +1073,13 @@ WITH "source" AS (
     "latest"."id" AS "t_id",
     "latest"."name" AS "t_name",
     "latest"."price" AS "t_price",
-    "latest"."test_updated_at" AS "t_test_updated_at",
+    "latest"."test_UPDATED_at" AS "t_test_UPDATED_at",
     "latest"."test_valid_from" AS "t_test_valid_from",
     "latest"."test_valid_to" AS "t_test_valid_to",
     "source"."id" AS "id",
     "source"."name" AS "name",
     "source"."price" AS "price",
-    "source"."test_updated_at" AS "test_updated_at"
+    "source"."test_UPDATED_at" AS "test_UPDATED_at"
   FROM "latest"
   LEFT JOIN "source"
     ON (
@@ -1094,13 +1094,13 @@ WITH "source" AS (
     "latest"."id" AS "t_id",
     "latest"."name" AS "t_name",
     "latest"."price" AS "t_price",
-    "latest"."test_updated_at" AS "t_test_updated_at",
+    "latest"."test_UPDATED_at" AS "t_test_UPDATED_at",
     "latest"."test_valid_from" AS "t_test_valid_from",
     "latest"."test_valid_to" AS "t_test_valid_to",
     "source"."id" AS "id",
     "source"."name" AS "name",
     "source"."price" AS "price",
-    "source"."test_updated_at" AS "test_updated_at"
+    "source"."test_UPDATED_at" AS "test_UPDATED_at"
   FROM "latest"
   RIGHT JOIN "source"
     ON (
@@ -1114,21 +1114,21 @@ WITH "source" AS (
     COALESCE("joined"."t_id", "joined"."id") AS "id",
     COALESCE("joined"."t_name", "joined"."name") AS "name",
     COALESCE("joined"."t_price", "joined"."price") AS "price",
-    COALESCE("joined"."t_test_updated_at", "joined"."test_updated_at") AS "test_updated_at",
+    COALESCE("joined"."t_test_UPDATED_at", "joined"."test_UPDATED_at") AS "test_UPDATED_at",
     CASE
       WHEN "t_test_valid_from" IS NULL AND NOT "latest_deleted"."_exists" IS NULL
       THEN CASE
-        WHEN "latest_deleted"."test_valid_to" > "test_updated_at"
+        WHEN "latest_deleted"."test_valid_to" > "test_UPDATED_at"
         THEN "latest_deleted"."test_valid_to"
-        ELSE "test_updated_at"
+        ELSE "test_UPDATED_at"
       END
       WHEN "t_test_valid_from" IS NULL
       THEN CAST('1970-01-01 00:00:00' AS TIMESTAMP)
       ELSE "t_test_valid_from"
     END AS "test_valid_from",
     CASE
-      WHEN "test_updated_at" > "t_test_updated_at"
-      THEN "test_updated_at"
+      WHEN "test_UPDATED_at" > "t_test_UPDATED_at"
+      THEN "test_UPDATED_at"
       WHEN "joined"."_exists" IS NULL
       THEN CAST('2020-01-01 00:00:00' AS TIMESTAMP)
       ELSE "t_test_valid_to"
@@ -1144,18 +1144,18 @@ WITH "source" AS (
     "id",
     "name",
     "price",
-    "test_updated_at",
-    "test_updated_at" AS "test_valid_from",
+    "test_UPDATED_at",
+    "test_UPDATED_at" AS "test_valid_from",
     CAST(NULL AS TIMESTAMP) AS "test_valid_to"
   FROM "joined"
   WHERE
-    "test_updated_at" > "t_test_updated_at"
+    "test_UPDATED_at" > "t_test_UPDATED_at"
 )
 SELECT
   "id",
   "name",
   "price",
-  "test_updated_at",
+  "test_UPDATED_at",
   "test_valid_from",
   "test_valid_to"
 FROM "static"
@@ -1164,7 +1164,7 @@ SELECT
   "id",
   "name",
   "price",
-  "test_updated_at",
+  "test_UPDATED_at",
   "test_valid_from",
   "test_valid_to"
 FROM "updated_rows"
@@ -1173,7 +1173,7 @@ SELECT
   "id",
   "name",
   "price",
-  "test_updated_at",
+  "test_UPDATED_at",
   "test_valid_from",
   "test_valid_to"
 FROM "inserted_rows"
@@ -1191,9 +1191,9 @@ def test_scd_type_2_by_time_no_invalidate_hard_deletes(make_mocked_engine_adapte
             exp.Select, parse_one("SELECT id, name, price, test_updated_at FROM source")
         ),
         unique_key=[exp.func("COALESCE", "id", "''")],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
-        updated_at_name="test_updated_at",
+        valid_from_col=exp.column("test_valid_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
+        updated_at_col=exp.column("test_updated_at", quoted=True),
         invalidate_hard_deletes=False,
         columns_to_types={
             "id": exp.DataType.build("INT"),
@@ -1385,9 +1385,9 @@ def test_merge_scd_type_2_pandas(make_mocked_engine_adapter: t.Callable):
         target_table="target",
         source_table=df,
         unique_key=[exp.column("id1"), exp.column("id2")],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
-        updated_at_name="test_updated_at",
+        valid_from_col=exp.column("test_valid_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
+        updated_at_col=exp.column("test_updated_at", quoted=True),
         columns_to_types={
             "id1": exp.DataType.build("INT"),
             "id2": exp.DataType.build("INT"),
@@ -1594,14 +1594,14 @@ def test_scd_type_2_by_column(make_mocked_engine_adapter: t.Callable):
         target_table="target",
         source_table=t.cast(exp.Select, parse_one("SELECT id, name, price FROM source")),
         unique_key=[exp.column("id")],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
+        valid_from_col=exp.column("test_VALID_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
         check_columns=[exp.column("name"), exp.column("price")],
         columns_to_types={
             "id": exp.DataType.build("INT"),
             "name": exp.DataType.build("VARCHAR"),
             "price": exp.DataType.build("DOUBLE"),
-            "test_valid_from": exp.DataType.build("TIMESTAMP"),
+            "test_VALID_from": exp.DataType.build("TIMESTAMP"),
             "test_valid_to": exp.DataType.build("TIMESTAMP"),
         },
         execution_time=datetime(2020, 1, 1, 0, 0, 0),
@@ -1630,7 +1630,7 @@ WITH "source" AS (
     "id",
     "name",
     "price",
-    "test_valid_from",
+    "test_VALID_from",
     "test_valid_to"
   FROM "target"
   WHERE
@@ -1640,7 +1640,7 @@ WITH "source" AS (
     "id",
     "name",
     "price",
-    "test_valid_from",
+    "test_VALID_from",
     "test_valid_to"
   FROM "target"
   WHERE
@@ -1650,7 +1650,7 @@ WITH "source" AS (
     "static"."id",
     "static"."name",
     "static"."price",
-    "static"."test_valid_from",
+    "static"."test_VALID_from",
     "static"."test_valid_to"
   FROM "static"
   LEFT JOIN "latest"
@@ -1671,7 +1671,7 @@ WITH "source" AS (
     "latest"."id" AS "t_id",
     "latest"."name" AS "t_name",
     "latest"."price" AS "t_price",
-    "latest"."test_valid_from" AS "t_test_valid_from",
+    "latest"."test_VALID_from" AS "t_test_VALID_from",
     "latest"."test_valid_to" AS "t_test_valid_to",
     "source"."id" AS "id",
     "source"."name" AS "name",
@@ -1685,7 +1685,7 @@ WITH "source" AS (
     "latest"."id" AS "t_id",
     "latest"."name" AS "t_name",
     "latest"."price" AS "t_price",
-    "latest"."test_valid_from" AS "t_test_valid_from",
+    "latest"."test_VALID_from" AS "t_test_VALID_from",
     "latest"."test_valid_to" AS "t_test_valid_to",
     "source"."id" AS "id",
     "source"."name" AS "name",
@@ -1698,7 +1698,7 @@ WITH "source" AS (
     COALESCE("joined"."t_id", "joined"."id") AS "id",
     COALESCE("joined"."t_name", "joined"."name") AS "name",
     COALESCE("joined"."t_price", "joined"."price") AS "price",
-    COALESCE("t_test_valid_from", CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AS "test_valid_from",
+    COALESCE("t_test_VALID_from", CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AS "test_VALID_from",
     CASE
       WHEN "joined"."_exists" IS NULL
       OR (
@@ -1733,7 +1733,7 @@ WITH "source" AS (
     "id",
     "name",
     "price",
-    CAST('2020-01-01 00:00:00' AS TIMESTAMP) AS "test_valid_from",
+    CAST('2020-01-01 00:00:00' AS TIMESTAMP) AS "test_VALID_from",
     CAST(NULL AS TIMESTAMP) AS "test_valid_to"
   FROM "joined"
   WHERE
@@ -1761,7 +1761,7 @@ SELECT
   "id",
   "name",
   "price",
-  "test_valid_from",
+  "test_VALID_from",
   "test_valid_to"
 FROM "static"
 UNION ALL
@@ -1769,7 +1769,7 @@ SELECT
   "id",
   "name",
   "price",
-  "test_valid_from",
+  "test_VALID_from",
   "test_valid_to"
 FROM "updated_rows"
 UNION ALL
@@ -1777,7 +1777,7 @@ SELECT
   "id",
   "name",
   "price",
-  "test_valid_from",
+  "test_VALID_from",
   "test_valid_to"
 FROM "inserted_rows"
     """
@@ -1792,8 +1792,8 @@ def test_scd_type_2_truncate(make_mocked_engine_adapter: t.Callable):
         target_table="target",
         source_table=t.cast(exp.Select, parse_one("SELECT id, name, price FROM source")),
         unique_key=[exp.column("id")],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
+        valid_from_col=exp.column("test_valid_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
         check_columns=[exp.column("name"), exp.column("price")],
         columns_to_types={
             "id": exp.DataType.build("INT"),
@@ -1993,8 +1993,8 @@ def test_scd_type_2_by_column_star_check(make_mocked_engine_adapter: t.Callable)
         target_table="target",
         source_table=t.cast(exp.Select, parse_one("SELECT id, name, price FROM source")),
         unique_key=[exp.column("id")],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
+        valid_from_col=exp.column("test_valid_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
         check_columns=exp.Star(),
         columns_to_types={
             "id": exp.DataType.build("INT"),
@@ -2205,8 +2205,8 @@ def test_scd_type_2_by_column_no_invalidate_hard_deletes(make_mocked_engine_adap
         target_table="target",
         source_table=t.cast(exp.Select, parse_one("SELECT id, name, price FROM source")),
         unique_key=[exp.column("id")],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
+        valid_from_col=exp.column("test_valid_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
         invalidate_hard_deletes=False,
         check_columns=[exp.column("name"), exp.column("price")],
         columns_to_types={
diff --git a/tests/core/engine_adapter/test_integration.py b/tests/core/engine_adapter/test_integration.py
index 9fc51b38..6d8231cf 100644
--- a/tests/core/engine_adapter/test_integration.py
+++ b/tests/core/engine_adapter/test_integration.py
@@ -1197,9 +1197,9 @@ def test_scd_type_2_by_time(ctx: TestContext):
         table,
         ctx.input_data(input_data, input_schema),
         unique_key=[parse_one("COALESCE(id, -1)")],
-        valid_from_name="valid_from",
-        valid_to_name="valid_to",
-        updated_at_name="updated_at",
+        valid_from_col=exp.column("valid_from", quoted=True),
+        valid_to_col=exp.column("valid_to", quoted=True),
+        updated_at_col=exp.column("updated_at", quoted=True),
         execution_time="2023-01-01",
         updated_at_as_valid_from=False,
         columns_to_types=input_schema,
@@ -1257,9 +1257,9 @@ def test_scd_type_2_by_time(ctx: TestContext):
         table,
         ctx.input_data(current_data, input_schema),
         unique_key=[exp.to_column("id")],
-        valid_from_name="valid_from",
-        valid_to_name="valid_to",
-        updated_at_name="updated_at",
+        valid_from_col=exp.column("valid_from", quoted=True),
+        valid_to_col=exp.column("valid_to", quoted=True),
+        updated_at_col=exp.column("updated_at", quoted=True),
         execution_time="2023-01-05",
         updated_at_as_valid_from=False,
         columns_to_types=input_schema,
@@ -1342,8 +1342,8 @@ def test_scd_type_2_by_column(ctx: TestContext):
         ctx.input_data(input_data, input_schema),
         unique_key=[exp.to_column("id")],
         check_columns=[exp.to_column("name"), exp.to_column("status")],
-        valid_from_name="valid_from",
-        valid_to_name="valid_to",
+        valid_from_col=exp.column("valid_from", quoted=True),
+        valid_to_col=exp.column("valid_to", quoted=True),
         execution_time="2023-01-01",
         execution_time_as_valid_from=False,
         columns_to_types=input_schema,
@@ -1411,8 +1411,8 @@ def test_scd_type_2_by_column(ctx: TestContext):
         ctx.input_data(current_data, input_schema),
         unique_key=[exp.to_column("id")],
         check_columns=[exp.to_column("name"), exp.to_column("status")],
-        valid_from_name="valid_from",
-        valid_to_name="valid_to",
+        valid_from_col=exp.column("valid_from", quoted=True),
+        valid_to_col=exp.column("valid_to", quoted=True),
         execution_time="2023-01-05",
         execution_time_as_valid_from=False,
         columns_to_types=input_schema,
diff --git a/tests/core/engine_adapter/test_spark.py b/tests/core/engine_adapter/test_spark.py
index 9d80c12d..57db7431 100644
--- a/tests/core/engine_adapter/test_spark.py
+++ b/tests/core/engine_adapter/test_spark.py
@@ -585,9 +585,9 @@ def test_scd_type_2_by_time(
             exp.Select, parse_one("SELECT id, name, price, test_updated_at FROM db.source")
         ),
         unique_key=[exp.func("COALESCE", "id", "''")],
-        valid_from_name="test_valid_from",
-        valid_to_name="test_valid_to",
-        updated_at_name="test_updated_at",
+        valid_from_col=exp.column("test_valid_from", quoted=True),
+        valid_to_col=exp.column("test_valid_to", quoted=True),
+        updated_at_col=exp.column("test_updated_at", quoted=True),
         columns_to_types={
             "id": exp.DataType.build("INT"),
             "name": exp.DataType.build("VARCHAR"),
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index fb54ddcd..0529d181 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -2672,9 +2672,9 @@ def test_scd_type_2_by_time_defaults():
         "valid_from": exp.DataType.build("TIMESTAMP"),
         "valid_to": exp.DataType.build("TIMESTAMP"),
     }
-    assert scd_type_2_model.kind.updated_at_name == "updated_at"
-    assert scd_type_2_model.kind.valid_from_name == "valid_from"
-    assert scd_type_2_model.kind.valid_to_name == "valid_to"
+    assert scd_type_2_model.kind.updated_at_name == exp.column("updated_at", quoted=True)
+    assert scd_type_2_model.kind.valid_from_name == exp.column("valid_from", quoted=True)
+    assert scd_type_2_model.kind.valid_to_name == exp.column("valid_to", quoted=True)
     assert not scd_type_2_model.kind.updated_at_as_valid_from
     assert scd_type_2_model.kind.is_scd_type_2_by_time
     assert scd_type_2_model.kind.is_scd_type_2
@@ -2699,6 +2699,7 @@ def test_scd_type_2_by_time_overrides():
                 disable_restatement False,
                 invalidate_hard_deletes False,
             ),
+            dialect snowflake
         );
         SELECT
             1 as "iD",
@@ -2712,15 +2713,15 @@ def test_scd_type_2_by_time_overrides():
     scd_type_2_model = load_sql_based_model(model_def)
     assert scd_type_2_model.unique_key == [
         exp.column("iD", quoted=True),
-        exp.column("ds", quoted=True),
+        exp.column("DS", quoted=True),
     ]
     assert scd_type_2_model.managed_columns == {
-        "test_valid_from": exp.DataType.build("TIMESTAMPTZ"),
-        "test_valid_to": exp.DataType.build("TIMESTAMPTZ"),
+        "TEST_VALID_FROM": exp.DataType.build("TIMESTAMPTZ"),
+        "TEST_VALID_TO": exp.DataType.build("TIMESTAMPTZ"),
     }
-    assert scd_type_2_model.kind.updated_at_name == "test_updated_at"
-    assert scd_type_2_model.kind.valid_from_name == "test_valid_from"
-    assert scd_type_2_model.kind.valid_to_name == "test_valid_to"
+    assert scd_type_2_model.kind.updated_at_name == exp.column("TEST_UPDATED_AT", quoted=True)
+    assert scd_type_2_model.kind.valid_from_name == exp.column("TEST_VALID_FROM", quoted=True)
+    assert scd_type_2_model.kind.valid_to_name == exp.column("TEST_VALID_TO", quoted=True)
     assert scd_type_2_model.kind.updated_at_as_valid_from
     assert scd_type_2_model.kind.is_scd_type_2_by_time
     assert scd_type_2_model.kind.is_scd_type_2
@@ -2764,8 +2765,8 @@ def test_scd_type_2_by_column_defaults():
         "valid_from": exp.DataType.build("TIMESTAMP"),
         "valid_to": exp.DataType.build("TIMESTAMP"),
     }
-    assert scd_type_2_model.kind.valid_from_name == "valid_from"
-    assert scd_type_2_model.kind.valid_to_name == "valid_to"
+    assert scd_type_2_model.kind.valid_from_name == exp.column("valid_from", quoted=True)
+    assert scd_type_2_model.kind.valid_to_name == exp.column("valid_to", quoted=True)
     assert not scd_type_2_model.kind.execution_time_as_valid_from
     assert scd_type_2_model.kind.is_scd_type_2_by_column
     assert scd_type_2_model.kind.is_scd_type_2
@@ -2807,8 +2808,8 @@ def test_scd_type_2_by_column_overrides():
         "test_valid_from": exp.DataType.build("TIMESTAMPTZ"),
         "test_valid_to": exp.DataType.build("TIMESTAMPTZ"),
     }
-    assert scd_type_2_model.kind.valid_from_name == "test_valid_from"
-    assert scd_type_2_model.kind.valid_to_name == "test_valid_to"
+    assert scd_type_2_model.kind.valid_from_name == exp.column("test_valid_from", quoted=True)
+    assert scd_type_2_model.kind.valid_to_name == exp.column("test_valid_to", quoted=True)
     assert scd_type_2_model.kind.execution_time_as_valid_from
     assert scd_type_2_model.kind.is_scd_type_2_by_column
     assert scd_type_2_model.kind.is_scd_type_2
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py
index 64b1cff2..2cf67dfd 100644
--- a/tests/core/test_snapshot_evaluator.py
+++ b/tests/core/test_snapshot_evaluator.py
@@ -1307,9 +1307,9 @@ def test_insert_into_scd_type_2_by_time(
             "valid_to": exp.DataType.build("TIMESTAMP"),
         },
         unique_key=[exp.to_column("id", quoted=True)],
-        valid_from_name="valid_from",
-        valid_to_name="valid_to",
-        updated_at_name="updated_at",
+        valid_from_col=exp.column("valid_from", quoted=True),
+        valid_to_col=exp.column("valid_to", quoted=True),
+        updated_at_col=exp.column("updated_at", quoted=True),
         start="2020-01-01",
         end="2020-01-02",
         execution_time="2020-01-02",
@@ -1475,8 +1475,8 @@ def test_insert_into_scd_type_2_by_column(
         },
         unique_key=[exp.to_column("id", quoted=True)],
         check_columns=exp.Star(),
-        valid_from_name="valid_from",
-        valid_to_name="valid_to",
+        valid_from_col=exp.column("valid_from", quoted=True),
+        valid_to_col=exp.column("valid_to", quoted=True),
         start="2020-01-01",
         end="2020-01-02",
         execution_time="2020-01-02",