PR Feedback 1

Author: VaggelisD

sqlmesh/core/context.py

@@ -274,6 +274,7 @@ def __init__(
         deployability_index: t.Optional[DeployabilityIndex] = None,
         default_dialect: t.Optional[str] = None,
         default_catalog: t.Optional[str] = None,
+        is_restatement_plan: t.Optional[bool] = None,
         variables: t.Optional[t.Dict[str, t.Any]] = None,
         blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
     ):
@@ -284,6 +285,7 @@ def __init__(
         self._default_dialect = default_dialect
         self._variables = variables or {}
         self._blueprint_variables = blueprint_variables or {}
+        self._is_restatement_plan = is_restatement_plan
 
     @property
     def default_dialect(self) -> t.Optional[str]:
@@ -308,6 +310,10 @@ def gateway(self) -> t.Optional[str]:
         """Returns the gateway name."""
         return self.var(c.GATEWAY)
 
+    @property
+    def is_restatement_plan(self) -> t.Optional[bool]:
+        return self._is_restatement_plan
+
     def var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
         """Returns a variable value."""
         return self._variables.get(var_name.lower(), default)
@@ -328,6 +334,7 @@ def with_variables(
             self.deployability_index,
             self._default_dialect,
             self._default_catalog,
+            self._is_restatement_plan,
             variables=variables,
             blueprint_variables=blueprint_variables,
         )

sqlmesh/core/engine_adapter/base.py

@@ -119,7 +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
+    SUPPORTS_METADATA_TABLE_LAST_MODIFIED_TS = False
 
     def __init__(
         self,
@@ -2928,7 +2928,7 @@ 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]:
+    def get_table_last_modified_ts(self, table_names: t.List[TableName]) -> t.List[int]:
         raise NotImplementedError()
 
 

sqlmesh/core/engine_adapter/bigquery.py

@@ -754,7 +754,7 @@ 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]:
+    def get_table_last_modified_ts(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)

sqlmesh/core/engine_adapter/snowflake.py

@@ -54,7 +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
+    SUPPORTS_METADATA_TABLE_LAST_MODIFIED_TS = True
     SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["DATABASE", "SCHEMA", "TABLE"]
     SCHEMA_DIFFER_KWARGS = {
         "parameterized_type_defaults": {
@@ -668,7 +668,7 @@ def close(self) -> t.Any:
 
         return super().close()
 
-    def get_external_model_freshness(self, table_names: t.List[TableName]) -> t.List[int]:
+    def get_table_last_modified_ts(self, table_names: t.List[TableName]) -> t.List[int]:
         from sqlmesh.utils.date import to_timestamp
 
         num_tables = len(table_names)

sqlmesh/core/scheduler.py

@@ -54,8 +54,6 @@
 
 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]
@@ -190,46 +188,6 @@ 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,
@@ -413,6 +371,7 @@ def batch_intervals(
                 deployability_index,
                 default_dialect=adapter.dialect,
                 default_catalog=self.default_catalog,
+                is_restatement_plan=is_restatement_plan,
             )
 
             intervals = self._check_ready_intervals(
@@ -989,25 +948,13 @@ def _check_ready_intervals(
 
         signal_names = signals.signals_to_kwargs.keys()
 
-        if (
-            is_restatement_plan
-            and len(signal_names) == 1
-            and next(iter(signal_names)) == "freshness"
-        ):
-            # Freshness signal is not checked for restatement plans to allow users
-            # for an escape hatch in reevaluating models
-            return intervals
-
         self.console.start_signal_progress(
             snapshot,
             self.default_catalog,
             environment_naming_info or EnvironmentNamingInfo(),
         )
 
         for signal_idx, (signal_name, kwargs) in enumerate(signals.signals_to_kwargs.items()):
-            if is_restatement_plan and signal_name == "freshness":
-                continue
-
             # Capture intervals before signal check for display
             intervals_to_check = merge_intervals(intervals)
 

sqlmesh/core/signal.py

@@ -42,10 +42,13 @@ class signal(registry_decorator):
 
 @signal()
 def freshness(batch: DatetimeRanges, snapshot: Snapshot, context: ExecutionContext) -> bool:
+    if context.is_restatement_plan:
+        return True
+
     deployability_index = context.deployability_index
     adapter = context.engine_adapter
 
-    if not deployability_index or not adapter.SUPPORTS_EXTERNAL_MODEL_FRESHNESS:
+    if not deployability_index or not adapter.SUPPORTS_METADATA_TABLE_LAST_MODIFIED_TS:
         return True
 
     last_altered_ts = (
@@ -67,7 +70,7 @@ def freshness(batch: DatetimeRanges, snapshot: Snapshot, context: ExecutionConte
     # since the last time the model was evaluated
     upstream_dep_has_new_data = any(
         upstream_last_altered_ts > last_altered_ts
-        for upstream_last_altered_ts in adapter.get_external_model_freshness(
+        for upstream_last_altered_ts in adapter.get_table_last_modified_ts(
             [p.name for p in parent_snapshots]
         )
     )

sqlmesh/core/snapshot/definition.py

@@ -207,11 +207,11 @@ 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 update_last_altered_ts(self, last_altered_ts: t.Optional[int]) -> None:
+        self._update_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 update_dev_last_altered_ts(self, last_altered_ts: t.Optional[int]) -> None:
+        self._update_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")
@@ -232,12 +232,12 @@ 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(
+    def _update_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))
+            setattr(self, last_altered_attr, max(existing_last_altered_ts or 0, last_altered_ts))
 
     def _remove_interval(self, start: int, end: int, interval_attr: str) -> None:
         target_intervals = getattr(self, interval_attr)
@@ -978,7 +978,7 @@ def merge_intervals(self, other: t.Union[Snapshot, SnapshotIntervals]) -> None:
                 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)
+            self.last_altered_ts = max(self.last_altered_ts or 0, other.last_altered_ts)
 
         if self.dev_version == other.dev_version:
             # Merge dev intervals if the dev versions match which would mean
@@ -988,7 +988,7 @@ def merge_intervals(self, other: t.Union[Snapshot, SnapshotIntervals]) -> None:
 
             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.dev_last_altered_ts or 0, other.dev_last_altered_ts
                 )
 
         self.pending_restatement_intervals = merge_intervals(

sqlmesh/core/state_sync/db/interval.py

@@ -331,10 +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)
+                        intervals[merge_key].update_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)
+                        intervals[merge_key].update_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

@@ -3845,8 +3845,8 @@ def _assert_mview_value(value: int):
 @use_terminal_console
 def test_external_model_freshness(ctx: TestContext, mocker: MockerFixture, tmp_path: pathlib.Path):
     adapter = ctx.engine_adapter
-    if not adapter.SUPPORTS_EXTERNAL_MODEL_FRESHNESS:
-        pytest.skip("This test only runs for engines that support external model freshness")
+    if not adapter.SUPPORTS_METADATA_TABLE_LAST_MODIFIED_TS:
+        pytest.skip("This test only runs for engines that support metadata-based freshness")
 
     def _assert_snapshot_last_altered_ts(
         context: Context,
@@ -3880,15 +3880,7 @@ def _assert_model_evaluation(lambda_func, was_evaluated, day_delta=0):
 
         evaluate_function_called = spy.call_count == 1
         signal_was_checked = "Checking signals for" in output.stdout
-        restatement_plan = isinstance(plan_or_run_result, Plan) and plan_or_run_result.restatements
-        if restatement_plan:
-            # Restatement plans exclude this signal so we expect the actual evaluation
-            # to happen but not through the signal
-            assert evaluate_function_called
-            assert not signal_was_checked
-            return
 
-        # All other cases (e.g normal plans or runs) will check the freshness signal
         assert signal_was_checked
         if was_evaluated:
             assert "All ready" in output.stdout