Fix: Iterate through the whole DAG before raising an error when creating physical tables (#4305)

Author: izeigerman

sqlmesh/core/plan/evaluator.py

@@ -33,6 +33,7 @@
     SnapshotId,
     SnapshotInfoLike,
     SnapshotTableInfo,
+    SnapshotCreationFailedError,
 )
 from sqlmesh.utils import CompletionStatus
 from sqlmesh.core.state_sync import StateSync
@@ -291,12 +292,14 @@ def _should_create(s: Snapshot) -> bool:
                 ),
                 on_complete=self.console.update_creation_progress,
             )
-        except NodeExecutionFailedError as ex:
+        except SnapshotCreationFailedError as ex:
             self.console.stop_creation_progress(success=False)
             progress_stopped = True
 
-            logger.info(str(ex), exc_info=ex)
-            self.console.log_failed_models([ex])
+            for error in ex.errors:
+                logger.info(str(error), exc_info=error)
+
+            self.console.log_failed_models(ex.errors)
 
             raise PlanError("Plan application failed.")
         finally:

sqlmesh/core/snapshot/__init__.py

@@ -26,4 +26,7 @@
     table_name as table_name,
     to_table_mapping as to_table_mapping,
 )
-from sqlmesh.core.snapshot.evaluator import SnapshotEvaluator as SnapshotEvaluator
+from sqlmesh.core.snapshot.evaluator import (
+    SnapshotEvaluator as SnapshotEvaluator,
+    SnapshotCreationFailedError as SnapshotCreationFailedError,
+)

sqlmesh/core/snapshot/evaluator.py

@@ -66,6 +66,7 @@
 from sqlmesh.utils.concurrency import (
     concurrent_apply_to_snapshots,
     concurrent_apply_to_values,
+    NodeExecutionFailedError,
 )
 from sqlmesh.utils.date import TimeLike, now, time_like_to_str
 from sqlmesh.utils.errors import (
@@ -87,6 +88,13 @@
 logger = logging.getLogger(__name__)
 
 
+class SnapshotCreationFailedError(SQLMeshError):
+    def __init__(self, errors: t.List[NodeExecutionFailedError[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
+
+
 class SnapshotEvaluator:
     """Evaluates a snapshot given runtime arguments through an arbitrary EngineAdapter.
 
@@ -397,7 +405,7 @@ def _create_snapshots(
     ) -> None:
         """Internal method to create tables in parallel."""
         with self.concurrent_context():
-            concurrent_apply_to_snapshots(
+            errors, _ = concurrent_apply_to_snapshots(
                 snapshots_to_create,
                 lambda s: self._create_snapshot(
                     s,
@@ -408,7 +416,10 @@ def _create_snapshots(
                     allow_destructive_snapshots=allow_destructive_snapshots,
                 ),
                 self.ddl_concurrent_tasks,
+                raise_on_error=False,
             )
+            if errors:
+                raise SnapshotCreationFailedError(errors)
 
     def migrate(
         self,