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

@@ -103,6 +103,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,14 @@
+from __future__ import annotations
+
+
+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
@@ -126,6 +127,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

@@ -2307,7 +2307,12 @@ def _run_janitor(self, ignore_ttl: bool = False) -> None:
 
     def _cleanup_environments(self) -> None:
         expired_environments = self.state_sync.delete_expired_environments()
-        cleanup_expired_views(self.engine_adapter, expired_environments, console=self.console)
+        cleanup_expired_views(
+            self.engine_adapter,
+            expired_environments,
+            warn_on_delete_failure=self.config.janitor.warn_on_delete_failure,
+            console=self.console,
+        )
 
     def _try_connection(self, connection_name: str, validator: t.Callable[[], None]) -> None:
         connection_name = connection_name.capitalize()

sqlmesh/core/state_sync/common.py

@@ -21,7 +21,10 @@
 
 
 def cleanup_expired_views(
-    adapter: EngineAdapter, environments: t.List[Environment], console: t.Optional[Console] = None
+    adapter: EngineAdapter,
+    environments: t.List[Environment],
+    warn_on_delete_failure: bool = False,
+    console: t.Optional[Console] = None,
 ) -> None:
     expired_schema_environments = [
         environment for environment in environments if environment.suffix_target.is_schema
@@ -52,9 +55,11 @@ def cleanup_expired_views(
             if console:
                 console.update_cleanup_progress(schema.sql(dialect=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
     for expired_view in {
         snapshot.qualified_view_name.for_environment(
             environment.naming_info, dialect=adapter.dialect
@@ -68,9 +73,11 @@ def cleanup_expired_views(
             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: