Feat: Add a janitor configuration that allows to warn instead of failing if it fails to delete an expired environment schema / view (#4150)

Author: izeigerman

docs/reference/configuration.md

@@ -104,6 +104,16 @@ Formatting settings for the `sqlmesh format` command and UI.
 | `append_newline`      | Whether to append a newline to the end of the file (Default: False)                            | boolean |    N     |
 | `no_rewrite_casts`    | Preserve the existing casts, without rewriting them to use the :: syntax. (Default: False)                  | boolean |    N     |
 
+
+## Janitor
+
+Configuration for the `sqlmesh janitor` command.
+
+| Option                   | Description                                                                                                                | Type    | Required |
+|--------------------------|----------------------------------------------------------------------------------------------------------------------------|:-------:|:--------:|
+| `warn_on_delete_failure` | Whether to warn instead of erroring if the janitor fails to delete the expired environment schema / views (Default: False) | boolean | N        |
+
+
 ## UI
 
 SQLMesh UI settings.

sqlmesh/core/config/janitor.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+import typing as t
+
+from sqlmesh.core.config.base import BaseConfig
+
+
+class JanitorConfig(BaseConfig):
+    """The configuration for the janitor.
+
+    Args:
+        warn_on_delete_failure: Whether to warn instead of erroring if the janitor fails to delete the expired environment schema / views.
+    """
+
+    warn_on_delete_failure: bool = False

sqlmesh/core/config/root.py

@@ -25,6 +25,7 @@
 from sqlmesh.core.config.feature_flag import FeatureFlag
 from sqlmesh.core.config.format import FormatConfig
 from sqlmesh.core.config.gateway import GatewayConfig
+from sqlmesh.core.config.janitor import JanitorConfig
 from sqlmesh.core.config.migration import MigrationConfig
 from sqlmesh.core.config.model import ModelDefaultsConfig
 from sqlmesh.core.config.naming import NameInferenceConfig as NameInferenceConfig
@@ -128,6 +129,7 @@ class Config(BaseConfig):
     before_all: t.Optional[t.List[str]] = None
     after_all: t.Optional[t.List[str]] = None
     linter: LinterConfig = LinterConfig()
+    janitor: JanitorConfig = JanitorConfig()
 
     _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {
         "gateways": UpdateStrategy.NESTED_UPDATE,

sqlmesh/core/context.py

@@ -2292,6 +2292,7 @@ def _cleanup_environments(self) -> None:
             default_adapter=self.engine_adapter,
             engine_adapters=self.engine_adapters,
             environments=expired_environments,
+            warn_on_delete_failure=self.config.janitor.warn_on_delete_failure,
             console=self.console,
         )
 

sqlmesh/core/state_sync/common.py

@@ -26,6 +26,7 @@ def cleanup_expired_views(
     default_adapter: EngineAdapter,
     engine_adapters: t.Dict[str, EngineAdapter],
     environments: t.List[Environment],
+    warn_on_delete_failure: bool = False,
     console: t.Optional[Console] = None,
 ) -> None:
     expired_schema_environments = [
@@ -66,9 +67,11 @@ def get_adapter(gateway_managed: bool, gateway: t.Optional[str] = None) -> Engin
             if console:
                 console.update_cleanup_progress(schema.sql(dialect=engine_adapter.dialect))
         except Exception as e:
-            raise SQLMeshError(
-                f"Failed to drop the expired environment schema '{schema}': {e}"
-            ) from e
+            message = f"Failed to drop the expired environment schema '{schema}': {e}"
+            if warn_on_delete_failure:
+                logger.warning(message)
+            else:
+                raise SQLMeshError(message) from e
 
     # Drop the views for the expired environments
     for engine_adapter, expired_view in {
@@ -87,9 +90,11 @@ def get_adapter(gateway_managed: bool, gateway: t.Optional[str] = None) -> Engin
             if console:
                 console.update_cleanup_progress(expired_view)
         except Exception as e:
-            raise SQLMeshError(
-                f"Failed to drop the expired environment view '{expired_view}': {e}"
-            ) from e
+            message = f"Failed to drop the expired environment view '{expired_view}': {e}"
+            if warn_on_delete_failure:
+                logger.warning(message)
+            else:
+                raise SQLMeshError(message) from e
 
 
 def transactional() -> t.Callable[[t.Callable], t.Callable]:

tests/core/state_sync/test_state_sync.py

@@ -2651,6 +2651,43 @@ def test_cleanup_expired_views(
     ]
 
 
+@pytest.mark.parametrize(
+    "suffix_target", [EnvironmentSuffixTarget.SCHEMA, EnvironmentSuffixTarget.TABLE]
+)
+def test_cleanup_expired_environment_schema_warn_on_delete_failure(
+    mocker: MockerFixture, make_snapshot: t.Callable, suffix_target: EnvironmentSuffixTarget
+):
+    adapter = mocker.MagicMock()
+    adapter.dialect = None
+    adapter.drop_schema.side_effect = Exception("Failed to drop the schema")
+    adapter.drop_view.side_effect = Exception("Failed to drop the view")
+
+    snapshot = make_snapshot(
+        SqlModel(name="test_catalog.test_schema.test_model", query=parse_one("select 1, ds"))
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    schema_environment = Environment(
+        name="test_environment",
+        suffix_target=suffix_target,
+        snapshots=[snapshot.table_info],
+        start_at="2022-01-01",
+        end_at="2022-01-01",
+        plan_id="test_plan_id",
+        previous_plan_id="test_plan_id",
+        catalog_name_override="catalog_override",
+    )
+
+    with pytest.raises(SQLMeshError, match="Failed to drop the expired environment .*"):
+        cleanup_expired_views(adapter, {}, [schema_environment], warn_on_delete_failure=False)
+
+    cleanup_expired_views(adapter, {}, [schema_environment], warn_on_delete_failure=True)
+
+    if suffix_target == EnvironmentSuffixTarget.SCHEMA:
+        assert adapter.drop_schema.called
+    else:
+        assert adapter.drop_view.called
+
+
 def test_max_interval_end_per_model(
     state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
 ) -> None: