add force-delete flag to delete from state even on failures

Signed-off-by: Themis Valtinos <73662635+themisvaltinos@users.noreply.github.com>

Author: themisvaltinos

sqlmesh/cli/main.py

@@ -631,16 +631,22 @@ def invalidate(ctx: click.Context, environment: str, **kwargs: t.Any) -> None:
     is_flag=True,
     help="Cleanup snapshots that are not referenced in any environment, regardless of when they're set to expire",
 )
+@click.option(
+    "--force-delete",
+    is_flag=True,
+    help="Delete expired environment and snapshot state records even when the physical table or view drops fail. "
+    "Any objects that could not be dropped become orphaned and must be removed manually.",
+)
 @click.pass_context
 @error_handler
 @cli_analytics
-def janitor(ctx: click.Context, ignore_ttl: bool, **kwargs: t.Any) -> None:
+def janitor(ctx: click.Context, ignore_ttl: bool, force_delete: bool, **kwargs: t.Any) -> None:
     """
     Run the janitor process on-demand.
 
     The janitor cleans up old environments and expired snapshots.
     """
-    ctx.obj.run_janitor(ignore_ttl, **kwargs)
+    ctx.obj.run_janitor(ignore_ttl, force_delete=force_delete, **kwargs)
 
 
 @cli.command("destroy")

sqlmesh/core/context.py

@@ -887,12 +887,12 @@ def _has_environment_changed() -> bool:
         return completion_status
 
     @python_api_analytics
-    def run_janitor(self, ignore_ttl: bool) -> bool:
+    def run_janitor(self, ignore_ttl: bool, force_delete: bool = False) -> bool:
         success = False
 
         if self.console.start_cleanup(ignore_ttl):
             try:
-                self._run_janitor(ignore_ttl)
+                self._run_janitor(ignore_ttl, force_delete=force_delete)
                 success = True
             finally:
                 self.console.stop_cleanup(success=success)
@@ -2896,19 +2896,22 @@ def _destroy(self) -> bool:
 
         return True
 
-    def _run_janitor(self, ignore_ttl: bool = False) -> None:
+    def _run_janitor(self, ignore_ttl: bool = False, force_delete: bool = False) -> None:
         current_ts = now_timestamp()
         failures: t.List[str] = []
 
         # Clean up expired environments by removing their views and schemas
-        failures.extend(self._cleanup_environments(current_ts=current_ts))
+        failures.extend(
+            self._cleanup_environments(current_ts=current_ts, force_delete=force_delete)
+        )
 
         failures.extend(
             delete_expired_snapshots(
                 self.state_sync,
                 self.snapshot_evaluator,
                 current_ts=current_ts,
                 ignore_ttl=ignore_ttl,
+                force_delete=force_delete,
                 console=self.console,
                 batch_size=self.config.janitor.expired_snapshots_batch_size,
             )
@@ -2918,12 +2921,16 @@ def _run_janitor(self, ignore_ttl: bool = False) -> None:
         if failures:
             failure_string = "\n  - ".join(failures)
             summary = f"Janitor completed with failures:\n  {failure_string}"
+            if force_delete:
+                summary += "\nState records have been deleted, but the underlying objects may still exist in the database.\nPlease investigate and clean up manually the above if necessary."
             if self.config.janitor.warn_on_delete_failure:
                 self.console.log_warning(summary)
             else:
                 raise SQLMeshError(summary)
 
-    def _cleanup_environments(self, current_ts: t.Optional[int] = None) -> t.List[str]:
+    def _cleanup_environments(
+        self, current_ts: t.Optional[int] = None, force_delete: bool = False
+    ) -> t.List[str]:
         current_ts = current_ts or now_timestamp()
         failures: t.List[str] = []
 
@@ -2944,8 +2951,9 @@ def _cleanup_environments(self, current_ts: t.Optional[int] = None) -> t.List[st
                     )
                 )
 
-        # we want to retry on the next janitor pass if drops failed
-        if not failures:
+        # we want to retry on the next janitor pass if drops failed, unless
+        # force_delete is set in which case we purge state records regardless
+        if not failures or force_delete:
             self.state_sync.delete_expired_environments(current_ts=current_ts)
         return failures
 

sqlmesh/core/janitor.py

@@ -125,6 +125,7 @@ def delete_expired_snapshots(
     *,
     current_ts: int,
     ignore_ttl: bool = False,
+    force_delete: bool = False,
     batch_size: t.Optional[int] = None,
     console: t.Optional[Console] = None,
 ) -> t.List[str]:
@@ -138,6 +139,7 @@ def delete_expired_snapshots(
         snapshot_evaluator: SnapshotEvaluator instance to clean up tables associated with snapshots.
         current_ts: Timestamp used to evaluate expiration.
         ignore_ttl: If True, include snapshots regardless of TTL (only checks if unreferenced).
+        force_delete: If True, delete snapshot state records even when physical table cleanup fails.
         batch_size: Maximum number of snapshots to fetch per batch.
         console: Optional console for reporting progress.
 
@@ -162,23 +164,32 @@ def delete_expired_snapshots(
             len(batch.expired_snapshot_ids),
             end_info,
         )
+        cleanup_succeeded = True
         try:
             snapshot_evaluator.cleanup(
                 target_snapshots=batch.cleanup_tasks,
                 on_complete=console.update_cleanup_progress if console else None,
             )
-            state_sync.delete_expired_snapshots(
-                batch_range=ExpiredBatchRange(
-                    start=RowBoundary.lowest_boundary(),
-                    end=batch.batch_range.end,
-                ),
-                ignore_ttl=ignore_ttl,
-            )
-            logger.info("Cleaned up expired snapshots batch")
-            num_expired_snapshots += len(batch.expired_snapshot_ids)
-        except Exception as e:
-            message = f"Failed to clean up an expired snapshots batch: {e}"
+        except Exception as failed_drops:
+            message = f"Failed to clean up: {failed_drops}"
             logger.warning(message)
             failures.append(message)
+            cleanup_succeeded = False
+
+        if cleanup_succeeded or force_delete:
+            try:
+                state_sync.delete_expired_snapshots(
+                    batch_range=ExpiredBatchRange(
+                        start=RowBoundary.lowest_boundary(),
+                        end=batch.batch_range.end,
+                    ),
+                    ignore_ttl=ignore_ttl,
+                )
+                logger.info("Cleaned up expired snapshots batch")
+                num_expired_snapshots += len(batch.expired_snapshot_ids)
+            except Exception as e:
+                message = f"Failed to delete expired snapshot state records: {e}"
+                logger.warning(message)
+                failures.append(message)
     logger.info("Cleaned up %s expired snapshots", num_expired_snapshots)
     return failures

sqlmesh/core/snapshot/evaluator.py

@@ -547,7 +547,7 @@ def cleanup(
             t.snapshot.snapshot_id: t.dev_table_only for t in filtered_targets
         }
         with self.concurrent_context():
-            concurrent_apply_to_snapshots(
+            errors, _ = concurrent_apply_to_snapshots(
                 [t.snapshot for t in filtered_targets],
                 lambda s: self._cleanup_snapshot(
                     s,
@@ -557,7 +557,11 @@ def cleanup(
                 ),
                 self.ddl_concurrent_tasks,
                 reverse_order=True,
+                raise_on_error=False,
             )
+        if errors:
+            errored_snapshots = "\n".join(f"  {e.node.name}: {e.__cause__}" for e in errors)
+            raise SQLMeshError(f"\n{errored_snapshots}")
 
     def audit(
         self,

tests/core/integration/test_aux_commands.py

@@ -224,6 +224,79 @@ def test_janitor_warn_on_delete_failure_downgrades_aggregated_error(
     assert "Janitor completed with failures" in warn_spy.call_args[0][0]
 
 
+def test_janitor_force_delete_removes_environment_state_despite_drop_failure(
+    mocker: MockerFixture, tmp_path: Path
+):
+    models_dir = tmp_path / "models"
+    models_dir.mkdir()
+    (models_dir / "model1.sql").write_text("MODEL(name test.model1, kind FULL); SELECT 1 AS col")
+
+    ctx = Context(
+        paths=[tmp_path],
+        config=Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")),
+    )
+    ctx.plan("dev", no_prompts=True, auto_apply=True)
+    ctx.invalidate_environment("dev")
+
+    mocker.patch(
+        "sqlmesh.core.context.cleanup_expired_views",
+        return_value=["view drop error"],
+    )
+    mocker.patch(
+        "sqlmesh.core.janitor.iter_expired_snapshot_batches",
+        return_value=iter([]),
+    )
+
+    # without force_delete the environment is retained for retry
+    with pytest.raises(SQLMeshError):
+        ctx._run_janitor(ignore_ttl=True, force_delete=False)
+    assert ctx.state_sync.get_environment("dev") is not None
+
+    # with force_delete the environment state is purged even though drops failed
+    with pytest.raises(SQLMeshError):
+        ctx._run_janitor(ignore_ttl=True, force_delete=True)
+    assert ctx.state_sync.get_environment("dev") is None
+
+
+def test_janitor_force_delete_removes_snapshot_state_despite_cleanup_failure(
+    mocker: MockerFixture, tmp_path: Path
+):
+    models_dir = tmp_path / "models"
+    models_dir.mkdir()
+    model1_path = models_dir / "model1.sql"
+    model1_path.write_text("MODEL(name test.model1, kind FULL); SELECT 1 AS col")
+
+    # using warn_on_delete_failure so the janitor completes and we can inspect the state after
+    ctx = Context(
+        paths=[tmp_path],
+        config=Config(
+            model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+            janitor=JanitorConfig(warn_on_delete_failure=True),
+        ),
+    )
+    ctx.plan("dev", no_prompts=True, auto_apply=True)
+    model1_snapshot = ctx.get_snapshot("test.model1")
+
+    # simulating a zombie snapshot
+    model1_path.unlink()
+    ctx.load()
+    ctx.plan("dev", no_prompts=True, auto_apply=True)
+    ctx.invalidate_environment("dev")
+
+    mocker.patch(
+        "sqlmesh.core.snapshot.evaluator.SnapshotEvaluator.cleanup",
+        side_effect=Exception("table cleanup error"),
+    )
+
+    # without force_delete the snapshot state is retained for retry
+    ctx._run_janitor(ignore_ttl=True, force_delete=False)
+    assert ctx.state_sync.get_snapshots([model1_snapshot.snapshot_id])  # type: ignore
+
+    # with force_delete the snapshot state record is purged even though cleanup failed
+    ctx._run_janitor(ignore_ttl=True, force_delete=True)
+    assert not ctx.state_sync.get_snapshots([model1_snapshot.snapshot_id])  # type: ignore
+
+
 @use_terminal_console
 def test_destroy(copy_to_temp_path):
     # Testing project with two gateways to verify cleanup is performed across engines