Fix latest date (#633)

* wip - try fix latest

* refactor

* add test for now

* more fixes

* fmt

* fmt

---------

Co-authored-by: tobymao <toby.mao@gmail.com>

Author: eakmanrq

sqlmesh/core/plan/definition.py

@@ -23,6 +23,7 @@
 from sqlmesh.utils.date import (
     TimeLike,
     make_inclusive,
+    make_inclusive_end,
     now,
     to_date,
     to_ds,
@@ -163,24 +164,35 @@ def set_start(self, new_start: TimeLike) -> None:
         self._start = new_start
         self.__missing_intervals = None
 
+    def _get_end_date(self, end_and_units: t.List[t.Tuple[int, IntervalUnit]]) -> TimeLike:
+        if end_and_units:
+            end, unit = max(end_and_units)
+
+            if unit == IntervalUnit.DAY:
+                return to_date(make_inclusive_end(end))
+            return end
+        return now()
+
     @property
     def end(self) -> TimeLike:
         """Returns the end of the plan or now."""
         if not self._end or not self.override_end:
             if self._missing_intervals:
-                end, interval_unit = max(
+                return self._get_end_date(
                     [
                         (end, snapshot.model.interval_unit())
                         for snapshot in self.snapshots
                         if snapshot.version_get_or_generate() in self._missing_intervals
                         for _, end in self._missing_intervals[snapshot.version_get_or_generate()]
                     ]
                 )
-                if interval_unit == IntervalUnit.DAY:
-                    return to_date(make_inclusive(self.start, end)[1])
-                return end
-            else:
-                return scheduler.latest_end_date(self.snapshots)
+            return self._get_end_date(
+                [
+                    (snapshot.intervals[-1][1], snapshot.model.interval_unit())
+                    for snapshot in self.snapshots
+                    if snapshot.intervals
+                ]
+            )
         return self._end
 
     @end.setter

sqlmesh/core/scheduler.py

@@ -2,10 +2,9 @@
 
 import logging
 import typing as t
-from datetime import date, datetime
+from datetime import datetime
 
 from sqlmesh.core.console import Console, get_console
-from sqlmesh.core.model.meta import IntervalUnit
 from sqlmesh.core.snapshot import (
     Snapshot,
     SnapshotEvaluator,
@@ -18,9 +17,7 @@
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
     TimeLike,
-    make_inclusive,
     now,
-    to_date,
     to_datetime,
     validate_date_range,
     yesterday,
@@ -381,27 +378,6 @@ def earliest_start_date(snapshots: t.Iterable[Snapshot]) -> datetime:
     return yesterday()
 
 
-def latest_end_date(snapshots: t.Iterable[Snapshot]) -> datetime | date:
-    """Get the latest end date from a collection of snapshots.
-
-    Args:
-        snapshots: Snapshots to find latest end date.
-    Returns:
-        The latest end date or now if none is found.
-    """
-    end_date_and_intervals = [
-        (snapshot.intervals[-1][1], snapshot.model.interval_unit())
-        for snapshot in snapshots
-        if snapshot.intervals
-    ]
-    if end_date_and_intervals:
-        end_date, interval = max(end_date_and_intervals)
-        if interval == IntervalUnit.DAY:
-            return to_date(make_inclusive(end_date, end_date)[1])
-        return to_datetime(end_date)
-    return now()
-
-
 def _batched_intervals(params: SnapshotToBatches) -> SnapshotToBatches:
     batches = {}
 

sqlmesh/core/snapshot/definition.py

@@ -22,7 +22,7 @@
 from sqlmesh.core.model.meta import HookCall
 from sqlmesh.utils.date import (
     TimeLike,
-    make_inclusive,
+    is_date,
     now,
     now_timestamp,
     to_datetime,
@@ -446,9 +446,10 @@ def remove_interval(self, start: TimeLike, end: TimeLike) -> None:
         self.dev_intervals = remove_interval(self.dev_intervals, *interval)
 
     def _inclusive_exclusive(self, start: TimeLike, end: TimeLike) -> t.Tuple[int, int]:
-        start_dt, end_dt = make_inclusive(start, end)
-        start_ts = to_timestamp(self.model.cron_floor(start_dt))
-        end_ts = to_timestamp(self.model.cron_next(end_dt))
+        start_ts = to_timestamp(self.model.cron_floor(start))
+        end_ts = to_timestamp(
+            self.model.cron_next(end) if is_date(end) else self.model.cron_floor(end)
+        )
 
         if start_ts >= end_ts:
             raise ValueError("`end` must be >= `start`")
@@ -475,27 +476,31 @@ def missing_intervals(
         Args:
             start: The start date/time of the interval (inclusive)
             end: The end date/time of the interval (inclusive)
+            latest: The date/time to use for latest (inclusive)
 
         Returns:
             A list of all the missing intervals as epoch timestamps.
         """
         if self.is_embedded_kind:
             return []
 
-        start_dt, end_dt = make_inclusive(start, self.model.cron_floor(end))
-
         if self.is_full_kind or self.is_view_kind or self.is_seed_kind:
-            latest_dt = to_datetime(self.model.cron_floor(latest or now()))
-            latest_ts = to_timestamp(latest_dt)
+            latest = latest or now()
+
+            latest_start, latest_end = self._inclusive_exclusive(
+                latest if is_date(latest) else self.model.cron_prev(self.model.cron_floor(latest)),
+                latest,
+            )
             # if the latest ts is stored in the last interval, nothing is missing
             # else returns the latest ts with the exclusive end ts.
-            if self.intervals and self.intervals[-1][1] >= latest_ts:
+            if self.intervals and self.intervals[-1][1] >= latest_end:
                 return []
-            return [(to_timestamp(self.model.cron_prev(latest_dt)), latest_ts)]
+            return [(latest_start, latest_end)]
 
         missing = []
-        dates = list(croniter_range(start_dt, end_dt, self.model.normalized_cron()))
-        size = len(dates)
+        start_dt, end_dt = (to_datetime(ts) for ts in self._inclusive_exclusive(start, end))
+        dates = tuple(croniter_range(start_dt, end_dt, self.model.normalized_cron()))
+        size = len(dates) - 1
 
         for i in range(size):
             current_ts = to_timestamp(dates[i])

sqlmesh/utils/date.py

@@ -232,11 +232,14 @@ def make_inclusive(start: TimeLike, end: TimeLike) -> Interval:
     Returns:
         A tuple of inclusive datetime objects.
     """
-    start_dt = to_datetime(start)
+    return (to_datetime(start), make_inclusive_end(end))
+
+
+def make_inclusive_end(end: TimeLike) -> datetime:
     end_dt = to_datetime(end)
     if is_date(end):
         end_dt = end_dt + timedelta(days=1)
-    return (start_dt, end_dt - timedelta(milliseconds=1))
+    return end_dt - timedelta(milliseconds=1)
 
 
 def preserve_time_like_kind(input_value: TimeLike, output_value: TimeLike) -> TimeLike:

tests/core/test_plan.py

@@ -78,7 +78,7 @@ def test_forward_only_dev(make_snapshot, mocker: MockerFixture):
     yesterday_ds_mock = mocker.patch("sqlmesh.core.scheduler.yesterday")
     yesterday_ds_mock.return_value = expected_start
 
-    now_ds_mock = mocker.patch("sqlmesh.core.scheduler.now")
+    now_ds_mock = mocker.patch("sqlmesh.core.plan.definition.now")
     now_ds_mock.return_value = expected_end
     state_reader_mock.missing_intervals.return_value = {}
 

tests/core/test_snapshot.py

@@ -10,6 +10,8 @@
 from sqlmesh.core.model import (
     IncrementalByTimeRangeKind,
     Model,
+    ModelKind,
+    ModelKindName,
     SeedKind,
     SqlModel,
     create_seed_model,
@@ -51,6 +53,20 @@ def model():
     )
 
 
+@pytest.fixture
+def full_refresh_model():
+    return SqlModel(
+        name="fr_model",
+        kind=ModelKind(name=ModelKindName.FULL),
+        owner="owner",
+        dialect="spark",
+        cron="1 0 * * *",
+        batch_size=30,
+        start="2020-01-01",
+        query=parse_one("SELECT @EACH([1, 2], x -> x), ds FROM parent.tbl"),
+    )
+
+
 @pytest.fixture
 def snapshot(
     model: Model,
@@ -70,6 +86,25 @@ def snapshot(
     return snapshot
 
 
+@pytest.fixture
+def full_refresh_snapshot(
+    full_refresh_model: Model,
+    parent_model: Model,
+    monkeypatch: MonkeyPatch,
+    mocker: MockerFixture,
+    make_snapshot,
+):
+    mock = mocker.Mock()
+    mock.return_value = to_datetime("2022-09-23T00:12:53+00:00")
+    monkeypatch.setattr("sqlmesh.utils.date.now", mock)
+    full_refresh_snapshot = make_snapshot(
+        full_refresh_model,
+        models={parent_model.name: parent_model, full_refresh_model.name: full_refresh_model},
+    )
+    full_refresh_snapshot.version = full_refresh_snapshot.fingerprint
+    return full_refresh_snapshot
+
+
 def test_json(snapshot: Snapshot):
     assert json.loads(snapshot.json()) == {
         "created_ts": 1663891973000,
@@ -151,11 +186,11 @@ def test_add_interval(snapshot: Snapshot, make_snapshot):
     ]
     snapshot.add_interval("2019-01-01 00:00:00", "2020-01-31 00:00:01")
     assert snapshot.intervals == [
-        (to_timestamp("2019-01-01"), to_timestamp("2020-02-01")),
+        (to_timestamp("2019-01-01"), to_timestamp("2020-01-31")),
     ]
-    snapshot.add_interval("2018-12-31 23:59:59", "2020-01-31 00:00:01")
+    snapshot.add_interval("2018-12-31 23:59:59", "2020-01-31 12:00:01")
     assert snapshot.intervals == [
-        (to_timestamp("2018-12-31"), to_timestamp("2020-02-01")),
+        (to_timestamp("2018-12-31"), to_timestamp("2020-01-31")),
     ]
 
     new_snapshot = make_snapshot(snapshot.model)
@@ -207,6 +242,34 @@ def test_missing_intervals(snapshot: Snapshot):
     ]
 
 
+def test_missing_interval_latest(
+    full_refresh_snapshot: Snapshot, monkeypatch: MonkeyPatch, mocker: MockerFixture
+):
+    mock = mocker.Mock()
+    mock.return_value = to_datetime("2020-01-05T00:12:53+00:00")
+    monkeypatch.setattr("sqlmesh.core.snapshot.definition.now", mock)
+    full_refresh_snapshot.add_interval("2020-01-01", "2020-01-01")
+    assert full_refresh_snapshot.missing_intervals("2020-01-01", "2020-01-01", "2020-01-01") == []
+    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-02", "2020-01-02") == [
+        (to_timestamp("2020-01-02"), to_timestamp("2020-01-03"))
+    ]
+    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-03", "2020-01-03") == [
+        (to_timestamp("2020-01-03"), to_timestamp("2020-01-04"))
+    ]
+    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-03", "2020-01-04") == [
+        (to_timestamp("2020-01-04"), to_timestamp("2020-01-05"))
+    ]
+    assert full_refresh_snapshot.missing_intervals(
+        "2020-01-02", "2020-01-03", "2020-01-03 01:00:00"
+    ) == [(to_timestamp("2020-01-02"), to_timestamp("2020-01-03"))]
+    assert full_refresh_snapshot.missing_intervals(
+        "2020-01-02", "2020-01-03", "2020-01-04 23:59:59"
+    ) == [(to_timestamp("2020-01-03"), to_timestamp("2020-01-04"))]
+    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-03") == [
+        (to_timestamp("2020-01-04"), to_timestamp("2020-01-05"))
+    ]
+
+
 def test_remove_intervals(snapshot: Snapshot):
     snapshot.add_interval("2020-01-01", "2020-01-01")
     snapshot.remove_interval("2020-01-01", "2020-01-01")