diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py
index 178cd8d2e4..43feb86d63 100644
--- a/sqlmesh/core/plan/builder.py
+++ b/sqlmesh/core/plan/builder.py
@@ -726,17 +726,11 @@ def _get_orphaned_indirect_change_category(
                 # One of the new parents in the chain was breaking so this indirect snapshot is breaking
                 return SnapshotChangeCategory.INDIRECT_BREAKING
 
-            if SnapshotChangeCategory.FORWARD_ONLY in previous_parent_categories:
-                # One of the new parents in the chain was forward-only so this indirect snapshot is forward-only
-                indirect_category = SnapshotChangeCategory.FORWARD_ONLY
-            elif (
-                previous_parent_categories.intersection(
-                    {
-                        SnapshotChangeCategory.NON_BREAKING,
-                        SnapshotChangeCategory.INDIRECT_NON_BREAKING,
-                    }
-                )
-                and indirect_category != SnapshotChangeCategory.FORWARD_ONLY
+            if previous_parent_categories.intersection(
+                {
+                    SnapshotChangeCategory.NON_BREAKING,
+                    SnapshotChangeCategory.INDIRECT_NON_BREAKING,
+                }
             ):
                 # All changes in the chain were non-breaking so this indirect snapshot can be non-breaking too
                 indirect_category = SnapshotChangeCategory.INDIRECT_NON_BREAKING
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index e053e1e108..3fcad34187 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -1071,11 +1071,22 @@ def _cleanup_snapshot(
 
         evaluation_strategy = _evaluation_strategy(snapshot, adapter)
         for is_table_deployable, table_name in table_names:
-            evaluation_strategy.delete(
-                table_name,
-                is_table_deployable=is_table_deployable,
-                physical_schema=snapshot.physical_schema,
-            )
+            try:
+                evaluation_strategy.delete(
+                    table_name,
+                    is_table_deployable=is_table_deployable,
+                    physical_schema=snapshot.physical_schema,
+                )
+            except Exception:
+                # Use `get_data_object` to check if the table exists instead of `table_exists` since the former
+                # is based on `INFORMATION_SCHEMA` and avoids touching the table directly.
+                # This is important when the table name is malformed for some reason and running any statement
+                # that touches the table would result in an error.
+                if adapter.get_data_object(table_name) is not None:
+                    raise
+                logger.warning(
+                    "Skipping cleanup of table '%s' because it does not exist", table_name
+                )
 
             if on_complete is not None:
                 on_complete(table_name)
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py
index 4b028e148b..41d81202f3 100644
--- a/tests/core/test_snapshot_evaluator.py
+++ b/tests/core/test_snapshot_evaluator.py
@@ -438,12 +438,14 @@ def create_and_cleanup(name: str, dev_table_only: bool):
         return snapshot
 
     snapshot = create_and_cleanup("catalog.test_schema.test_model", True)
+    adapter_mock.get_data_object.assert_not_called()
     adapter_mock.drop_table.assert_called_once_with(
         f"catalog.sqlmesh__test_schema.test_schema__test_model__{snapshot.fingerprint.to_version()}__dev"
     )
     adapter_mock.reset_mock()
 
     snapshot = create_and_cleanup("test_schema.test_model", False)
+    adapter_mock.get_data_object.assert_not_called()
     adapter_mock.drop_table.assert_has_calls(
         [
             call(
@@ -455,6 +457,7 @@ def create_and_cleanup(name: str, dev_table_only: bool):
     adapter_mock.reset_mock()
 
     snapshot = create_and_cleanup("test_model", False)
+    adapter_mock.get_data_object.assert_not_called()
     adapter_mock.drop_table.assert_has_calls(
         [
             call(f"sqlmesh__default.test_model__{snapshot.fingerprint.to_version()}__dev"),
@@ -463,6 +466,59 @@ def create_and_cleanup(name: str, dev_table_only: bool):
     )
 
 
+def test_cleanup_fails(adapter_mock, make_snapshot):
+    adapter_mock.drop_table.side_effect = RuntimeError("test_error")
+
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = SqlModel(
+        name="catalog.test_schema.test_model",
+        kind=IncrementalByTimeRangeKind(time_column="a"),
+        storage_format="parquet",
+        query=parse_one("SELECT a FROM tbl WHERE ds BETWEEN @start_ds and @end_ds"),
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
+    snapshot.version = "test_version"
+
+    evaluator.promote([snapshot], EnvironmentNamingInfo(name="test_env"))
+    with pytest.raises(NodeExecutionFailedError) as exc_info:
+        evaluator.cleanup(
+            [SnapshotTableCleanupTask(snapshot=snapshot.table_info, dev_table_only=True)]
+        )
+
+    assert str(exc_info.value.__cause__) == "test_error"
+
+
+def test_cleanup_skip_missing_table(adapter_mock, make_snapshot):
+    adapter_mock.get_data_object.return_value = None
+    adapter_mock.drop_table.side_effect = RuntimeError("fail")
+
+    evaluator = SnapshotEvaluator(adapter_mock)
+
+    model = SqlModel(
+        name="catalog.test_schema.test_model",
+        kind=IncrementalByTimeRangeKind(time_column="a"),
+        storage_format="parquet",
+        query=parse_one("SELECT a FROM tbl WHERE ds BETWEEN @start_ds and @end_ds"),
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
+    snapshot.version = "test_version"
+
+    evaluator.promote([snapshot], EnvironmentNamingInfo(name="test_env"))
+    evaluator.cleanup([SnapshotTableCleanupTask(snapshot=snapshot.table_info, dev_table_only=True)])
+
+    adapter_mock.get_data_object.assert_called_once_with(
+        f"catalog.sqlmesh__test_schema.test_schema__test_model__{snapshot.fingerprint.to_version()}__dev"
+    )
+    adapter_mock.drop_table.assert_called_once_with(
+        f"catalog.sqlmesh__test_schema.test_schema__test_model__{snapshot.fingerprint.to_version()}__dev"
+    )
+
+
 def test_cleanup_external_model(mocker: MockerFixture, adapter_mock, make_snapshot):
     evaluator = SnapshotEvaluator(adapter_mock)