fix: Missing Intervals less than Interval Unit (#1584)

* fix: missing intervals less than interval unit

* fix: missing intervals less than interval unit

Author: eakmanrq

sqlmesh/core/node.py

@@ -135,6 +135,10 @@ def cron_floor(self, value: TimeLike) -> TimeLike:
     def seconds(self) -> int:
         return INTERVAL_SECONDS[self]
 
+    @property
+    def milliseconds(self) -> int:
+        return self.seconds * 1000
+
 
 INTERVAL_SECONDS = {
     IntervalUnit.YEAR: 60 * 60 * 24 * 365,

sqlmesh/core/snapshot/definition.py

@@ -27,6 +27,7 @@
     to_datetime,
     to_ds,
     to_timestamp,
+    validate_date_range,
     yesterday,
 )
 from sqlmesh.utils.errors import SQLMeshError
@@ -632,7 +633,7 @@ def inclusive_exclusive(
         """
         interval_unit = self.node.interval_unit
         start_ts = to_timestamp(interval_unit.cron_floor(start))
-        if start_ts < to_timestamp(start):
+        if start_ts < to_timestamp(start) and not self.model.allow_partials:
             start_ts = to_timestamp(interval_unit.cron_next(start_ts))
 
         if is_date(end):
@@ -691,6 +692,15 @@ def missing_intervals(
         Returns:
             A list of all the missing intervals as epoch timestamps.
         """
+        # If the amount of time being checked is less than the size of a single interval then we
+        # know that there can't being missing intervals within that range and return
+        validate_date_range(start, end)
+        if (
+            not is_date(end)
+            and not (self.is_model and self.model.allow_partials)
+            and to_timestamp(end) - to_timestamp(start) < self.node.interval_unit.milliseconds
+        ):
+            return []
         intervals = self.dev_intervals if is_dev and self.is_paused_forward_only else self.intervals
 
         if self.is_symbolic or (self.is_seed and intervals):

tests/core/test_snapshot.py

@@ -384,6 +384,78 @@ def test_seed_intervals(make_snapshot):
     assert snapshot_a.missing_intervals("2020-01-02", "2020-01-02") == []
 
 
+def test_missing_interval_smaller_than_interval_unit(make_snapshot):
+    snapshot_daily = make_snapshot(
+        SqlModel(
+            name="name",
+            kind=IncrementalByTimeRangeKind(time_column="ds", batch_size=30),
+            owner="owner",
+            dialect="spark",
+            cron="@daily",
+            start="2020-01-01",
+            query=parse_one("SELECT @EACH([1, 2], x -> x), ds FROM parent.tbl"),
+        )
+    )
+
+    assert snapshot_daily.missing_intervals("2020-01-01 00:00:05", "2020-01-01 23:59:59") == []
+    assert snapshot_daily.missing_intervals("2020-01-01 00:00:00", "2020-01-02 00:00:00") == [
+        (to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))
+    ]
+
+    snapshot_hourly = make_snapshot(
+        SqlModel(
+            name="name",
+            kind=IncrementalByTimeRangeKind(time_column="ds", batch_size=30),
+            owner="owner",
+            dialect="spark",
+            cron="@hourly",
+            start="2020-01-01",
+            query=parse_one("SELECT @EACH([1, 2], x -> x), ds FROM parent.tbl"),
+        )
+    )
+
+    assert snapshot_hourly.missing_intervals("2020-01-01 00:00:00", "2020-01-01 00:59:00") == []
+    assert snapshot_hourly.missing_intervals("2020-01-01 00:00:00", "2020-01-01 01:00:00") == [
+        (to_timestamp("2020-01-01"), to_timestamp("2020-01-01 01:00:00"))
+    ]
+
+    snapshot_end_categorical = make_snapshot(
+        SqlModel(
+            name="name",
+            kind=IncrementalByTimeRangeKind(time_column="ds", batch_size=30),
+            owner="owner",
+            dialect="spark",
+            cron="@daily",
+            start="2020-01-01",
+            query=parse_one("SELECT @EACH([1, 2], x -> x), ds FROM parent.tbl"),
+        )
+    )
+
+    assert snapshot_end_categorical.missing_intervals("2020-01-01 00:00:00", "2020-01-01") == [
+        (to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))
+    ]
+
+    snapshot_partial = make_snapshot(
+        SqlModel(
+            name="name",
+            kind=IncrementalByTimeRangeKind(time_column="ds", batch_size=30),
+            owner="owner",
+            dialect="spark",
+            cron="@daily",
+            start="2020-01-01",
+            query=parse_one("SELECT @EACH([1, 2], x -> x), ds FROM parent.tbl"),
+            allow_partials=True,
+        )
+    )
+
+    assert snapshot_partial.missing_intervals("2020-01-01 00:00:05", "2020-01-01 23:59:59") == [
+        (to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))
+    ]
+    assert snapshot_partial.missing_intervals("2020-01-01 00:00:00", "2020-01-02 00:00:00") == [
+        (to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))
+    ]
+
+
 def test_remove_intervals(snapshot: Snapshot):
     snapshot.add_interval("2020-01-01", "2020-01-01")
     snapshot.remove_interval(snapshot.get_removal_interval("2020-01-01", "2020-01-01"))