Feat: Add destroy command to remove entire project resources

docs/reference/cli.md

@@ -23,6 +23,7 @@ Commands:
   create_external_models  Create a schema file containing external model...
   create_test             Generate a unit test fixture for a given model.
   dag                     Render the DAG as an html file.
+  destroy                 The destroy command removes all project resources.
   diff                    Show the diff between the local state and the...
   dlt_refresh             Attaches to a DLT pipeline with the option to...
   environments            Prints the list of SQLMesh environments with...
@@ -143,6 +144,17 @@ Options:
   --help               Show this message and exit.
 ```
 
+## destroy
+
+```
+Usage: sqlmesh destroy
+
+  Removes all project resources, including warehouse objects, state tables, the SQLMesh cache and any build artifacts.
+
+Options:
+  --help               Show this message and exit.
+```
+
 ## dlt_refresh
 
 ```

docs/reference/notebook.md

@@ -232,6 +232,13 @@ options:
   --file FILE, -f FILE  An optional file path to write the HTML output to.
 ```
 
+#### destroy
+```
+%destroy
+
+Removes all project resources, including warehouse objects, state tables, the SQLMesh cache and any build artifacts.
+```
+
 #### dlt_refresh
 ```
 %dlt_refresh PIPELINE [--table] TABLE [--force]

sqlmesh/cli/main.py

@@ -546,6 +546,19 @@ def janitor(ctx: click.Context, ignore_ttl: bool, **kwargs: t.Any) -> None:
     ctx.obj.run_janitor(ignore_ttl, **kwargs)
 
 
+@cli.command("destroy")
+@click.pass_context
+@error_handler
+@cli_analytics
+def destroy(ctx: click.Context, **kwargs: t.Any) -> None:
+    """
+    The destroy command removes all project resources.
+
+    This includes engine-managed objects, state tables, the SQLMesh cache and any build artifacts.
+    """
+    ctx.obj.destroy(**kwargs)
+
+
 @cli.command("dag")
 @click.argument("file", required=True)
 @click.option(

sqlmesh/core/console.py

@@ -181,6 +181,26 @@ def stop_cleanup(self, success: bool = True) -> None:
         """
 
 
+class DestroyConsole(abc.ABC):
+    """Console for describing a destroy operation"""
+
+    @abc.abstractmethod
+    def start_destroy(self) -> bool:
+        """Start a destroy operation.
+
+        Returns:
+            Whether or not the destroy operation should proceed
+        """
+
+    @abc.abstractmethod
+    def stop_destroy(self, success: bool = True) -> None:
+        """Indicates the destroy operation has ended
+
+        Args:
+            success: Whether or not the cleanup completed successfully
+        """
+
+
 class EnvironmentsConsole(abc.ABC):
     """Console for displaying environments"""
 
@@ -304,6 +324,7 @@ class Console(
     StateExporterConsole,
     StateImporterConsole,
     JanitorConsole,
+    DestroyConsole,
     EnvironmentsConsole,
     DifferenceConsole,
     TableDiffConsole,
@@ -744,6 +765,12 @@ def print_connection_config(
     ) -> None:
         pass
 
+    def start_destroy(self) -> bool:
+        return True
+
+    def stop_destroy(self, success: bool = True) -> None:
+        pass
+
 
 def make_progress_bar(
     message: str,
@@ -1092,6 +1119,25 @@ def stop_cleanup(self, success: bool = False) -> None:
         else:
             self.log_error("Cleanup failed!")
 
+    def start_destroy(self) -> bool:
+        self.log_warning(
+            (
+                "This will permanently delete all engine-managed objects, state tables and SQLMesh cache.\n"
+                "The operation is irreversible and may disrupt any currently running or scheduled plans.\n"
+                "Use this command only when you intend to fully reset the project."
+            )
+        )
+        if not self._confirm("Proceed?"):
+            self.log_error("Destroy aborted!")
+            return False
+        return True
+
+    def stop_destroy(self, success: bool = False) -> None:
+        if success:
+            self.log_success("Destroy completed successfully.")
+        else:
+            self.log_error("Destroy failed!")
+
     def start_promotion_progress(
         self,
         snapshots: t.List[SnapshotTableInfo],

sqlmesh/core/context.py

@@ -832,6 +832,19 @@ def run_janitor(self, ignore_ttl: bool) -> bool:
 
         return success
 
+    @python_api_analytics
+    def destroy(self) -> bool:
+        success = False
+
+        if self.console.start_destroy():
+            try:
+                self._destroy()
+                success = True
+            finally:
+                self.console.stop_destroy(success=success)
+
+        return success
+
     @t.overload
     def get_model(
         self, model_or_snapshot: ModelOrSnapshot, raise_if_missing: Literal[True] = True
@@ -2537,6 +2550,22 @@ def _context_diff(
             gateway_managed_virtual_layer=self.gateway_managed_virtual_layer,
         )
 
+    def _destroy(self) -> None:
+        # Invalidate all environments, including prod
+        for environment in self.state_reader.get_environments():
+            self.state_sync.invalidate_environment(name=environment.name, protect_prod=False)
+            self.console.log_success(f"Environment '{environment.name}' invalidated.")
+
+        # Run janitor to clean up all objects
+        self._run_janitor(ignore_ttl=True)
+
+        # Remove state tables, including backup tables
+        self.state_sync.remove_state(including_backup=True)
+        self.console.log_status_update("State tables removed.")
+
+        # Finally clear caches
+        self.clear_caches()
+
     def _run_janitor(self, ignore_ttl: bool = False) -> None:
         current_ts = now_timestamp()
 

sqlmesh/core/state_sync/base.py

@@ -347,13 +347,18 @@ def delete_expired_snapshots(
         """
 
     @abc.abstractmethod
-    def invalidate_environment(self, name: str) -> None:
+    def invalidate_environment(self, name: str, protect_prod: bool = True) -> None:
         """Invalidates the target environment by setting its expiration timestamp to now.
 
         Args:
             name: The name of the environment to invalidate.
+            protect_prod: If True, prevents invalidation of the production environment.
         """
 
+    @abc.abstractmethod
+    def remove_state(self, including_backup: bool = False) -> None:
+        """Removes the state store objects."""
+
     @abc.abstractmethod
     def remove_intervals(
         self,

sqlmesh/core/state_sync/db/environment.py

@@ -108,14 +108,15 @@ def update_environment_statements(
                 columns_to_types=self._environment_statements_columns_to_types,
             )
 
-    def invalidate_environment(self, name: str) -> None:
+    def invalidate_environment(self, name: str, protect_prod: bool = True) -> None:
         """Invalidates the environment.
 
         Args:
             name: The name of the environment
+            protect_prod: If True, prevents invalidation of the production environment.
         """
         name = name.lower()
-        if name == c.PROD:
+        if protect_prod and name == c.PROD:
             raise SQLMeshError("Cannot invalidate the production environment.")
 
         filter_expr = exp.column("name").eq(name)

sqlmesh/core/state_sync/db/facade.py

@@ -59,7 +59,7 @@
 from sqlmesh.core.state_sync.db.environment import EnvironmentState
 from sqlmesh.core.state_sync.db.snapshot import SnapshotState
 from sqlmesh.core.state_sync.db.version import VersionState
-from sqlmesh.core.state_sync.db.migrator import StateMigrator
+from sqlmesh.core.state_sync.db.migrator import StateMigrator, _backup_table_name
 from sqlmesh.utils.date import TimeLike, to_timestamp, time_like_to_str, now_timestamp
 from sqlmesh.utils.errors import ConflictingPlanError, SQLMeshError
 
@@ -270,8 +270,8 @@ def unpause_snapshots(
     ) -> None:
         self.snapshot_state.unpause_snapshots(snapshots, unpaused_dt, self.interval_state)
 
-    def invalidate_environment(self, name: str) -> None:
-        self.environment_state.invalidate_environment(name)
+    def invalidate_environment(self, name: str, protect_prod: bool = True) -> None:
+        self.environment_state.invalidate_environment(name, protect_prod)
 
     def get_expired_snapshots(
         self, current_ts: int, ignore_ttl: bool = False
@@ -313,18 +313,26 @@ def snapshots_exist(self, snapshot_ids: t.Iterable[SnapshotIdLike]) -> t.Set[Sna
     def nodes_exist(self, names: t.Iterable[str], exclude_external: bool = False) -> t.Set[str]:
         return self.snapshot_state.nodes_exist(names, exclude_external)
 
-    def reset(self, default_catalog: t.Optional[str]) -> None:
-        """Resets the state store to the state when it was first initialized."""
+    def remove_state(self, including_backup: bool = False) -> None:
+        """Removes the state store objects."""
         for table in (
             self.snapshot_state.snapshots_table,
             self.snapshot_state.auto_restatements_table,
             self.environment_state.environments_table,
+            self.environment_state.environment_statements_table,
             self.interval_state.intervals_table,
             self.plan_dags_table,
             self.version_state.versions_table,
         ):
             self.engine_adapter.drop_table(table)
+            if including_backup:
+                self.engine_adapter.drop_table(_backup_table_name(table))
+
         self.snapshot_state.clear_cache()
+
+    def reset(self, default_catalog: t.Optional[str]) -> None:
+        """Resets the state store to the state when it was first initialized."""
+        self.remove_state()
         self.migrate(default_catalog)
 
     @transactional()

sqlmesh/magics.py

@@ -1111,6 +1111,13 @@ def lint(self, context: Context, line: str) -> None:
         args = parse_argstring(self.lint, line)
         context.lint_models(args.models)
 
+    @magic_arguments()
+    @line_magic
+    @pass_sqlmesh_context
+    def destroy(self, context: Context, line: str) -> None:
+        """Removes all project resources, engine-managed objects, state tables and clears the SQLMesh cache."""
+        context.destroy()
+
 
 def register_magics() -> None:
     try: