Feat: Allow specifying a minimum number of missing intervals to check

Author: erindru

examples/multi/repo_1/linter/__init__.py

@@ -513,6 +513,11 @@ def diff(ctx: click.Context, environment: t.Optional[str] = None) -> None:
     help="Explain the plan instead of applying it.",
     default=None,
 )
+@click.option(
+    "--min-intervals",
+    default=0,
+    help="For every model, ensure at least this many intervals are covered by a missing intervals check regardless of the plan start date",
+)
 @opt.verbose
 @click.pass_context
 @error_handler

sqlmesh/cli/main.py

@@ -1215,6 +1215,7 @@ def plan(
         diff_rendered: t.Optional[bool] = None,
         skip_linter: t.Optional[bool] = None,
         explain: t.Optional[bool] = None,
+        min_intervals: t.Optional[int] = None,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1261,6 +1262,8 @@ def plan(
             diff_rendered: Whether the diff should compare raw vs rendered models
             skip_linter: Linter runs by default so this will skip it if enabled
             explain: Whether to explain the plan instead of applying it.
+            min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
+                on every model when checking for missing intervals
 
         Returns:
             The populated Plan object.
@@ -1289,6 +1292,7 @@ def plan(
             diff_rendered=diff_rendered,
             skip_linter=skip_linter,
             explain=explain,
+            min_intervals=min_intervals,
         )
 
         plan = plan_builder.build()
@@ -1338,6 +1342,7 @@ def plan_builder(
         diff_rendered: t.Optional[bool] = None,
         skip_linter: t.Optional[bool] = None,
         explain: t.Optional[bool] = None,
+        min_intervals: t.Optional[int] = None,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1374,6 +1379,8 @@ def plan_builder(
             enable_preview: Indicates whether to enable preview for forward-only models in development environments.
             run: Whether to run latest intervals as part of the plan application.
             diff_rendered: Whether the diff should compare raw vs rendered models
+            min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
+                on every model when checking for missing intervals
 
         Returns:
             The plan builder.
@@ -1401,6 +1408,7 @@ def plan_builder(
             "run": run,
             "diff_rendered": diff_rendered,
             "skip_linter": skip_linter,
+            "min_intervals": min_intervals,
         }
         user_provided_flags: t.Dict[str, UserProvidedFlags] = {
             k: v for k, v in kwargs.items() if v is not None
@@ -1557,6 +1565,7 @@ def plan_builder(
             console=self.console,
             user_provided_flags=user_provided_flags,
             explain=explain or False,
+            min_intervals=min_intervals,
         )
 
     def apply(

sqlmesh/core/context.py

@@ -31,6 +31,9 @@ class IntervalUnit(str, Enum):
     IntervalUnit can be one of 5 types, YEAR, MONTH, DAY, HOUR, MINUTE. The unit is inferred
     based on the cron schedule of a node. The minimum time delta between a sample set of dates
     is used to determine which unit a node's schedule is.
+
+    It's designed to align with common partitioning schemes, hence why there is no WEEK unit
+    because generally tables are not partitioned by week
     """
 
     YEAR = "year"

sqlmesh/core/node.py

@@ -87,6 +87,8 @@ class PlanBuilder:
             the environment is not finalized.
         interval_end_per_model: The mapping from model FQNs to target end dates.
         explain: Whether to explain the plan instead of applying it.
+        min_intervals: Adjust the plan start date on a per-model basis in order to ensure at least this many intervals are covered
+                on every model when checking for missing intervals
     """
 
     def __init__(
@@ -120,6 +122,7 @@ def __init__(
         interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
         console: t.Optional[PlanBuilderConsole] = None,
         user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None,
+        min_intervals: t.Optional[int] = None,
     ):
         self._context_diff = context_diff
         self._no_gaps = no_gaps
@@ -176,6 +179,7 @@ def __init__(
         )
 
         self._latest_plan: t.Optional[Plan] = None
+        self._min_intervals = min_intervals
 
     @property
     def is_start_and_end_allowed(self) -> bool:
@@ -330,6 +334,7 @@ def build(self) -> Plan:
             end_bounded=self._end_bounded,
             ensure_finalized_snapshots=self._ensure_finalized_snapshots,
             user_provided_flags=self._user_provided_flags,
+            min_intervals=self._min_intervals,
         )
         self._latest_plan = plan
         return plan

sqlmesh/core/plan/builder.py

@@ -58,6 +58,7 @@ class Plan(PydanticModel, frozen=True):
     deployability_index: DeployabilityIndex
     restatements: t.Dict[SnapshotId, Interval]
     interval_end_per_model: t.Optional[t.Dict[str, int]]
+    min_intervals: t.Optional[int] = None
 
     selected_models_to_backfill: t.Optional[t.Set[str]] = None
     """Models that have been explicitly selected for backfill by a user."""
@@ -179,6 +180,7 @@ def missing_intervals(self) -> t.List[SnapshotIntervals]:
                 deployability_index=self.deployability_index,
                 interval_end_per_model=self.interval_end_per_model,
                 end_bounded=self.end_bounded,
+                min_intervals=self.min_intervals,
             ).items()
             if snapshot.is_model and missing
         ]

sqlmesh/core/plan/definition.py

@@ -1790,6 +1790,7 @@ def missing_intervals(
     interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
     ignore_cron: bool = False,
     end_bounded: bool = False,
+    min_intervals: t.Optional[int] = None,
 ) -> t.Dict[Snapshot, Intervals]:
     """Returns all missing intervals given a collection of snapshots."""
     if not isinstance(snapshots, dict):
@@ -1798,11 +1799,12 @@ def missing_intervals(
     missing = {}
     cache: t.Dict[str, datetime] = {}
     end_date = end or now_timestamp()
-    start_dt = (
-        to_datetime(start)
-        if start
-        else earliest_start_date(snapshots, cache=cache, relative_to=end_date)
+    execution_time_dt = (
+        to_datetime(execution_time) if execution_time is not None else execution_time
     )
+    fallback_start_dt = earliest_start_date(snapshots, cache=cache, relative_to=end_date)
+    start_dt = to_datetime(start, relative_base=execution_time_dt) if start else fallback_start_dt
+    end_date = end or execution_time_dt or now_timestamp()
     restatements = restatements or {}
     interval_end_per_model = interval_end_per_model or {}
     deployability_index = deployability_index or DeployabilityIndex.all_deployable()
@@ -1833,10 +1835,24 @@ def missing_intervals(
         if node_end_date and (to_datetime(node_end_date) < to_datetime(snapshot_end_date)):
             missing_interval_end_date = node_end_date
 
+        missing_interval_start_date = snapshot_start_date
+        if min_intervals:
+            # we need to ensure that the time range submitted to snapshot.missing_intervals covers at least :min_intervals intervals.
+            # To do that, we can wind back the model cron :min_interval intervals and check if the start date we were going to send
+            # is early enough to cover those intervals. If it isnt, we widen it, otherwise we keep it as-is
+            min_interval_start = snapshot.node.cron_floor(missing_interval_end_date)
+            for _ in range(min_intervals):
+                min_interval_start = snapshot.node.cron_prev(min_interval_start)
+
+            if min_interval_start < snapshot_start_date:
+                missing_interval_start_date = min_interval_start
+
+        inferred_start_date = start_date(snapshot, snapshots, cache, relative_to=snapshot_end_date)
+
         intervals = snapshot.missing_intervals(
             max(
-                to_datetime(snapshot_start_date),
-                to_datetime(start_date(snapshot, snapshots, cache, relative_to=snapshot_end_date)),
+                missing_interval_start_date,
+                inferred_start_date,
             ),
             missing_interval_end_date,
             execution_time=execution_time,
@@ -1889,7 +1905,7 @@ def compute_missing_intervals(
     Returns:
         A list of all timestamps in this range.
     """
-    if start_ts == end_ts:
+    if start_ts >= end_ts:
         return []
 
     timestamps = expand_range(start_ts, end_ts, interval_unit)

sqlmesh/core/snapshot/definition.py

@@ -2,7 +2,7 @@
 import pathlib
 import typing as t
 import re
-from datetime import date, timedelta
+from datetime import date, timedelta, datetime
 from tempfile import TemporaryDirectory
 from unittest.mock import PropertyMock, call, patch
 
@@ -36,6 +36,7 @@
 from sqlmesh.core.dialect import parse, schema_
 from sqlmesh.core.engine_adapter.duckdb import DuckDBEngineAdapter
 from sqlmesh.core.environment import Environment, EnvironmentNamingInfo, EnvironmentStatements
+from sqlmesh.core.plan.definition import Plan
 from sqlmesh.core.macros import MacroEvaluator, RuntimeStage
 from sqlmesh.core.model import load_sql_based_model, model, SqlModel, Model
 from sqlmesh.core.model.cache import OptimizedQueryCache
@@ -2094,6 +2095,7 @@ def test_plan_audit_intervals(tmp_path: pathlib.Path, caplog):
     plan = ctx.plan(
         environment="dev", auto_apply=True, no_prompts=True, start="2025-02-01", end="2025-02-01"
     )
+    assert plan.missing_intervals
 
     date_snapshot = next(s for s in plan.new_snapshots if "date_example" in s.name)
     timestamp_snapshot = next(s for s in plan.new_snapshots if "timestamp_example" in s.name)
@@ -2304,3 +2306,114 @@ def test_dev_environment_virtual_update_with_environment_statements(tmp_path: Pa
         updated_statements[0].before_all[1]
         == "CREATE TABLE IF NOT EXISTS metrics (metric_name VARCHAR(50), value INT)"
     )
+
+
+def test_plan_min_intervals(tmp_path: Path):
+    init_example_project(tmp_path, engine_type="duckdb", dialect="duckdb")
+
+    context = Context(
+        paths=tmp_path, config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))
+    )
+
+    current_time = to_datetime("2020-02-01 00:00:01")
+
+    # initial state of example project
+    context.plan(auto_apply=True, execution_time=current_time)
+
+    (tmp_path / "models" / "daily_model.sql").write_text("""
+    MODEL (
+      name sqlmesh_example.daily_model,
+      kind INCREMENTAL_BY_TIME_RANGE (
+        time_column event_date
+      ),
+      start '2020-01-01',
+      cron '@daily'
+    );                        
+
+    select * from sqlmesh_example.incremental_model where event_date between @start_ds and @end_ds                            
+    """)
+
+    (tmp_path / "models" / "weekly_model.sql").write_text("""
+    MODEL (
+      name sqlmesh_example.weekly_model,
+      kind INCREMENTAL_BY_TIME_RANGE (
+        time_column event_date
+      ),
+      start '2020-01-01',
+      cron '@weekly'
+    );                        
+
+    select * from sqlmesh_example.incremental_model where event_date between @start_ds and @end_ds                            
+    """)
+
+    (tmp_path / "models" / "monthly_model.sql").write_text("""
+    MODEL (
+      name sqlmesh_example.monthly_model,
+      kind INCREMENTAL_BY_TIME_RANGE (
+        time_column event_date
+      ),
+      start '2020-01-01',
+      cron '@monthly'
+    );                        
+
+    select * from sqlmesh_example.incremental_model where event_date between @start_ds and @end_ds                            
+    """)
+
+    context.load()
+
+    # initial state - backfill from 2020-01-01 -> now() (2020-01-02 00:00:01) on new models
+    plan = context.plan(execution_time=current_time)
+
+    assert to_datetime(plan.start) == to_datetime("2020-01-01 00:00:00")
+    assert to_datetime(plan.end) == to_datetime("2020-02-01 00:00:00")
+    assert to_datetime(plan.execution_time) == to_datetime("2020-02-01 00:00:01")
+
+    def _get_missing_intervals(plan: Plan, name: str) -> t.List[t.Tuple[datetime, datetime]]:
+        snapshot_id = context.get_snapshot(name, raise_if_missing=True).snapshot_id
+        snapshot_intervals = next(
+            si for si in plan.missing_intervals if si.snapshot_id == snapshot_id
+        )
+        return [(to_datetime(s), to_datetime(e)) for s, e in snapshot_intervals.merged_intervals]
+
+    # check initial intervals - should be full time range between start and execution time
+    assert len(plan.missing_intervals) == 3
+
+    assert _get_missing_intervals(plan, "sqlmesh_example.daily_model") == [
+        (to_datetime("2020-01-01 00:00:00"), to_datetime("2020-02-01 00:00:00"))
+    ]
+    assert _get_missing_intervals(plan, "sqlmesh_example.weekly_model") == [
+        (
+            to_datetime("2020-01-01 00:00:00"),
+            to_datetime("2020-01-26 00:00:00"),
+        )  # last week in 2020-01 hasnt fully elapsed yet
+    ]
+    assert _get_missing_intervals(plan, "sqlmesh_example.monthly_model") == [
+        (to_datetime("2020-01-01 00:00:00"), to_datetime("2020-02-01 00:00:00"))
+    ]
+
+    # now, create a dev env for "1 day ago"
+    plan = context.plan(
+        environment="pr_env",
+        start="1 day ago",
+        execution_time=current_time,
+        min_intervals=1,
+    )
+
+    # this should pick up last day for daily model, last week for weekly model and last month for the monthly model
+    assert len(plan.missing_intervals) == 3
+
+    assert _get_missing_intervals(plan, "sqlmesh_example.daily_model") == [
+        (to_datetime("2020-01-31 00:00:00"), to_datetime("2020-02-01 00:00:00"))
+    ]
+    assert _get_missing_intervals(plan, "sqlmesh_example.weekly_model") == [
+        (
+            to_datetime("2020-01-19 00:00:00"),  # last completed week
+            to_datetime("2020-01-26 00:00:00"),
+        )
+    ]
+    assert _get_missing_intervals(plan, "sqlmesh_example.monthly_model") == [
+        (
+            to_datetime("2020-01-01 00:00:00"),  # last completed month
+            to_datetime("2020-02-01 00:00:00"),
+        )
+    ]

tests/core/test_context.py

@@ -3289,3 +3289,34 @@ def test_environment_statements_change_allows_dev_environment_creation(make_snap
     assert plan is not None
     assert plan.context_diff.has_environment_statements_changes
     assert plan.context_diff.environment_statements == environment_statements
+
+
+def test_min_intervals(make_snapshot):
+    snapshot_a = make_snapshot(
+        SqlModel(name="a", query=parse_one("select 1, ds"), dialect="duckdb")
+    )
+
+    context_diff = ContextDiff(
+        environment="test_environment",
+        is_new_environment=True,
+        is_unfinalized_environment=False,
+        normalize_environment_name=True,
+        create_from="prod",
+        create_from_env_exists=True,
+        added={snapshot_a.snapshot_id},
+        removed_snapshots={},
+        modified_snapshots={},
+        snapshots={},
+        new_snapshots={snapshot_a.snapshot_id: snapshot_a},
+        previous_plan_id=None,
+        previously_promoted_snapshot_ids=set(),
+        previous_finalized_snapshots=None,
+        previous_gateway_managed_virtual_layer=False,
+        gateway_managed_virtual_layer=False,
+    )
+
+    prod_plan = PlanBuilder(context_diff, min_intervals=1).build()
+    dev_plan = PlanBuilder(context_diff, start="1 day ago", is_dev=True, min_intervals=1).build()
+
+    assert prod_plan.min_intervals == 1
+    assert dev_plan.min_intervals == 1