Repository: TobikoData/sqlmesh
Pull request #4646 (2025-06-04 00:17:18)

fix: cicd bot correctly surface plan errors
Prior to this PR, plan errors would not be surfaced back to the user and would give a confusing `action_required` check title. Now the title communicates a plan error and then the plan error is shown in the summary. 


Patch:
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index 8b05973c..032c1da4 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -93,8 +93,10 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin, ClusteredByMixi
             return
 
         self.execute(f"USE WAREHOUSE {warehouse_sql}")
-        yield
-        self.execute(f"USE WAREHOUSE {current_warehouse_sql}")
+        try:
+            yield
+        finally:
+            self.execute(f"USE WAREHOUSE {current_warehouse_sql}")
 
     @property
     def _current_warehouse(self) -> exp.Identifier:
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index ee0af270..e7df3157 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -164,6 +164,13 @@ class Loader(abc.ABC):
     """Abstract base class to load macros and models for a context"""
 
     def __init__(self, context: GenericContext, path: Path) -> None:
+        # This ensures pandas is imported before any model loading happens in the forked process
+        # to avoid macOS fork() safety issues, see https://stackoverflow.com/a/52230415. Without
+        # it, the following error was observerd in a macOS 15.5 system:
+        #
+        # "+[NSMutableString initialize] may have been in progress in another thread when fork() was called."
+        import pandas as pd  # noqa
+
         from sqlmesh.core.console import get_console
 
         self._path_mtimes: t.Dict[Path, float] = {}
diff --git a/sqlmesh/integrations/github/cicd/command.py b/sqlmesh/integrations/github/cicd/command.py
index 9ccb8f56..01a9236e 100644
--- a/sqlmesh/integrations/github/cicd/command.py
+++ b/sqlmesh/integrations/github/cicd/command.py
@@ -185,9 +185,11 @@ def _deploy_production(controller: GithubController) -> bool:
             skip_reason=str(e),
         )
         return False
-    except PlanError:
+    except PlanError as e:
         controller.update_prod_environment_check(
-            status=GithubCheckStatus.COMPLETED, conclusion=GithubCheckConclusion.ACTION_REQUIRED
+            status=GithubCheckStatus.COMPLETED,
+            conclusion=GithubCheckConclusion.ACTION_REQUIRED,
+            plan_error=e,
         )
         return False
     except Exception:
diff --git a/sqlmesh/integrations/github/cicd/controller.py b/sqlmesh/integrations/github/cicd/controller.py
index eadb34e3..2afccc48 100644
--- a/sqlmesh/integrations/github/cicd/controller.py
+++ b/sqlmesh/integrations/github/cicd/controller.py
@@ -977,6 +977,7 @@ class GithubController:
         status: GithubCheckStatus,
         conclusion: t.Optional[GithubCheckConclusion] = None,
         skip_reason: t.Optional[str] = None,
+        plan_error: t.Optional[PlanError] = None,
     ) -> None:
         """
         Updates the status of the merge commit for the prod environment.
@@ -990,6 +991,7 @@ class GithubController:
                 GithubCheckConclusion.CANCELLED: "Cancelled deploying to prod",
                 GithubCheckConclusion.SKIPPED: skip_reason,
                 GithubCheckConclusion.FAILURE: "Failed to deploy to prod",
+                GithubCheckConclusion.ACTION_REQUIRED: "Failed due to error applying plan",
             }
             title = (
                 conclusion_to_title.get(conclusion)
@@ -1002,6 +1004,11 @@ class GithubController:
                 summary = (
                     captured_errors or f"{title}\n\n**Error:**\n```\n{traceback.format_exc()}\n```"
                 )
+            elif conclusion.is_action_required:
+                if plan_error:
+                    summary = f"**Plan error:**\n```\n{plan_error}\n```"
+                else:
+                    summary = "Got an action required conclusion but no plan error was provided. This is unexpected."
             else:
                 summary = "**Generated Prod Plan**\n" + self.get_plan_summary(self.prod_plan)
             return conclusion, title, summary


Test patch:
diff --git a/tests/cli/test_integration_cli.py b/tests/cli/test_integration_cli.py
index 1b803aa7..7d8802d1 100644
--- a/tests/cli/test_integration_cli.py
+++ b/tests/cli/test_integration_cli.py
@@ -7,8 +7,6 @@ from sqlmesh.utils import yaml
 import shutil
 import site
 import uuid
-import os
-import platform
 
 pytestmark = pytest.mark.slow
 
@@ -32,11 +30,6 @@ def invoke_cli(tmp_path: Path) -> InvokeCliType:
     ).stdout.strip()
 
     def _invoke(sqlmesh_args: t.List[str], **kwargs: t.Any) -> subprocess.CompletedProcess:
-        # Set up environment to handle macOS fork safety, see https://stackoverflow.com/a/52230415
-        env = os.environ.copy()
-        if platform.system() == "Darwin":
-            env["OBJC_DISABLE_INITIALIZE_FORK_SAFETY"] = "YES"
-
         return subprocess.run(
             args=[sqlmesh_bin] + sqlmesh_args,
             # set the working directory to the isolated temp dir for this test
@@ -46,7 +39,6 @@ def invoke_cli(tmp_path: Path) -> InvokeCliType:
             # combine stdout/stderr into a single stream
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
-            env=env,
             **kwargs,
         )
 
diff --git a/tests/core/engine_adapter/test_snowflake.py b/tests/core/engine_adapter/test_snowflake.py
index c6101542..f0a47b33 100644
--- a/tests/core/engine_adapter/test_snowflake.py
+++ b/tests/core/engine_adapter/test_snowflake.py
@@ -98,6 +98,7 @@ def test_session(
     adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
     adapter.cursor.fetchone.return_value = (current_warehouse,)
 
+    # Test normal execution
     with adapter.session({"warehouse": configured_warehouse}):
         pass
 
@@ -114,6 +115,27 @@ def test_session(
 
     assert to_sql_calls(adapter) == expected_calls
 
+    # Test exception handling - warehouse should still be reset
+    if should_change:
+        adapter.cursor.execute.reset_mock()
+        adapter.cursor.fetchone.return_value = (current_warehouse,)
+
+        try:
+            with adapter.session({"warehouse": configured_warehouse}):
+                adapter.execute("SELECT 1")
+                raise RuntimeError("Test exception")
+        except RuntimeError:
+            pass
+
+        expected_exception_calls = [
+            "SELECT CURRENT_WAREHOUSE()",
+            f"USE WAREHOUSE {configured_warehouse_exp}",
+            "SELECT 1",
+            f"USE WAREHOUSE {current_warehouse_exp}",
+        ]
+
+        assert to_sql_calls(adapter) == expected_exception_calls
+
 
 def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
     adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
diff --git a/tests/integrations/github/cicd/test_github_commands.py b/tests/integrations/github/cicd/test_github_commands.py
index 0d2d303f..4d628389 100644
--- a/tests/integrations/github/cicd/test_github_commands.py
+++ b/tests/integrations/github/cicd/test_github_commands.py
@@ -884,6 +884,15 @@ def make_test_prod_update_failure_case(
     assert GithubCheckStatus(prod_checks_runs[1]["status"]).is_in_progress
     assert GithubCheckStatus(prod_checks_runs[2]["status"]).is_completed
     assert GithubCheckConclusion(prod_checks_runs[2]["conclusion"]) == expect_prod_sync_conclusion
+    if expect_prod_sync_conclusion.is_action_required:
+        assert prod_checks_runs[2]["output"]["title"] == "Failed due to error applying plan"
+        assert (
+            prod_checks_runs[2]["output"]["summary"]
+            == f"""**Plan error:**
+```
+{to_raise_on_prod_plan}
+```"""
+        )
 
     assert "SQLMesh - Run Unit Tests" in controller._check_run_mapping
     test_checks_runs = controller._check_run_mapping["SQLMesh - Run Unit Tests"].all_kwargs