Feat: Skip evaluation if upstream external model has not changed

Author: VaggelisD

sqlmesh/core/engine_adapter/base.py

@@ -119,6 +119,7 @@ class EngineAdapter:
     MAX_IDENTIFIER_LENGTH: t.Optional[int] = None
     ATTACH_CORRELATION_ID = True
     SUPPORTS_QUERY_EXECUTION_TRACKING = False
+    SUPPORTS_EXTERNAL_MODEL_FRESHNESS = False
 
     def __init__(
         self,
@@ -2873,6 +2874,9 @@ def _check_identifier_length(self, expression: exp.Expression) -> None:
                     f"Identifier name '{name}' (length {name_length}) exceeds {self.dialect.capitalize()}'s max identifier limit of {self.MAX_IDENTIFIER_LENGTH} characters"
                 )
 
+    def get_external_model_freshness(self, table_names: t.List[TableName]) -> t.List[int]:
+        raise NotImplementedError()
+
 
 class EngineAdapterWithIndexSupport(EngineAdapter):
     SUPPORTS_INDEXES = True

sqlmesh/core/engine_adapter/bigquery.py

@@ -67,6 +67,7 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin, ClusteredByMixin, Row
     MAX_TABLE_COMMENT_LENGTH = 1024
     MAX_COLUMN_COMMENT_LENGTH = 1024
     SUPPORTS_QUERY_EXECUTION_TRACKING = True
+    SUPPORTS_EXTERNAL_MODEL_FRESHNESS = True
     SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["SCHEMA"]
 
     SCHEMA_DIFFER_KWARGS = {
@@ -753,6 +754,28 @@ def table_exists(self, table_name: TableName) -> bool:
         except NotFound:
             return False
 
+    def get_external_model_freshness(self, table_names: t.List[TableName]) -> t.List[int]:
+        from sqlmesh.utils.date import to_timestamp
+
+        datasets_to_tables: t.DefaultDict[str, t.List[str]] = defaultdict(list)
+        for table_name in table_names:
+            table = exp.to_table(table_name)
+            datasets_to_tables[table.db].append(table.name)
+
+        results = []
+
+        for dataset, tables in datasets_to_tables.items():
+            query = (
+                f"SELECT TIMESTAMP_MILLIS(last_modified_time) FROM `{dataset}.__TABLES__` WHERE "
+            )
+            for i, table_name in enumerate(tables):
+                query += f"TABLE_ID = '{table_name}'"
+                if i < len(tables) - 1:
+                    query += " OR "
+            results.extend(self.fetchall(query))
+
+        return [to_timestamp(row[0]) for row in results]
+
     def _get_table(self, table_name: TableName) -> BigQueryTable:
         """
         Returns a BigQueryTable object for the given table name.

sqlmesh/core/engine_adapter/snowflake.py

@@ -54,6 +54,7 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin, ClusteredByMixi
     SUPPORTS_MANAGED_MODELS = True
     CURRENT_CATALOG_EXPRESSION = exp.func("current_database")
     SUPPORTS_CREATE_DROP_CATALOG = True
+    SUPPORTS_EXTERNAL_MODEL_FRESHNESS = True
     SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["DATABASE", "SCHEMA", "TABLE"]
     SCHEMA_DIFFER_KWARGS = {
         "parameterized_type_defaults": {
@@ -665,3 +666,18 @@ def close(self) -> t.Any:
             self._connection_pool.set_attribute(self.SNOWPARK, None)
 
         return super().close()
+
+    def get_external_model_freshness(self, table_names: t.List[TableName]) -> t.List[int]:
+        from sqlmesh.utils.date import to_timestamp
+
+        num_tables = len(table_names)
+
+        query = "SELECT LAST_ALTERED FROM INFORMATION_SCHEMA.TABLES WHERE"
+        for i, table_name in enumerate(table_names):
+            table = exp.to_table(table_name)
+            query += f"""(TABLE_NAME = '{table.name}' AND TABLE_SCHEMA = '{table.db}' AND TABLE_CATALOG = '{table.catalog}')"""
+            if i < num_tables - 1:
+                query += " OR "
+
+        result = self.fetchall(query)
+        return [to_timestamp(row[0]) for row in result]

sqlmesh/core/plan/evaluator.py

@@ -257,6 +257,7 @@ def visit_backfill_stage(self, stage: stages.BackfillStage, plan: EvaluatablePla
             allow_destructive_snapshots=plan.allow_destructive_models,
             allow_additive_snapshots=plan.allow_additive_models,
             selected_snapshot_ids=stage.selected_snapshot_ids,
+            is_restatement_plan=bool(plan.restatements),
         )
         if errors:
             raise PlanError("Plan application failed.")

sqlmesh/core/scheduler.py

@@ -54,6 +54,8 @@
 
 if t.TYPE_CHECKING:
     from sqlmesh.core.context import ExecutionContext
+    from sqlmesh.core._typing import TableName
+    from sqlmesh.core.engine_adapter import EngineAdapter
 
 logger = logging.getLogger(__name__)
 SnapshotToIntervals = t.Dict[Snapshot, Intervals]
@@ -188,6 +190,46 @@ def merged_missing_intervals(
             }
         return snapshots_to_intervals
 
+    def can_skip_evaluation(self, snapshot: Snapshot, snapshots: t.Dict[str, Snapshot]) -> bool:
+        if not snapshot.last_altered_ts:
+            return False
+
+        from collections import defaultdict
+
+        parent_snapshots = {p for p in snapshots.values() if p.name != snapshot.name}
+        if len(parent_snapshots) != len(snapshot.node.depends_on):
+            # The mismatch can happen if e.g an external model is not registered in the project
+            return False
+
+        adapter_to_parent_snapshots: t.Dict[EngineAdapter, t.List[Snapshot]] = defaultdict(list)
+
+        for parent_snapshot in parent_snapshots:
+            if not parent_snapshot.is_external:
+                return False
+
+            adapter = self.snapshot_evaluator.get_adapter(parent_snapshot.model_gateway)
+            if not adapter.SUPPORTS_EXTERNAL_MODEL_FRESHNESS:
+                return False
+
+            adapter_to_parent_snapshots[adapter].append(parent_snapshot)
+
+        if not adapter_to_parent_snapshots:
+            return False
+
+        external_models_freshness: t.List[int] = []
+
+        for adapter, adapter_snapshots in adapter_to_parent_snapshots.items():
+            table_names: t.List[TableName] = [
+                exp.to_table(parent_snapshot.name, parent_snapshot.node.dialect)
+                for parent_snapshot in adapter_snapshots
+            ]
+            external_models_freshness.extend(adapter.get_external_model_freshness(table_names))
+
+        return all(
+            snapshot.last_altered_ts > external_model_freshness
+            for external_model_freshness in external_models_freshness
+        )
+
     def evaluate(
         self,
         snapshot: Snapshot,
@@ -200,6 +242,7 @@ def evaluate(
         allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
         allow_additive_snapshots: t.Optional[t.Set[str]] = None,
         target_table_exists: t.Optional[bool] = None,
+        is_restatement_plan: bool = False,
         **kwargs: t.Any,
     ) -> t.List[AuditResult]:
         """Evaluate a snapshot and add the processed interval to the state sync.
@@ -224,6 +267,14 @@ def evaluate(
 
         snapshots = parent_snapshots_by_name(snapshot, self.snapshots)
 
+        if not is_restatement_plan and self.can_skip_evaluation(snapshot, snapshots):
+            logger.info(f"""
+            Skipping evaluation for snapshot {snapshot.name} as it depends on external models 
+            that have not been updated since the last run.
+            """)
+
+            return []
+
         is_deployable = deployability_index.is_deployable(snapshot)
 
         wap_id = self.snapshot_evaluator.evaluate(
@@ -251,7 +302,9 @@ def evaluate(
             **kwargs,
         )
 
-        self.state_sync.add_interval(snapshot, start, end, is_dev=not is_deployable)
+        self.state_sync.add_interval(
+            snapshot, start, end, is_dev=not is_deployable, last_altered_ts=now_timestamp()
+        )
         return audit_results
 
     def run(
@@ -421,6 +474,7 @@ def run_merged_intervals(
         run_environment_statements: bool = False,
         audit_only: bool = False,
         auto_restatement_triggers: t.Dict[SnapshotId, t.List[SnapshotId]] = {},
+        is_restatement_plan: bool = False,
     ) -> t.Tuple[t.List[NodeExecutionFailedError[SchedulingUnit]], t.List[SchedulingUnit]]:
         """Runs precomputed batches of missing intervals.
 
@@ -526,6 +580,7 @@ def run_node(node: SchedulingUnit) -> None:
                             allow_destructive_snapshots=allow_destructive_snapshots,
                             allow_additive_snapshots=allow_additive_snapshots,
                             target_table_exists=snapshot.snapshot_id not in snapshots_to_create,
+                            is_restatement_plan=is_restatement_plan,
                         )
 
                     evaluation_duration_ms = now_timestamp() - execution_start_ts

sqlmesh/core/snapshot/definition.py

@@ -185,6 +185,7 @@ class SnapshotIntervals(PydanticModel):
     intervals: Intervals = []
     dev_intervals: Intervals = []
     pending_restatement_intervals: Intervals = []
+    last_altered_ts: t.Optional[int] = None
 
     @property
     def snapshot_id(self) -> t.Optional[SnapshotId]:
@@ -652,6 +653,9 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
     dev_table_suffix: str = "dev"
     table_naming_convention: TableNamingConvention = TableNamingConvention.default
     forward_only: bool = False
+    # 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
 
     @field_validator("ttl")
     @classmethod
@@ -690,6 +694,12 @@ def hydrate_with_intervals_by_version(
             )
             for interval in snapshot_intervals:
                 snapshot.merge_intervals(interval)
+
+                if interval.last_altered_ts:
+                    snapshot.last_altered_ts = max(
+                        snapshot.last_altered_ts or -1, interval.last_altered_ts
+                    )
+
             result.append(snapshot)
 
         return result

sqlmesh/core/state_sync/base.py

@@ -477,6 +477,7 @@ def add_interval(
         start: TimeLike,
         end: TimeLike,
         is_dev: bool = False,
+        last_altered_ts: t.Optional[int] = None,
     ) -> None:
         """Add an interval to a snapshot and sync it to the store.
 
@@ -485,6 +486,7 @@ def add_interval(
             start: The start of the interval to add.
             end: The end of the interval to add.
             is_dev: Indicates whether the given interval is being added while in development mode
+            last_altered_ts: The timestamp of the last modification of the physical table
         """
         start_ts, end_ts = snapshot.inclusive_exclusive(start, end, strict=False, expand=False)
         if not snapshot.version:
@@ -497,6 +499,7 @@ def add_interval(
             dev_version=snapshot.dev_version,
             intervals=intervals if not is_dev else [],
             dev_intervals=intervals if is_dev else [],
+            last_altered_ts=last_altered_ts,
         )
         self.add_snapshots_intervals([snapshot_intervals])
 

sqlmesh/core/state_sync/db/interval.py

@@ -60,6 +60,7 @@ def __init__(
             "is_removed": exp.DataType.build("boolean"),
             "is_compacted": exp.DataType.build("boolean"),
             "is_pending_restatement": exp.DataType.build("boolean"),
+            "last_altered_ts": exp.DataType.build("bigint"),
         }
 
     def add_snapshots_intervals(self, snapshots_intervals: t.Sequence[SnapshotIntervals]) -> None:
@@ -215,13 +216,23 @@ def _push_snapshot_intervals(
             for start_ts, end_ts in snapshot.intervals:
                 new_intervals.append(
                     _interval_to_df(
-                        snapshot, start_ts, end_ts, is_dev=False, is_compacted=is_compacted
+                        snapshot,
+                        start_ts,
+                        end_ts,
+                        is_dev=False,
+                        is_compacted=is_compacted,
+                        last_altered_ts=snapshot.last_altered_ts,
                     )
                 )
             for start_ts, end_ts in snapshot.dev_intervals:
                 new_intervals.append(
                     _interval_to_df(
-                        snapshot, start_ts, end_ts, is_dev=True, is_compacted=is_compacted
+                        snapshot,
+                        start_ts,
+                        end_ts,
+                        is_dev=True,
+                        is_compacted=is_compacted,
+                        last_altered_ts=snapshot.last_altered_ts,
                     )
                 )
 
@@ -236,6 +247,7 @@ def _push_snapshot_intervals(
                         is_dev=False,
                         is_compacted=is_compacted,
                         is_pending_restatement=True,
+                        last_altered_ts=snapshot.last_altered_ts,
                     )
                 )
 
@@ -284,6 +296,7 @@ def _get_snapshot_intervals(
                 is_dev,
                 is_removed,
                 is_pending_restatement,
+                last_altered_ts,
             ) in rows:
                 interval_ids.add(interval_id)
                 merge_key = (name, version, dev_version, identifier)
@@ -296,6 +309,12 @@ 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:
@@ -340,6 +359,7 @@ def _get_snapshot_intervals_query(self, uncompacted_only: bool) -> exp.Select:
                 "is_dev",
                 "is_removed",
                 "is_pending_restatement",
+                "last_altered_ts",
             )
             .from_(exp.to_table(self.intervals_table).as_("intervals"))
             .order_by(
@@ -458,6 +478,7 @@ def _interval_to_df(
     is_removed: bool = False,
     is_compacted: bool = False,
     is_pending_restatement: bool = False,
+    last_altered_ts: t.Optional[int] = None,
 ) -> t.Dict[str, t.Any]:
     return {
         "id": random_id(),
@@ -472,4 +493,5 @@ def _interval_to_df(
         "is_removed": is_removed,
         "is_compacted": is_compacted,
         "is_pending_restatement": is_pending_restatement,
+        "last_altered_ts": last_altered_ts,
     }

sqlmesh/migrations/v0093_add_last_altered_to_intervals.py

@@ -0,0 +1,23 @@
+"""Add dev version to the intervals table."""
+
+from sqlglot import exp
+
+
+def migrate(state_sync, **kwargs):  # type: ignore
+    engine_adapter = state_sync.engine_adapter
+    schema = state_sync.schema
+    intervals_table = "_intervals"
+    if schema:
+        intervals_table = f"{schema}.{intervals_table}"
+
+    alter_table_exp = exp.Alter(
+        this=exp.to_table(intervals_table),
+        kind="TABLE",
+        actions=[
+            exp.ColumnDef(
+                this=exp.to_column("last_altered_ts"),
+                kind=exp.DataType.build("BIGINT", dialect=engine_adapter.dialect),
+            )
+        ],
+    )
+    engine_adapter.execute(alter_table_exp)