PR Feedback 2

Author: VaggelisD

sqlmesh/core/context.py

@@ -274,7 +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,
+        is_restatement: t.Optional[bool] = None,
         variables: t.Optional[t.Dict[str, t.Any]] = None,
         blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
     ):
@@ -285,7 +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
+        self._is_restatement = is_restatement
 
     @property
     def default_dialect(self) -> t.Optional[str]:
@@ -311,8 +311,8 @@ def gateway(self) -> t.Optional[str]:
         return self.var(c.GATEWAY)
 
     @property
-    def is_restatement_plan(self) -> t.Optional[bool]:
-        return self._is_restatement_plan
+    def is_restatement(self) -> t.Optional[bool]:
+        return self._is_restatement
 
     def var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
         """Returns a variable value."""
@@ -334,7 +334,7 @@ def with_variables(
             self.deployability_index,
             self._default_dialect,
             self._default_catalog,
-            self._is_restatement_plan,
+            self._is_restatement,
             variables=variables,
             blueprint_variables=blueprint_variables,
         )

sqlmesh/core/plan/evaluator.py

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

sqlmesh/core/scheduler.py

@@ -200,7 +200,6 @@ 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.
@@ -338,7 +337,7 @@ def batch_intervals(
         deployability_index: t.Optional[DeployabilityIndex],
         environment_naming_info: EnvironmentNamingInfo,
         dag: t.Optional[DAG[SnapshotId]] = None,
-        is_restatement_plan: bool = False,
+        is_restatement: bool = False,
     ) -> t.Dict[Snapshot, Intervals]:
         dag = dag or snapshots_to_dag(merged_intervals)
 
@@ -371,15 +370,14 @@ def batch_intervals(
                 deployability_index,
                 default_dialect=adapter.dialect,
                 default_catalog=self.default_catalog,
-                is_restatement_plan=is_restatement_plan,
+                is_restatement=is_restatement,
             )
 
             intervals = self._check_ready_intervals(
                 snapshot,
                 intervals,
                 context,
                 environment_naming_info,
-                is_restatement_plan=is_restatement_plan,
             )
             unready -= set(intervals)
 
@@ -428,7 +426,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,
+        is_restatement: bool = False,
     ) -> t.Tuple[t.List[NodeExecutionFailedError[SchedulingUnit]], t.List[SchedulingUnit]]:
         """Runs precomputed batches of missing intervals.
 
@@ -466,7 +464,7 @@ def run_merged_intervals(
             deployability_index,
             environment_naming_info,
             dag=snapshot_dag,
-            is_restatement_plan=is_restatement_plan,
+            is_restatement=is_restatement,
         )
         self.console.start_evaluation_progress(
             batched_intervals,
@@ -552,7 +550,6 @@ def run_node(node: SchedulingUnit) -> None:
                             allow_additive_snapshots=allow_additive_snapshots,
                             target_table_exists=snapshot.snapshot_id not in snapshots_to_create,
                             selected_models=selected_models,
-                            is_restatement_plan=is_restatement_plan,
                         )
 
                     evaluation_duration_ms = now_timestamp() - execution_start_ts
@@ -926,7 +923,6 @@ def _check_ready_intervals(
         intervals: Intervals,
         context: ExecutionContext,
         environment_naming_info: EnvironmentNamingInfo,
-        is_restatement_plan: bool = False,
     ) -> Intervals:
         """Checks if the intervals are ready for evaluation for the given snapshot.
 
@@ -948,8 +944,6 @@ def _check_ready_intervals(
         if not (signals and signals.signals_to_kwargs):
             return intervals
 
-        signal_names = signals.signals_to_kwargs.keys()
-
         self.console.start_signal_progress(
             snapshot,
             self.default_catalog,

sqlmesh/core/signal.py

@@ -7,6 +7,7 @@
     from sqlmesh.core.context import ExecutionContext
     from sqlmesh.core.snapshot.definition import Snapshot
     from sqlmesh.utils.date import DatetimeRanges
+    from sqlmesh.core.snapshot.definition import DeployabilityIndex
 
 
 class signal(registry_decorator):
@@ -42,15 +43,12 @@ 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_METADATA_TABLE_LAST_MODIFIED_TS:
+    if context.is_restatement or not adapter.SUPPORTS_METADATA_TABLE_LAST_MODIFIED_TS:
         return True
 
+    deployability_index = context.deployability_index or DeployabilityIndex.all_deployable()
+
     last_altered_ts = (
         snapshot.last_altered_ts
         if deployability_index.is_deployable(snapshot)