Repository: TobikoData/sqlmesh
Pull request #3005 (2024-08-13 21:03:10)

Chore!: Improve the data model and performance of the state sync
This update contains the following performance improvements to the state sync:
- Add dev intervals in bulk for new forward-only snapshots. Previously they were added one interval at a time.
- Update the `updated_ts` property using the `UPDATE` statement for all impacted snapshots at once when demoting snapshots from an environment.
- Only fetch the `plan_id` attribute and only update the `finalized_ts` attribute when finalizing an environment. Previously we serialized / deserialized the full environment object to perform this operation.
- Use the `UPDATE` statement to update the `unpaused_ts` attribute for all impacted snapshots at once when unpausing.


Patch:
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index 3d9b148e..cd2bb279 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -1760,7 +1760,7 @@ class EngineAdapter:
         query: t.Union[exp.Expression, str],
         ignore_unsupported_errors: bool = False,
         quote_identifiers: bool = False,
-    ) -> t.Tuple:
+    ) -> t.Optional[t.Tuple]:
         with self.transaction():
             self.execute(
                 query,
diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py
index 9ef82274..0d8d0ecd 100644
--- a/sqlmesh/core/engine_adapter/bigquery.py
+++ b/sqlmesh/core/engine_adapter/bigquery.py
@@ -233,7 +233,7 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
         query: t.Union[exp.Expression, str],
         ignore_unsupported_errors: bool = False,
         quote_identifiers: bool = False,
-    ) -> t.Tuple:
+    ) -> t.Optional[t.Tuple]:
         """
         BigQuery's `fetchone` method doesn't call execute and therefore would not benefit from the execute
         configuration we have in place. Therefore this implementation calls execute instead.
@@ -246,7 +246,7 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
         try:
             return next(self._query_data)
         except StopIteration:
-            return ()
+            return None
 
     def fetchall(
         self,
diff --git a/sqlmesh/core/engine_adapter/mysql.py b/sqlmesh/core/engine_adapter/mysql.py
index b885f1d6..ccedb784 100644
--- a/sqlmesh/core/engine_adapter/mysql.py
+++ b/sqlmesh/core/engine_adapter/mysql.py
@@ -137,7 +137,8 @@ class MySQLEngineAdapter(
         # self.columns() only returns the column types so doesn't allow us to fully/correctly replace a column definition.
         # To get the full column definition we retrieve and parse the table's CREATE TABLE statement.
         create_table_exp = parse_one(
-            self.fetchone(f"SHOW CREATE TABLE {table_sql}")[1], dialect=self.dialect
+            self.fetchone(f"SHOW CREATE TABLE {table_sql}")[1],  # type: ignore
+            dialect=self.dialect,
         )
         col_def_exps = {
             col_def.name: col_def.copy()
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index 55afe89e..26ce82d8 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -92,7 +92,7 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin):
 
     @property
     def _current_warehouse(self) -> exp.Identifier:
-        current_warehouse_str = self.fetchone("SELECT CURRENT_WAREHOUSE()")[0]
+        current_warehouse_str = self.fetchone("SELECT CURRENT_WAREHOUSE()")[0]  # type: ignore
         # The warehouse value returned by Snowflake is already normalized, so only quoting is needed.
         return quote_identifiers(exp.to_identifier(current_warehouse_str), dialect=self.dialect)
 
diff --git a/sqlmesh/core/engine_adapter/spark.py b/sqlmesh/core/engine_adapter/spark.py
index cf04f033..d1f752b2 100644
--- a/sqlmesh/core/engine_adapter/spark.py
+++ b/sqlmesh/core/engine_adapter/spark.py
@@ -356,7 +356,7 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
     def get_current_database(self) -> str:
         if self._use_spark_session:
             return self.spark.catalog.currentDatabase()
-        return self.fetchone(exp.select(exp.func("current_database")))[0]
+        return self.fetchone(exp.select(exp.func("current_database")))[0]  # type: ignore
 
     def create_state_table(
         self,
diff --git a/sqlmesh/core/engine_adapter/trino.py b/sqlmesh/core/engine_adapter/trino.py
index 47619a34..986b5597 100644
--- a/sqlmesh/core/engine_adapter/trino.py
+++ b/sqlmesh/core/engine_adapter/trino.py
@@ -75,8 +75,11 @@ class TrinoEngineAdapter(
     def get_catalog_type(self, catalog: t.Optional[str]) -> str:
         row: t.Tuple = tuple()
         if catalog:
-            row = self.fetchone(
-                f"select connector_name from system.metadata.catalogs where catalog_name='{catalog}'"
+            row = (
+                self.fetchone(
+                    f"select connector_name from system.metadata.catalogs where catalog_name='{catalog}'"
+                )
+                or ()
             )
         return seq_get(row, 0) or self.DEFAULT_CATALOG_TYPE
 
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 5d408154..ceb37c2d 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -27,7 +27,7 @@ from sqlmesh.core.notification_target import (
 )
 from sqlmesh.core.plan.definition import Plan
 from sqlmesh.core.scheduler import Scheduler, SignalFactory
-from sqlmesh.core.snapshot import DeployabilityIndex, Snapshot, SnapshotEvaluator
+from sqlmesh.core.snapshot import DeployabilityIndex, Snapshot, SnapshotEvaluator, SnapshotIntervals
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.core.state_sync.base import PromotionResult
 from sqlmesh.core.user import User
@@ -296,7 +296,7 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         if not plan.restatements:
             return
 
-        self.state_sync.remove_interval(
+        self.state_sync.remove_intervals(
             [
                 (plan.context_diff.snapshots[s_id], interval)
                 for s_id, interval in plan.restatements.items()
@@ -526,8 +526,13 @@ class MWAAPlanEvaluator(StateBasedAirflowPlanEvaluator):
 def update_intervals_for_new_snapshots(
     snapshots: t.Collection[Snapshot], state_sync: StateSync
 ) -> None:
+    snapshots_intervals: t.List[SnapshotIntervals] = []
     for snapshot in state_sync.refresh_snapshot_intervals(snapshots):
         if snapshot.is_forward_only:
             snapshot.dev_intervals = snapshot.intervals.copy()
-            for start, end in snapshot.dev_intervals:
-                state_sync.add_interval(snapshot, start, end, is_dev=True)
+            snapshot_intervals = snapshot.snapshot_intervals
+            snapshot_intervals.intervals.clear()
+            snapshots_intervals.append(snapshot_intervals)
+
+    if snapshots_intervals:
+        state_sync.add_snapshots_intervals(snapshots_intervals)
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 6cbda5ed..048c2975 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1185,7 +1185,15 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
 
     @property
     def expiration_ts(self) -> int:
-        return to_timestamp(self.ttl, relative_base=to_datetime(self.updated_ts))
+        return to_timestamp(
+            self.ttl,
+            relative_base=to_datetime(self.updated_ts),
+            check_categorical_relative_expression=False,
+        )
+
+    @property
+    def ttl_ms(self) -> int:
+        return self.expiration_ts - self.updated_ts
 
     @property
     def custom_materialization(self) -> t.Optional[str]:
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 7426095d..7a4c0cbc 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -856,7 +856,7 @@ class SnapshotEvaluator:
         count, *_ = self.adapter.fetchone(
             select("COUNT(*)").from_(query.subquery("audit")),
             quote_identifiers=True,
-        )
+        )  # type: ignore
         if count and raise_exception:
             audit_error = AuditError(
                 audit_name=audit.name,
diff --git a/sqlmesh/core/state_sync/__init__.py b/sqlmesh/core/state_sync/__init__.py
index 8b78e574..4626de3c 100644
--- a/sqlmesh/core/state_sync/__init__.py
+++ b/sqlmesh/core/state_sync/__init__.py
@@ -20,8 +20,5 @@ from sqlmesh.core.state_sync.base import (
     Versions as Versions,
 )
 from sqlmesh.core.state_sync.cache import CachingStateSync as CachingStateSync
-from sqlmesh.core.state_sync.common import (
-    CommonStateSyncMixin as CommonStateSyncMixin,
-    cleanup_expired_views as cleanup_expired_views,
-)
+from sqlmesh.core.state_sync.common import cleanup_expired_views as cleanup_expired_views
 from sqlmesh.core.state_sync.engine_adapter import EngineAdapterStateSync as EngineAdapterStateSync
diff --git a/sqlmesh/core/state_sync/base.py b/sqlmesh/core/state_sync/base.py
index 08724d27..00e1aac2 100644
--- a/sqlmesh/core/state_sync/base.py
+++ b/sqlmesh/core/state_sync/base.py
@@ -312,7 +312,7 @@ class StateSync(StateReader, abc.ABC):
         """
 
     @abc.abstractmethod
-    def remove_interval(
+    def remove_intervals(
         self,
         snapshot_intervals: t.Sequence[t.Tuple[SnapshotInfoLike, Interval]],
         remove_shared_versions: bool = False,
@@ -323,10 +323,8 @@ class StateSync(StateReader, abc.ABC):
         can also grab all snapshots tied to the passed in version.
 
         Args:
-            snapshots: The snapshot info like object to remove intervals from.
-            start: The start of the interval to add.
-            end: The end of the interval to add.
-            all_snapshots: All snapshots can be passed in to skip fetching matching snapshot versions.
+            snapshot_intervals: The snapshot intervals to remove.
+            remove_shared_versions: Whether to remove intervals for snapshots that share the same version with the target snapshots.
         """
 
     @abc.abstractmethod
@@ -417,11 +415,11 @@ class StateSync(StateReader, abc.ABC):
         """Rollback to previous backed up state."""
 
     @abc.abstractmethod
-    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
+    def add_snapshots_intervals(self, snapshots_intervals: t.Sequence[SnapshotIntervals]) -> None:
         """Add snapshot intervals to state
 
         Args:
-            snapshot_intervals: The snapshot intervals to add.
+            snapshots_intervals: The intervals to add.
         """
 
     def add_interval(
@@ -450,7 +448,7 @@ class StateSync(StateReader, abc.ABC):
             intervals=intervals if not is_dev else [],
             dev_intervals=intervals if is_dev else [],
         )
-        self._add_snapshot_intervals(snapshot_intervals)
+        self.add_snapshots_intervals([snapshot_intervals])
 
 
 class DelegatingStateSync(StateSync):
diff --git a/sqlmesh/core/state_sync/cache.py b/sqlmesh/core/state_sync/cache.py
index bf59eeed..153af117 100644
--- a/sqlmesh/core/state_sync/cache.py
+++ b/sqlmesh/core/state_sync/cache.py
@@ -122,18 +122,19 @@ class CachingStateSync(DelegatingStateSync):
         self.snapshot_cache.clear()
         return self.state_sync.delete_expired_snapshots(ignore_ttl=ignore_ttl)
 
-    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
-        self.snapshot_cache.pop(snapshot_intervals.snapshot_id, None)
-        self.state_sync._add_snapshot_intervals(snapshot_intervals)
+    def add_snapshots_intervals(self, snapshots_intervals: t.Sequence[SnapshotIntervals]) -> None:
+        for snapshot_intervals in snapshots_intervals:
+            self.snapshot_cache.pop(snapshot_intervals.snapshot_id, None)
+        self.state_sync.add_snapshots_intervals(snapshots_intervals)
 
-    def remove_interval(
+    def remove_intervals(
         self,
         snapshot_intervals: t.Sequence[t.Tuple[SnapshotInfoLike, Interval]],
         remove_shared_versions: bool = False,
     ) -> None:
         for s, _ in snapshot_intervals:
             self.snapshot_cache.pop(s.snapshot_id, None)
-        self.state_sync.remove_interval(snapshot_intervals, remove_shared_versions)
+        self.state_sync.remove_intervals(snapshot_intervals, remove_shared_versions)
 
     def unpause_snapshots(
         self, snapshots: t.Collection[SnapshotInfoLike], unpaused_dt: TimeLike
diff --git a/sqlmesh/core/state_sync/common.py b/sqlmesh/core/state_sync/common.py
index 12da604d..6a18a387 100644
--- a/sqlmesh/core/state_sync/common.py
+++ b/sqlmesh/core/state_sync/common.py
@@ -1,26 +1,12 @@
 from __future__ import annotations
 
-import abc
 import logging
 import typing as t
-from datetime import datetime
 from functools import wraps
 
 from sqlmesh.core.console import Console
 from sqlmesh.core.dialect import schema_
 from sqlmesh.core.environment import Environment
-from sqlmesh.core.snapshot import (
-    Snapshot,
-    SnapshotId,
-    SnapshotIdLike,
-    SnapshotInfoLike,
-    SnapshotNameVersionLike,
-    SnapshotTableInfo,
-    start_date,
-)
-from sqlmesh.core.state_sync.base import PromotionResult, StateSync
-from sqlmesh.utils.date import TimeLike, now, now_timestamp, to_timestamp
-from sqlmesh.utils.errors import SQLMeshError
 
 if t.TYPE_CHECKING:
     from sqlmesh.core.engine_adapter.base import EngineAdapter
@@ -88,285 +74,3 @@ def transactional() -> t.Callable[[t.Callable], t.Callable]:
         return wrapper
 
     return decorator
-
-
-class CommonStateSyncMixin(StateSync):
-    def get_snapshots(
-        self,
-        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
-    ) -> t.Dict[SnapshotId, Snapshot]:
-        return self._get_snapshots(snapshot_ids)
-
-    def get_environment(self, environment: str) -> t.Optional[Environment]:
-        return self._get_environment(environment)
-
-    @transactional()
-    def promote(
-        self,
-        environment: Environment,
-        no_gaps_snapshot_names: t.Optional[t.Set[str]] = None,
-    ) -> PromotionResult:
-        """Update the environment to reflect the current state.
-
-        This method verifies that snapshots have been pushed.
-
-        Args:
-            environment: The environment to promote.
-            no_gaps_snapshot_names: A set of snapshot names to check for data gaps. If None,
-                all snapshots will be checked. The data gap check ensures that models that are already a
-                part of the target environment have no data gaps when compared against previous
-                snapshots for same models.
-
-        Returns:
-           A tuple of (added snapshot table infos, removed snapshot table infos, and environment target suffix for the removed table infos)
-        """
-        logger.info("Promoting environment '%s'", environment.name)
-
-        missing = {s.snapshot_id for s in environment.snapshots} - self.snapshots_exist(
-            environment.snapshots
-        )
-        if missing:
-            raise SQLMeshError(
-                f"Missing snapshots {missing}. Make sure to push and backfill your snapshots."
-            )
-
-        existing_environment = self._get_environment(environment.name, lock_for_update=True)
-
-        existing_table_infos = (
-            {table_info.name: table_info for table_info in existing_environment.promoted_snapshots}
-            if existing_environment
-            else {}
-        )
-        table_infos = {table_info.name: table_info for table_info in environment.promoted_snapshots}
-        views_that_changed_location: t.Set[SnapshotTableInfo] = set()
-        if existing_environment:
-            views_that_changed_location = {
-                existing_table_info
-                for name, existing_table_info in existing_table_infos.items()
-                if name in table_infos
-                and existing_table_info.qualified_view_name.for_environment(
-                    existing_environment.naming_info
-                )
-                != table_infos[name].qualified_view_name.for_environment(environment.naming_info)
-            }
-            if environment.previous_plan_id != existing_environment.plan_id:
-                raise SQLMeshError(
-                    f"Plan '{environment.plan_id}' is no longer valid for the target environment '{environment.name}'. "
-                    f"Expected previous plan ID: '{environment.previous_plan_id}', actual previous plan ID: '{existing_environment.plan_id}'. "
-                    "Please recreate the plan and try again"
-                )
-            if no_gaps_snapshot_names != set():
-                snapshots = self._get_snapshots(environment.snapshots).values()
-                self._ensure_no_gaps(
-                    snapshots,
-                    existing_environment,
-                    no_gaps_snapshot_names,
-                )
-            demoted_snapshots = set(existing_environment.snapshots) - set(environment.snapshots)
-            for demoted_snapshot in self._get_snapshots(demoted_snapshots).values():
-                # Update the updated_at attribute.
-                self._update_snapshot(demoted_snapshot)
-
-        missing_models = set(existing_table_infos) - {
-            snapshot.name for snapshot in environment.promoted_snapshots
-        }
-
-        added_table_infos = set(table_infos.values())
-        if existing_environment and existing_environment.finalized_ts:
-            # Only promote new snapshots.
-            added_table_infos -= set(existing_environment.promoted_snapshots)
-
-        self._update_environment(environment)
-
-        removed = {existing_table_infos[name] for name in missing_models}.union(
-            views_that_changed_location
-        )
-
-        return PromotionResult(
-            added=sorted(added_table_infos),
-            removed=list(removed),
-            removed_environment_naming_info=(
-                existing_environment.naming_info if removed and existing_environment else None
-            ),
-        )
-
-    @transactional()
-    def finalize(self, environment: Environment) -> None:
-        """Finalize the target environment, indicating that this environment has been
-        fully promoted and is ready for use.
-
-        Args:
-            environment: The target environment to finalize.
-        """
-        logger.info("Finalizing environment '%s'", environment.name)
-
-        stored_environment = self._get_environment(environment.name, lock_for_update=True)
-        if stored_environment and stored_environment.plan_id != environment.plan_id:
-            raise SQLMeshError(
-                f"Plan '{environment.plan_id}' is no longer valid for the target environment '{environment.name}'. "
-                f"Stored plan ID: '{stored_environment.plan_id}'. Please recreate the plan and try again"
-            )
-
-        environment.finalized_ts = now_timestamp()
-        self._update_environment(environment)
-
-    @transactional()
-    def unpause_snapshots(
-        self, snapshots: t.Collection[SnapshotInfoLike], unpaused_dt: TimeLike
-    ) -> None:
-        current_ts = now()
-
-        target_snapshot_ids = {s.snapshot_id for s in snapshots}
-        snapshots = self._get_snapshots_with_same_version(snapshots, lock_for_update=True)
-        target_snapshots_by_version = {
-            (s.name, s.version): s for s in snapshots if s.snapshot_id in target_snapshot_ids
-        }
-
-        for snapshot in snapshots:
-            is_target_snapshot = snapshot.snapshot_id in target_snapshot_ids
-            if is_target_snapshot and not snapshot.unpaused_ts:
-                logger.info("Unpausing snapshot %s", snapshot.snapshot_id)
-                snapshot.set_unpaused_ts(unpaused_dt)
-                self._update_snapshot(snapshot)
-            elif not is_target_snapshot:
-                target_snapshot = target_snapshots_by_version[(snapshot.name, snapshot.version)]
-                if target_snapshot.normalized_effective_from_ts:
-                    # Making sure that there are no overlapping intervals.
-                    effective_from_ts = target_snapshot.normalized_effective_from_ts
-                    logger.info(
-                        "Removing all intervals after '%s' for snapshot %s, superseded by snapshot %s",
-                        target_snapshot.effective_from,
-                        snapshot.snapshot_id,
-                        target_snapshot.snapshot_id,
-                    )
-                    self.remove_interval(
-                        [(snapshot, snapshot.get_removal_interval(effective_from_ts, current_ts))]
-                    )
-
-                update_required = False
-
-                if snapshot.unpaused_ts:
-                    logger.info("Pausing snapshot %s", snapshot.snapshot_id)
-                    snapshot.set_unpaused_ts(None)
-                    update_required = True
-
-                if (
-                    not snapshot.is_forward_only
-                    and target_snapshot.is_forward_only
-                    and not snapshot.unrestorable
-                ):
-                    logger.info("Marking snapshot %s as unrestorable", snapshot.snapshot_id)
-                    snapshot.unrestorable = True
-                    update_required = True
-
-                if update_required:
-                    self._update_snapshot(snapshot)
-
-    def _ensure_no_gaps(
-        self,
-        target_snapshots: t.Iterable[Snapshot],
-        target_environment: Environment,
-        snapshot_names: t.Optional[t.Set[str]],
-    ) -> None:
-        target_snapshots_by_name = {s.name: s for s in target_snapshots}
-
-        changed_version_prev_snapshots_by_name = {
-            s.name: s
-            for s in target_environment.snapshots
-            if s.name in target_snapshots_by_name
-            and target_snapshots_by_name[s.name].version != s.version
-        }
-
-        prev_snapshots = self._get_snapshots(
-            changed_version_prev_snapshots_by_name.values()
-        ).values()
-        cache: t.Dict[str, datetime] = {}
-
-        for prev_snapshot in prev_snapshots:
-            target_snapshot = target_snapshots_by_name[prev_snapshot.name]
-            if (
-                (snapshot_names is None or prev_snapshot.name in snapshot_names)
-                and target_snapshot.is_incremental
-                and prev_snapshot.is_incremental
-                and prev_snapshot.intervals
-            ):
-                start = to_timestamp(
-                    start_date(target_snapshot, target_snapshots_by_name.values(), cache)
-                )
-                end = prev_snapshot.intervals[-1][1]
-
-                if start < end:
-                    missing_intervals = target_snapshot.missing_intervals(
-                        start, end, end_bounded=True
-                    )
-
-                    if missing_intervals:
-                        raise SQLMeshError(
-                            f"Detected gaps in snapshot {target_snapshot.snapshot_id}: {missing_intervals}"
-                        )
-
-    @abc.abstractmethod
-    def _update_environment(self, environment: Environment) -> None:
-        """Overwrites the target environment with a given environment.
-
-        Args:
-            environment: The new environment.
-        """
-
-    @abc.abstractmethod
-    def _update_snapshot(self, snapshot: Snapshot) -> None:
-        """Updates the target snapshot.
-
-        Args:
-            snapshot: The target snapshot.
-        """
-
-    @abc.abstractmethod
-    def _get_snapshots(
-        self,
-        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]] = None,
-        lock_for_update: bool = False,
-        hydrate_intervals: bool = True,
-    ) -> t.Dict[SnapshotId, Snapshot]:
-        """Fetches specified snapshots.
-
-        Args:
-            snapshot_ids: The collection of IDs of snapshots to fetch
-            lock_for_update: Lock the snapshot rows for future update
-            hydrate_intervals: Whether to hydrate result snapshots with intervals.
-
-        Returns:
-            A dictionary of snapshot ids to snapshots for ones that could be found.
-        """
-
-    @abc.abstractmethod
-    def _get_snapshots_with_same_version(
-        self,
-        snapshots: t.Collection[SnapshotNameVersionLike],
-        lock_for_update: bool = False,
-    ) -> t.List[Snapshot]:
-        """Fetches all snapshots that share the same version as the snapshots.
-
-        The output includes the snapshots with the specified version.
-
-        Args:
-            snapshots: The collection of target name / version pairs.
-            lock_for_update: Lock the snapshot rows for future update
-
-        Returns:
-            The list of Snapshot objects.
-        """
-
-    @abc.abstractmethod
-    def _get_environment(
-        self, environment: str, lock_for_update: bool = False
-    ) -> t.Optional[Environment]:
-        """Fetches the environment if it exists.
-
-        Args:
-            environment: The target environment name.
-            lock_for_update: Lock the snapshot rows for future update
-
-        Returns:
-            The target environment.
-        """
diff --git a/sqlmesh/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index 4fcfeef9..4a92122e 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -24,6 +24,7 @@ import typing as t
 from collections import defaultdict
 from copy import deepcopy
 from pathlib import Path
+from datetime import datetime
 
 import pandas as pd
 from sqlglot import __version__ as SQLGLOT_VERSION
@@ -51,6 +52,7 @@ from sqlmesh.core.snapshot import (
     SnapshotTableCleanupTask,
     SnapshotTableInfo,
     fingerprint_from_node,
+    start_date,
 )
 from sqlmesh.core.snapshot.definition import (
     Interval,
@@ -58,11 +60,17 @@ from sqlmesh.core.snapshot.definition import (
     merge_intervals,
     remove_interval,
 )
-from sqlmesh.core.state_sync.base import MIGRATIONS, SCHEMA_VERSION, StateSync, Versions
-from sqlmesh.core.state_sync.common import CommonStateSyncMixin, transactional
+from sqlmesh.core.state_sync.base import (
+    MIGRATIONS,
+    SCHEMA_VERSION,
+    PromotionResult,
+    StateSync,
+    Versions,
+)
+from sqlmesh.core.state_sync.common import transactional
 from sqlmesh.utils import major_minor, random_id, unique
 from sqlmesh.utils.dag import DAG
-from sqlmesh.utils.date import TimeLike, now_timestamp, time_like_to_str
+from sqlmesh.utils.date import TimeLike, now, now_timestamp, time_like_to_str, to_timestamp
 from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils.pydantic import parse_obj_as
 
@@ -84,7 +92,7 @@ except ImportError:
     )
 
 
-class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
+class EngineAdapterStateSync(StateSync):
     """Manages state of nodes and snapshot with an existing engine adapter.
 
     This state sync is convenient to use because it requires no additional setup.
@@ -125,7 +133,10 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             "version": exp.DataType.build("text"),
             "snapshot": exp.DataType.build("text"),
             "kind_name": exp.DataType.build("text"),
-            "expiration_ts": exp.DataType.build("bigint"),
+            "updated_ts": exp.DataType.build("bigint"),
+            "unpaused_ts": exp.DataType.build("bigint"),
+            "ttl_ms": exp.DataType.build("bigint"),
+            "unrestorable": exp.DataType.build("boolean"),
         }
 
         self._environment_columns_to_types = {
@@ -163,7 +174,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             "sqlmesh_version": exp.DataType.build("text"),
         }
 
-    def _fetchone(self, query: t.Union[exp.Expression, str]) -> t.Tuple:
+    def _fetchone(self, query: t.Union[exp.Expression, str]) -> t.Optional[t.Tuple]:
         return self.engine_adapter.fetchone(
             query, ignore_unsupported_errors=True, quote_identifiers=True
         )
@@ -221,6 +232,237 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             columns_to_types=self._snapshot_columns_to_types,
         )
 
+    @transactional()
+    def promote(
+        self,
+        environment: Environment,
+        no_gaps_snapshot_names: t.Optional[t.Set[str]] = None,
+    ) -> PromotionResult:
+        """Update the environment to reflect the current state.
+
+        This method verifies that snapshots have been pushed.
+
+        Args:
+            environment: The environment to promote.
+            no_gaps_snapshot_names: A set of snapshot names to check for data gaps. If None,
+                all snapshots will be checked. The data gap check ensures that models that are already a
+                part of the target environment have no data gaps when compared against previous
+                snapshots for same models.
+
+        Returns:
+           A tuple of (added snapshot table infos, removed snapshot table infos, and environment target suffix for the removed table infos)
+        """
+        logger.info("Promoting environment '%s'", environment.name)
+
+        missing = {s.snapshot_id for s in environment.snapshots} - self.snapshots_exist(
+            environment.snapshots
+        )
+        if missing:
+            raise SQLMeshError(
+                f"Missing snapshots {missing}. Make sure to push and backfill your snapshots."
+            )
+
+        existing_environment = self._get_environment(environment.name, lock_for_update=True)
+
+        existing_table_infos = (
+            {table_info.name: table_info for table_info in existing_environment.promoted_snapshots}
+            if existing_environment
+            else {}
+        )
+        table_infos = {table_info.name: table_info for table_info in environment.promoted_snapshots}
+        views_that_changed_location: t.Set[SnapshotTableInfo] = set()
+        if existing_environment:
+            views_that_changed_location = {
+                existing_table_info
+                for name, existing_table_info in existing_table_infos.items()
+                if name in table_infos
+                and existing_table_info.qualified_view_name.for_environment(
+                    existing_environment.naming_info
+                )
+                != table_infos[name].qualified_view_name.for_environment(environment.naming_info)
+            }
+            if environment.previous_plan_id != existing_environment.plan_id:
+                raise SQLMeshError(
+                    f"Plan '{environment.plan_id}' is no longer valid for the target environment '{environment.name}'. "
+                    f"Expected previous plan ID: '{environment.previous_plan_id}', actual previous plan ID: '{existing_environment.plan_id}'. "
+                    "Please recreate the plan and try again"
+                )
+            if no_gaps_snapshot_names != set():
+                snapshots = self._get_snapshots(environment.snapshots).values()
+                self._ensure_no_gaps(
+                    snapshots,
+                    existing_environment,
+                    no_gaps_snapshot_names,
+                )
+            demoted_snapshots = set(existing_environment.snapshots) - set(environment.snapshots)
+            # Update the updated_at attribute.
+            self._update_snapshots(demoted_snapshots)
+
+        missing_models = set(existing_table_infos) - {
+            snapshot.name for snapshot in environment.promoted_snapshots
+        }
+
+        added_table_infos = set(table_infos.values())
+        if existing_environment and existing_environment.finalized_ts:
+            # Only promote new snapshots.
+            added_table_infos -= set(existing_environment.promoted_snapshots)
+
+        self._update_environment(environment)
+
+        removed = {existing_table_infos[name] for name in missing_models}.union(
+            views_that_changed_location
+        )
+
+        return PromotionResult(
+            added=sorted(added_table_infos),
+            removed=list(removed),
+            removed_environment_naming_info=(
+                existing_environment.naming_info if removed and existing_environment else None
+            ),
+        )
+
+    def _ensure_no_gaps(
+        self,
+        target_snapshots: t.Iterable[Snapshot],
+        target_environment: Environment,
+        snapshot_names: t.Optional[t.Set[str]],
+    ) -> None:
+        target_snapshots_by_name = {s.name: s for s in target_snapshots}
+
+        changed_version_prev_snapshots_by_name = {
+            s.name: s
+            for s in target_environment.snapshots
+            if s.name in target_snapshots_by_name
+            and target_snapshots_by_name[s.name].version != s.version
+        }
+
+        prev_snapshots = self._get_snapshots(
+            changed_version_prev_snapshots_by_name.values()
+        ).values()
+        cache: t.Dict[str, datetime] = {}
+
+        for prev_snapshot in prev_snapshots:
+            target_snapshot = target_snapshots_by_name[prev_snapshot.name]
+            if (
+                (snapshot_names is None or prev_snapshot.name in snapshot_names)
+                and target_snapshot.is_incremental
+                and prev_snapshot.is_incremental
+                and prev_snapshot.intervals
+            ):
+                start = to_timestamp(
+                    start_date(target_snapshot, target_snapshots_by_name.values(), cache)
+                )
+                end = prev_snapshot.intervals[-1][1]
+
+                if start < end:
+                    missing_intervals = target_snapshot.missing_intervals(
+                        start, end, end_bounded=True
+                    )
+
+                    if missing_intervals:
+                        raise SQLMeshError(
+                            f"Detected gaps in snapshot {target_snapshot.snapshot_id}: {missing_intervals}"
+                        )
+
+    @transactional()
+    def finalize(self, environment: Environment) -> None:
+        """Finalize the target environment, indicating that this environment has been
+        fully promoted and is ready for use.
+
+        Args:
+            environment: The target environment to finalize.
+        """
+        logger.info("Finalizing environment '%s'", environment.name)
+
+        environment_filter = exp.column("name").eq(exp.Literal.string(environment.name))
+
+        stored_plan_id_query = (
+            exp.select("plan_id")
+            .from_(self.environments_table)
+            .where(environment_filter, copy=False)
+            .lock(copy=False)
+        )
+        stored_plan_id_row = self._fetchone(stored_plan_id_query)
+
+        if not stored_plan_id_row:
+            raise SQLMeshError(f"Missing environment '{environment.name}' can't be finalized")
+
+        stored_plan_id = stored_plan_id_row[0]
+        if stored_plan_id != environment.plan_id:
+            raise SQLMeshError(
+                f"Plan '{environment.plan_id}' is no longer valid for the target environment '{environment.name}'. "
+                f"Stored plan ID: '{stored_plan_id}'. Please recreate the plan and try again"
+            )
+
+        environment.finalized_ts = now_timestamp()
+        self.engine_adapter.update_table(
+            self.environments_table,
+            {"finalized_ts": environment.finalized_ts},
+            where=environment_filter,
+        )
+
+    @transactional()
+    def unpause_snapshots(
+        self, snapshots: t.Collection[SnapshotInfoLike], unpaused_dt: TimeLike
+    ) -> None:
+        current_ts = now()
+
+        target_snapshot_ids = {s.snapshot_id for s in snapshots}
+        snapshots = self._get_snapshots_with_same_version(snapshots, lock_for_update=True)
+        target_snapshots_by_version = {
+            (s.name, s.version): s for s in snapshots if s.snapshot_id in target_snapshot_ids
+        }
+
+        unpaused_snapshots: t.Dict[int, t.List[SnapshotId]] = defaultdict(list)
+        paused_snapshots: t.List[SnapshotId] = []
+        unrestorable_snapshots: t.List[SnapshotId] = []
+
+        for snapshot in snapshots:
+            is_target_snapshot = snapshot.snapshot_id in target_snapshot_ids
+            if is_target_snapshot and not snapshot.unpaused_ts:
+                logger.info("Unpausing snapshot %s", snapshot.snapshot_id)
+                snapshot.set_unpaused_ts(unpaused_dt)
+                assert snapshot.unpaused_ts is not None
+                unpaused_snapshots[snapshot.unpaused_ts].append(snapshot.snapshot_id)
+            elif not is_target_snapshot:
+                target_snapshot = target_snapshots_by_version[(snapshot.name, snapshot.version)]
+                if target_snapshot.normalized_effective_from_ts:
+                    # Making sure that there are no overlapping intervals.
+                    effective_from_ts = target_snapshot.normalized_effective_from_ts
+                    logger.info(
+                        "Removing all intervals after '%s' for snapshot %s, superseded by snapshot %s",
+                        target_snapshot.effective_from,
+                        snapshot.snapshot_id,
+                        target_snapshot.snapshot_id,
+                    )
+                    self.remove_intervals(
+                        [(snapshot, snapshot.get_removal_interval(effective_from_ts, current_ts))]
+                    )
+
+                if snapshot.unpaused_ts:
+                    logger.info("Pausing snapshot %s", snapshot.snapshot_id)
+                    snapshot.set_unpaused_ts(None)
+                    paused_snapshots.append(snapshot.snapshot_id)
+
+                if (
+                    not snapshot.is_forward_only
+                    and target_snapshot.is_forward_only
+                    and not snapshot.unrestorable
+                ):
+                    logger.info("Marking snapshot %s as unrestorable", snapshot.snapshot_id)
+                    snapshot.unrestorable = True
+                    unrestorable_snapshots.append(snapshot.snapshot_id)
+
+        if unpaused_snapshots:
+            for unpaused_ts, snapshot_ids in unpaused_snapshots.items():
+                self._update_snapshots(snapshot_ids, unpaused_ts=unpaused_ts)
+
+        if paused_snapshots:
+            self._update_snapshots(paused_snapshots, unpaused_ts=None)
+
+        if unrestorable_snapshots:
+            self._update_snapshots(unrestorable_snapshots, unrestorable=True)
+
     def _update_versions(
         self,
         schema_version: int = SCHEMA_VERSION,
@@ -265,7 +507,9 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         expired_query = exp.select("name", "identifier", "version").from_(self.snapshots_table)
 
         if not ignore_ttl:
-            expired_query = expired_query.where(exp.column("expiration_ts") <= current_ts)
+            expired_query = expired_query.where(
+                (exp.column("updated_ts") + exp.column("ttl_ms")) <= current_ts
+            )
 
         expired_candidates = {
             SnapshotId(name=name, identifier=identifier): SnapshotNameVersion(
@@ -393,15 +637,24 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             columns_to_types=self._environment_columns_to_types,
         )
 
-    def _update_snapshot(self, snapshot: Snapshot) -> None:
-        snapshot.updated_ts = now_timestamp()
-        for where in self._snapshot_id_filter([snapshot.snapshot_id]):
+    def _update_snapshots(
+        self,
+        snapshots: t.Iterable[SnapshotIdLike],
+        **kwargs: t.Any,
+    ) -> None:
+        properties = kwargs
+        properties["updated_ts"] = now_timestamp()
+
+        for where in self._snapshot_id_filter(snapshots):
             self.engine_adapter.update_table(
                 self.snapshots_table,
-                {"snapshot": _snapshot_to_json(snapshot), "expiration_ts": snapshot.expiration_ts},
+                properties,
                 where=where,
             )
 
+    def get_environment(self, environment: str) -> t.Optional[Environment]:
+        return self._get_environment(environment)
+
     def get_environments(self) -> t.List[Environment]:
         """Fetches all environments.
 
@@ -429,30 +682,11 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
             return query.lock(copy=False)
         return query
 
-    def _get_snapshots_expressions(
+    def get_snapshots(
         self,
-        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]] = None,
-        lock_for_update: bool = False,
-        batch_size: t.Optional[int] = None,
-    ) -> t.Iterator[exp.Expression]:
-        for where in (
-            [None]
-            if snapshot_ids is None
-            else self._snapshot_id_filter(snapshot_ids, alias="snapshots", batch_size=batch_size)
-        ):
-            query = (
-                exp.select(
-                    "snapshots.snapshot",
-                    "snapshots.name",
-                    "snapshots.identifier",
-                    "snapshots.version",
-                )
-                .from_(exp.to_table(self.snapshots_table).as_("snapshots"))
-                .where(where)
-            )
-            if lock_for_update:
-                query = query.lock(copy=False)
-            yield query
+        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]],
+    ) -> t.Dict[SnapshotId, Snapshot]:
+        return self._get_snapshots(snapshot_ids)
 
     def _get_snapshots(
         self,
@@ -475,12 +709,23 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         model_cache = ModelCache(self._context_path / c.CACHE)
 
         for query in self._get_snapshots_expressions(snapshot_ids, lock_for_update):
-            for serialized_snapshot, name, identifier, _ in self._fetchall(query):
+            for (
+                serialized_snapshot,
+                name,
+                identifier,
+                _,
+                updated_ts,
+                unpaused_ts,
+                unrestorable,
+            ) in self._fetchall(query):
                 snapshot = parse_snapshot(
                     model_cache,
                     serialized_snapshot=serialized_snapshot,
                     name=name,
                     identifier=identifier,
+                    updated_ts=updated_ts,
+                    unpaused_ts=unpaused_ts,
+                    unrestorable=unrestorable,
                 )
                 snapshot_id = snapshot.snapshot_id
                 if snapshot_id in snapshots:
@@ -502,6 +747,34 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
 
         return snapshots
 
+    def _get_snapshots_expressions(
+        self,
+        snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]] = None,
+        lock_for_update: bool = False,
+        batch_size: t.Optional[int] = None,
+    ) -> t.Iterator[exp.Expression]:
+        for where in (
+            [None]
+            if snapshot_ids is None
+            else self._snapshot_id_filter(snapshot_ids, alias="snapshots", batch_size=batch_size)
+        ):
+            query = (
+                exp.select(
+                    "snapshots.snapshot",
+                    "snapshots.name",
+                    "snapshots.identifier",
+                    "snapshots.version",
+                    "snapshots.updated_ts",
+                    "snapshots.unpaused_ts",
+                    "snapshots.unrestorable",
+                )
+                .from_(exp.to_table(self.snapshots_table).as_("snapshots"))
+                .where(where)
+            )
+            if lock_for_update:
+                query = query.lock(copy=False)
+            yield query
+
     def _get_snapshots_with_same_version(
         self,
         snapshots: t.Collection[SnapshotNameVersionLike],
@@ -525,7 +798,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
 
         for where in self._snapshot_name_version_filter(snapshots):
             query = (
-                exp.select("snapshot")
+                exp.select("snapshot", "updated_ts", "unpaused_ts", "unrestorable")
                 .from_(exp.to_table(self.snapshots_table).as_("snapshots"))
                 .where(where)
             )
@@ -534,7 +807,17 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
 
             snapshot_rows.extend(self._fetchall(query))
 
-        return [Snapshot(**json.loads(row[0])) for row in snapshot_rows]
+        return [
+            Snapshot(
+                **{
+                    **json.loads(snapshot),
+                    "updated_ts": updated_ts,
+                    "unpaused_ts": unpaused_ts,
+                    "unrestorable": unrestorable,
+                }
+            )
+            for snapshot, updated_ts, unpaused_ts, unrestorable in snapshot_rows
+        ]
 
     def _get_versions(self, lock_for_update: bool = False) -> Versions:
         no_version = Versions()
@@ -593,7 +876,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         super().add_interval(snapshot, start, end, is_dev)
 
     @transactional()
-    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
+    def add_snapshots_intervals(self, snapshots_intervals: t.Sequence[SnapshotIntervals]) -> None:
         def remove_partial_intervals(
             intervals: t.List[Interval], snapshot_id: SnapshotId, *, is_dev: bool
         ) -> t.List[Interval]:
@@ -613,30 +896,36 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                         "Skipping partial interval (%s, %s) for snapshot %s",
                         start_ts,
                         end_ts,
-                        snapshot_intervals.snapshot_id,
+                        snapshot_id,
                     )
             return results
 
-        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.intervals and not snapshot_intervals.dev_intervals:
-            return
-        self.engine_adapter.insert_append(
-            self.intervals_table,
-            _snapshot_interval_to_df(snapshot_intervals, is_removed=False),
-            columns_to_types=self._interval_columns_to_types,
-        )
+        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:
+                intervals_to_insert.append(snapshot_intervals)
+
+        if intervals_to_insert:
+            self.engine_adapter.insert_append(
+                self.intervals_table,
+                _snapshots_intervals_to_df(intervals_to_insert, is_removed=False),
+                columns_to_types=self._interval_columns_to_types,
+            )
 
     @transactional()
-    def remove_interval(
+    def remove_intervals(
         self,
         snapshot_intervals: t.Sequence[t.Tuple[SnapshotInfoLike, Interval]],
         remove_shared_versions: bool = False,
@@ -722,7 +1011,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                 .from_(exp.to_table(self.intervals_table).as_("intervals"))
                 .where(where, copy=False)
                 .where(exp.to_column("is_dev").not_(), copy=False),
-            )[0]
+            )[0]  # type: ignore
 
             if max_end is None:
                 max_end = end
@@ -770,7 +1059,7 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
                 max_end_subquery.subquery(alias="max_ends")
             )
 
-            end = self._fetchone(query)[0]
+            end = self._fetchone(query)[0]  # type: ignore
 
             if greatest_common_end is None:
                 greatest_common_end = end
@@ -1318,8 +1607,8 @@ def _intervals_to_df(
     )
 
 
-def _snapshot_interval_to_df(
-    snapshot_intervals: SnapshotIntervals,
+def _snapshots_intervals_to_df(
+    snapshots_intervals: t.Sequence[SnapshotIntervals],
     is_removed: bool = False,
 ) -> pd.DataFrame:
     return pd.DataFrame(
@@ -1331,6 +1620,7 @@ def _snapshot_interval_to_df(
                 is_dev=is_dev,
                 is_removed=is_removed,
             )
+            for snapshot_intervals in snapshots_intervals
             for is_dev in (False, True)
             for start_ts, end_ts in getattr(
                 snapshot_intervals, "dev_intervals" if is_dev else "intervals"
@@ -1370,7 +1660,10 @@ def _snapshots_to_df(snapshots: t.Iterable[Snapshot]) -> pd.DataFrame:
                 "version": snapshot.version,
                 "snapshot": _snapshot_to_json(snapshot),
                 "kind_name": snapshot.model_kind_name.value if snapshot.model_kind_name else None,
-                "expiration_ts": snapshot.expiration_ts,
+                "updated_ts": snapshot.updated_ts,
+                "unpaused_ts": snapshot.unpaused_ts,
+                "ttl_ms": snapshot.ttl_ms,
+                "unrestorable": snapshot.unrestorable,
             }
             for snapshot in snapshots
         ]
@@ -1421,7 +1714,9 @@ def _backup_table_name(table_name: TableName) -> exp.Table:
 
 
 def _snapshot_to_json(snapshot: Snapshot) -> str:
-    return snapshot.json(exclude={"intervals", "dev_intervals"})
+    return snapshot.json(
+        exclude={"intervals", "dev_intervals", "updated_ts", "unpaused_ts", "unrestorable"}
+    )
 
 
 def parse_snapshot(
@@ -1429,6 +1724,9 @@ def parse_snapshot(
     serialized_snapshot: str,
     name: str,
     identifier: str,
+    updated_ts: int,
+    unpaused_ts: t.Optional[int],
+    unrestorable: bool,
 ) -> Snapshot:
     payload = json.loads(serialized_snapshot)
 
@@ -1439,7 +1737,14 @@ def parse_snapshot(
     node._data_hash = payload["fingerprint"]["data_hash"]
     node._metadata_hash = payload["fingerprint"]["metadata_hash"]
     payload["node"] = node
-    snapshot = Snapshot(**payload)
+    snapshot = Snapshot(
+        **{
+            **payload,
+            "updated_ts": updated_ts,
+            "unpaused_ts": unpaused_ts,
+            "unrestorable": unrestorable,
+        }
+    )
 
     return snapshot
 
diff --git a/sqlmesh/migrations/v0055_add_updated_ts_unpaused_ts_ttl_ms_unrestorable_to_snapshot.py b/sqlmesh/migrations/v0055_add_updated_ts_unpaused_ts_ttl_ms_unrestorable_to_snapshot.py
new file mode 100644
index 00000000..b46895ca
--- /dev/null
+++ b/sqlmesh/migrations/v0055_add_updated_ts_unpaused_ts_ttl_ms_unrestorable_to_snapshot.py
@@ -0,0 +1,102 @@
+"""Add updated_ts, unpaused_ts, ttl_ms, and unrestorable columns to the snapshots table."""
+
+import json
+
+import pandas as pd
+from sqlglot import exp
+
+from sqlmesh.utils.date import to_datetime, to_timestamp
+from sqlmesh.utils.migration import index_text_type
+
+
+def migrate(state_sync, **kwargs):  # type: ignore
+    engine_adapter = state_sync.engine_adapter
+    schema = state_sync.schema
+    snapshots_table = "_snapshots"
+    if schema:
+        snapshots_table = f"{schema}.{snapshots_table}"
+
+    index_type = index_text_type(engine_adapter.dialect)
+
+    add_column_exps = [
+        exp.AlterTable(
+            this=exp.to_table(snapshots_table),
+            actions=[
+                exp.ColumnDef(
+                    this=exp.to_column(column_name),
+                    kind=exp.DataType.build("bigint"),
+                )
+            ],
+        )
+        for column_name in ["updated_ts", "unpaused_ts", "ttl_ms"]
+    ] + [
+        exp.AlterTable(
+            this=exp.to_table(snapshots_table),
+            actions=[
+                exp.ColumnDef(
+                    this=exp.to_column("unrestorable"),
+                    kind=exp.DataType.build("boolean"),
+                )
+            ],
+        )
+    ]
+    engine_adapter.execute(add_column_exps)
+
+    drop_column_exp = exp.AlterTable(
+        this=exp.to_table(snapshots_table),
+        actions=[exp.Drop(this=exp.to_column("expiration_ts"), kind="COLUMN")],
+    )
+    engine_adapter.execute(drop_column_exp)
+
+    new_snapshots = []
+
+    for name, identifier, version, snapshot, kind_name in engine_adapter.fetchall(
+        exp.select("name", "identifier", "version", "snapshot", "kind_name").from_(snapshots_table),
+        quote_identifiers=True,
+    ):
+        parsed_snapshot = json.loads(snapshot)
+        updated_ts = parsed_snapshot.pop("updated_ts")
+        unpaused_ts = parsed_snapshot.pop("unpaused_ts", None)
+        ttl_ms = max(
+            to_timestamp(
+                parsed_snapshot["ttl"],
+                relative_base=to_datetime(updated_ts),
+                check_categorical_relative_expression=False,
+            )
+            - updated_ts,
+            0,
+        )
+        unrestorable = parsed_snapshot.pop("unrestorable", False)
+
+        new_snapshots.append(
+            {
+                "name": name,
+                "identifier": identifier,
+                "version": version,
+                "snapshot": snapshot,
+                "kind_name": kind_name,
+                "updated_ts": updated_ts,
+                "unpaused_ts": unpaused_ts,
+                "ttl_ms": ttl_ms,
+                "unrestorable": unrestorable,
+            }
+        )
+
+    if new_snapshots:
+        engine_adapter.delete_from(snapshots_table, "TRUE")
+
+        engine_adapter.insert_append(
+            snapshots_table,
+            pd.DataFrame(new_snapshots),
+            columns_to_types={
+                "name": exp.DataType.build(index_type),
+                "identifier": exp.DataType.build(index_type),
+                "version": exp.DataType.build(index_type),
+                "snapshot": exp.DataType.build("text"),
+                "kind_name": exp.DataType.build(index_type),
+                "updated_ts": exp.DataType.build("bigint"),
+                "unpaused_ts": exp.DataType.build("bigint"),
+                "ttl_ms": exp.DataType.build("bigint"),
+                "unrestorable": exp.DataType.build("boolean"),
+            },
+        )
diff --git a/sqlmesh/schedulers/airflow/plan.py b/sqlmesh/schedulers/airflow/plan.py
index e5941f54..7d437b4e 100644
--- a/sqlmesh/schedulers/airflow/plan.py
+++ b/sqlmesh/schedulers/airflow/plan.py
@@ -113,7 +113,7 @@ def create_plan_dag_spec(
             for s in all_snapshots.values()
             if s.snapshot_id in restatements and s.snapshot_id not in new_snapshots
         ]
-        state_sync.remove_interval(
+        state_sync.remove_intervals(
             intervals_to_remove,
             remove_shared_versions=not request.is_dev,
         )
diff --git a/sqlmesh/schedulers/airflow/state_sync.py b/sqlmesh/schedulers/airflow/state_sync.py
index 1cb89cbd..9bb9cd7e 100644
--- a/sqlmesh/schedulers/airflow/state_sync.py
+++ b/sqlmesh/schedulers/airflow/state_sync.py
@@ -219,10 +219,10 @@ class HttpStateSync(StateSync):
         """
         raise NotImplementedError("Adding intervals is not supported by the Airflow state sync.")
 
-    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
+    def add_snapshots_intervals(self, snapshots_intervals: t.Sequence[SnapshotIntervals]) -> None:
         raise NotImplementedError("Adding intervals is not supported by the Airflow state sync.")
 
-    def remove_interval(
+    def remove_intervals(
         self,
         snapshot_intervals: t.Sequence[t.Tuple[SnapshotInfoLike, Interval]],
         remove_shared_versions: bool = False,
diff --git a/sqlmesh/utils/date.py b/sqlmesh/utils/date.py
index 4e75dff1..d9aefdb7 100644
--- a/sqlmesh/utils/date.py
+++ b/sqlmesh/utils/date.py
@@ -115,27 +115,44 @@ def yesterday_timestamp() -> int:
     return to_timestamp(yesterday())
 
 
-def to_timestamp(value: TimeLike, relative_base: t.Optional[datetime] = None) -> int:
+def to_timestamp(
+    value: TimeLike,
+    relative_base: t.Optional[datetime] = None,
+    check_categorical_relative_expression: bool = True,
+) -> int:
     """
     Converts a value into an epoch millis timestamp.
 
     Args:
         value: A variety of date formats. If value is a string, it must be in iso format.
         relative_base: The datetime to reference for time expressions that are using relative terms
+        check_categorical_relative_expression: If True, takes into account the relative expressions that are categorical.
 
     Returns:
         Epoch millis timestamp.
     """
-    return int(to_datetime(value, relative_base=relative_base).timestamp() * 1000)
+    return int(
+        to_datetime(
+            value,
+            relative_base=relative_base,
+            check_categorical_relative_expression=check_categorical_relative_expression,
+        ).timestamp()
+        * 1000
+    )
 
 
 @ttl_cache()
-def to_datetime(value: TimeLike, relative_base: t.Optional[datetime] = None) -> datetime:
+def to_datetime(
+    value: TimeLike,
+    relative_base: t.Optional[datetime] = None,
+    check_categorical_relative_expression: bool = True,
+) -> datetime:
     """Converts a value into a UTC datetime object.
 
     Args:
         value: A variety of date formats. If the value is number-like, it is assumed to be millisecond epochs.
-        relative_base: The datetime to reference for time expressions that are using relative terms
+        relative_base: The datetime to reference for time expressions that are using relative terms.
+        check_categorical_relative_expression: If True, takes into account the relative expressions that are categorical.
 
     Raises:
         ValueError if value cannot be converted to a datetime.
@@ -158,7 +175,9 @@ def to_datetime(value: TimeLike, relative_base: t.Optional[datetime] = None) ->
         if epoch is None:
             relative_base = relative_base or now()
             expression = str(value)
-            if is_catagorical_relative_expression(expression):
+            if check_categorical_relative_expression and is_categorical_relative_expression(
+                expression
+            ):
                 relative_base = relative_base.replace(hour=0, minute=0, second=0, microsecond=0)
             dt = dateparser.parse(expression, settings={"RELATIVE_BASE": relative_base})
         else:
@@ -315,7 +334,7 @@ def time_like_to_str(time_like: TimeLike) -> str:
     return to_ts(time_like)
 
 
-def is_catagorical_relative_expression(expression: str) -> bool:
+def is_categorical_relative_expression(expression: str) -> bool:
     if expression.strip().lower() in DAY_SHORTCUT_EXPRESSIONS:
         return True
     grain_kwargs = freshness_date_parser.get_kwargs(expression)


Test patch:
diff --git a/tests/core/test_plan_evaluator.py b/tests/core/test_plan_evaluator.py
index 6726139b..54dc8566 100644
--- a/tests/core/test_plan_evaluator.py
+++ b/tests/core/test_plan_evaluator.py
@@ -172,7 +172,7 @@ def test_update_intervals_for_new_snapshots(
         query=parse_one("SELECT 1::INT AS one"),
     )
     snapshot = make_snapshot(model)
-    snapshot.change_category = change_category
+    snapshot.categorize_as(change_category)
 
     snapshot.add_interval("2023-01-01", "2023-01-01")
 
@@ -185,12 +185,12 @@ def test_update_intervals_for_new_snapshots(
 
     if change_category == SnapshotChangeCategory.FORWARD_ONLY:
         assert snapshot.dev_intervals == [(to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))]
-        state_sync_mock.add_interval.assert_called_once_with(
-            snapshot, to_timestamp("2023-01-01"), to_timestamp("2023-01-02"), is_dev=True
-        )
+        expected_intervals = snapshot.snapshot_intervals
+        expected_intervals.intervals.clear()
+        state_sync_mock.add_snapshots_intervals.assert_called_once_with([expected_intervals])
     else:
         assert not snapshot.dev_intervals
-        state_sync_mock.add_interval.assert_not_called()
+        state_sync_mock.add_snapshots_intervals.assert_not_called()
 
 
 def test_state_based_airflow_evaluator_with_restatements(
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 7da0a95b..4dbefc9f 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -1906,3 +1906,14 @@ def test_custom_model_kind(make_snapshot):
 
     parsed_table_info = SnapshotTableInfo.parse_raw(table_info.json())
     assert parsed_table_info.custom_materialization == "MyCustomStrategy"
+
+
+def test_ttl_ms(make_snapshot):
+    snapshot = make_snapshot(
+        SqlModel(
+            name="test_model_name",
+            query=parse_one("SELECT 1"),
+        ),
+        ttl="in 1 week",
+    )
+    assert snapshot.ttl_ms == 604800000
diff --git a/tests/core/test_state_sync.py b/tests/core/test_state_sync.py
index d12c5072..19521502 100644
--- a/tests/core/test_state_sync.py
+++ b/tests/core/test_state_sync.py
@@ -314,14 +314,14 @@ def test_remove_interval(state_sync: EngineAdapterStateSync, make_snapshot: t.Ca
 
     num_of_removals = 4
     for _ in range(num_of_removals):
-        state_sync.remove_interval(
+        state_sync.remove_intervals(
             [(snapshot_a, snapshot_a.inclusive_exclusive("2020-01-15", "2020-01-17"))],
             remove_shared_versions=True,
         )
 
     remove_records_count = state_sync.engine_adapter.fetchone(
         "SELECT COUNT(*) FROM sqlmesh._intervals WHERE name = '\"a\"' AND version = 'a' AND is_removed"
-    )[0]
+    )[0]  # type: ignore
     assert (
         remove_records_count == num_of_removals * 4
     )  # (1 dev record + 1 prod record) * 2 snapshots
@@ -369,7 +369,7 @@ def test_remove_interval_missing_snapshot(
         (to_timestamp("2020-01-01"), to_timestamp("2020-01-31")),
     ]
 
-    state_sync.remove_interval(
+    state_sync.remove_intervals(
         [(snapshot_a, snapshot_a.inclusive_exclusive("2020-01-15", "2020-01-17"))],
         remove_shared_versions=True,
     )
@@ -464,11 +464,11 @@ def test_compact_intervals(
 
     state_sync.add_interval(snapshot, "2020-01-01", "2020-01-10")
     state_sync.add_interval(snapshot, "2020-01-11", "2020-01-15")
-    state_sync.remove_interval(
+    state_sync.remove_intervals(
         [(snapshot, snapshot.inclusive_exclusive("2020-01-05", "2020-01-12"))]
     )
     state_sync.add_interval(snapshot, "2020-01-12", "2020-01-16")
-    state_sync.remove_interval(
+    state_sync.remove_intervals(
         [(snapshot, snapshot.inclusive_exclusive("2020-01-14", "2020-01-16"))]
     )
 
@@ -1623,8 +1623,8 @@ def test_migrate(state_sync: EngineAdapterStateSync, mocker: MockerFixture, tmp_
 
     assert (
         state_sync.engine_adapter.fetchone(
-            "SELECT COUNT(*) FROM sqlmesh._snapshots WHERE expiration_ts IS NULL"
-        )[0]
+            "SELECT COUNT(*) FROM sqlmesh._snapshots WHERE ttl_ms IS NULL"
+        )[0]  # type: ignore
         == 0
     )
 
@@ -1923,7 +1923,7 @@ def test_cache(state_sync, make_snapshot, mocker):
         mock.assert_called()
 
     # clear the cache by removing intervals
-    cache.remove_interval([(snapshot, snapshot.inclusive_exclusive("2020-01-01", "2020-01-01"))])
+    cache.remove_intervals([(snapshot, snapshot.inclusive_exclusive("2020-01-01", "2020-01-01"))])
 
     # prime the cache
     assert cache.get_snapshots([snapshot.snapshot_id]) == {snapshot.snapshot_id: snapshot}
@@ -2273,6 +2273,9 @@ def test_snapshot_batching(state_sync, mocker, make_snapshot):
                 "a",
                 "1",
                 "1",
+                1,
+                1,
+                False,
             ],
             [
                 make_snapshot(
@@ -2281,6 +2284,9 @@ def test_snapshot_batching(state_sync, mocker, make_snapshot):
                 "a",
                 "2",
                 "2",
+                1,
+                1,
+                False,
             ],
         ],
         [
@@ -2291,6 +2297,9 @@ def test_snapshot_batching(state_sync, mocker, make_snapshot):
                 "a",
                 "3",
                 "3",
+                1,
+                1,
+                False,
             ],
         ],
     ]
diff --git a/tests/dbt/test_integration.py b/tests/dbt/test_integration.py
index a7838897..4c68f2f7 100644
--- a/tests/dbt/test_integration.py
+++ b/tests/dbt/test_integration.py
@@ -242,7 +242,7 @@ test_config = config"""
         return df
 
     def _get_duckdb_now(self, adapter: DuckDBEngineAdapter) -> datetime.datetime:
-        return adapter.fetchone("SELECT now()")[0]
+        return adapter.fetchone("SELECT now()")[0]  # type: ignore
 
     def _init_test(
         self,
diff --git a/tests/schedulers/airflow/test_plan.py b/tests/schedulers/airflow/test_plan.py
index 0714d86e..62deca9c 100644
--- a/tests/schedulers/airflow/test_plan.py
+++ b/tests/schedulers/airflow/test_plan.py
@@ -308,7 +308,7 @@ def test_restatement(
     state_sync_mock.get_environment.assert_called_once()
     state_sync_mock.refresh_snapshot_intervals.assert_called_once()
 
-    state_sync_mock.remove_interval.assert_called_once_with(
+    state_sync_mock.remove_intervals.assert_called_once_with(
         [(the_snapshot, (to_timestamp("2022-01-02"), to_timestamp("2022-01-04")))],
         remove_shared_versions=True,
     )
diff --git a/tests/utils/test_date.py b/tests/utils/test_date.py
index 0f1a9c42..fca5d565 100644
--- a/tests/utils/test_date.py
+++ b/tests/utils/test_date.py
@@ -9,7 +9,7 @@ from sqlmesh.utils.date import (
     UTC,
     TimeLike,
     date_dict,
-    is_catagorical_relative_expression,
+    is_categorical_relative_expression,
     make_inclusive,
     to_datetime,
     to_time_column,
@@ -120,7 +120,7 @@ def test_make_inclusive(start_in, end_in, start_out, end_out) -> None:
     ],
 )
 def test_is_catagorical_relative_expression(expression, result):
-    assert is_catagorical_relative_expression(expression) == result
+    assert is_categorical_relative_expression(expression) == result
 
 
 def test_to_ts():