Fix: Make plan dates relative to --execution-time and not the current time

Author: erindru

sqlmesh/core/context.py

@@ -1513,7 +1513,11 @@ def plan_builder(
             # If no end date is specified, use the max interval end from prod
             # to prevent unintended evaluation of the entire DAG.
             default_start, default_end = self._get_plan_default_start_end(
-                snapshots, max_interval_end_per_model, backfill_models, modified_model_names
+                snapshots,
+                max_interval_end_per_model,
+                backfill_models,
+                modified_model_names,
+                execution_time,
             )
 
             # Refresh snapshot intervals to ensure that they are up to date with values reflected in the max_interval_end_per_model.
@@ -2818,6 +2822,7 @@ def _get_plan_default_start_end(
         max_interval_end_per_model: t.Dict[str, int],
         backfill_models: t.Optional[t.Set[str]],
         modified_model_names: t.Set[str],
+        execution_time: t.Optional[TimeLike] = None,
     ) -> t.Tuple[t.Optional[int], t.Optional[int]]:
         if not max_interval_end_per_model:
             return None, None
@@ -2843,6 +2848,12 @@ def _get_plan_default_start_end(
                     )
                 ),
             )
+
+        if execution_time and to_timestamp(default_end) > to_timestamp(execution_time):
+            # the end date can't be in the future, which can happen if a specific `execution_time` is set and prod intervals
+            # are newer than it
+            default_end = to_timestamp(execution_time)
+
         return default_start, default_end
 
     def _get_max_interval_end_per_model(

sqlmesh/core/plan/builder.py

@@ -41,6 +41,7 @@
     to_datetime,
     yesterday_ds,
     to_timestamp,
+    time_like_to_str,
 )
 from sqlmesh.utils.errors import NoChangesPlanError, PlanError
 
@@ -55,6 +56,7 @@ class PlanBuilder:
         start: The start time to backfill data.
         end: The end time to backfill data.
         execution_time: The date/time time reference to use for execution time. Defaults to now.
+            If :start or :end are relative time expressions, they are interpreted as relative to the :execution_time
         apply: The callback to apply the plan.
         restate_models: A list of models for which the data should be restated for the time range
             specified in this plan. Note: models defined outside SQLMesh (external) won't be a part
@@ -172,6 +174,22 @@ def is_start_and_end_allowed(self) -> bool:
         """Indicates whether this plan allows to set the start and end dates."""
         return self._is_dev or bool(self._restate_models)
 
+    @property
+    def start(self) -> t.Optional[TimeLike]:
+        if self._start and self._execution_time:
+            return to_datetime(self._start, relative_base=to_datetime(self._execution_time))
+        return self._start
+
+    @property
+    def end(self) -> t.Optional[TimeLike]:
+        if self._end and self._execution_time:
+            return to_datetime(self._end, relative_base=to_datetime(self._execution_time))
+        return self._end
+
+    @property
+    def execution_time(self) -> TimeLike:
+        return self._execution_time or now()
+
     def set_start(self, new_start: TimeLike) -> PlanBuilder:
         self._start = new_start
         self.override_start = True
@@ -256,7 +274,7 @@ def build(self) -> Plan:
         )
 
         restatements = self._build_restatements(
-            dag, earliest_interval_start(self._context_diff.snapshots.values())
+            dag, earliest_interval_start(self._context_diff.snapshots.values(), self.execution_time)
         )
         models_to_backfill = self._build_models_to_backfill(dag, restatements)
 
@@ -269,8 +287,8 @@ def build(self) -> Plan:
         plan = Plan(
             context_diff=self._context_diff,
             plan_id=self._plan_id,
-            provided_start=self._start,
-            provided_end=self._end,
+            provided_start=self.start,
+            provided_end=self.end,
             is_dev=self._is_dev,
             skip_backfill=self._skip_backfill,
             empty_backfill=self._empty_backfill,
@@ -289,7 +307,7 @@ def build(self) -> Plan:
             selected_models_to_backfill=self._backfill_models,
             models_to_backfill=models_to_backfill,
             effective_from=self._effective_from,
-            execution_time=self._execution_time,
+            execution_time=self.execution_time,
             end_bounded=self._end_bounded,
             ensure_finalized_snapshots=self._ensure_finalized_snapshots,
             user_provided_flags=self._user_provided_flags,
@@ -739,6 +757,18 @@ def _ensure_valid_date_range(self) -> None:
                 "The start and end dates can't be set for a production plan without restatements."
             )
 
+        if (start := self.start) and (end := self.end):
+            if to_datetime(start) > to_datetime(end):
+                raise PlanError(
+                    f"Plan end date: '{time_like_to_str(end)}' must be after the plan start date: '{time_like_to_str(start)}'"
+                )
+
+        if end := self.end:
+            if to_datetime(end) > to_datetime(self.execution_time):
+                raise PlanError(
+                    f"Plan end date: '{time_like_to_str(end)}' cannot be in the future (execution time: '{time_like_to_str(self.execution_time)}')"
+                )
+
     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/definition.py

@@ -5,6 +5,7 @@
 from datetime import datetime
 from enum import Enum
 from functools import cached_property
+from pydantic import Field, BeforeValidator
 
 from sqlmesh.core.context_diff import ContextDiff
 from sqlmesh.core.environment import Environment, EnvironmentNamingInfo, EnvironmentStatements
@@ -29,6 +30,7 @@
 
 SnapshotMapping = t.Dict[SnapshotId, t.Set[SnapshotId]]
 UserProvidedFlags = t.Union[TimeLike, str, bool, t.List[str]]
+ExecutionTimeValidator = BeforeValidator(lambda execution_time: execution_time or now())
 
 
 class Plan(PydanticModel, frozen=True):
@@ -63,7 +65,7 @@ class Plan(PydanticModel, frozen=True):
     models_to_backfill: t.Optional[t.Set[str]] = None
     """All models that should be backfilled as part of this plan."""
     effective_from: t.Optional[TimeLike] = None
-    execution_time: t.Optional[TimeLike] = None
+    execution_time_: t.Optional[TimeLike] = Field(default=None, alias="execution_time")
 
     user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None
 
@@ -80,7 +82,11 @@ def start(self) -> TimeLike:
 
     @cached_property
     def end(self) -> TimeLike:
-        return self.provided_end or now()
+        return self.provided_end or self.execution_time
+
+    @property
+    def execution_time(self) -> TimeLike:
+        return self.execution_time_ or now()
 
     @property
     def previous_plan_id(self) -> t.Optional[str]:
@@ -271,7 +277,7 @@ def to_evaluatable(self) -> EvaluatablePlan:
 
     @cached_property
     def _earliest_interval_start(self) -> datetime:
-        return earliest_interval_start(self.snapshots.values())
+        return earliest_interval_start(self.snapshots.values(), self.execution_time)
 
 
 class EvaluatablePlan(PydanticModel):
@@ -345,8 +351,10 @@ def format_intervals(self, unit: t.Optional[IntervalUnit] = None) -> str:
         return format_intervals(self.merged_intervals, unit)
 
 
-def earliest_interval_start(snapshots: t.Collection[Snapshot]) -> datetime:
-    earliest_start = earliest_start_date(snapshots)
+def earliest_interval_start(
+    snapshots: t.Collection[Snapshot], execution_time: t.Optional[TimeLike] = None
+) -> datetime:
+    earliest_start = earliest_start_date(snapshots, relative_to=execution_time)
     earliest_interval_starts = [s.intervals[0][0] for s in snapshots if s.intervals]
     return (
         min(earliest_start, to_datetime(min(earliest_interval_starts)))

sqlmesh/core/snapshot/definition.py

@@ -1995,7 +1995,12 @@ def earliest_start_date(
             start_date(snapshot, snapshots, cache=cache, relative_to=relative_to)
             for snapshot in snapshots.values()
         )
-    return yesterday()
+
+    relative_base = None
+    if relative_to is not None:
+        relative_base = to_datetime(relative_to)
+
+    return yesterday(relative_base=relative_base)
 
 
 def start_date(

sqlmesh/utils/date.py

@@ -87,34 +87,34 @@ def now_ds() -> str:
     return to_ds(now())
 
 
-def yesterday() -> datetime:
+def yesterday(relative_base: t.Optional[datetime] = None) -> datetime:
     """
     Yesterday utc datetime.
 
     Returns:
         A datetime object with tz utc representing yesterday's date
     """
-    return to_datetime("yesterday")
+    return to_datetime("yesterday", relative_base=relative_base)
 
 
-def yesterday_ds() -> str:
+def yesterday_ds(relative_base: t.Optional[datetime] = None) -> str:
     """
     Yesterday utc ds.
 
     Returns:
         Yesterday's ds string.
     """
-    return to_ds("yesterday")
+    return to_ds("yesterday", relative_base=relative_base)
 
 
-def yesterday_timestamp() -> int:
+def yesterday_timestamp(relative_base: t.Optional[datetime] = None) -> int:
     """
     Yesterday utc timestamp.
 
     Returns:
         UTC epoch millis timestamp of yesterday
     """
-    return to_timestamp(yesterday())
+    return to_timestamp(yesterday(relative_base=relative_base))
 
 
 def to_timestamp(
@@ -265,19 +265,19 @@ def date_dict(
     return kwargs
 
 
-def to_ds(obj: TimeLike) -> str:
+def to_ds(obj: TimeLike, relative_base: t.Optional[datetime] = None) -> str:
     """Converts a TimeLike object into YYYY-MM-DD formatted string."""
-    return to_ts(obj)[0:10]
+    return to_ts(obj, relative_base=relative_base)[0:10]
 
 
-def to_ts(obj: TimeLike) -> str:
+def to_ts(obj: TimeLike, relative_base: t.Optional[datetime] = None) -> str:
     """Converts a TimeLike object into YYYY-MM-DD HH:MM:SS formatted string."""
-    return to_datetime(obj).replace(tzinfo=None).isoformat(sep=" ")
+    return to_datetime(obj, relative_base=relative_base).replace(tzinfo=None).isoformat(sep=" ")
 
 
-def to_tstz(obj: TimeLike) -> str:
+def to_tstz(obj: TimeLike, relative_base: t.Optional[datetime] = None) -> str:
     """Converts a TimeLike object into YYYY-MM-DD HH:MM:SS+00:00 formatted string."""
-    return to_datetime(obj).isoformat(sep=" ")
+    return to_datetime(obj, relative_base=relative_base).isoformat(sep=" ")
 
 
 def is_date(obj: TimeLike) -> bool:

tests/core/test_context.py

@@ -49,6 +49,7 @@
     make_inclusive_end,
     now,
     to_date,
+    to_datetime,
     to_timestamp,
     yesterday_ds,
 )
@@ -438,6 +439,99 @@ def test_plan_execution_time():
     )
 
 
+def test_plan_execution_time_start_end():
+    context = Context(config=Config())
+    context.upsert_model(
+        load_sql_based_model(
+            parse(
+                """
+                MODEL(
+                    name db.x,
+                    start '2020-01-01',
+                    kind INCREMENTAL_BY_TIME_RANGE (
+                        time_column ds
+                    ),
+                    cron '@daily'
+                );
+
+                SELECT id, ds FROM (VALUES
+                    ('1', '2020-01-01'),
+                    ('2', '2021-01-01'),
+                    ('3', '2022-01-01'),
+                    ('4', '2023-01-01'),
+                    ('5', '2024-01-01')
+                ) data(id, ds)
+                WHERE ds BETWEEN @start_ds AND @end_ds
+                """
+            )
+        )
+    )
+
+    # prod plan - no fixed execution time so it defaults to now() and reads all the data
+    prod_plan = context.plan(auto_apply=True)
+
+    assert len(prod_plan.new_snapshots) == 1
+
+    context.upsert_model(
+        load_sql_based_model(
+            parse(
+                """
+                MODEL(
+                    name db.x,
+                    start '2020-01-01',
+                    kind INCREMENTAL_BY_TIME_RANGE (
+                        time_column ds
+                    ),
+                    cron '@daily'
+                );
+
+                SELECT id, ds, 'changed' as a FROM (VALUES
+                    ('1', '2020-01-01'),
+                    ('2', '2021-01-01'),
+                    ('3', '2022-01-01'),
+                    ('4', '2023-01-01'),
+                    ('5', '2024-01-01')
+                ) data(id, ds)
+                WHERE ds BETWEEN @start_ds AND @end_ds
+                """
+            )
+        )
+    )
+
+    # dev plan with an execution time in the past and no explicit start/end specified
+    # the plan end should be bounded to it and not exceed it even though in prod the last interval (used as a default end)
+    # is newer than the execution time
+    dev_plan = context.plan("dev", execution_time="2020-01-05")
+
+    assert to_datetime(dev_plan.start) == to_datetime(
+        "2020-01-01"
+    )  # default start is the earliest prod interval
+    assert to_datetime(dev_plan.execution_time) == to_datetime("2020-01-05")
+    assert to_datetime(dev_plan.end) == to_datetime(
+        "2020-01-05"
+    )  # end should not be greater than execution_time
+
+    # same as above but with a relative start
+    dev_plan = context.plan("dev", start="1 day ago", execution_time="2020-01-05")
+
+    assert to_datetime(dev_plan.start) == to_datetime(
+        "2020-01-04"
+    )  # start relative to execution_time
+    assert to_datetime(dev_plan.execution_time) == to_datetime("2020-01-05")
+    assert to_datetime(dev_plan.end) == to_datetime(
+        "2020-01-05"
+    )  # end should not be greater than execution_time
+
+    # same as above but with a relative start and a relative end
+    dev_plan = context.plan("dev", start="2 days ago", execution_time="2020-01-05", end="1 day ago")
+
+    assert to_datetime(dev_plan.start) == to_datetime(
+        "2020-01-03"
+    )  # start relative to execution_time
+    assert to_datetime(dev_plan.execution_time) == to_datetime("2020-01-05")
+    assert to_datetime(dev_plan.end) == to_datetime("2020-01-04")  # end relative to execution_time
+
+
 def test_override_builtin_audit_blocking_mode():
     context = Context(config=Config())
     context.upsert_model(