Differentiate last_altered_ts from dev_last_altered_ts

Author: VaggelisD

sqlmesh/core/signal.py

@@ -42,11 +42,20 @@ class signal(registry_decorator):
 
 @signal()
 def freshness(batch: DatetimeRanges, snapshot: Snapshot, context: ExecutionContext) -> bool:
+    deployability_index = context.deployability_index
     adapter = context.engine_adapter
-    if not snapshot.last_altered_ts or not adapter.SUPPORTS_EXTERNAL_MODEL_FRESHNESS:
+
+    if not deployability_index or not adapter.SUPPORTS_EXTERNAL_MODEL_FRESHNESS:
+        return True
+
+    last_altered_ts = (
+        snapshot.last_altered_ts
+        if deployability_index.is_deployable(snapshot)
+        else snapshot.dev_last_altered_ts
+    )
+    if not last_altered_ts:
         return True
 
-    adapter = context.engine_adapter
     parent_snapshots = {context.snapshots[p.name] for p in snapshot.parents}
     if len(parent_snapshots) != len(snapshot.node.depends_on) or not all(
         p.is_external for p in parent_snapshots
@@ -57,7 +66,7 @@ def freshness(batch: DatetimeRanges, snapshot: Snapshot, context: ExecutionConte
     # Finding new data means that the upstream depedencies have been altered
     # since the last time the model was evaluated
     upstream_dep_has_new_data = any(
-        upstream_last_altered_ts > snapshot.last_altered_ts
+        upstream_last_altered_ts > last_altered_ts
         for upstream_last_altered_ts in adapter.get_external_model_freshness(
             [p.name for p in parent_snapshots]
         )

sqlmesh/core/snapshot/definition.py

@@ -186,6 +186,7 @@ class SnapshotIntervals(PydanticModel):
     dev_intervals: Intervals = []
     pending_restatement_intervals: Intervals = []
     last_altered_ts: t.Optional[int] = None
+    dev_last_altered_ts: t.Optional[int] = None
 
     @property
     def snapshot_id(self) -> t.Optional[SnapshotId]:
@@ -206,6 +207,12 @@ def add_dev_interval(self, start: int, end: int) -> None:
     def add_pending_restatement_interval(self, start: int, end: int) -> None:
         self._add_interval(start, end, "pending_restatement_intervals")
 
+    def add_last_altered_ts(self, last_altered_ts: t.Optional[int]) -> None:
+        self._add_last_altered_ts(last_altered_ts, "last_altered_ts")
+
+    def add_dev_last_altered_ts(self, last_altered_ts: t.Optional[int]) -> None:
+        self._add_last_altered_ts(last_altered_ts, "dev_last_altered_ts")
+
     def remove_interval(self, start: int, end: int) -> None:
         self._remove_interval(start, end, "intervals")
 
@@ -225,6 +232,13 @@ def _add_interval(self, start: int, end: int, interval_attr: str) -> None:
         target_intervals = merge_intervals([*target_intervals, (start, end)])
         setattr(self, interval_attr, target_intervals)
 
+    def _add_last_altered_ts(
+        self, last_altered_ts: t.Optional[int], last_altered_attr: str
+    ) -> None:
+        if last_altered_ts:
+            existing_last_altered_ts = getattr(self, last_altered_attr)
+            setattr(self, last_altered_attr, max(existing_last_altered_ts or -1, last_altered_ts))
+
     def _remove_interval(self, start: int, end: int, interval_attr: str) -> None:
         target_intervals = getattr(self, interval_attr)
         target_intervals = remove_interval(target_intervals, start, end)
@@ -687,6 +701,7 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
     # Physical table last modified timestamp, not to be confused with the "updated_ts" field
     # which is for the snapshot record itself
     last_altered_ts: t.Optional[int] = None
+    dev_last_altered_ts: t.Optional[int] = None
 
     @field_validator("ttl")
     @classmethod
@@ -726,13 +741,6 @@ def hydrate_with_intervals_by_version(
             for interval in snapshot_intervals:
                 snapshot.merge_intervals(interval)
 
-                # Differentiate last_altered_ts between snapshots with shared version but
-                # different dev versions e.g prod vs FORWARD_ONLY dev
-                if snapshot.dev_version == interval.dev_version and interval.last_altered_ts:
-                    snapshot.last_altered_ts = max(
-                        snapshot.last_altered_ts or -1, interval.last_altered_ts
-                    )
-
             result.append(snapshot)
 
         return result
@@ -939,12 +947,20 @@ def merge_intervals(self, other: t.Union[Snapshot, SnapshotIntervals]) -> None:
             if not apply_effective_from or end <= effective_from_ts:
                 self.add_interval(start, end)
 
+        if other.last_altered_ts:
+            self.last_altered_ts = max(self.last_altered_ts or -1, other.last_altered_ts)
+
         if self.dev_version == other.dev_version:
             # Merge dev intervals if the dev versions match which would mean
             # that this and the other snapshot are pointing to the same dev table.
             for start, end in other.dev_intervals:
                 self.add_interval(start, end, is_dev=True)
 
+            if other.dev_last_altered_ts:
+                self.dev_last_altered_ts = max(
+                    self.dev_last_altered_ts or -1, other.dev_last_altered_ts
+                )
+
         self.pending_restatement_intervals = merge_intervals(
             [*self.pending_restatement_intervals, *other.pending_restatement_intervals]
         )

sqlmesh/core/state_sync/db/interval.py

@@ -309,12 +309,6 @@ def _get_snapshot_intervals(
                         identifier=identifier,
                         version=version,
                         dev_version=dev_version,
-                        last_altered_ts=last_altered_ts,
-                    )
-
-                if last_altered_ts:
-                    intervals[merge_key].last_altered_ts = max(
-                        intervals[merge_key].last_altered_ts or 0, last_altered_ts
                     )
 
                 if pending_restatement_interval_merge_key not in intervals:
@@ -337,8 +331,10 @@ def _get_snapshot_intervals(
                 else:
                     if is_dev:
                         intervals[merge_key].add_dev_interval(start, end)
+                        intervals[merge_key].add_dev_last_altered_ts(last_altered_ts)
                     else:
                         intervals[merge_key].add_interval(start, end)
+                        intervals[merge_key].add_last_altered_ts(last_altered_ts)
                         # Remove all pending restatement intervals recorded before the current interval has been added
                         intervals[
                             pending_restatement_interval_merge_key

tests/core/engine_adapter/integration/test_integration.py

@@ -3835,13 +3835,24 @@ def test_external_model_freshness(ctx: TestContext, mocker: MockerFixture, tmp_p
     if not adapter.SUPPORTS_EXTERNAL_MODEL_FRESHNESS:
         pytest.skip("This test only runs for engines that support external model freshness")
 
-    def _assert_snapshot_last_altered_ts(context: Context, snapshot_id: str, timestamp: datetime):
+    def _assert_snapshot_last_altered_ts(
+        context: Context,
+        snapshot_id: str,
+        last_altered_ts: datetime,
+        dev_last_altered_ts: t.Optional[datetime] = None,
+    ):
         from sqlmesh.utils.date import to_datetime
 
         snapshot = context.state_sync.get_snapshots([snapshot_id])[snapshot_id]
-        assert to_datetime(snapshot.last_altered_ts).replace(microsecond=0) == timestamp.replace(
+
+        assert to_datetime(snapshot.last_altered_ts).replace(
             microsecond=0
-        )
+        ) == last_altered_ts.replace(microsecond=0)
+
+        if dev_last_altered_ts:
+            assert to_datetime(snapshot.dev_last_altered_ts).replace(
+                microsecond=0
+            ) == dev_last_altered_ts.replace(microsecond=0)
 
     import sqlmesh
 
@@ -3931,14 +3942,14 @@ def _set_config(gateway: str, config: Config) -> None:
     )
 
     prod_snapshot_id = next(iter(prod_plan.context_diff.new_snapshots))
-    _assert_snapshot_last_altered_ts(context, prod_snapshot_id, prod_plan_ts)
+    _assert_snapshot_last_altered_ts(context, prod_snapshot_id, last_altered_ts=prod_plan_ts)
 
     # Case 2: Model is NOT evaluated on run if external models are not fresh
     _assert_model_evaluation(lambda: context.run(), was_evaluated=False, day_delta=1)
 
     # Case 3: Differentiate last_altered_ts between snapshots with shared version
     # For instance, creating a FORWARD_ONLY change in dev (reusing the version but creating a dev preview) should not cause
-    # the prod snapshot's last_altered_ts to be updated when fetched from the state sync
+    # any side effects to the prod snapshot's last_altered_ts hydration
     model_path.write_text(model_path.read_text().replace("col1 * col2", "col1 + col2"))
     context.load()
     dev_plan_ts = now(minute_floor=False) + timedelta(days=2)
@@ -3949,8 +3960,13 @@ def _set_config(gateway: str, config: Config) -> None:
 
     context.state_sync.clear_cache()
     dev_snapshot_id = next(iter(dev_plan.context_diff.new_snapshots))
-    _assert_snapshot_last_altered_ts(context, dev_snapshot_id, dev_plan_ts)
-    _assert_snapshot_last_altered_ts(context, prod_snapshot_id, prod_plan_ts)
+    _assert_snapshot_last_altered_ts(
+        context,
+        dev_snapshot_id,
+        last_altered_ts=prod_plan_ts,
+        dev_last_altered_ts=dev_plan_ts,
+    )
+    _assert_snapshot_last_altered_ts(context, prod_snapshot_id, last_altered_ts=prod_plan_ts)
 
     # Case 4: Model is evaluated on run if any external model is fresh
     adapter.execute(f"INSERT INTO {external_table2} (col2) VALUES (3)", quote_identifiers=False)