Revert plan-related triggers

Author: treysp

sqlmesh/core/console.py

@@ -3830,10 +3830,6 @@ def update_snapshot_evaluation_progress(
                 message += f" | auto_restatement_triggers={','.join(trigger.name for trigger in snapshot_evaluation_triggers.auto_restatement_triggers)}"
             if snapshot_evaluation_triggers.select_snapshot_triggers:
                 message += f" | select_snapshot_triggers={','.join(trigger.name for trigger in snapshot_evaluation_triggers.select_snapshot_triggers)}"
-            if snapshot_evaluation_triggers.directly_modified_triggers:
-                message += f" | directly_modified_triggers={','.join(trigger.name for trigger in snapshot_evaluation_triggers.directly_modified_triggers)}"
-            if snapshot_evaluation_triggers.restatement_triggers:
-                message += f" | restatement_triggers={','.join(trigger.name for trigger in snapshot_evaluation_triggers.restatement_triggers)}"
 
         if audit_only:
             message = f"Audited {snapshot.name} duration={duration_ms}ms | num_audits_passed={num_audits_passed} | num_audits_failed={num_audits_failed}"

sqlmesh/core/plan/builder.py

@@ -293,7 +293,7 @@ def build(self) -> Plan:
             else DeployabilityIndex.all_deployable()
         )
 
-        restatements, restatement_triggers = self._build_restatements(
+        restatements = self._build_restatements(
             dag,
             earliest_interval_start(self._context_diff.snapshots.values(), self.execution_time),
         )
@@ -330,7 +330,6 @@ def build(self) -> Plan:
             indirectly_modified=indirectly_modified,
             deployability_index=deployability_index,
             restatements=restatements,
-            restatement_triggers=restatement_triggers,
             start_override_per_model=self._start_override_per_model,
             end_override_per_model=end_override_per_model,
             selected_models_to_backfill=self._backfill_models,
@@ -353,14 +352,14 @@ def _build_dag(self) -> DAG[SnapshotId]:
 
     def _build_restatements(
         self, dag: DAG[SnapshotId], earliest_interval_start: TimeLike
-    ) -> t.Tuple[t.Dict[SnapshotId, Interval], t.Dict[SnapshotId, t.List[SnapshotId]]]:
+    ) -> t.Dict[SnapshotId, Interval]:
         restate_models = self._restate_models
         if restate_models == set():
             # This is a warning but we print this as error since the Console is lacking API for warnings.
             self._console.log_error(
                 "Provided restated models do not match any models. No models will be included in plan."
             )
-            return {}, {}
+            return {}
 
         restatements: t.Dict[SnapshotId, Interval] = {}
         forward_only_preview_needed = self._forward_only_preview_needed
@@ -384,7 +383,7 @@ def _build_restatements(
             is_preview = True
 
         if not restate_models:
-            return {}, {}
+            return {}
 
         start = self._start or earliest_interval_start
         end = self._end or now()
@@ -394,7 +393,6 @@ def _build_restatements(
             if model_fqn not in self._model_fqn_to_snapshot:
                 raise PlanError(f"Cannot restate model '{model_fqn}'. Model does not exist.")
 
-        restatement_triggers: t.Dict[SnapshotId, t.List[SnapshotId]] = {}
         # Get restatement intervals for all restated snapshots and make sure that if an incremental snapshot expands it's
         # restatement range that it's downstream dependencies all expand their restatement ranges as well.
         for s_id in dag:
@@ -430,13 +428,6 @@ def _build_restatements(
                     logger.info("Skipping restatement for model '%s'", snapshot.name)
                     continue
 
-            if snapshot.name in restate_models:
-                restatement_triggers[s_id] = [s_id]
-            if restating_parents:
-                restatement_triggers[s_id] = restatement_triggers.get(s_id, []) + [
-                    s.snapshot_id for s in restating_parents
-                ]
-
             possible_intervals = {
                 restatements[p.snapshot_id] for p in restating_parents if p.is_incremental
             }
@@ -465,7 +456,7 @@ def _build_restatements(
 
             restatements[s_id] = (snapshot_start, snapshot_end)
 
-        return restatements, restatement_triggers
+        return restatements
 
     def _build_directly_and_indirectly_modified(
         self, dag: DAG[SnapshotId]

sqlmesh/core/plan/definition.py

@@ -58,7 +58,6 @@ class Plan(PydanticModel, frozen=True):
 
     deployability_index: DeployabilityIndex
     restatements: t.Dict[SnapshotId, Interval]
-    restatement_triggers: t.Dict[SnapshotId, t.List[SnapshotId]] = {}
     start_override_per_model: t.Optional[t.Dict[str, datetime]]
     end_override_per_model: t.Optional[t.Dict[str, datetime]]
 
@@ -257,7 +256,6 @@ def to_evaluatable(self) -> EvaluatablePlan:
             skip_backfill=self.skip_backfill,
             empty_backfill=self.empty_backfill,
             restatements={s.name: i for s, i in self.restatements.items()},
-            restatement_triggers=self.restatement_triggers,
             is_dev=self.is_dev,
             allow_destructive_models=self.allow_destructive_models,
             forward_only=self.forward_only,
@@ -300,7 +298,6 @@ class EvaluatablePlan(PydanticModel):
     skip_backfill: bool
     empty_backfill: bool
     restatements: t.Dict[str, Interval]
-    restatement_triggers: t.Dict[SnapshotId, t.List[SnapshotId]] = {}
     is_dev: bool
     allow_destructive_models: t.Set[str]
     forward_only: bool

sqlmesh/core/plan/evaluator.py

@@ -37,7 +37,6 @@
     SnapshotCreationFailedError,
     SnapshotNameVersion,
 )
-from sqlmesh.core.snapshot.definition import SnapshotEvaluationTriggers
 from sqlmesh.utils import to_snake_case
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.utils import CorrelationId
@@ -245,27 +244,6 @@ def visit_backfill_stage(self, stage: stages.BackfillStage, plan: EvaluatablePla
             self.console.log_success("SKIP: No model batches to execute")
             return
 
-        directly_modified_triggers: t.Dict[SnapshotId, t.List[SnapshotId]] = {}
-        for parent, children in plan.indirectly_modified_snapshots.items():
-            parent_id = stage.all_snapshots[parent].snapshot_id
-            directly_modified_triggers[parent_id] = directly_modified_triggers.get(
-                parent_id, []
-            ) + [parent_id]
-            for child in children:
-                directly_modified_triggers[child] = directly_modified_triggers.get(child, []) + [
-                    parent_id
-                ]
-        directly_modified_triggers = {
-            k: list(dict.fromkeys(v)) for k, v in directly_modified_triggers.items()
-        }
-        snapshot_evaluation_triggers = {
-            s_id: SnapshotEvaluationTriggers(
-                directly_modified_triggers=directly_modified_triggers.get(s_id, []),
-                restatement_triggers=plan.restatement_triggers.get(s_id, []),
-            )
-            for s_id in [s.snapshot_id for s in stage.all_snapshots.values()]
-        }
-
         scheduler = self.create_scheduler(stage.all_snapshots.values(), self.snapshot_evaluator)
         errors, _ = scheduler.run_merged_intervals(
             merged_intervals=stage.snapshot_to_intervals,

sqlmesh/core/snapshot/definition.py

@@ -332,8 +332,6 @@ class SnapshotEvaluationTriggers(PydanticModel):
     cron_ready: t.Optional[bool] = None
     auto_restatement_triggers: t.List[SnapshotId] = []
     select_snapshot_triggers: t.List[SnapshotId] = []
-    directly_modified_triggers: t.List[SnapshotId] = []
-    restatement_triggers: t.List[SnapshotId] = []
 
 
 class SnapshotInfoMixin(ModelKindMixin):

tests/core/test_integration.py

@@ -26,7 +26,6 @@
 
 
 from sqlmesh import CustomMaterialization
-import sqlmesh
 from sqlmesh.cli.project_init import init_example_project
 from sqlmesh.core import constants as c
 from sqlmesh.core import dialect as d
@@ -1868,27 +1867,6 @@ def test_snapshot_triggers(init_and_plan_context: t.Callable, mocker: MockerFixt
     context, plan = init_and_plan_context("examples/sushi")
     context.apply(plan)
 
-    # modify 3 models
-    # - 2 breaking changes for testing plan directly modified triggers
-    # - 1 adding an auto-restatement for subsequent `run` test
-    marketing = context.get_model("sushi.marketing")
-    marketing_kwargs = {
-        **marketing.dict(),
-        "query": d.parse_one(
-            f"{marketing.query.sql(dialect='duckdb')} ORDER BY customer_id", dialect="duckdb"
-        ),
-    }
-    context.upsert_model(SqlModel.parse_obj(marketing_kwargs))
-
-    customers = context.get_model("sushi.customers")
-    customers_kwargs = {
-        **customers.dict(),
-        "query": d.parse_one(
-            f"{customers.query.sql(dialect='duckdb')} ORDER BY customer_id", dialect="duckdb"
-        ),
-    }
-    context.upsert_model(SqlModel.parse_obj(customers_kwargs))
-
     # add auto restatement to orders
     orders = context.get_model("sushi.orders")
     orders_kind = {
@@ -1901,67 +1879,8 @@ def test_snapshot_triggers(init_and_plan_context: t.Callable, mocker: MockerFixt
     }
     context.upsert_model(PythonModel.parse_obj(orders_kwargs))
 
-    spy = mocker.spy(sqlmesh.core.scheduler.Scheduler, "run_merged_intervals")
-
     context.plan(auto_apply=True, no_prompts=True, categorizer_config=CategorizerConfig.all_full())
 
-    # PLAN: directly modified triggers
-    actual_triggers = spy.call_args.kwargs["snapshot_evaluation_triggers"]
-    actual_triggers_name = {
-        k.name: sorted([s.name for s in v.directly_modified_triggers])
-        for k, v in actual_triggers.items()
-        if v.directly_modified_triggers
-    }
-    marketing_name = '"memory"."sushi"."marketing"'
-    customers_name = '"memory"."sushi"."customers"'
-    marketing_customers_names = sorted([marketing_name, customers_name])
-    children_names = [
-        f'"memory"."sushi"."{model}"'
-        for model in {
-            "waiter_as_customer_by_day",
-            "active_customers",
-            "count_customers_active",
-            "count_customers_inactive",
-        }
-    ]
-    assert actual_triggers_name == {
-        marketing_name: [marketing_name],
-        customers_name: [customers_name],
-        **{k: marketing_customers_names for k in children_names},
-    }
-
-    # PLAN: restatement triggers
-    spy.reset_mock()
-    context.plan(
-        restate_models=[
-            '"memory"."sushi"."marketing"',
-            '"memory"."sushi"."order_items"',
-            '"memory"."sushi"."waiter_revenue_by_day"',
-        ],
-        auto_apply=True,
-        no_prompts=True,
-    )
-
-    order_items_name = '"memory"."sushi"."order_items"'
-    waiter_revenue_by_day_name = '"memory"."sushi"."waiter_revenue_by_day"'
-    actual_triggers = spy.call_args.kwargs["snapshot_evaluation_triggers"]
-    actual_triggers_name = {
-        k.name: sorted([s.name for s in v.restatement_triggers])
-        for k, v in actual_triggers.items()
-        if v.restatement_triggers
-    }
-
-    assert sorted(actual_triggers_name[waiter_revenue_by_day_name]) == sorted(
-        [waiter_revenue_by_day_name, order_items_name]
-    )
-    assert actual_triggers_name[order_items_name] == [order_items_name]
-    assert actual_triggers_name['"memory"."sushi"."top_waiters"'] == [waiter_revenue_by_day_name]
-    assert actual_triggers_name['"memory"."sushi"."customer_revenue_by_day"'] == [order_items_name]
-    assert actual_triggers_name['"memory"."sushi"."customer_revenue_lifetime"'] == [
-        order_items_name
-    ]
-
-    # RUN: select and auto-restatement triggers
     # User selects top_waiters and waiter_revenue_by_day, others added as auto-upstream
     selected_models = {"top_waiters", "waiter_revenue_by_day"}
     selected_models_auto_upstream = {"order_items", "orders", "items"}