Feat: include dropped column name in destructive change message (#3367)

sqlmesh/core/plan/builder.py

@@ -17,7 +17,7 @@
 from sqlmesh.core.context_diff import ContextDiff
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.plan.definition import Plan, SnapshotMapping, earliest_interval_start
-from sqlmesh.core.schema_diff import SchemaDiffer, has_drop_alteration
+from sqlmesh.core.schema_diff import SchemaDiffer, has_drop_alteration, get_dropped_column_names
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
     Snapshot,
@@ -464,12 +464,21 @@ def _check_destructive_changes(self, directly_modified: t.Set[SnapshotId]) -> No
                 )
 
                 if has_drop_alteration(schema_diff):
-                    warning_msg = f"Plan results in a destructive change to forward-only model '{snapshot.name}'s schema"
+                    dropped_column_names = get_dropped_column_names(schema_diff)
+                    dropped_column_str = (
+                        "', '".join(dropped_column_names) if dropped_column_names else None
+                    )
+                    dropped_column_msg = (
+                        f" that drops column{'s' if dropped_column_names and len(dropped_column_names) > 1 else ''} '{dropped_column_str}'"
+                        if dropped_column_str
+                        else ""
+                    )
+                    warning_msg = f"Plan results in a destructive change to forward-only model '{snapshot.name}'s schema{dropped_column_msg}."
                     if snapshot.model.on_destructive_change.is_warn:
                         logger.warning(warning_msg)
                     else:
                         raise PlanError(
-                            f"{warning_msg}. To allow this, change the model's `on_destructive_change` setting to `warn` or `allow` or include it in the plan's `--allow-destructive-model` option."
+                            f"{warning_msg} To allow this, change the model's `on_destructive_change` setting to `warn` or `allow` or include it in the plan's `--allow-destructive-model` option."
                         )
 
     def _categorize_snapshots(

sqlmesh/core/schema_diff.py

@@ -687,5 +687,15 @@ def has_drop_alteration(alter_expressions: t.List[exp.Alter]) -> bool:
     )
 
 
+def get_dropped_column_names(alter_expressions: t.List[exp.Alter]) -> t.List[str]:
+    dropped_columns = []
+    for actions in alter_expressions:
+        for action in actions.args.get("actions", []):
+            if isinstance(action, exp.Drop):
+                if action.kind == "COLUMN":
+                    dropped_columns.append(action.alias_or_name)
+    return dropped_columns
+
+
 def _get_name_and_type(struct: exp.ColumnDef) -> t.Tuple[exp.Identifier, exp.DataType]:
     return struct.this, struct.args["kind"]

sqlmesh/core/snapshot/evaluator.py

@@ -49,7 +49,7 @@
     SCDType2ByTimeKind,
     ViewKind,
 )
-from sqlmesh.core.schema_diff import has_drop_alteration
+from sqlmesh.core.schema_diff import has_drop_alteration, get_dropped_column_names
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
     Intervals,
@@ -1855,9 +1855,14 @@ def _check_destructive_schema_change(
     if snapshot.needs_destructive_check(allow_destructive_snapshots) and has_drop_alteration(
         alter_expressions
     ):
-        warning_msg = (
-            f"Plan results in a destructive change to forward-only table '{snapshot.name}'s schema."
+        dropped_column_names = get_dropped_column_names(alter_expressions)
+        dropped_column_str = "', '".join(dropped_column_names) if dropped_column_names else None
+        dropped_column_msg = (
+            f" that drops column{'s' if dropped_column_names and len(dropped_column_names) > 1 else ''} '{dropped_column_str}'"
+            if dropped_column_str
+            else ""
         )
+        warning_msg = f"Plan results in a destructive change to forward-only table '{snapshot.name}'s schema{dropped_column_msg}."
         if snapshot.model.on_destructive_change.is_warn:
             logger.warning(warning_msg)
             return

tests/conftest.py

@@ -366,8 +366,8 @@ def _make_function(node: Node, version: t.Optional[str] = None, **kwargs) -> Sna
 def make_snapshot_on_destructive_change(make_snapshot: t.Callable) -> t.Callable:
     def _make_function(
         name: str = "a",
-        old_query: str = "select '1' as one, '2022-01-01' ds",
-        new_query: str = "select 1 as one, '2022-01-01' ds",
+        old_query: str = "select '1' as one, '2' as two, '2022-01-01' ds",
+        new_query: str = "select 1 as one, 2 as two, '2022-01-01' ds",
         on_destructive_change: OnDestructiveChange = OnDestructiveChange.ERROR,
     ) -> t.Tuple[Snapshot, Snapshot]:
         snapshot_old = make_snapshot(

tests/core/test_plan.py

@@ -484,7 +484,7 @@ def test_forward_only_model_on_destructive_change(
 
     with pytest.raises(
         PlanError,
-        match="""Plan results in a destructive change to forward-only model '"a"'s schema.""",
+        match="""Plan results in a destructive change to forward-only model '"a"'s schema that drops columns 'one', 'two'.""",
     ):
         PlanBuilder(context_diff_1, schema_differ).build()
 

tests/core/test_snapshot_evaluator.py

@@ -1550,7 +1550,7 @@ def columns(table_name):
         evaluator.migrate([snapshot], {})
         assert (
             mock_logger.call_args[0][0]
-            == """Plan results in a destructive change to forward-only table '"test_schema"."test_model"'s schema."""
+            == """Plan results in a destructive change to forward-only table '"test_schema"."test_model"'s schema that drops column 'b'."""
         )
 
     # allow destructive