fix: dbt time column serialization (#5336)

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_manifest.py

@@ -5,7 +5,6 @@
 import pytest
 
 from sqlmesh.core.config import ModelDefaultsConfig
-from sqlmesh.core.model import TimeColumn
 from sqlmesh.dbt.basemodel import Dependencies
 from sqlmesh.dbt.common import ModelAttrs
 from sqlmesh.dbt.context import DbtContext
@@ -82,9 +81,9 @@ def test_manifest_helper(caplog):
         macros=[MacroReference(name="ref")],
     )
     assert waiter_as_customer_by_day_config.materialized == "incremental"
-    assert waiter_as_customer_by_day_config.incremental_strategy == "delete+insert"
+    assert waiter_as_customer_by_day_config.incremental_strategy == "incremental_by_time_range"
     assert waiter_as_customer_by_day_config.cluster_by == ["ds"]
-    assert waiter_as_customer_by_day_config.time_column == TimeColumn.create("ds", "duckdb")
+    assert waiter_as_customer_by_day_config.time_column == "ds"
 
     if DBT_VERSION >= (1, 5, 0):
         waiter_revenue_by_day_config = models["waiter_revenue_by_day_v2"]
@@ -104,9 +103,9 @@ def test_manifest_helper(caplog):
         has_dynamic_var_names=True,
     )
     assert waiter_revenue_by_day_config.materialized == "incremental"
-    assert waiter_revenue_by_day_config.incremental_strategy == "delete+insert"
+    assert waiter_revenue_by_day_config.incremental_strategy == "incremental_by_time_range"
     assert waiter_revenue_by_day_config.cluster_by == ["ds"]
-    assert waiter_revenue_by_day_config.time_column == TimeColumn.create("ds", "duckdb")
+    assert waiter_revenue_by_day_config.time_column == "ds"
     assert waiter_revenue_by_day_config.dialect_ == "bigquery"
 
     assert helper.models("customers")["customers"].dependencies == Dependencies(

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"