Fix: Add ignored snapshots for invalid start date (#1324)

* add ignored snapshots

* add tests

* cleanup

* cleanup

* add refresh dag to setting end

* feedback

Author: eakmanrq

examples/sushi/models/items.py

@@ -50,7 +50,7 @@
 @model(
     "sushi.items",
     kind=IncrementalByTimeRangeKind(time_column="ds", batch_size=30),
-    start="3 months ago",
+    start="1 week ago",
     cron="@daily",
     columns={
         "id": "int",

examples/sushi/models/orders.py

@@ -17,7 +17,7 @@
     "sushi.orders",
     description="Table of sushi orders.",
     kind=IncrementalByTimeRangeKind(time_column="ds", batch_size=30),
-    start="3 months ago",
+    start="1 week ago",
     cron="@daily",
     grains=[
         "id AS order_id",

sqlmesh/core/console.py

@@ -24,7 +24,7 @@
 from rich.tree import Tree
 
 from sqlmesh.core.environment import EnvironmentNamingInfo
-from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory
+from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory, start_date
 from sqlmesh.core.test import ModelTest
 from sqlmesh.utils import rich as srich
 from sqlmesh.utils.date import to_date, yesterday_ds
@@ -111,7 +111,10 @@ def stop_migration_progress(self, success: bool = True) -> None:
 
     @abc.abstractmethod
     def show_model_difference_summary(
-        self, context_diff: ContextDiff, detailed: bool = False
+        self,
+        context_diff: ContextDiff,
+        detailed: bool = False,
+        ignored_snapshot_names: t.Optional[t.Set[str]] = None,
     ) -> None:
         """Displays a summary of differences for the given models."""
 
@@ -386,14 +389,19 @@ def stop_migration_progress(self, success: bool = True) -> None:
                 self.log_success("The migration has been completed successfully")
 
     def show_model_difference_summary(
-        self, context_diff: ContextDiff, detailed: bool = False
+        self,
+        context_diff: ContextDiff,
+        detailed: bool = False,
+        ignored_snapshot_names: t.Optional[t.Set[str]] = None,
     ) -> None:
         """Shows a summary of the differences.
 
         Args:
             context_diff: The context diff to use to print the summary
             detailed: Show the actual SQL differences if True.
+            ignored_snapshot_names: A set of snapshot names that are ignored
         """
+        ignored_snapshot_names = ignored_snapshot_names or set()
         if context_diff.is_new_environment:
             self._print(
                 Tree(
@@ -408,40 +416,44 @@ def show_model_difference_summary(
             return
 
         tree = Tree(f"[bold]Summary of differences against `{context_diff.environment}`:")
-
-        if context_diff.added:
+        added_model_names = context_diff.added - ignored_snapshot_names
+        if added_model_names:
             added_tree = Tree(f"[bold][added]Added Models:")
-            for model in context_diff.added:
-                added_tree.add(f"[added]{model}")
+            for model_name in added_model_names:
+                added_tree.add(f"[added]{model_name}")
             tree.add(added_tree)
 
-        if context_diff.removed:
+        removed_model_names = context_diff.removed - ignored_snapshot_names
+        if removed_model_names:
             removed_tree = Tree(f"[bold][removed]Removed Models:")
-            for model in context_diff.removed:
-                removed_tree.add(f"[removed]{model}")
+            for model_name in removed_model_names:
+                removed_tree.add(f"[removed]{model_name}")
             tree.add(removed_tree)
 
-        if context_diff.modified_snapshots:
+        modified_model_names = context_diff.modified_snapshots.keys() - ignored_snapshot_names
+        if modified_model_names:
             direct = Tree(f"[bold][direct]Directly Modified:")
             indirect = Tree(f"[bold][indirect]Indirectly Modified:")
             metadata = Tree(f"[bold][metadata]Metadata Updated:")
-            for model in context_diff.modified_snapshots:
-                if context_diff.directly_modified(model):
+            for model_name in modified_model_names:
+                if context_diff.directly_modified(model_name):
                     direct.add(
-                        Syntax(f"{model}\n{context_diff.text_diff(model)}", "sql")
+                        Syntax(f"{model_name}\n{context_diff.text_diff(model_name)}", "sql")
                         if detailed
-                        else f"[direct]{model}"
+                        else f"[direct]{model_name}"
                     )
-                elif context_diff.indirectly_modified(model):
-                    indirect.add(f"[indirect]{model}")
-                elif context_diff.metadata_updated(model):
-                    metadata.add(f"[metadata]{model}")
+                elif context_diff.indirectly_modified(model_name):
+                    indirect.add(f"[indirect]{model_name}")
+                elif context_diff.metadata_updated(model_name):
+                    metadata.add(f"[metadata]{model_name}")
             if direct.children:
                 tree.add(direct)
             if indirect.children:
                 tree.add(indirect)
             if metadata.children:
                 tree.add(metadata)
+        if ignored_snapshot_names:
+            tree.add(self._get_ignored_tree(ignored_snapshot_names, context_diff.snapshots))
         self._print(tree)
 
     def plan(self, plan: Plan, auto_apply: bool) -> None:
@@ -460,22 +472,35 @@ def plan(self, plan: Plan, auto_apply: bool) -> None:
         if auto_apply:
             plan.apply()
 
+    def _get_ignored_tree(
+        self, ignored_snapshot_names: t.Set[str], snapshots: t.Dict[str, Snapshot]
+    ) -> Tree:
+        ignored = Tree(f"[bold][ignored]Ignored Models (Expected Plan Start):")
+        for model in ignored_snapshot_names:
+            snapshot = snapshots[model]
+            ignored.add(
+                f"[ignored]{model} ({snapshot.get_latest(start_date(snapshot, snapshots.values()))})"
+            )
+        return ignored
+
     def _show_options_after_categorization(self, plan: Plan, auto_apply: bool) -> None:
         if plan.forward_only and plan.new_snapshots:
             self._prompt_effective_from(plan, auto_apply)
 
         if plan.requires_backfill:
             self._show_missing_dates(plan)
             self._prompt_backfill(plan, auto_apply)
-        elif plan.context_diff.has_changes and not auto_apply:
+        elif plan.has_changes and not auto_apply:
             self._prompt_promote(plan)
         elif plan.has_unmodified_unpromoted and not auto_apply:
             self.log_status_update("\n[bold]Virtually updating unmodified models\n")
             self._prompt_promote(plan)
 
     def _prompt_categorize(self, plan: Plan, auto_apply: bool) -> None:
         """Get the user's change category for the directly modified models."""
-        self.show_model_difference_summary(plan.context_diff)
+        self.show_model_difference_summary(
+            plan.context_diff, ignored_snapshot_names=plan.ignored_snapshot_names
+        )
 
         self._show_categorized_snapshots(plan)
 
@@ -565,7 +590,10 @@ def _prompt_backfill(self, plan: Plan, auto_apply: bool) -> None:
                 )
                 if end:
                     plan.end = end
-
+        if plan.ignored_snapshot_names:
+            self._print(
+                self._get_ignored_tree(plan.ignored_snapshot_names, plan.context_diff.snapshots)
+            )
         if not auto_apply and self._confirm(f"Apply - {backfill_or_preview.capitalize()} Tables"):
             plan.apply()
 
@@ -1040,14 +1068,19 @@ class MarkdownConsole(CaptureTerminalConsole):
     """
 
     def show_model_difference_summary(
-        self, context_diff: ContextDiff, detailed: bool = False
+        self,
+        context_diff: ContextDiff,
+        detailed: bool = False,
+        ignored_snapshot_names: t.Optional[t.Set[str]] = None,
     ) -> None:
         """Shows a summary of the differences.
 
         Args:
             context_diff: The context diff to use to print the summary.
             detailed: Show the actual SQL differences if True.
+            ignored_snapshot_names: A set of snapshot names that are ignored
         """
+        ignored_snapshot_names = ignored_snapshot_names or set()
         if context_diff.is_new_environment:
             self._print(
                 f"**New environment `{context_diff.environment}` will be created from `{context_diff.create_from}`**\n\n"
@@ -1061,46 +1094,57 @@ def show_model_difference_summary(
 
         self._print(f"**Summary of differences against `{context_diff.environment}`:**\n\n")
 
-        if context_diff.added:
+        added_model_names = context_diff.added - ignored_snapshot_names
+        if added_model_names:
             self._print(f"**Added Models:**\n")
-            for model in context_diff.added:
-                self._print(f"- {model}\n")
+            for model_name in added_model_names:
+                self._print(f"- {model_name}\n")
             self._print("\n")
 
-        if context_diff.removed:
+        removed_model_names = context_diff.removed - ignored_snapshot_names
+        if removed_model_names:
             self._print(f"**Removed Models:**\n")
-            for model in context_diff.removed:
-                self._print(f"- {model}\n")
+            for model_name in removed_model_names:
+                self._print(f"- {model_name}\n")
             self._print("\n")
 
-        if context_diff.modified_snapshots:
+        modified_model_names = context_diff.modified_snapshots.keys() - ignored_snapshot_names
+        if modified_model_names:
             directly_modified = []
             indirectly_modified = []
             metadata_modified = []
-            for model in context_diff.modified_snapshots:
-                if context_diff.directly_modified(model):
-                    directly_modified.append(model)
-                elif context_diff.indirectly_modified(model):
-                    indirectly_modified.append(model)
-                elif context_diff.metadata_updated(model):
-                    metadata_modified.append(model)
+            for model_name in modified_model_names:
+                if context_diff.directly_modified(model_name):
+                    directly_modified.append(model_name)
+                elif context_diff.indirectly_modified(model_name):
+                    indirectly_modified.append(model_name)
+                elif context_diff.metadata_updated(model_name):
+                    metadata_modified.append(model_name)
             if directly_modified:
                 self._print(f"**Directly Modified:**\n")
-                for model in directly_modified:
-                    self._print(f"- `{model}`\n")
+                for model_name in directly_modified:
+                    self._print(f"- `{model_name}`\n")
                     if detailed:
-                        self._print(f"```diff\n{context_diff.text_diff(model)}\n```\n")
+                        self._print(f"```diff\n{context_diff.text_diff(model_name)}\n```\n")
                 self._print("\n")
             if indirectly_modified:
                 self._print(f"**Indirectly Modified:**\n")
-                for model in indirectly_modified:
-                    self._print(f"- `{model}`\n")
+                for model_name in indirectly_modified:
+                    self._print(f"- `{model_name}`\n")
                 self._print("\n")
             if metadata_modified:
                 self._print(f"**Metadata Updated:**\n")
-                for model in metadata_modified:
-                    self._print(f"- `{model}`\n")
+                for model_name in metadata_modified:
+                    self._print(f"- `{model_name}`\n")
                 self._print("\n")
+        if ignored_snapshot_names:
+            self._print(f"**Ignored Models (Expected Plan Start):**\n")
+            for model_name in ignored_snapshot_names:
+                snapshot = context_diff.snapshots[model_name]
+                self._print(
+                    f"- `{model_name}` ({snapshot.get_latest(start_date(snapshot, context_diff.snapshots.values()))})\n"
+                )
+            self._print("\n")
 
     def _show_missing_dates(self, plan: Plan) -> None:
         """Displays the models with missing dates."""

sqlmesh/core/node.py

@@ -48,6 +48,26 @@ def from_cron(klass, cron: str, sample_size: int = 10) -> IntervalUnit:
     def is_date_granularity(self) -> bool:
         return self in (IntervalUnit.YEAR, IntervalUnit.MONTH, IntervalUnit.DAY)
 
+    @property
+    def is_year(self) -> bool:
+        return self == IntervalUnit.YEAR
+
+    @property
+    def is_month(self) -> bool:
+        return self == IntervalUnit.MONTH
+
+    @property
+    def is_day(self) -> bool:
+        return self == IntervalUnit.DAY
+
+    @property
+    def is_hour(self) -> bool:
+        return self == IntervalUnit.HOUR
+
+    @property
+    def is_minute(self) -> bool:
+        return self == IntervalUnit.MINUTE
+
     @property
     def _cron_expr(self) -> str:
         if self == IntervalUnit.MINUTE: