Repository: TobikoData/sqlmesh
Pull request #3118 (2024-09-10 23:50:52)

Chore!: Introduce the evaluatable version of a plan object
This update introduces a light-weight serializable version of the plan object which is used specifically for evaluation. This allows us to keep a consistent interface across various variances of plan evaluation (eg. airflow / builtin).


Patch:
diff --git a/sqlmesh/core/analytics/collector.py b/sqlmesh/core/analytics/collector.py
index fdb2b667..bf9b3542 100644
--- a/sqlmesh/core/analytics/collector.py
+++ b/sqlmesh/core/analytics/collector.py
@@ -17,7 +17,7 @@ from sqlmesh.utils.yaml import load as yaml_load
 
 if t.TYPE_CHECKING:
     from sqlmesh.cicd.config import CICDBotConfig
-    from sqlmesh.core.plan import Plan
+    from sqlmesh.core.plan import EvaluatablePlan
     from sqlmesh.core.snapshot import Snapshot
 
 
@@ -146,7 +146,7 @@ class AnalyticsCollector:
     def on_plan_apply_start(
         self,
         *,
-        plan: Plan,
+        plan: EvaluatablePlan,
         engine_type: t.Optional[str],
         state_sync_type: t.Optional[str],
         scheduler_type: str,
@@ -172,11 +172,15 @@ class AnalyticsCollector:
                 "forward_only": plan.forward_only,
                 "ensure_finalized_snapshots": plan.ensure_finalized_snapshots,
                 "has_restatements": bool(plan.restatements),
-                "directly_modified_count": len(plan.directly_modified),
+                "directly_modified_count": len(plan.directly_modified_snapshots),
                 "indirectly_modified_count": len(
-                    {s_id for s_ids in plan.indirectly_modified.values() for s_id in s_ids}
+                    {
+                        s_id
+                        for s_ids in plan.indirectly_modified_snapshots.values()
+                        for s_id in s_ids
+                    }
                 ),
-                "environment_name_hash": _anonymize(plan.environment_naming_info.name),
+                "environment_name_hash": _anonymize(plan.environment.name),
             },
         )
 
diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py
index 5561f696..0aeb9ce8 100644
--- a/sqlmesh/core/console.py
+++ b/sqlmesh/core/console.py
@@ -40,7 +40,7 @@ if t.TYPE_CHECKING:
 
     from sqlglot.dialects.dialect import DialectType
     from sqlmesh.core.context_diff import ContextDiff
-    from sqlmesh.core.plan import Plan, PlanBuilder
+    from sqlmesh.core.plan import Plan, EvaluatablePlan, PlanBuilder
     from sqlmesh.core.table_diff import RowDiff, SchemaDiff
 
     LayoutWidget = t.TypeVar("LayoutWidget", bound=t.Union[widgets.VBox, widgets.HBox])
@@ -64,7 +64,7 @@ class Console(abc.ABC):
     INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD = 10
 
     @abc.abstractmethod
-    def start_plan_evaluation(self, plan: Plan) -> None:
+    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
         """Indicates that a new evaluation has begun."""
 
     @abc.abstractmethod
@@ -335,7 +335,7 @@ class TerminalConsole(Console):
     def _confirm(self, message: str, **kwargs: t.Any) -> bool:
         return Confirm.ask(message, console=self.console, **kwargs)
 
-    def start_plan_evaluation(self, plan: Plan) -> None:
+    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
         pass
 
     def stop_plan_evaluation(self) -> None:
@@ -1985,7 +1985,7 @@ class DebuggerTerminalConsole(TerminalConsole):
     def _write(self, msg: t.Any, *args: t.Any, **kwargs: t.Any) -> None:
         self.console.log(msg, *args, **kwargs)
 
-    def start_plan_evaluation(self, plan: Plan) -> None:
+    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
         self._write("Starting plan", plan.plan_id)
 
     def stop_plan_evaluation(self) -> None:
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 3f4fb43a..446c047c 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -1828,7 +1828,9 @@ class GenericContext(BaseContext, t.Generic[C]):
         return success
 
     def _apply(self, plan: Plan, circuit_breaker: t.Optional[t.Callable[[], bool]]) -> None:
-        self._scheduler.create_plan_evaluator(self).evaluate(plan, circuit_breaker=circuit_breaker)
+        self._scheduler.create_plan_evaluator(self).evaluate(
+            plan.to_evaluatable(), circuit_breaker=circuit_breaker
+        )
 
     @python_api_analytics
     def table_name(self, model_name: str, dev: bool) -> str:
diff --git a/sqlmesh/core/plan/__init__.py b/sqlmesh/core/plan/__init__.py
index 139979e2..c686e79d 100644
--- a/sqlmesh/core/plan/__init__.py
+++ b/sqlmesh/core/plan/__init__.py
@@ -1,6 +1,7 @@
 from sqlmesh.core.plan.builder import PlanBuilder as PlanBuilder
 from sqlmesh.core.plan.definition import (
     Plan as Plan,
+    EvaluatablePlan as EvaluatablePlan,
     PlanStatus as PlanStatus,
     SnapshotIntervals as SnapshotIntervals,
 )
diff --git a/sqlmesh/core/plan/definition.py b/sqlmesh/core/plan/definition.py
index 833acf0b..f1128838 100644
--- a/sqlmesh/core/plan/definition.py
+++ b/sqlmesh/core/plan/definition.py
@@ -230,11 +230,67 @@ class Plan(PydanticModel, frozen=True):
         """Returns True if a model with the given FQN should be backfilled as part of this plan."""
         return self.models_to_backfill is None or model_fqn in self.models_to_backfill
 
+    def to_evaluatable(self) -> EvaluatablePlan:
+        return EvaluatablePlan(
+            start=self.start,
+            end=self.end,
+            new_snapshots=self.new_snapshots,
+            environment=self.environment,
+            no_gaps=self.no_gaps,
+            skip_backfill=self.skip_backfill,
+            restatements={s.name: i for s, i in self.restatements.items()},
+            is_dev=self.is_dev,
+            allow_destructive_models=self.allow_destructive_models,
+            forward_only=self.forward_only,
+            end_bounded=self.end_bounded,
+            ensure_finalized_snapshots=self.ensure_finalized_snapshots,
+            directly_modified_snapshots=sorted(self.directly_modified),
+            indirectly_modified_snapshots={
+                s.name: sorted(snapshot_ids) for s, snapshot_ids in self.indirectly_modified.items()
+            },
+            removed_snapshots=sorted(self.context_diff.removed_snapshots),
+            requires_backfill=self.requires_backfill,
+            models_to_backfill=self.models_to_backfill,
+            interval_end_per_model=self.interval_end_per_model,
+            execution_time=self.execution_time,
+        )
+
     @cached_property
     def _earliest_interval_start(self) -> datetime:
         return earliest_interval_start(self.snapshots.values())
 
 
+class EvaluatablePlan(PydanticModel):
+    """A serializable version of a plan that can be evaluated."""
+
+    start: TimeLike
+    end: TimeLike
+    new_snapshots: t.List[Snapshot]
+    environment: Environment
+    no_gaps: bool
+    skip_backfill: bool
+    restatements: t.Dict[str, Interval]
+    is_dev: bool
+    allow_destructive_models: t.Set[str]
+    forward_only: bool
+    end_bounded: bool
+    ensure_finalized_snapshots: bool
+    directly_modified_snapshots: t.List[SnapshotId]
+    indirectly_modified_snapshots: t.Dict[str, t.List[SnapshotId]]
+    removed_snapshots: t.List[SnapshotId]
+    requires_backfill: bool
+    models_to_backfill: t.Optional[t.Set[str]] = None
+    interval_end_per_model: t.Optional[t.Dict[str, int]] = None
+    execution_time: t.Optional[TimeLike] = None
+
+    def is_selected_for_backfill(self, model_fqn: str) -> bool:
+        return self.models_to_backfill is None or model_fqn in self.models_to_backfill
+
+    @property
+    def plan_id(self) -> str:
+        return self.environment.plan_id
+
+
 class PlanStatus(str, Enum):
     STARTED = "started"
     FINISHED = "finished"
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 5cf57c1b..3d65fe4d 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -24,9 +24,15 @@ from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.notification_target import (
     NotificationTarget,
 )
-from sqlmesh.core.plan.definition import Plan
+from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.scheduler import Scheduler
-from sqlmesh.core.snapshot import DeployabilityIndex, Snapshot, SnapshotEvaluator, SnapshotIntervals
+from sqlmesh.core.snapshot import (
+    DeployabilityIndex,
+    Snapshot,
+    SnapshotEvaluator,
+    SnapshotIntervals,
+    SnapshotId,
+)
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.core.state_sync.base import PromotionResult
 from sqlmesh.core.user import User
@@ -41,7 +47,7 @@ logger = logging.getLogger(__name__)
 class PlanEvaluator(abc.ABC):
     @abc.abstractmethod
     def evaluate(
-        self, plan: Plan, circuit_breaker: t.Optional[t.Callable[[], bool]] = None
+        self, plan: EvaluatablePlan, circuit_breaker: t.Optional[t.Callable[[], bool]] = None
     ) -> None:
         """Evaluates a plan by pushing snapshots and backfilling data.
 
@@ -72,7 +78,7 @@ class BuiltInPlanEvaluator(PlanEvaluator):
 
     def evaluate(
         self,
-        plan: Plan,
+        plan: EvaluatablePlan,
         circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
     ) -> None:
         self.console.start_plan_evaluation(plan)
@@ -84,10 +90,13 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         )
 
         try:
-            snapshots = plan.snapshots
-            all_names = {
-                s.name for s in snapshots.values() if plan.is_selected_for_backfill(s.name)
-            }
+            new_snapshots = {s.snapshot_id: s for s in plan.new_snapshots}
+            stored_snapshots = self.state_sync.get_snapshots(plan.environment.snapshots)
+            snapshots = {**new_snapshots, **stored_snapshots}
+            snapshots_by_name = {s.name: s for s in snapshots.values()}
+
+            all_names = {name for name in snapshots_by_name if plan.is_selected_for_backfill(name)}
+
             deployability_index_for_evaluation = DeployabilityIndex.create(snapshots)
             deployability_index_for_creation = deployability_index_for_evaluation
             if plan.is_dev:
@@ -103,23 +112,27 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                 after_promote_snapshots = all_names - before_promote_snapshots
                 deployability_index_for_evaluation = DeployabilityIndex.all_deployable()
 
-            self._push(plan, deployability_index_for_creation)
+            self._push(plan, snapshots, deployability_index_for_creation)
             update_intervals_for_new_snapshots(plan.new_snapshots, self.state_sync)
-            self._restate(plan)
+            self._restate(plan, snapshots_by_name)
             self._backfill(
                 plan,
+                snapshots_by_name,
                 before_promote_snapshots,
                 deployability_index_for_evaluation,
                 circuit_breaker=circuit_breaker,
             )
-            promotion_result = self._promote(plan, before_promote_snapshots)
+            promotion_result = self._promote(plan, snapshots, before_promote_snapshots)
             self._backfill(
                 plan,
+                snapshots_by_name,
                 after_promote_snapshots,
                 deployability_index_for_evaluation,
                 circuit_breaker=circuit_breaker,
             )
-            self._update_views(plan, promotion_result, deployability_index_for_evaluation)
+            self._update_views(
+                plan, snapshots, promotion_result, deployability_index_for_evaluation
+            )
 
             if not plan.requires_backfill:
                 self.console.log_success("Virtual Update executed successfully")
@@ -133,7 +146,8 @@ class BuiltInPlanEvaluator(PlanEvaluator):
 
     def _backfill(
         self,
-        plan: Plan,
+        plan: EvaluatablePlan,
+        snapshots_by_name: t.Dict[str, Snapshot],
         selected_snapshots: t.Set[str],
         deployability_index: DeployabilityIndex,
         circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
@@ -147,14 +161,16 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         if not plan.requires_backfill or not selected_snapshots:
             return
 
-        snapshots = plan.snapshots
-        scheduler = self.create_scheduler(snapshots.values())
+        scheduler = self.create_scheduler(snapshots_by_name.values())
         is_run_successful = scheduler.run(
-            plan.environment_naming_info,
+            plan.environment,
             plan.start,
             plan.end,
             execution_time=plan.execution_time,
-            restatements=plan.restatements,
+            restatements={
+                snapshots_by_name[name].snapshot_id: interval
+                for name, interval in plan.restatements.items()
+            },
             selected_snapshots=selected_snapshots,
             deployability_index=deployability_index,
             circuit_breaker=circuit_breaker,
@@ -164,7 +180,12 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         if not is_run_successful:
             raise SQLMeshError("Plan application failed.")
 
-    def _push(self, plan: Plan, deployability_index: t.Optional[DeployabilityIndex] = None) -> None:
+    def _push(
+        self,
+        plan: EvaluatablePlan,
+        snapshots: t.Dict[SnapshotId, Snapshot],
+        deployability_index: t.Optional[DeployabilityIndex] = None,
+    ) -> None:
         """Push the snapshots to the state sync.
 
         As a part of plan pushing, snapshot tables are created.
@@ -175,21 +196,21 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         """
         snapshots_to_create = [
             s
-            for s in plan.snapshots.values()
+            for s in snapshots.values()
             if s.is_model and not s.is_symbolic and plan.is_selected_for_backfill(s.name)
         ]
         snapshots_to_create_count = len(snapshots_to_create)
 
         if snapshots_to_create_count > 0:
             self.console.start_creation_progress(
-                snapshots_to_create_count, plan.environment_naming_info, self.default_catalog
+                snapshots_to_create_count, plan.environment, self.default_catalog
             )
 
         completed = False
         try:
             self.snapshot_evaluator.create(
                 snapshots_to_create,
-                plan.snapshots,
+                snapshots,
                 allow_destructive_snapshots=plan.allow_destructive_models,
                 deployability_index=deployability_index,
                 on_complete=self.console.update_creation_progress,
@@ -205,7 +226,10 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         )
 
     def _promote(
-        self, plan: Plan, no_gaps_snapshot_names: t.Optional[t.Set[str]] = None
+        self,
+        plan: EvaluatablePlan,
+        snapshots: t.Dict[SnapshotId, Snapshot],
+        no_gaps_snapshot_names: t.Optional[t.Set[str]] = None,
     ) -> PromotionResult:
         """Promote a plan.
 
@@ -221,8 +245,8 @@ class BuiltInPlanEvaluator(PlanEvaluator):
 
         if not plan.is_dev:
             self.snapshot_evaluator.migrate(
-                [s for s in plan.snapshots.values() if s.is_paused],
-                plan.snapshots,
+                [s for s in snapshots.values() if s.is_paused],
+                snapshots,
                 plan.allow_destructive_models,
             )
             if not plan.ensure_finalized_snapshots:
@@ -235,7 +259,8 @@ class BuiltInPlanEvaluator(PlanEvaluator):
 
     def _update_views(
         self,
-        plan: Plan,
+        plan: EvaluatablePlan,
+        snapshots: t.Dict[SnapshotId, Snapshot],
         promotion_result: PromotionResult,
         deployability_index: t.Optional[DeployabilityIndex] = None,
     ) -> None:
@@ -263,7 +288,7 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         completed = False
         try:
             self.snapshot_evaluator.promote(
-                [plan.context_diff.snapshots[s.snapshot_id] for s in promotion_result.added],
+                [snapshots[s.snapshot_id] for s in promotion_result.added],
                 environment.naming_info,
                 deployability_index=deployability_index,
                 on_complete=lambda s: self.console.update_promotion_progress(s, True),
@@ -279,15 +304,12 @@ class BuiltInPlanEvaluator(PlanEvaluator):
         finally:
             self.console.stop_promotion_progress(success=completed)
 
-    def _restate(self, plan: Plan) -> None:
+    def _restate(self, plan: EvaluatablePlan, snapshots_by_name: t.Dict[str, Snapshot]) -> None:
         if not plan.restatements:
             return
 
         self.state_sync.remove_intervals(
-            [
-                (plan.context_diff.snapshots[s_id], interval)
-                for s_id, interval in plan.restatements.items()
-            ],
+            [(snapshots_by_name[name], interval) for name, interval in plan.restatements.items()],
             remove_shared_versions=not plan.is_dev,
         )
 
@@ -308,7 +330,7 @@ class BaseAirflowPlanEvaluator(PlanEvaluator):
         self.console = console or get_console()
 
     def evaluate(
-        self, plan: Plan, circuit_breaker: t.Optional[t.Callable[[], bool]] = None
+        self, plan: EvaluatablePlan, circuit_breaker: t.Optional[t.Callable[[], bool]] = None
     ) -> None:
         plan_request_id = plan.plan_id
         self._apply_plan(plan, plan_request_id)
@@ -322,7 +344,7 @@ class BaseAirflowPlanEvaluator(PlanEvaluator):
 
         if self.blocking:
             plan_application_dag_id = airflow_common.plan_application_dag_id(
-                plan.environment_naming_info.name, plan_request_id
+                plan.environment.name, plan_request_id
             )
 
             self.console.log_status_update(
@@ -354,7 +376,7 @@ class BaseAirflowPlanEvaluator(PlanEvaluator):
     def client(self) -> BaseAirflowClient:
         raise NotImplementedError
 
-    def _apply_plan(self, plan: Plan, plan_request_id: str) -> None:
+    def _apply_plan(self, plan: EvaluatablePlan, plan_request_id: str) -> None:
         raise NotImplementedError
 
 
@@ -364,34 +386,15 @@ class StateBasedAirflowPlanEvaluator(BaseAirflowPlanEvaluator):
     notification_targets: t.Optional[t.List[NotificationTarget]]
     users: t.Optional[t.List[User]]
 
-    def _apply_plan(self, plan: Plan, plan_request_id: str) -> None:
+    def _apply_plan(self, plan: EvaluatablePlan, plan_request_id: str) -> None:
         from sqlmesh.schedulers.airflow.plan import PlanDagState, create_plan_dag_spec
 
         plan_application_request = airflow_common.PlanApplicationRequest(
-            new_snapshots=plan.new_snapshots,
-            environment=plan.environment,
-            no_gaps=plan.no_gaps,
-            skip_backfill=plan.skip_backfill,
-            request_id=plan_request_id,
-            restatements={s.name: i for s, i in (plan.restatements or {}).items()},
+            plan=plan,
             notification_targets=self.notification_targets or [],
             backfill_concurrent_tasks=self.backfill_concurrent_tasks,
             ddl_concurrent_tasks=self.ddl_concurrent_tasks,
             users=self.users or [],
-            is_dev=plan.is_dev,
-            forward_only=plan.forward_only,
-            models_to_backfill=plan.models_to_backfill,
-            end_bounded=plan.end_bounded,
-            ensure_finalized_snapshots=plan.ensure_finalized_snapshots,
-            directly_modified_snapshots=list(plan.directly_modified),
-            indirectly_modified_snapshots={
-                change_source.name: list(snapshots)
-                for change_source, snapshots in plan.indirectly_modified.items()
-            },
-            removed_snapshots=list(plan.context_diff.removed_snapshots),
-            interval_end_per_model=plan.interval_end_per_model,
-            execution_time=plan.execution_time,
-            allow_destructive_snapshots=plan.allow_destructive_models,
         )
         plan_dag_spec = create_plan_dag_spec(plan_application_request, self.state_sync)
         PlanDagState.from_state_sync(self.state_sync).add_dag_spec(plan_dag_spec)
@@ -441,36 +444,17 @@ class AirflowPlanEvaluator(StateBasedAirflowPlanEvaluator):
             raise SQLMeshError("State Sync is not configured")
         return self._state_sync
 
-    def _apply_plan(self, plan: Plan, plan_request_id: str) -> None:
+    def _apply_plan(self, plan: EvaluatablePlan, plan_request_id: str) -> None:
         if self._state_sync is not None:
             super()._apply_plan(plan, plan_request_id)
             return
 
         self._airflow_client.apply_plan(
-            plan.new_snapshots,
-            plan.environment,
-            plan_request_id,
-            no_gaps=plan.no_gaps,
-            skip_backfill=plan.skip_backfill,
-            restatements=plan.restatements,
+            plan,
             notification_targets=self.notification_targets,
             backfill_concurrent_tasks=self.backfill_concurrent_tasks,
             ddl_concurrent_tasks=self.ddl_concurrent_tasks,
             users=self.users,
-            is_dev=plan.is_dev,
-            forward_only=plan.forward_only,
-            models_to_backfill=plan.models_to_backfill,
-            end_bounded=plan.end_bounded,
-            ensure_finalized_snapshots=plan.ensure_finalized_snapshots,
-            directly_modified_snapshots=list(plan.directly_modified),
-            indirectly_modified_snapshots={
-                change_source.name: list(snapshots)
-                for change_source, snapshots in plan.indirectly_modified.items()
-            },
-            removed_snapshots=list(plan.context_diff.removed_snapshots),
-            interval_end_per_model=plan.interval_end_per_model,
-            execution_time=plan.execution_time,
-            allow_destructive_snapshots=plan.allow_destructive_models,
         )
 
 
diff --git a/sqlmesh/schedulers/airflow/client.py b/sqlmesh/schedulers/airflow/client.py
index a51a8dd0..c616f715 100644
--- a/sqlmesh/schedulers/airflow/client.py
+++ b/sqlmesh/schedulers/airflow/client.py
@@ -9,12 +9,11 @@ import requests
 from sqlmesh.core.console import Console
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.notification_target import NotificationTarget
+from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.snapshot import Snapshot, SnapshotId
-from sqlmesh.core.snapshot.definition import Interval
 from sqlmesh.core.state_sync import Versions
 from sqlmesh.core.user import User
 from sqlmesh.schedulers.airflow import common, NO_DEFAULT_CATALOG
-from sqlmesh.utils.date import TimeLike
 from sqlmesh.utils.errors import (
     ApiServerError,
     NotFoundError,
@@ -182,50 +181,18 @@ class AirflowClient(BaseAirflowClient):
 
     def apply_plan(
         self,
-        new_snapshots: t.Iterable[Snapshot],
-        environment: Environment,
-        request_id: str,
-        no_gaps: bool = False,
-        skip_backfill: bool = False,
-        restatements: t.Optional[t.Dict[SnapshotId, Interval]] = None,
+        plan: EvaluatablePlan,
         notification_targets: t.Optional[t.List[NotificationTarget]] = None,
         backfill_concurrent_tasks: int = 1,
         ddl_concurrent_tasks: int = 1,
         users: t.Optional[t.List[User]] = None,
-        is_dev: bool = False,
-        allow_destructive_snapshots: t.Set[str] = set(),
-        forward_only: bool = False,
-        models_to_backfill: t.Optional[t.Set[str]] = None,
-        end_bounded: bool = False,
-        ensure_finalized_snapshots: bool = False,
-        directly_modified_snapshots: t.Optional[t.List[SnapshotId]] = None,
-        indirectly_modified_snapshots: t.Optional[t.Dict[str, t.List[SnapshotId]]] = None,
-        removed_snapshots: t.Optional[t.List[SnapshotId]] = None,
-        interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
-        execution_time: t.Optional[TimeLike] = None,
     ) -> None:
         request = common.PlanApplicationRequest(
-            new_snapshots=list(new_snapshots),
-            environment=environment,
-            no_gaps=no_gaps,
-            skip_backfill=skip_backfill,
-            request_id=request_id,
-            restatements={s.name: i for s, i in (restatements or {}).items()},
+            plan=plan,
             notification_targets=notification_targets or [],
             backfill_concurrent_tasks=backfill_concurrent_tasks,
             ddl_concurrent_tasks=ddl_concurrent_tasks,
             users=users or [],
-            is_dev=is_dev,
-            allow_destructive_snapshots=allow_destructive_snapshots,
-            forward_only=forward_only,
-            models_to_backfill=models_to_backfill,
-            end_bounded=end_bounded,
-            ensure_finalized_snapshots=ensure_finalized_snapshots,
-            directly_modified_snapshots=directly_modified_snapshots or [],
-            indirectly_modified_snapshots=indirectly_modified_snapshots or {},
-            removed_snapshots=removed_snapshots or [],
-            interval_end_per_model=interval_end_per_model,
-            execution_time=execution_time,
         )
         self._post(PLANS_PATH, request.json())
 
diff --git a/sqlmesh/schedulers/airflow/common.py b/sqlmesh/schedulers/airflow/common.py
index d360dfd9..3e781ae5 100644
--- a/sqlmesh/schedulers/airflow/common.py
+++ b/sqlmesh/schedulers/airflow/common.py
@@ -5,6 +5,7 @@ import typing as t
 from sqlmesh.core import constants as c
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.notification_target import NotificationTarget
+from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.scheduler import Interval
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
@@ -14,7 +15,6 @@ from sqlmesh.core.snapshot import (
     SnapshotIntervals,
     SnapshotTableInfo,
 )
-from sqlmesh.core.snapshot.definition import Interval as SnapshotInterval
 from sqlmesh.core.user import User
 from sqlmesh.utils import sanitize_name
 from sqlmesh.utils.date import TimeLike
@@ -37,30 +37,11 @@ SQLMESH_API_BASE_PATH: str = f"{c.SQLMESH}/api/v1"
 
 
 class PlanApplicationRequest(PydanticModel):
-    request_id: str
-    new_snapshots: t.List[Snapshot]
-    environment: Environment
-    no_gaps: bool
-    skip_backfill: bool
-    restatements: t.Dict[str, SnapshotInterval]
+    plan: EvaluatablePlan
     notification_targets: t.List[NotificationTarget]
     backfill_concurrent_tasks: int
     ddl_concurrent_tasks: int
     users: t.List[User]
-    is_dev: bool
-    allow_destructive_snapshots: t.Set[str] = set()
-    forward_only: bool
-    models_to_backfill: t.Optional[t.Set[str]] = None
-    end_bounded: bool
-    ensure_finalized_snapshots: bool
-    directly_modified_snapshots: t.List[SnapshotId]
-    indirectly_modified_snapshots: t.Dict[str, t.List[SnapshotId]]
-    removed_snapshots: t.List[SnapshotId]
-    interval_end_per_model: t.Optional[t.Dict[str, int]] = None
-    execution_time: t.Optional[TimeLike] = None
-
-    def is_selected_for_backfill(self, model_fqn: str) -> bool:
-        return self.models_to_backfill is None or model_fqn in self.models_to_backfill
 
 
 class BackfillIntervalsPerSnapshot(PydanticModel):
diff --git a/sqlmesh/schedulers/airflow/plan.py b/sqlmesh/schedulers/airflow/plan.py
index 2a72c8ac..636d4ceb 100644
--- a/sqlmesh/schedulers/airflow/plan.py
+++ b/sqlmesh/schedulers/airflow/plan.py
@@ -83,14 +83,15 @@ class PlanDagState:
 def create_plan_dag_spec(
     request: common.PlanApplicationRequest, state_sync: StateSync
 ) -> common.PlanDagSpec:
-    new_snapshots = {s.snapshot_id: s for s in request.new_snapshots}
-    stored_snapshots = state_sync.get_snapshots([*new_snapshots, *request.environment.snapshots])
+    plan = request.plan
+    new_snapshots = {s.snapshot_id: s for s in plan.new_snapshots}
+    stored_snapshots = state_sync.get_snapshots([*new_snapshots, *plan.environment.snapshots])
     all_snapshots = {**new_snapshots, **stored_snapshots}
 
     all_snaphots_by_name = {s.name: s for s in all_snapshots.values()}
     restatements = {
         all_snaphots_by_name[n].snapshot_id: i
-        for n, i in request.restatements.items()
+        for n, i in plan.restatements.items()
         if n in all_snaphots_by_name
     }
 
@@ -104,10 +105,10 @@ def create_plan_dag_spec(
     update_intervals_for_new_snapshots(new_snapshots.values(), state_sync)
 
     now_dt = now()
-    end = request.environment.end_at or now_dt
-    unpaused_dt = end if not request.is_dev and not request.restatements else None
+    end = plan.environment.end_at or now_dt
+    unpaused_dt = end if not plan.is_dev and not plan.restatements else None
 
-    if request.restatements:
+    if plan.restatements:
         intervals_to_remove = [
             (s, restatements[s.snapshot_id])
             for s in all_snapshots.values()
@@ -115,26 +116,26 @@ def create_plan_dag_spec(
         ]
         state_sync.remove_intervals(
             intervals_to_remove,
-            remove_shared_versions=not request.is_dev,
+            remove_shared_versions=not plan.is_dev,
         )
         for s, interval in intervals_to_remove:
             all_snapshots[s.snapshot_id].remove_interval(interval)
 
     deployability_index_for_creation = DeployabilityIndex.create(all_snapshots)
     deployability_index_for_evaluation = (
-        deployability_index_for_creation if request.is_dev else DeployabilityIndex.all_deployable()
+        deployability_index_for_creation if plan.is_dev else DeployabilityIndex.all_deployable()
     )
 
-    if not request.skip_backfill:
+    if not plan.skip_backfill:
         backfill_batches = scheduler.compute_interval_params(
-            [s for s in all_snapshots.values() if request.is_selected_for_backfill(s.name)],
-            start=request.environment.start_at,
+            [s for s in all_snapshots.values() if plan.is_selected_for_backfill(s.name)],
+            start=plan.environment.start_at,
             end=end,
-            execution_time=request.execution_time or now(),
+            execution_time=plan.execution_time or now(),
             deployability_index=deployability_index_for_evaluation,
             restatements=restatements,
-            interval_end_per_model=request.interval_end_per_model,
-            end_bounded=request.end_bounded,
+            interval_end_per_model=plan.interval_end_per_model,
+            end_bounded=plan.end_bounded,
             signal_factory=None,
         )
     else:
@@ -144,7 +145,7 @@ def create_plan_dag_spec(
         common.BackfillIntervalsPerSnapshot(
             snapshot_id=s.snapshot_id,
             intervals=intervals,
-            before_promote=request.is_dev or deployability_index_for_creation.is_representative(s),
+            before_promote=plan.is_dev or deployability_index_for_creation.is_representative(s),
         )
         for s, intervals in backfill_batches.items()
     ]
@@ -155,37 +156,37 @@ def create_plan_dag_spec(
             for s in all_snapshots.values()
             if deployability_index_for_creation.is_representative(s)
         }
-        if request.no_gaps and not request.is_dev
+        if plan.no_gaps and not plan.is_dev
         else None
-        if request.no_gaps
+        if plan.no_gaps
         else set()
     )
 
     return common.PlanDagSpec(
-        request_id=request.request_id,
-        environment=request.environment,
-        new_snapshots=request.new_snapshots,
+        request_id=plan.plan_id,
+        environment=plan.environment,
+        new_snapshots=plan.new_snapshots,
         backfill_intervals_per_snapshot=backfill_intervals_per_snapshot,
-        demoted_snapshots=_get_demoted_snapshots(request.environment, state_sync),
+        demoted_snapshots=_get_demoted_snapshots(plan.environment, state_sync),
         unpaused_dt=unpaused_dt,
-        no_gaps=request.no_gaps,
+        no_gaps=plan.no_gaps,
         notification_targets=request.notification_targets,
         backfill_concurrent_tasks=request.backfill_concurrent_tasks,
         ddl_concurrent_tasks=request.ddl_concurrent_tasks,
         users=request.users,
-        is_dev=request.is_dev,
-        allow_destructive_snapshots=request.allow_destructive_snapshots,
-        forward_only=request.forward_only,
+        is_dev=plan.is_dev,
+        allow_destructive_snapshots=plan.allow_destructive_models,
+        forward_only=plan.forward_only,
         dag_start_ts=to_timestamp(now_dt),
         deployability_index=deployability_index_for_evaluation,
         deployability_index_for_creation=deployability_index_for_creation,
         no_gaps_snapshot_names=no_gaps_snapshot_names,
-        models_to_backfill=request.models_to_backfill,
-        ensure_finalized_snapshots=request.ensure_finalized_snapshots,
-        directly_modified_snapshots=request.directly_modified_snapshots,
-        indirectly_modified_snapshots=request.indirectly_modified_snapshots,
-        removed_snapshots=request.removed_snapshots,
-        execution_time=request.execution_time,
+        models_to_backfill=plan.models_to_backfill,
+        ensure_finalized_snapshots=plan.ensure_finalized_snapshots,
+        directly_modified_snapshots=plan.directly_modified_snapshots,
+        indirectly_modified_snapshots=plan.indirectly_modified_snapshots,
+        removed_snapshots=plan.removed_snapshots,
+        execution_time=plan.execution_time,
     )
 
 
diff --git a/web/server/console.py b/web/server/console.py
index ae578892..3112de3f 100644
--- a/web/server/console.py
+++ b/web/server/console.py
@@ -10,7 +10,7 @@ from sse_starlette.sse import ServerSentEvent
 
 from sqlmesh.core.console import TerminalConsole
 from sqlmesh.core.environment import EnvironmentNamingInfo
-from sqlmesh.core.plan.definition import Plan
+from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.snapshot import Snapshot, SnapshotInfoLike
 from sqlmesh.core.test import ModelTest
 from sqlmesh.utils.date import now_timestamp
@@ -28,7 +28,7 @@ class ApiConsole(TerminalConsole):
         self.current_task_status: t.Dict[str, t.Dict[str, t.Any]] = {}
         self.queue: asyncio.Queue = asyncio.Queue()
 
-    def start_plan_evaluation(self, plan: Plan) -> None:
+    def start_plan_evaluation(self, plan: EvaluatablePlan) -> None:
         self.plan_apply_stage_tracker = (
             self.plan_apply_stage_tracker
             or models.PlanApplyStageTracker(environment=plan.environment.name)


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index f7e1c37c..50e0a19b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -234,9 +234,9 @@ def push_plan(context: Context, plan: Plan) -> None:
         context.create_scheduler,
         context.default_catalog,
     )
-    plan_evaluator._push(plan)
-    promotion_result = plan_evaluator._promote(plan)
-    plan_evaluator._update_views(plan, promotion_result)
+    plan_evaluator._push(plan.to_evaluatable(), plan.snapshots)
+    promotion_result = plan_evaluator._promote(plan.to_evaluatable(), plan.snapshots)
+    plan_evaluator._update_views(plan.to_evaluatable(), plan.snapshots, promotion_result)
 
 
 @pytest.fixture()
diff --git a/tests/core/analytics/test_collector.py b/tests/core/analytics/test_collector.py
index 7c39081c..a1528b60 100644
--- a/tests/core/analytics/test_collector.py
+++ b/tests/core/analytics/test_collector.py
@@ -169,7 +169,10 @@ def test_on_plan_apply(
 
     plan_id = plan.plan_id
     collector.on_plan_apply_start(
-        plan=plan, engine_type="bigquery", state_sync_type="mysql", scheduler_type="builtin"
+        plan=plan.to_evaluatable(),
+        engine_type="bigquery",
+        state_sync_type="mysql",
+        scheduler_type="builtin",
     )
     collector.on_plan_apply_end(plan_id=plan_id)
     collector.on_plan_apply_end(plan_id=plan_id, error=SQLMeshError("test_error"))
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index 22a034a9..c7069b89 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -224,8 +224,8 @@ def test_diff(sushi_context: Context, mocker: MockerFixture):
     # https://github.com/pydantic/pydantic/issues/8016
     assert str(plan) != ""
 
-    promotion_result = plan_evaluator._promote(plan)
-    plan_evaluator._update_views(plan, promotion_result)
+    promotion_result = plan_evaluator._promote(plan.to_evaluatable(), plan.snapshots)
+    plan_evaluator._update_views(plan.to_evaluatable(), plan.snapshots, promotion_result)
 
     sushi_context.upsert_model("sushi.customers", query=parse_one("select 1 as customer_id"))
     sushi_context.diff("test")
diff --git a/tests/core/test_plan_evaluator.py b/tests/core/test_plan_evaluator.py
index bec11bba..18aba9d6 100644
--- a/tests/core/test_plan_evaluator.py
+++ b/tests/core/test_plan_evaluator.py
@@ -71,7 +71,7 @@ def test_builtin_evaluator_push(sushi_context: Context, make_snapshot):
         sushi_context.default_catalog,
         console=sushi_context.console,
     )
-    evaluator._push(plan)
+    evaluator._push(plan.to_evaluatable(), plan.snapshots)
 
     assert (
         len(sushi_context.state_sync.get_snapshots([new_model_snapshot, new_view_model_snapshot]))
@@ -86,31 +86,17 @@ def test_airflow_evaluator(sushi_plan: Plan, mocker: MockerFixture):
     airflow_client_mock.wait_for_dag_run_completion.return_value = True
     airflow_client_mock.wait_for_first_dag_run.return_value = "test_plan_application_dag_run_id"
 
+    evaluatable_plan = sushi_plan.to_evaluatable()
+
     evaluator = AirflowPlanEvaluator(airflow_client_mock)
-    evaluator.evaluate(sushi_plan)
+    evaluator.evaluate(evaluatable_plan)
 
     airflow_client_mock.apply_plan.assert_called_once_with(
-        sushi_plan.new_snapshots,
-        sushi_plan.environment,
-        mocker.ANY,
-        no_gaps=False,
+        evaluatable_plan,
         notification_targets=[],
-        restatements={},
         backfill_concurrent_tasks=1,
         ddl_concurrent_tasks=1,
-        skip_backfill=False,
         users=[],
-        is_dev=True,
-        allow_destructive_snapshots=set(),
-        forward_only=False,
-        models_to_backfill=None,
-        end_bounded=False,
-        ensure_finalized_snapshots=False,
-        directly_modified_snapshots=[],
-        indirectly_modified_snapshots={},
-        removed_snapshots=[],
-        execution_time=None,
-        interval_end_per_model=None,
     )
 
     airflow_client_mock.wait_for_dag_run_completion.assert_called_once()
@@ -125,7 +111,7 @@ def test_airflow_evaluator_plan_application_dag_fails(sushi_plan: Plan, mocker:
     evaluator = AirflowPlanEvaluator(airflow_client_mock)
 
     with pytest.raises(SQLMeshError):
-        evaluator.evaluate(sushi_plan)
+        evaluator.evaluate(sushi_plan.to_evaluatable())
 
     airflow_client_mock.apply_plan.assert_called_once()
     airflow_client_mock.wait_for_dag_run_completion.assert_called_once()
@@ -152,7 +138,7 @@ def test_mwaa_evaluator(sushi_plan: Plan, mocker: MockerFixture):
     )
 
     evaluator = MWAAPlanEvaluator(mwaa_client_mock, state_sync_mock)
-    evaluator.evaluate(sushi_plan)
+    evaluator.evaluate(sushi_plan.to_evaluatable())
 
     plan_dag_state_mock.add_dag_spec.assert_called_once_with(plan_dag_spec_mock)
 
@@ -226,7 +212,7 @@ def test_state_based_airflow_evaluator_with_restatements(
     )
 
     evaluator = MWAAPlanEvaluator(mwaa_client_mock, state_sync_mock)
-    evaluator.evaluate(plan)
+    evaluator.evaluate(plan.to_evaluatable())
 
     plan_application_request = create_plan_dag_spec_mock.call_args[0][0]
-    assert plan_application_request.restatements.keys() == {model_fqn, downstream_model_fqn}
+    assert plan_application_request.plan.restatements.keys() == {model_fqn, downstream_model_fqn}
diff --git a/tests/schedulers/airflow/test_client.py b/tests/schedulers/airflow/test_client.py
index 5f6ca588..4f991762 100644
--- a/tests/schedulers/airflow/test_client.py
+++ b/tests/schedulers/airflow/test_client.py
@@ -9,6 +9,7 @@ from sqlmesh.core.config import EnvironmentSuffixTarget
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.model import IncrementalByTimeRangeKind, SqlModel
 from sqlmesh.core.node import NodeType
+from sqlmesh.core.plan import EvaluatablePlan
 from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory
 from sqlmesh.schedulers.airflow import common
 from sqlmesh.schedulers.airflow.client import AirflowClient
@@ -57,125 +58,140 @@ def test_apply_plan(mocker: MockerFixture, snapshot: Snapshot):
         promoted_snapshot_ids=[snapshot.snapshot_id],
     )
 
-    request_id = "test_request_id"
-
-    client = AirflowClient(airflow_url=common.AIRFLOW_LOCAL_URL, session=requests.Session())
-    client.apply_plan(
-        [snapshot],
-        environment,
-        request_id,
-        models_to_backfill={'"test_model"'},
+    plan = EvaluatablePlan(
+        start=environment.start_at,
+        end=environment.end_at,
+        new_snapshots=[snapshot],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        restatements={snapshot.name: (to_timestamp("2024-01-01"), to_timestamp("2024-01-02"))},
+        is_dev=False,
+        allow_destructive_models=set(),
+        forward_only=False,
+        end_bounded=False,
+        ensure_finalized_snapshots=False,
         directly_modified_snapshots=[snapshot.snapshot_id],
-        restatements={
-            snapshot.snapshot_id: (to_timestamp("2024-01-01"), to_timestamp("2024-01-02"))
-        },
+        indirectly_modified_snapshots={},
+        removed_snapshots=[],
+        requires_backfill=True,
+        models_to_backfill={'"test_model"'},
     )
 
+    client = AirflowClient(airflow_url=common.AIRFLOW_LOCAL_URL, session=requests.Session())
+    client.apply_plan(plan)
+
     apply_plan_mock.assert_called_once()
     args, data = apply_plan_mock.call_args_list[0]
 
     assert args[0] == "http://localhost:8080/sqlmesh/api/v1/plans"
     assert data["headers"] == {"Content-Type": "application/json"}
     assert json.loads(data["data"]) == {
-        "new_snapshots": [
-            {
-                "created_ts": 1665014400000,
-                "ttl": "in 1 week",
-                "fingerprint": snapshot.fingerprint.dict(),
-                "intervals": [],
-                "dev_intervals": [],
-                "node": {
-                    "audits": [],
-                    "clustered_by": [],
-                    "cron": "@daily",
-                    "dialect": "spark",
-                    "pre_statements": ["@DEF(key, " "'value')"],
-                    "kind": {
-                        "name": "INCREMENTAL_BY_TIME_RANGE",
-                        "time_column": {"column": "`ds`"},
-                        "forward_only": False,
-                        "on_destructive_change": "ERROR",
-                        "disable_restatement": False,
-                        "dialect": "spark",
-                    },
-                    "mapping_schema": {},
-                    "inline_audits": {},
-                    "name": "test_model",
-                    "partitioned_by": ["`a`"],
-                    "query": "SELECT a, ds FROM tbl",
-                    "references": [],
-                    "project": "",
-                    "storage_format": "parquet",
-                    "jinja_macros": {
-                        "create_builtins_module": "sqlmesh.utils.jinja",
-                        "global_objs": {},
-                        "packages": {},
-                        "root_macros": {},
-                        "top_level_packages": [],
-                    },
-                    "source_type": "sql",
-                    "tags": [],
-                    "grains": [],
-                    "allow_partials": False,
-                    "signals": [],
-                    "enabled": True,
-                },
-                "audits": [],
-                "name": '"test_model"',
-                "parents": [],
-                "previous_versions": [],
-                "updated_ts": 1665014400000,
-                "version": snapshot.version,
-                "change_category": snapshot.change_category,
-                "migrated": False,
-                "unrestorable": False,
-            }
-        ],
-        "environment": {
-            "name": "test_env",
-            "snapshots": [
+        "plan": {
+            "start": "2022-01-01",
+            "end": "2022-01-01",
+            "new_snapshots": [
                 {
+                    "created_ts": 1665014400000,
+                    "ttl": "in 1 week",
                     "fingerprint": snapshot.fingerprint.dict(),
+                    "intervals": [],
+                    "dev_intervals": [],
+                    "node": {
+                        "audits": [],
+                        "clustered_by": [],
+                        "cron": "@daily",
+                        "dialect": "spark",
+                        "pre_statements": ["@DEF(key, " "'value')"],
+                        "kind": {
+                            "name": "INCREMENTAL_BY_TIME_RANGE",
+                            "time_column": {"column": "`ds`"},
+                            "forward_only": False,
+                            "on_destructive_change": "ERROR",
+                            "disable_restatement": False,
+                            "dialect": "spark",
+                        },
+                        "mapping_schema": {},
+                        "inline_audits": {},
+                        "name": "test_model",
+                        "partitioned_by": ["`a`"],
+                        "query": "SELECT a, ds FROM tbl",
+                        "references": [],
+                        "project": "",
+                        "storage_format": "parquet",
+                        "jinja_macros": {
+                            "create_builtins_module": "sqlmesh.utils.jinja",
+                            "global_objs": {},
+                            "packages": {},
+                            "root_macros": {},
+                            "top_level_packages": [],
+                        },
+                        "source_type": "sql",
+                        "tags": [],
+                        "grains": [],
+                        "allow_partials": False,
+                        "signals": [],
+                        "enabled": True,
+                    },
+                    "audits": [],
                     "name": '"test_model"',
-                    "node_type": NodeType.MODEL,
+                    "parents": [],
                     "previous_versions": [],
+                    "updated_ts": 1665014400000,
                     "version": snapshot.version,
-                    "physical_schema": "sqlmesh__default",
                     "change_category": snapshot.change_category,
-                    "parents": [],
-                    "kind_name": "INCREMENTAL_BY_TIME_RANGE",
+                    "migrated": False,
+                    "unrestorable": False,
                 }
             ],
-            "start_at": "2022-01-01",
-            "end_at": "2022-01-01",
-            "plan_id": "test_plan_id",
-            "previous_plan_id": "previous_plan_id",
-            "promoted_snapshot_ids": [
-                {
-                    "name": '"test_model"',
-                    "identifier": snapshot.identifier,
-                }
-            ],
-            "suffix_target": "schema",
-            "normalize_name": True,
+            "environment": {
+                "name": "test_env",
+                "snapshots": [
+                    {
+                        "fingerprint": snapshot.fingerprint.dict(),
+                        "name": '"test_model"',
+                        "node_type": NodeType.MODEL,
+                        "previous_versions": [],
+                        "version": snapshot.version,
+                        "physical_schema": "sqlmesh__default",
+                        "change_category": snapshot.change_category,
+                        "parents": [],
+                        "kind_name": "INCREMENTAL_BY_TIME_RANGE",
+                    }
+                ],
+                "start_at": "2022-01-01",
+                "end_at": "2022-01-01",
+                "plan_id": "test_plan_id",
+                "previous_plan_id": "previous_plan_id",
+                "promoted_snapshot_ids": [
+                    {
+                        "name": '"test_model"',
+                        "identifier": snapshot.identifier,
+                    }
+                ],
+                "suffix_target": "schema",
+                "normalize_name": True,
+            },
+            "no_gaps": False,
+            "skip_backfill": False,
+            "is_dev": False,
+            "forward_only": False,
+            "allow_destructive_models": [],
+            "models_to_backfill": ['"test_model"'],
+            "end_bounded": False,
+            "ensure_finalized_snapshots": False,
+            "directly_modified_snapshots": [{"identifier": "844700562", "name": '"test_model"'}],
+            "indirectly_modified_snapshots": {},
+            "removed_snapshots": [],
+            "restatements": {
+                '"test_model"': [to_timestamp("2024-01-01"), to_timestamp("2024-01-02")]
+            },
+            "requires_backfill": True,
         },
-        "no_gaps": False,
-        "skip_backfill": False,
         "notification_targets": [],
-        "request_id": request_id,
         "backfill_concurrent_tasks": 1,
         "ddl_concurrent_tasks": 1,
         "users": [],
-        "is_dev": False,
-        "forward_only": False,
-        "allow_destructive_snapshots": [],
-        "models_to_backfill": ['"test_model"'],
-        "end_bounded": False,
-        "ensure_finalized_snapshots": False,
-        "directly_modified_snapshots": [{"identifier": "844700562", "name": '"test_model"'}],
-        "indirectly_modified_snapshots": {},
-        "removed_snapshots": [],
-        "restatements": {'"test_model"': [to_timestamp("2024-01-01"), to_timestamp("2024-01-02")]},
     }
 
     common.PlanApplicationRequest.parse_raw(data["data"])
diff --git a/tests/schedulers/airflow/test_integration.py b/tests/schedulers/airflow/test_integration.py
index aac7ffaa..77c7c72f 100644
--- a/tests/schedulers/airflow/test_integration.py
+++ b/tests/schedulers/airflow/test_integration.py
@@ -8,11 +8,12 @@ from tenacity import retry, stop_after_attempt, wait_fixed
 from sqlmesh.core import constants as c
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.model import IncrementalByTimeRangeKind, Model, SqlModel
+from sqlmesh.core.plan import EvaluatablePlan
 from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory
 from sqlmesh.schedulers.airflow import common
 from sqlmesh.schedulers.airflow.client import AirflowClient
 from sqlmesh.utils import random_id
-from sqlmesh.utils.date import yesterday
+from sqlmesh.utils.date import yesterday, now
 from sqlmesh.utils.errors import SQLMeshError
 
 pytestmark = [
@@ -61,7 +62,7 @@ def test_apply_plan_create_backfill_promote(
 
     # Make sure that the same Snapshot can't be added again.
     with pytest.raises(SQLMeshError, match=r"Snapshots.*already exist.*"):
-        airflow_client.apply_plan([snapshot], environment, random_name())
+        airflow_client.apply_plan(_create_evaluatable_plan([snapshot], environment))
 
     # Verify full environment demotion.
     environment.snapshots_ = []
@@ -77,13 +78,10 @@ def _apply_plan_and_block(
     environment: Environment,
     is_dev: t.Optional[bool] = None,
 ) -> None:
-    if is_dev is None:
-        is_dev = environment.name != c.PROD
-
-    plan_request_id = random_id()
-    airflow_client.apply_plan(new_snapshots, environment, plan_request_id, is_dev=is_dev)
+    plan = _create_evaluatable_plan(new_snapshots, environment, is_dev)
+    airflow_client.apply_plan(plan)
 
-    plan_application_dag_id = common.plan_application_dag_id(environment.name, plan_request_id)
+    plan_application_dag_id = common.plan_application_dag_id(environment.name, plan.plan_id)
     plan_application_dag_run_id = airflow_client.wait_for_first_dag_run(
         plan_application_dag_id, DAG_CREATION_WAIT_INTERVAL, DAG_CREATION_RETRY_ATTEMPTS
     )
@@ -92,6 +90,33 @@ def _apply_plan_and_block(
     )
 
 
+def _create_evaluatable_plan(
+    new_snapshots: t.List[Snapshot],
+    environment: Environment,
+    is_dev: t.Optional[bool] = None,
+) -> EvaluatablePlan:
+    if is_dev is None:
+        is_dev = environment.name != c.PROD
+    return EvaluatablePlan(
+        start=environment.start_at,
+        end=environment.end_at or now(),
+        new_snapshots=new_snapshots,
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=False,
+        restatements={},
+        is_dev=is_dev,
+        allow_destructive_models=set(),
+        forward_only=False,
+        end_bounded=True,
+        ensure_finalized_snapshots=False,
+        directly_modified_snapshots=[],
+        indirectly_modified_snapshots={},
+        removed_snapshots=[],
+        requires_backfill=True,
+    )
+
+
 @retry(wait=wait_fixed(3), stop=stop_after_attempt(5), reraise=True)
 def _get_snapshot_dag(
     airflow_client: AirflowClient, model_name: str, version: str
diff --git a/tests/schedulers/airflow/test_plan.py b/tests/schedulers/airflow/test_plan.py
index 6d382506..e5489ba8 100644
--- a/tests/schedulers/airflow/test_plan.py
+++ b/tests/schedulers/airflow/test_plan.py
@@ -17,6 +17,7 @@ from sqlmesh.core.model import (
     create_sql_model,
 )
 from sqlmesh.core.node import NodeType
+from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
     Snapshot,
@@ -26,7 +27,7 @@ from sqlmesh.core.snapshot import (
 )
 from sqlmesh.schedulers.airflow import common
 from sqlmesh.schedulers.airflow.plan import PlanDagState, create_plan_dag_spec
-from sqlmesh.utils.date import to_datetime, to_timestamp
+from sqlmesh.utils.date import to_datetime, to_timestamp, now
 from sqlmesh.utils.errors import SQLMeshError
 
 pytestmark = pytest.mark.airflow
@@ -104,17 +105,14 @@ def test_create_plan_dag_spec(
         catalog_name_override="test_catalog",
     )
 
-    plan_request = common.PlanApplicationRequest(
-        request_id="test_request_id",
+    plan = EvaluatablePlan(
+        start=new_environment.start_at,
+        end=new_environment.end_at,
         new_snapshots=[the_snapshot],
         environment=new_environment,
         no_gaps=True,
         skip_backfill=False,
         restatements={},
-        notification_targets=[],
-        backfill_concurrent_tasks=1,
-        ddl_concurrent_tasks=1,
-        users=[],
         is_dev=False,
         forward_only=True,
         models_to_backfill=None,
@@ -124,6 +122,16 @@ def test_create_plan_dag_spec(
         indirectly_modified_snapshots={},
         removed_snapshots=[],
         interval_end_per_model=None,
+        allow_destructive_models=set(),
+        requires_backfill=True,
+    )
+
+    plan_request = common.PlanApplicationRequest(
+        plan=plan,
+        notification_targets=[],
+        backfill_concurrent_tasks=1,
+        ddl_concurrent_tasks=1,
+        users=[],
     )
 
     deleted_snapshot = SnapshotTableInfo(
@@ -160,7 +168,7 @@ def test_create_plan_dag_spec(
         plan_spec = create_plan_dag_spec(plan_request, state_sync_mock)
 
     assert plan_spec == common.PlanDagSpec(
-        request_id="test_request_id",
+        request_id="test_plan_id",
         environment=new_environment,
         new_snapshots=[the_snapshot],
         backfill_intervals_per_snapshot=[
@@ -234,8 +242,9 @@ def test_restatement(
 
     the_snapshot.add_interval("2022-01-01", "2022-01-07")
 
-    plan_request = common.PlanApplicationRequest(
-        request_id="test_request_id",
+    plan = EvaluatablePlan(
+        start=new_environment.start_at,
+        end=new_environment.end_at,
         new_snapshots=[],
         environment=new_environment,
         no_gaps=True,
@@ -246,10 +255,6 @@ def test_restatement(
                 to_timestamp("2022-01-04"),
             )
         },
-        notification_targets=[],
-        backfill_concurrent_tasks=1,
-        ddl_concurrent_tasks=1,
-        users=[],
         is_dev=False,
         forward_only=True,
         models_to_backfill=None,
@@ -259,6 +264,16 @@ def test_restatement(
         indirectly_modified_snapshots={},
         removed_snapshots=[],
         interval_end_per_model=None,
+        allow_destructive_models=set(),
+        requires_backfill=True,
+    )
+
+    plan_request = common.PlanApplicationRequest(
+        plan=plan,
+        notification_targets=[],
+        backfill_concurrent_tasks=1,
+        ddl_concurrent_tasks=1,
+        users=[],
     )
     old_environment = Environment(
         name=environment_name,
@@ -280,7 +295,7 @@ def test_restatement(
         plan_spec = create_plan_dag_spec(plan_request, state_sync_mock)
 
     assert plan_spec == common.PlanDagSpec(
-        request_id="test_request_id",
+        request_id="test_plan_id",
         environment=new_environment,
         new_snapshots=[],
         backfill_intervals_per_snapshot=[
@@ -352,17 +367,14 @@ def test_select_models_for_backfill(mocker: MockerFixture, random_name, make_sna
         suffix_target=EnvironmentSuffixTarget.TABLE,
     )
 
-    plan_request = common.PlanApplicationRequest(
-        request_id="test_request_id",
+    plan = EvaluatablePlan(
+        start=new_environment.start_at,
+        end=new_environment.end_at,
         new_snapshots=[snapshot_a, snapshot_b],
         environment=new_environment,
         no_gaps=True,
         skip_backfill=False,
         restatements={},
-        notification_targets=[],
-        backfill_concurrent_tasks=1,
-        ddl_concurrent_tasks=1,
-        users=[],
         is_dev=False,
         forward_only=True,
         models_to_backfill={snapshot_b.name},
@@ -372,6 +384,16 @@ def test_select_models_for_backfill(mocker: MockerFixture, random_name, make_sna
         indirectly_modified_snapshots={},
         removed_snapshots=[],
         interval_end_per_model=None,
+        allow_destructive_models=set(),
+        requires_backfill=True,
+    )
+
+    plan_request = common.PlanApplicationRequest(
+        plan=plan,
+        notification_targets=[],
+        backfill_concurrent_tasks=1,
+        ddl_concurrent_tasks=1,
+        users=[],
     )
 
     state_sync_mock = mocker.Mock()
@@ -387,7 +409,7 @@ def test_select_models_for_backfill(mocker: MockerFixture, random_name, make_sna
         plan_spec = create_plan_dag_spec(plan_request, state_sync_mock)
 
     assert plan_spec == common.PlanDagSpec(
-        request_id="test_request_id",
+        request_id="test_plan_id",
         environment=new_environment,
         new_snapshots=[snapshot_a, snapshot_b],
         backfill_intervals_per_snapshot=[
@@ -429,17 +451,14 @@ def test_create_plan_dag_spec_duplicated_snapshot(
         plan_id="test_plan_id",
     )
 
-    plan_request = common.PlanApplicationRequest(
-        request_id="test_request_id",
+    plan = EvaluatablePlan(
+        start=new_environment.start_at,
+        end=new_environment.end_at,
         new_snapshots=[snapshot],
         environment=new_environment,
         no_gaps=False,
         skip_backfill=False,
         restatements={},
-        notification_targets=[],
-        backfill_concurrent_tasks=1,
-        ddl_concurrent_tasks=1,
-        users=[],
         is_dev=False,
         forward_only=False,
         models_to_backfill=None,
@@ -449,6 +468,16 @@ def test_create_plan_dag_spec_duplicated_snapshot(
         indirectly_modified_snapshots={},
         removed_snapshots=[],
         interval_end_per_model=None,
+        allow_destructive_models=set(),
+        requires_backfill=True,
+    )
+
+    plan_request = common.PlanApplicationRequest(
+        plan=plan,
+        notification_targets=[],
+        backfill_concurrent_tasks=1,
+        ddl_concurrent_tasks=1,
+        users=[],
     )
 
     dag_run_mock = mocker.Mock()
@@ -483,17 +512,14 @@ def test_create_plan_dag_spec_unbounded_end(
         plan_id="test_plan_id",
     )
 
-    plan_request = common.PlanApplicationRequest(
-        request_id="test_request_id",
+    plan = EvaluatablePlan(
+        start=new_environment.start_at,
+        end=now(),
         new_snapshots=[],
         environment=new_environment,
         no_gaps=True,
         skip_backfill=False,
         restatements={},
-        notification_targets=[],
-        backfill_concurrent_tasks=1,
-        ddl_concurrent_tasks=1,
-        users=[],
         is_dev=False,
         forward_only=False,
         models_to_backfill=None,
@@ -503,6 +529,16 @@ def test_create_plan_dag_spec_unbounded_end(
         indirectly_modified_snapshots={},
         removed_snapshots=[],
         interval_end_per_model=None,
+        allow_destructive_models=set(),
+        requires_backfill=True,
+    )
+
+    plan_request = common.PlanApplicationRequest(
+        plan=plan,
+        notification_targets=[],
+        backfill_concurrent_tasks=1,
+        ddl_concurrent_tasks=1,
+        users=[],
     )
 
     state_sync_mock = mocker.Mock()
@@ -531,7 +567,7 @@ def test_plan_dag_state(snapshot: Snapshot, sushi_context: Context, random_name)
         plan_id="test_plan_id",
     )
     plan_dag_spec = common.PlanDagSpec(
-        request_id="test_request_id",
+        request_id="test_plan_id",
         environment=environment,
         new_snapshots=[],
         backfill_intervals_per_snapshot=[],
@@ -565,6 +601,6 @@ def test_plan_dag_state(snapshot: Snapshot, sushi_context: Context, random_name)
     assert get_hydrated_dag_specs() == [plan_dag_spec]
 
     plan_dag_state.delete_dag_specs(
-        [common.plan_application_dag_id(environment_name, "test_request_id")]
+        [common.plan_application_dag_id(environment_name, "test_plan_id")]
     )
     assert not plan_dag_state.get_dag_specs()