Fix: Invalid fallback start date for snapshots with missing start (#1504)

Author: izeigerman

sqlmesh/core/context.py

@@ -38,6 +38,7 @@
 import traceback
 import typing as t
 import unittest.result
+from datetime import timedelta
 from io import StringIO
 from pathlib import Path
 from types import MappingProxyType
@@ -89,7 +90,7 @@
 from sqlmesh.core.user import User
 from sqlmesh.utils import UniqueKeyDict, env_vars, sys_path
 from sqlmesh.utils.dag import DAG
-from sqlmesh.utils.date import TimeLike, now_ds
+from sqlmesh.utils.date import TimeLike, now_ds, to_date
 from sqlmesh.utils.errors import (
     ConfigError,
     MissingDependencyError,
@@ -775,6 +776,7 @@ def plan(
         # If no end date is specified, use the max interval end from prod
         # to prevent unintended evaluation of the entire DAG.
         default_end = self.state_sync.max_interval_end_for_environment(c.PROD) if is_dev else None
+        default_start = to_date(default_end) - timedelta(days=1) if default_end else None
 
         plan = Plan(
             context_diff=self._context_diff(
@@ -797,6 +799,7 @@ def plan(
             auto_categorization_enabled=not no_auto_categorization,
             effective_from=effective_from,
             include_unmodified=include_unmodified,
+            default_start=default_start,
             default_end=default_end,
         )
 

sqlmesh/core/plan/definition.py

@@ -71,6 +71,7 @@ class Plan:
         effective_from: The effective date from which to apply forward-only changes on production.
         include_unmodified: Indicates whether to include unmodified nodes in the target development environment.
         environment_suffix_target: Indicates whether to append the environment name to the schema or table name.
+        default_start: The default plan start to use if not specified.
         default_end: The default plan end to use if not specified.
     """
 
@@ -92,6 +93,7 @@ def __init__(
         auto_categorization_enabled: bool = True,
         effective_from: t.Optional[TimeLike] = None,
         include_unmodified: bool = False,
+        default_start: t.Optional[TimeLike] = None,
         default_end: t.Optional[TimeLike] = None,
     ):
         self.context_diff = context_diff
@@ -109,7 +111,9 @@ def __init__(
         self.include_unmodified = include_unmodified
         self._restate_models = set(restate_models or [])
         self._effective_from: t.Optional[TimeLike] = None
-        self._start = start if start or not (is_dev and forward_only) else yesterday_ds()
+        self._start = (
+            start if start or not (is_dev and forward_only) else (default_start or yesterday_ds())
+        )
         self._end = end if end or not is_dev else (default_end or now())
         self._execution_time = execution_time or now()
         self._apply = apply

sqlmesh/core/snapshot/definition.py

@@ -1253,9 +1253,6 @@ def earliest_start_date(
     snapshots = list(snapshots)
     earliest = to_datetime(yesterday().date())
     if snapshots:
-        for snapshot in snapshots:
-            if not snapshot.parents and not snapshot.node.start:
-                cache[snapshot.name] = earliest
         return min(start_date(snapshot, snapshots, cache) or earliest for snapshot in snapshots)
     return earliest
 

tests/core/test_context.py

@@ -1,5 +1,5 @@
 import pathlib
-from datetime import date
+from datetime import date, timedelta
 from tempfile import TemporaryDirectory
 from unittest.mock import call
 
@@ -19,7 +19,7 @@
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.model import load_sql_based_model
 from sqlmesh.core.plan import BuiltInPlanEvaluator, Plan
-from sqlmesh.utils.date import yesterday_ds
+from sqlmesh.utils.date import now, to_date, yesterday_ds
 from sqlmesh.utils.errors import ConfigError
 from tests.utils.test_filesystem import create_temp_file
 
@@ -425,3 +425,22 @@ def test_janitor(sushi_context, mocker: MockerFixture) -> None:
         ],
         any_order=True,
     )
+
+
+def test_plan_default_end(sushi_context_pre_scheduling: Context):
+    prod_plan = sushi_context_pre_scheduling.plan("prod", no_prompts=True)
+    # Simulate that the prod is 3 days behind.
+    plan_end = to_date(now()) - timedelta(days=3)
+    prod_plan._end = plan_end
+    sushi_context_pre_scheduling.apply(prod_plan)
+
+    dev_plan = sushi_context_pre_scheduling.plan(
+        "test_env", no_prompts=True, include_unmodified=True, skip_backfill=True, auto_apply=True
+    )
+    assert dev_plan.end == plan_end
+
+    forward_only_dev_plan = sushi_context_pre_scheduling.plan(
+        "test_env_forward_only", no_prompts=True, include_unmodified=True, forward_only=True
+    )
+    assert forward_only_dev_plan.end == plan_end
+    assert forward_only_dev_plan._start == plan_end

tests/core/test_snapshot.py

@@ -1,7 +1,7 @@
 import json
 import typing as t
 from copy import deepcopy
-from datetime import timedelta
+from datetime import datetime, timedelta
 from pathlib import Path
 
 import pytest
@@ -11,6 +11,7 @@
 
 from sqlmesh.core.audit import StandaloneAudit
 from sqlmesh.core.config import AutoCategorizationMode, CategorizerConfig
+from sqlmesh.core.context import Context
 from sqlmesh.core.dialect import parse, parse_one
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.model import (
@@ -30,6 +31,7 @@
     SnapshotChangeCategory,
     SnapshotFingerprint,
     categorize_change,
+    earliest_start_date,
     fingerprint_from_node,
     has_paused_forward_only,
 )
@@ -1239,3 +1241,14 @@ def test_multi_interval_merge(make_snapshot):
 
     assert b_start == to_timestamp("2023-01-01 00:15:00")
     assert b_end == to_timestamp("2023-01-01 00:45:00")
+
+
+def test_earliest_start_date(sushi_context: Context):
+    model_name = "sushi.waiter_names"
+    assert sushi_context.snapshots[model_name].node.start is None
+
+    cache: t.Dict[str, datetime] = {}
+    earliest_start_date(sushi_context.snapshots.values(), cache)
+
+    # Make sure that the default value for a snapshot with a missing start is not cached.
+    assert model_name not in cache