Chore: Move removal of partial intervals out of the interval state and into the facade (#3932)

Author: izeigerman

sqlmesh/core/state_sync/db/facade.py

@@ -52,7 +52,7 @@
 from sqlmesh.core.state_sync.db.snapshot import SnapshotState
 from sqlmesh.core.state_sync.db.version import VersionState
 from sqlmesh.core.state_sync.db.migrator import StateMigrator
-from sqlmesh.utils.date import TimeLike, to_timestamp
+from sqlmesh.utils.date import TimeLike, to_timestamp, time_like_to_str
 from sqlmesh.utils.errors import ConflictingPlanError, SQLMeshError
 
 logger = logging.getLogger(__name__)
@@ -358,7 +358,24 @@ def add_interval(
 
     @transactional()
     def add_snapshots_intervals(self, snapshots_intervals: t.Sequence[SnapshotIntervals]) -> None:
-        self.interval_state.add_snapshots_intervals(snapshots_intervals)
+        intervals_to_insert = []
+        for snapshot_intervals in snapshots_intervals:
+            snapshot_intervals = snapshot_intervals.copy(
+                update={
+                    "intervals": _remove_partial_intervals(
+                        snapshot_intervals.intervals, snapshot_intervals.snapshot_id, is_dev=False
+                    ),
+                    "dev_intervals": _remove_partial_intervals(
+                        snapshot_intervals.dev_intervals,
+                        snapshot_intervals.snapshot_id,
+                        is_dev=True,
+                    ),
+                }
+            )
+            if not snapshot_intervals.is_empty():
+                intervals_to_insert.append(snapshot_intervals)
+        if intervals_to_insert:
+            self.interval_state.add_snapshots_intervals(intervals_to_insert)
 
     @transactional()
     def remove_intervals(
@@ -478,3 +495,27 @@ def _ensure_no_gaps(
     def _transaction(self) -> t.Iterator[None]:
         with self.engine_adapter.transaction():
             yield
+
+
+def _remove_partial_intervals(
+    intervals: t.List[Interval], snapshot_id: t.Optional[SnapshotId], *, is_dev: bool
+) -> t.List[Interval]:
+    results = []
+    for start_ts, end_ts in intervals:
+        if start_ts < end_ts:
+            logger.info(
+                "Adding %s (%s, %s) for snapshot %s",
+                "dev interval" if is_dev else "interval",
+                time_like_to_str(start_ts),
+                time_like_to_str(end_ts),
+                snapshot_id,
+            )
+            results.append((start_ts, end_ts))
+        else:
+            logger.info(
+                "Skipping partial interval (%s, %s) for snapshot %s",
+                start_ts,
+                end_ts,
+                snapshot_id,
+            )
+    return results

sqlmesh/core/state_sync/db/interval.py

@@ -21,12 +21,11 @@
     SnapshotNameVersion,
     SnapshotInfoLike,
     Snapshot,
-    SnapshotId,
 )
 from sqlmesh.core.snapshot.definition import Interval
 from sqlmesh.utils.migration import index_text_type
 from sqlmesh.utils import random_id
-from sqlmesh.utils.date import now_timestamp, time_like_to_str
+from sqlmesh.utils.date import now_timestamp
 
 
 logger = logging.getLogger(__name__)
@@ -62,52 +61,8 @@ def __init__(
         }
 
     def add_snapshots_intervals(self, snapshots_intervals: t.Sequence[SnapshotIntervals]) -> None:
-        def remove_partial_intervals(
-            intervals: t.List[Interval], snapshot_id: t.Optional[SnapshotId], *, is_dev: bool
-        ) -> t.List[Interval]:
-            results = []
-            for start_ts, end_ts in intervals:
-                if start_ts < end_ts:
-                    logger.info(
-                        "Adding %s (%s, %s) for snapshot %s",
-                        "dev interval" if is_dev else "interval",
-                        time_like_to_str(start_ts),
-                        time_like_to_str(end_ts),
-                        snapshot_id,
-                    )
-                    results.append((start_ts, end_ts))
-                else:
-                    logger.info(
-                        "Skipping partial interval (%s, %s) for snapshot %s",
-                        start_ts,
-                        end_ts,
-                        snapshot_id,
-                    )
-            return results
-
-        intervals_to_insert = []
-        for snapshot_intervals in snapshots_intervals:
-            snapshot_intervals = snapshot_intervals.copy(
-                update={
-                    "intervals": remove_partial_intervals(
-                        snapshot_intervals.intervals, snapshot_intervals.snapshot_id, is_dev=False
-                    ),
-                    "dev_intervals": remove_partial_intervals(
-                        snapshot_intervals.dev_intervals,
-                        snapshot_intervals.snapshot_id,
-                        is_dev=True,
-                    ),
-                }
-            )
-            if (
-                snapshot_intervals.intervals
-                or snapshot_intervals.dev_intervals
-                or snapshot_intervals.pending_restatement_intervals
-            ):
-                intervals_to_insert.append(snapshot_intervals)
-
-        if intervals_to_insert:
-            self._push_snapshot_intervals(intervals_to_insert)
+        if snapshots_intervals:
+            self._push_snapshot_intervals(snapshots_intervals)
 
     def remove_intervals(
         self,