fix: dbt time column serialization

Author: eakmanrq

sqlmesh/dbt/model.py

@@ -31,7 +31,6 @@
     OnAdditiveChange,
     on_destructive_change_validator,
     on_additive_change_validator,
-    TimeColumn,
 )
 from sqlmesh.dbt.basemodel import BaseModelConfig, Materialization, SnapshotStrategy
 from sqlmesh.dbt.common import SqlStr, sql_str_validator
@@ -86,7 +85,7 @@ class ModelConfig(BaseModelConfig):
 
     # sqlmesh fields
     sql: SqlStr = SqlStr("")
-    time_column: t.Optional[TimeColumn] = None
+    time_column: t.Optional[t.Union[str, t.Dict[str, str]]] = None
     cron: t.Optional[str] = None
     interval_unit: t.Optional[str] = None
     batch_concurrency: t.Optional[int] = None
@@ -153,7 +152,6 @@ class ModelConfig(BaseModelConfig):
     _sql_validator = sql_str_validator
     _on_destructive_change_validator = on_destructive_change_validator
     _on_additive_change_validator = on_additive_change_validator
-    _time_column_validator = TimeColumn.validator()
 
     @field_validator(
         "unique_key",

tests/dbt/test_model.py

@@ -9,6 +9,7 @@
 from sqlmesh import Context
 from sqlmesh.core.model import TimeColumn, IncrementalByTimeRangeKind
 from sqlmesh.core.model.kind import OnDestructiveChange, OnAdditiveChange
+from sqlmesh.core.state_sync.db.snapshot import _snapshot_to_json
 from sqlmesh.dbt.common import Dependencies
 from sqlmesh.dbt.context import DbtContext
 from sqlmesh.dbt.model import ModelConfig
@@ -328,7 +329,8 @@ def test_load_incremental_time_range_strategy_required_only(
 
     snapshot_fqn = '"local"."main"."incremental_time_range"'
     context = Context(paths=project_dir)
-    model = context.snapshots[snapshot_fqn].model
+    snapshot = context.snapshots[snapshot_fqn]
+    model = snapshot.model
     # Validate model-level attributes
     assert model.start == "2025-01-01"
     assert model.interval_unit.is_day
@@ -342,6 +344,8 @@ def test_load_incremental_time_range_strategy_required_only(
     assert model.depends_on_self is False
     assert model.kind.auto_restatement_intervals is None
     assert model.kind.partition_by_time_column is True
+    # make sure the snapshot can be serialized to json
+    assert isinstance(_snapshot_to_json(snapshot), str)
 
 
 @pytest.mark.slow
@@ -381,7 +385,8 @@ def test_load_incremental_time_range_strategy_all_defined(
 
     snapshot_fqn = '"local"."main"."incremental_time_range"'
     context = Context(paths=project_dir)
-    model = context.snapshots[snapshot_fqn].model
+    snapshot = context.snapshots[snapshot_fqn]
+    model = snapshot.model
     # Validate model-level attributes
     assert model.start == "2025-01-01"
     assert model.interval_unit.is_day
@@ -402,6 +407,8 @@ def test_load_incremental_time_range_strategy_all_defined(
     assert model.kind.batch_size == 3
     assert model.kind.batch_concurrency == 2
     assert model.depends_on_self is False
+    # make sure the snapshot can be serialized to json
+    assert isinstance(_snapshot_to_json(snapshot), str)
 
 
 @pytest.mark.slow

tests/fixtures/dbt/sushi_test/models/waiter_as_customer_by_day.sql

@@ -1,7 +1,7 @@
 {{
     config(
         materialized='incremental',
-        incremental_strategy='delete+insert',
+        incremental_strategy='incremental_by_time_range',
         cluster_by=['ds'],
         time_column='ds',
     )

tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day.sql

@@ -1,7 +1,7 @@
 {{
     config(
         materialized='incremental',
-        incremental_strategy='delete+insert',
+        incremental_strategy='incremental_by_time_range',
         cluster_by=['ds'],
         time_column='ds',
         dialect="bigquery"

tests/fixtures/dbt/sushi_test/models/waiter_revenue_by_day_v1.sql

@@ -1,7 +1,7 @@
 {{
     config(
         materialized='incremental',
-        incremental_strategy='delete+insert',
+        incremental_strategy='incremental_by_time_range',
         cluster_by=['ds'],
         time_column='ds',
         dialect="bigquery"