Repository: TobikoData/sqlmesh
Pull request #2342 (2024-03-26 03:28:32)

fix!: fix time_data_type serialization/deserialization
The bug described here is what would happen prior to this PR: https://github.com/TobikoData/sqlmesh/pull/2318

This PR is expected to fix this behavior. I was able to repro the bugged behavior locally and it did fix that. I also verified scd type 2 column BQ integration still passes. The added test would fail with past behavior but now works.

Breaking because it depends on a minor bump of SQLGlot. 

Depends on: https://github.com/tobymao/sqlglot/pull/3220


Patch:
diff --git a/setup.py b/setup.py
index 0a53ef73..fa7e5e11 100644
--- a/setup.py
+++ b/setup.py
@@ -45,7 +45,7 @@ setup(
         "requests",
         "rich[jupyter]",
         "ruamel.yaml",
-        "sqlglot[rs]~=23.0.3",
+        "sqlglot[rs]~=23.2.0",
     ],
     extras_require={
         "bigquery": [
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 46c3e38e..56e2b1f6 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -755,6 +755,7 @@ class _Model(ModelMeta, frozen=True):
             json.dumps(self.column_descriptions, sort_keys=True),
             self.cron,
             str(self.start) if self.start else None,
+            str(self.end) if self.end else None,
             str(self.retention) if self.retention else None,
             str(self.batch_size) if self.batch_size is not None else None,
             json.dumps(self.mapping_schema, sort_keys=True),
diff --git a/sqlmesh/core/model/kind.py b/sqlmesh/core/model/kind.py
index 630ec9d4..7d1a2a01 100644
--- a/sqlmesh/core/model/kind.py
+++ b/sqlmesh/core/model/kind.py
@@ -370,12 +370,13 @@ class _SCDType2Kind(_ModelKind):
     valid_from_name: SQLGlotString = "valid_from"
     valid_to_name: SQLGlotString = "valid_to"
     invalidate_hard_deletes: SQLGlotBool = False
-    time_data_type: exp.DataType = exp.DataType.build("TIMESTAMP")
+    time_data_type: exp.DataType = Field(exp.DataType.build("TIMESTAMP"), validate_default=True)
 
     forward_only: SQLGlotBool = True
     disable_restatement: SQLGlotBool = True
 
-    @field_validator("time_data_type", mode="before")
+    # always=True can be removed once Pydantic 1 is deprecated
+    @field_validator("time_data_type", mode="before", always=True)
     @classmethod
     def _time_data_type_validator(
         cls, v: t.Union[str, exp.Expression], values: t.Any
@@ -383,7 +384,9 @@ class _SCDType2Kind(_ModelKind):
         values = values if isinstance(values, dict) else values.data
         if isinstance(v, exp.Expression) and not isinstance(v, exp.DataType):
             v = v.name
-        return exp.DataType.build(v, dialect=values.get("dialect"))
+        data_type = exp.DataType.build(v, dialect=values.get("dialect"))
+        data_type.meta["dialect"] = values.get("dialect")
+        return data_type
 
     @property
     def managed_columns(self) -> t.Dict[str, exp.DataType]:
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index 634d4d3b..d5a2fcaa 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -5,6 +5,7 @@ import traceback
 import typing as t
 from datetime import datetime
 
+from sqlmesh.core import constants as c
 from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.model import SeedModel
@@ -243,7 +244,11 @@ class Scheduler:
         else:
             environment_naming_info = environment
 
-        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
+        deployability_index = deployability_index or (
+            DeployabilityIndex.create(self.snapshots.values())
+            if environment_naming_info.name != c.PROD
+            else DeployabilityIndex.all_deployable()
+        )
         execution_time = execution_time or now()
         batches = self.batches(
             start,


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 74ce12a3..fa76bd07 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -444,6 +444,40 @@ def test_json_serde():
     assert deserialized_model == model
 
 
+def test_scd_type_2_by_col_serde():
+    expressions = parse(
+        """
+        MODEL(
+            name test_model,
+            KIND SCD_TYPE_2_BY_COLUMN (
+                unique_key a,
+                columns *,
+            ),
+            cron '@daily',
+            owner 'reakman',
+            grain a,
+            dialect bigquery
+        );
+
+        SELECT a, ds FROM `tbl`
+    """,
+        default_dialect="bigquery",
+    )
+
+    model = load_sql_based_model(expressions)
+
+    model_json = model.json()
+    model_json_parsed = json.loads(model.json())
+    assert model_json_parsed["kind"]["dialect"] == "bigquery"
+    assert model_json_parsed["kind"]["unique_key"] == ["`a`"]
+    assert model_json_parsed["kind"]["columns"] == "*"
+    # Bigquery converts TIMESTAMP -> DATETIME
+    assert model_json_parsed["kind"]["time_data_type"] == "DATETIME"
+
+    deserialized_model = SqlModel.parse_raw(model_json)
+    assert deserialized_model.dict() == model.dict()
+
+
 def test_column_descriptions(sushi_context, assert_exp_eq):
     assert sushi_context.models[
         '"memory"."sushi"."customer_revenue_by_day"'
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 8ad01eb8..88ae4a78 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -606,7 +606,7 @@ def test_fingerprint(model: Model, parent_model: Model):
 
     original_fingerprint = SnapshotFingerprint(
         data_hash="372035512",
-        metadata_hash="3575901791",
+        metadata_hash="490962550",
     )
 
     assert fingerprint == original_fingerprint
@@ -666,7 +666,7 @@ def test_fingerprint_seed_model():
 
     expected_fingerprint = SnapshotFingerprint(
         data_hash="2156038176",
-        metadata_hash="3058233394",
+        metadata_hash="3304326900",
     )
 
     model = load_sql_based_model(expressions, path=Path("./examples/sushi/models/test_model.sql"))
@@ -705,7 +705,7 @@ def test_fingerprint_jinja_macros(model: Model):
     )
     original_fingerprint = SnapshotFingerprint(
         data_hash="900027747",
-        metadata_hash="3575901791",
+        metadata_hash="490962550",
     )
 
     fingerprint = fingerprint_from_node(model, nodes={})