make startdate relative so that development is faster (#638)

Author: tobymao

examples/sushi/models/items.py

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

examples/sushi/models/orders.py

@@ -17,7 +17,7 @@
     "sushi.orders",
     description="Table of sushi orders.",
     kind=IncrementalByTimeRangeKind(time_column="ds"),
-    start="2022-01-01",
+    start="3 months ago",
     cron="@daily",
     batch_size=30,
     columns={

sqlmesh/core/snapshot/definition.py

@@ -768,7 +768,7 @@ def _model_metadata_hash(model: Model, audits: t.Dict[str, Audit]) -> str:
         model.dialect,
         model.owner,
         model.description,
-        str(to_timestamp(model.start)) if model.start else None,
+        str(model.start) if model.start else None,
         str(model.batch_size) if model.batch_size is not None else None,
     ]
 

sqlmesh/core/state_sync/common.py

@@ -218,6 +218,8 @@ def unpause_snapshots(
     def _ensure_no_gaps(
         self, target_snapshots: t.Iterable[Snapshot], target_environment: Environment
     ) -> None:
+        from sqlmesh.core.scheduler import start_date
+
         target_snapshots_by_name = {s.name: s for s in target_snapshots}
 
         changed_version_prev_snapshots_by_name = {
@@ -257,7 +259,8 @@ def _ensure_no_gaps(
                 and prev_snapshot.intervals
             ):
                 missing_intervals = target_snapshot.missing_intervals(
-                    prev_snapshot.intervals[0][0],
+                    start_date(target_snapshot, target_snapshots_by_name.values())
+                    or prev_snapshot.intervals[0][0],
                     prev_snapshot.intervals[-1][1],
                 )
                 if missing_intervals:

tests/conftest.py

@@ -27,21 +27,36 @@ def duck_conn() -> duckdb.DuckDBPyConnection:
     return duckdb.connect()
 
 
+def push_plan(context: Context, plan: Plan) -> None:
+    plan_evaluator = BuiltInPlanEvaluator(context.state_sync, context.snapshot_evaluator)
+    plan_evaluator._push(plan)
+    plan_evaluator._promote(plan)
+
+
 @pytest.fixture()
 def sushi_context_pre_scheduling(mocker: MockerFixture) -> Context:
     context, plan = init_and_plan_sushi_context("examples/sushi", mocker)
+    push_plan(context, plan)
+    return context
 
-    plan_evaluator = BuiltInPlanEvaluator(context.state_sync, context.snapshot_evaluator)
-    plan_evaluator._push(plan)
-    plan_evaluator._promote(plan)
 
+@pytest.fixture()
+def sushi_context_fixed_date(mocker: MockerFixture) -> Context:
+    context, plan = init_and_plan_sushi_context("examples/sushi", mocker)
+
+    for model in context.models.values():
+        if model.start:
+            context.upsert_model(model.name, start="2022-01-01")
+
+    plan = context.plan("prod")
+    plan.set_start("1 week ago")
+    push_plan(context, plan)
     return context
 
 
 @pytest.fixture()
 def sushi_context(mocker: MockerFixture) -> Context:
     context, plan = init_and_plan_sushi_context("examples/sushi", mocker)
-
     context.apply(plan)
     return context
 
@@ -71,10 +86,6 @@ def init_and_plan_sushi_context(
     path: str, mocker: MockerFixture, start: TimeLike = "1 week ago"
 ) -> t.Tuple[Context, Plan]:
     sushi_context = Context(path=path, config="test_config")
-
-    for snapshot in sushi_context.snapshots.values():
-        snapshot.set_version()
-
     confirm = mocker.patch("sqlmesh.core.console.Confirm")
     confirm.ask.return_value = False
 

tests/core/test_context.py

@@ -11,6 +11,7 @@
 from sqlmesh.core.dialect import parse
 from sqlmesh.core.model import load_model
 from sqlmesh.core.plan import BuiltInPlanEvaluator, Plan
+from sqlmesh.utils.date import yesterday_ds
 from sqlmesh.utils.errors import ConfigError
 from tests.utils.test_filesystem import create_temp_file
 
@@ -217,7 +218,8 @@ def test_render(sushi_context, assert_exp_eq):
 def test_diff(sushi_context: Context, mocker: MockerFixture):
     mock_console = mocker.Mock()
     sushi_context.console = mock_console
-    sushi_context.run(start="2022-01-01", end="2022-01-01", latest="2022-01-01")
+    yesterday = yesterday_ds()
+    sushi_context.run(start=yesterday, end=yesterday, latest=yesterday)
 
     plan_evaluator = BuiltInPlanEvaluator(
         sushi_context.state_sync, sushi_context.snapshot_evaluator
@@ -324,8 +326,8 @@ def test():
 def test_plan_apply(sushi_context) -> None:
     plan = sushi_context.plan(
         "dev",
-        start="2022-01-01",
-        end="2022-01-01",
+        start=yesterday_ds(),
+        end=yesterday_ds(),
     )
     sushi_context.apply(plan)
     assert sushi_context.state_reader.get_environment("dev")

tests/core/test_scheduler.py

@@ -9,19 +9,17 @@
 
 
 @pytest.fixture
-def scheduler(sushi_context_pre_scheduling: Context) -> Scheduler:
-    return sushi_context_pre_scheduling.scheduler()
+def scheduler(sushi_context_fixed_date: Context) -> Scheduler:
+    return sushi_context_fixed_date.scheduler()
 
 
 @pytest.fixture
-def orders(sushi_context_pre_scheduling: Context) -> Snapshot:
-    return sushi_context_pre_scheduling.snapshots["sushi.orders"]
+def orders(sushi_context_fixed_date: Context) -> Snapshot:
+    return sushi_context_fixed_date.snapshots["sushi.orders"]
 
 
-def test_interval_params(
-    scheduler: Scheduler, sushi_context_pre_scheduling: Context, orders: Snapshot
-):
-    waiter_revenue = sushi_context_pre_scheduling.snapshots["sushi.waiter_revenue_by_day"]
+def test_interval_params(scheduler: Scheduler, sushi_context_fixed_date: Context, orders: Snapshot):
+    waiter_revenue = sushi_context_fixed_date.snapshots["sushi.waiter_revenue_by_day"]
     start_ds = "2022-01-01"
     end_ds = "2022-02-05"
     assert scheduler._interval_params([orders, waiter_revenue], start_ds, end_ds) == {
@@ -53,8 +51,8 @@ def test_interval_params_nonconsecutive(scheduler: Scheduler, orders: Snapshot):
     }
 
 
-def test_interval_params_missing(scheduler: Scheduler, sushi_context_pre_scheduling: Context):
-    waiters = sushi_context_pre_scheduling.snapshots["sushi.waiter_as_customer_by_day"]
+def test_interval_params_missing(scheduler: Scheduler, sushi_context_fixed_date: Context):
+    waiters = sushi_context_fixed_date.snapshots["sushi.waiter_as_customer_by_day"]
 
     start_ds = "2022-01-01"
     end_ds = "2022-03-01"
@@ -66,34 +64,34 @@ def test_interval_params_missing(scheduler: Scheduler, sushi_context_pre_schedul
 
 
 def test_multi_version_snapshots(
-    sushi_context_pre_scheduling: Context, scheduler: Scheduler, make_snapshot
+    sushi_context_fixed_date: Context, scheduler: Scheduler, make_snapshot
 ):
     start_ds = "2022-01-01"
     end_ds = "2022-02-05"
 
-    model = sushi_context_pre_scheduling.models["sushi.waiter_as_customer_by_day"]
+    model = sushi_context_fixed_date.models["sushi.waiter_as_customer_by_day"]
 
     items_a = make_snapshot(
         model,
-        models=sushi_context_pre_scheduling.models,
+        models=sushi_context_fixed_date.models,
         version="1",
     )
     items_a.fingerprint = SnapshotFingerprint(data_hash="data", metadata_hash="metadata")
     items_a.add_interval("2022-01-10", "2022-01-15")
-    sushi_context_pre_scheduling.state_sync.push_snapshots([items_a])
+    sushi_context_fixed_date.state_sync.push_snapshots([items_a])
 
-    model = sushi_context_pre_scheduling.upsert_model(
+    model = sushi_context_fixed_date.upsert_model(
         model,
         query=parse_one("SELECT 1::INT, '2022-01-01'::TEXT AS ds"),
     )
 
     items_b = make_snapshot(
         model,
-        models=sushi_context_pre_scheduling.models,
+        models=sushi_context_fixed_date.models,
         version="1",
     )
     items_b.add_interval("2022-01-20", "2022-01-25")
-    sushi_context_pre_scheduling.state_sync.push_snapshots([items_b])
+    sushi_context_fixed_date.state_sync.push_snapshots([items_b])
 
     interval_params = scheduler._interval_params([items_a], start_ds, end_ds)
     assert len(interval_params) == 1
@@ -113,9 +111,9 @@ def test_multi_version_snapshots(
     ]
 
 
-def test_run(sushi_context_pre_scheduling: Context, scheduler: Scheduler):
-    adapter = sushi_context_pre_scheduling.engine_adapter
-    snapshot = sushi_context_pre_scheduling.snapshots["sushi.items"]
+def test_run(sushi_context_fixed_date: Context, scheduler: Scheduler):
+    adapter = sushi_context_fixed_date.engine_adapter
+    snapshot = sushi_context_fixed_date.snapshots["sushi.items"]
     scheduler.run(
         c.PROD,
         "2022-01-01",

tests/core/test_snapshot.py

@@ -284,7 +284,7 @@ def test_fingerprint(model: Model, parent_model: Model):
 
     original_fingerprint = SnapshotFingerprint(
         data_hash="3042895307",
-        metadata_hash="3589467163",
+        metadata_hash="2417444816",
     )
 
     assert fingerprint == original_fingerprint
@@ -363,7 +363,7 @@ def test_fingerprint_jinja_macros(model: Model):
 
     original_fingerprint = SnapshotFingerprint(
         data_hash="2665680291",
-        metadata_hash="3589467163",
+        metadata_hash="2417444816",
     )
 
     fingerprint = fingerprint_from_model(model, models={})

tests/core/test_state_sync.py

@@ -14,7 +14,7 @@
 )
 from sqlmesh.core.snapshot import Snapshot, SnapshotTableInfo
 from sqlmesh.core.state_sync import EngineAdapterStateSync
-from sqlmesh.utils.date import now_timestamp, to_datetime, to_timestamp
+from sqlmesh.utils.date import now_timestamp, to_datetime, to_ds, to_timestamp
 from sqlmesh.utils.errors import SQLMeshError
 
 
@@ -432,6 +432,41 @@ def test_promote_snapshots_no_gaps(state_sync: EngineAdapterStateSync, make_snap
     promote_snapshots(state_sync, [new_snapshot_same_interval], "prod", no_gaps=True)
 
 
+def test_start_date_gap(state_sync: EngineAdapterStateSync, make_snapshot: t.Callable):
+    model = SqlModel(
+        name="a",
+        query=parse_one("select 1, ds"),
+        start="2022-01-01",
+        kind=IncrementalByTimeRangeKind(time_column="ds"),
+        cron="@daily",
+    )
+
+    snapshot = make_snapshot(model, version="a")
+    snapshot.add_interval("2022-01-01", "2022-01-03")
+    state_sync.push_snapshots([snapshot])
+    promote_snapshots(state_sync, [snapshot], "prod")
+
+    model = SqlModel(
+        name="a",
+        query=parse_one("select 1, ds"),
+        start="2022-01-02",
+        kind=IncrementalByTimeRangeKind(time_column="ds"),
+        cron="@daily",
+    )
+
+    snapshot = make_snapshot(model, version="b")
+    snapshot.add_interval("2022-01-03", "2022-01-04")
+    state_sync.push_snapshots([snapshot])
+    with pytest.raises(
+        SQLMeshError,
+        match=r"Detected gaps in snapshot.*",
+    ):
+        promote_snapshots(state_sync, [snapshot], "prod", no_gaps=True)
+
+    state_sync.add_interval(snapshot, "2022-01-02", "2022-01-03")
+    promote_snapshots(state_sync, [snapshot], "prod", no_gaps=True)
+
+
 def test_delete_expired_environments(state_sync: EngineAdapterStateSync, make_snapshot: t.Callable):
     snapshot = make_snapshot(
         SqlModel(
@@ -472,10 +507,12 @@ def test_delete_expired_environments(state_sync: EngineAdapterStateSync, make_sn
 def test_missing_intervals(sushi_context_pre_scheduling: Context) -> None:
     sushi_context = sushi_context_pre_scheduling
     state_sync = sushi_context.state_reader
-    missing = state_sync.missing_intervals("prod", "2022-01-01", "2022-01-07", latest="2022-01-07")
+    start = to_ds("1 week ago")
+    end = to_ds("yesterday")
+    missing = state_sync.missing_intervals("prod", start, end, latest=end)
     assert missing
     assert missing == sushi_context.state_reader.missing_intervals(
-        sushi_context.snapshots.values(), "2022-01-01", "2022-01-07", "2022-01-07"
+        sushi_context.snapshots.values(), start, end, end
     )