Fix: Report models for which physical table creation was skipped due to a failure upstream (#4308)

Author: izeigerman

sqlmesh/core/plan/evaluator.py

@@ -290,6 +290,7 @@ def _should_create(s: Snapshot) -> bool:
             for error in ex.errors:
                 logger.info(str(error), exc_info=error)
 
+            self.console.log_skipped_models({s.name for s in ex.skipped})
             self.console.log_failed_models(ex.errors)
 
             raise PlanError("Plan application failed.")

sqlmesh/core/snapshot/evaluator.py

@@ -89,10 +89,13 @@
 
 
 class SnapshotCreationFailedError(SQLMeshError):
-    def __init__(self, errors: t.List[NodeExecutionFailedError[SnapshotId]]):
+    def __init__(
+        self, errors: t.List[NodeExecutionFailedError[SnapshotId]], skipped: t.List[SnapshotId]
+    ):
         messages = "\n\n".join(f"{error}\n  {error.__cause__}" for error in errors)
         super().__init__(f"Physical table creation failed:\n\n{messages}")
         self.errors = errors
+        self.skipped = skipped
 
 
 class SnapshotEvaluator:
@@ -380,7 +383,7 @@ def _create_snapshots(
     ) -> None:
         """Internal method to create tables in parallel."""
         with self.concurrent_context():
-            errors, _ = concurrent_apply_to_snapshots(
+            errors, skipped = concurrent_apply_to_snapshots(
                 snapshots_to_create,
                 lambda s: self._create_snapshot(
                     s,
@@ -394,7 +397,7 @@ def _create_snapshots(
                 raise_on_error=False,
             )
             if errors:
-                raise SnapshotCreationFailedError(errors)
+                raise SnapshotCreationFailedError(errors, skipped)
 
     def migrate(
         self,