Feat: Allow specifying a minimum number of intervals to include for dev plans with a relative start date

Author: erindru

examples/multi/repo_1/linter/__init__.py

@@ -470,6 +470,12 @@ def diff(ctx: click.Context, environment: t.Optional[str] = None) -> None:
     help="Explain the plan instead of applying it.",
     default=None,
 )
+@click.option(
+    "--relative-start-min-intervals",
+    default=0,
+    help="When creating a new development environment with a relative start date, eg '1 week ago', backfill at least this many intervals for each model. "
+    "This ensures that models with an interval unit larger than the relative time period still get included in the backfill.",
+)
 @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,
+        relative_start_min_intervals: t.Optional[int] = None,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1261,6 +1262,9 @@ 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.
+            relative_start_min_intervals: If the plan start is relative, eg `1 day ago` then if
+                no intervals are matched for the model, wind back the start on a per-model basis
+                until we get at least this many missing intervals
 
         Returns:
             The populated Plan object.
@@ -1289,6 +1293,7 @@ def plan(
             diff_rendered=diff_rendered,
             skip_linter=skip_linter,
             explain=explain,
+            relative_start_min_intervals=relative_start_min_intervals,
         )
 
         plan = plan_builder.build()
@@ -1338,6 +1343,7 @@ def plan_builder(
         diff_rendered: t.Optional[bool] = None,
         skip_linter: t.Optional[bool] = None,
         explain: t.Optional[bool] = None,
+        relative_start_min_intervals: t.Optional[int] = None,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1374,6 +1380,9 @@ 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
+            relative_start_min_intervals: If the plan start is relative, eg `1 day ago` then if
+                no intervals are matched for the model, wind back the start on a per-model basis
+                until we get at least this many missing intervals
 
         Returns:
             The plan builder.
@@ -1401,6 +1410,7 @@ def plan_builder(
             "run": run,
             "diff_rendered": diff_rendered,
             "skip_linter": skip_linter,
+            "relative_start_min_intervals": relative_start_min_intervals,
         }
         user_provided_flags: t.Dict[str, UserProvidedFlags] = {
             k: v for k, v in kwargs.items() if v is not None
@@ -1557,6 +1567,7 @@ def plan_builder(
             console=self.console,
             user_provided_flags=user_provided_flags,
             explain=explain or False,
+            relative_start_min_intervals=relative_start_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,9 @@ 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.
+        relative_start_min_intervals: If the plan start is relative, eg `1 day ago` then if
+            no intervals are matched for the model, wind back the start on a per-model basis
+            until we get at least this many missing intervals
     """
 
     def __init__(
@@ -120,6 +123,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,
+        relative_start_min_intervals: t.Optional[int] = None,
     ):
         self._context_diff = context_diff
         self._no_gaps = no_gaps
@@ -176,6 +180,7 @@ def __init__(
         )
 
         self._latest_plan: t.Optional[Plan] = None
+        self._relative_start_min_intervals = relative_start_min_intervals
 
     @property
     def is_start_and_end_allowed(self) -> bool:
@@ -330,6 +335,7 @@ def build(self) -> Plan:
             end_bounded=self._end_bounded,
             ensure_finalized_snapshots=self._ensure_finalized_snapshots,
             user_provided_flags=self._user_provided_flags,
+            relative_start_min_intervals=self._relative_start_min_intervals,
         )
         self._latest_plan = plan
         return plan
@@ -788,6 +794,11 @@ def _ensure_valid_date_range(self) -> None:
                     f"Plan end date: '{time_like_to_str(end)}' cannot be in the future (execution time: '{time_like_to_str(self.execution_time)}')"
                 )
 
+        if (start := self._start) and self._relative_start_min_intervals and not is_relative(start):
+            raise PlanError(
+                f"Plan start date '{time_like_to_str(start)}' must be a relative date in order to use --relative-start-min-intervals"
+            )
+
     def _ensure_no_forward_only_revert(self) -> None:
         """Ensures that a previously superseded breaking / non-breaking snapshot is not being
         used again to replace an existing forward-only snapshot with the same version.

sqlmesh/core/plan/builder.py

@@ -68,6 +68,8 @@ class Plan(PydanticModel, frozen=True):
 
     user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None
 
+    relative_start_min_intervals: t.Optional[int] = None
+
     @cached_property
     def start(self) -> TimeLike:
         if self.provided_start is not None:
@@ -179,6 +181,10 @@ 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,
+                # note: by the time relative_start_min_intervals is used here
+                # the PlanBuilder has validated that the start time was relative
+                # (above, self.provided_start has been resolved to an absolute time already)
+                min_intervals=self.relative_start_min_intervals,
             ).items()
             if snapshot.is_model and missing
         ]

sqlmesh/core/plan/definition.py

@@ -1783,26 +1783,40 @@ 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):
         # Make sure that the mapping is only constructed once
         snapshots = {snapshot.snapshot_id: snapshot for snapshot in snapshots}
     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
     )
+    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()
 
+    fallback_start = earliest_start_date(snapshots, cache=cache, relative_to=end_date)
+
     for snapshot in snapshots.values():
         if not snapshot.evaluatable:
             continue
+
+        start_relative_base = execution_time_dt
+        if min_intervals and snapshot.node.end:
+            # this is an optimisation to reduce the number of calls to Snapshot.missing_intervals() if the model end date is in the past
+            # if we make the start date relative to the model end date instead of the plan end date, we can reach min_intervals missing intervals more quickly
+            node_end_dt = to_datetime(snapshot.node.end)
+            if to_datetime(end_date) > node_end_dt:
+                start_relative_base = node_end_dt
+
+        start_dt = (
+            to_datetime(start, relative_base=start_relative_base) if start else fallback_start
+        )
+
         snapshot_start_date = start_dt
         snapshot_end_date: TimeLike = end_date
 
@@ -1826,17 +1840,35 @@ 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
 
-        intervals = snapshot.missing_intervals(
-            max(
-                to_datetime(snapshot_start_date),
-                to_datetime(start_date(snapshot, snapshots, cache, relative_to=snapshot_end_date)),
-            ),
-            missing_interval_end_date,
-            execution_time=execution_time,
-            deployability_index=deployability_index,
-            ignore_cron=ignore_cron,
-            end_bounded=end_bounded,
-        )
+        inferred_start_date = start_date(snapshot, snapshots, cache, relative_to=snapshot_end_date)
+        missing_interval_start_date = snapshot_start_date
+
+        def _get_missing_intervals() -> Intervals:
+            return snapshot.missing_intervals(
+                max(missing_interval_start_date, inferred_start_date),
+                missing_interval_end_date,
+                execution_time=execution_time,
+                deployability_index=deployability_index,
+                ignore_cron=ignore_cron,
+                end_bounded=end_bounded,
+            )
+
+        while True:
+            intervals = _get_missing_intervals()
+
+            if (
+                min_intervals
+                and len(intervals) < min_intervals
+                and missing_interval_start_date > inferred_start_date
+            ):
+                # keep winding back the start date until either:
+                # - we get min_intervals intervals
+                # - we've wound back beyond the model start date which means there are no missing intervals
+                missing_interval_start_date = snapshot.node.cron_prev(missing_interval_start_date)
+                continue
+
+            break
+
         if intervals:
             missing[snapshot] = intervals
 
@@ -1882,7 +1914,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

@@ -2088,6 +2088,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)
@@ -2218,3 +2219,136 @@ def test_plan_explain_skips_tests(sushi_context: Context, mocker: MockerFixture)
     spy = mocker.spy(sushi_context, "_run_plan_tests")
     sushi_context.plan(environment="dev", explain=True, no_prompts=True, include_unmodified=True)
     spy.assert_called_once_with(skip_tests=True)
+
+
+def test_plan_relative_start_min_intervals(tmp_path: Path):
+    init_example_project(tmp_path, 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")
+
+    # check initial intervals - should be full time range between start and execution time
+    assert len(plan.missing_intervals) == 3
+    assert (
+        plan.missing_intervals[0].snapshot_id
+        == context.get_snapshot("sqlmesh_example.daily_model", raise_if_missing=True).snapshot_id
+    )
+    assert [
+        (to_datetime(s), to_datetime(e)) for s, e in plan.missing_intervals[0].merged_intervals
+    ] == [(to_datetime("2020-01-01 00:00:00"), to_datetime("2020-02-01 00:00:00"))]
+    assert (
+        plan.missing_intervals[1].snapshot_id
+        == context.get_snapshot("sqlmesh_example.monthly_model", raise_if_missing=True).snapshot_id
+    )
+    assert [
+        (to_datetime(s), to_datetime(e)) for s, e in plan.missing_intervals[1].merged_intervals
+    ] == [(to_datetime("2020-01-01 00:00:00"), to_datetime("2020-02-01 00:00:00"))]
+    assert (
+        plan.missing_intervals[2].snapshot_id
+        == context.get_snapshot("sqlmesh_example.weekly_model", raise_if_missing=True).snapshot_id
+    )
+    assert [
+        (to_datetime(s), to_datetime(e)) for s, e in plan.missing_intervals[2].merged_intervals
+    ] == [
+        (
+            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
+    ]
+
+    # now, create a dev env for "1 day ago"
+    plan = context.plan(
+        environment="pr_env",
+        start="1 day ago",
+        execution_time=current_time,
+        relative_start_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 (
+        plan.missing_intervals[0].snapshot_id
+        == context.get_snapshot("sqlmesh_example.daily_model", raise_if_missing=True).snapshot_id
+    )
+    assert [
+        (to_datetime(s), to_datetime(e)) for s, e in plan.missing_intervals[0].merged_intervals
+    ] == [(to_datetime("2020-01-31 00:00:00"), to_datetime("2020-02-01 00:00:00"))]
+    assert (
+        plan.missing_intervals[1].snapshot_id
+        == context.get_snapshot("sqlmesh_example.monthly_model", raise_if_missing=True).snapshot_id
+    )
+    assert [
+        (to_datetime(s), to_datetime(e)) for s, e in plan.missing_intervals[1].merged_intervals
+    ] == [
+        (
+            to_datetime("2020-01-01 00:00:00"),  # last completed month
+            to_datetime("2020-02-01 00:00:00"),
+        )
+    ]
+    assert (
+        plan.missing_intervals[2].snapshot_id
+        == context.get_snapshot("sqlmesh_example.weekly_model", raise_if_missing=True).snapshot_id
+    )
+    assert [
+        (to_datetime(s), to_datetime(e)) for s, e in plan.missing_intervals[2].merged_intervals
+    ] == [
+        (
+            to_datetime("2020-01-19 00:00:00"),  # last completed week
+            to_datetime("2020-01-26 00:00:00"),
+        )
+    ]