PR feedback

Author: treysp

sqlmesh/core/scheduler.py

@@ -166,10 +166,8 @@ def evaluate(
         execution_time: TimeLike,
         deployability_index: DeployabilityIndex,
         batch_index: int,
-        environment_naming_info: EnvironmentNamingInfo,
-        default_catalog: t.Optional[str],
         **kwargs: t.Any,
-    ) -> t.List[AuditResult]:
+    ) -> t.Tuple[t.List[AuditResult], t.List[AuditError]]:
         """Evaluate a snapshot and add the processed interval to the state sync.
 
         Args:
@@ -183,7 +181,7 @@ def evaluate(
             kwargs: Additional kwargs to pass to the renderer.
 
         Returns:
-            List of audit results from the evaluation.
+            Tuple of list of all audit results from the evaluation and list of non-blocking audit errors to warn.
         """
         validate_date_range(start, end)
 
@@ -213,6 +211,7 @@ def evaluate(
         )
 
         audit_errors_to_raise: t.List[AuditError] = []
+        audit_errors_to_warn: t.List[AuditError] = []
         for audit_result in (result for result in audit_results if result.count):
             error = AuditError(
                 audit_name=audit_result.audit.name,
@@ -230,21 +229,13 @@ def evaluate(
             if audit_result.blocking:
                 audit_errors_to_raise.append(error)
             else:
-                display_name = snapshot.display_name(
-                    environment_naming_info,
-                    default_catalog,
-                    self.snapshot_evaluator.adapter.dialect,
-                )
-                self.console.log_warning(
-                    f"\n{display_name}: {error}.",
-                    f"{error}. Audit query:\n{error.query.sql(error.adapter_dialect)}",
-                )
+                audit_errors_to_warn.append(error)
 
         if audit_errors_to_raise:
             raise NodeAuditsErrors(audit_errors_to_raise)
 
         self.state_sync.add_interval(snapshot, start, end, is_dev=not is_deployable)
-        return audit_results
+        return audit_results, audit_errors_to_warn
 
     def run(
         self,
@@ -475,11 +466,12 @@ def evaluate_node(node: SchedulingUnit) -> None:
             execution_start_ts = now_timestamp()
             evaluation_duration_ms: t.Optional[int] = None
 
+            audit_results: t.List[AuditResult] = []
+            audit_errors_to_warn: t.List[AuditError] = []
             try:
                 assert execution_time  # mypy
                 assert deployability_index  # mypy
-                audit_results = []  # so it exists for finally if `evaluate` raises
-                audit_results = self.evaluate(
+                audit_results, audit_errors_to_warn = self.evaluate(
                     snapshot=snapshot,
                     start=start,
                     end=end,
@@ -489,6 +481,18 @@ def evaluate_node(node: SchedulingUnit) -> None:
                     environment_naming_info=environment_naming_info,
                     default_catalog=self.default_catalog,
                 )
+
+                for audit_error in audit_errors_to_warn:
+                    display_name = snapshot.display_name(
+                        environment_naming_info,
+                        self.default_catalog,
+                        self.snapshot_evaluator.adapter.dialect,
+                    )
+                    self.console.log_warning(
+                        f"\n{display_name}: {audit_error}.",
+                        f"{audit_error}. Audit query:\n{audit_error.query.sql(audit_error.adapter_dialect)}",
+                    )
+
                 evaluation_duration_ms = now_timestamp() - execution_start_ts
             finally:
                 num_audits = len(audit_results)

tests/core/test_scheduler.py

@@ -528,8 +528,6 @@ def _evaluate():
             to_datetime("2022-01-03"),
             DeployabilityIndex.all_deployable(),
             0,
-            EnvironmentNamingInfo(),
-            None,
         )
 
     evaluator_audit_mock.return_value = [