feat: add ignore destructive support

Author: eakmanrq

docs/concepts/models/overview.md

@@ -507,11 +507,15 @@ Some properties are only available in specific model kinds - see the [model conf
 :   Set this to true to indicate that all changes to this model should be [forward-only](../plans.md#forward-only-plans).
 
 ### on_destructive_change
-:   What should happen when a change to a [forward-only model](../../guides/incremental_time.md#forward-only-models) or incremental model in a [forward-only plan](../plans.md#forward-only-plans) causes a destructive modification to the table schema (i.e., requires dropping an existing column).
+:   What should happen when a change to a [forward-only model](../../guides/incremental_time.md#forward-only-models) or incremental model in a [forward-only plan](../plans.md#forward-only-plans) causes a destructive modification to the table schema (i.e., requires dropping an existing column or modifying column constraints in ways that could cause data loss).
 
     SQLMesh checks for destructive changes at plan time based on the model definition and run time based on the model's underlying physical tables.
 
-    Must be one of the following values: `allow`, `warn`, or `error` (default).
+    Must be one of the following values: `allow`, `warn`, `error` (default), or `ignore`.
+
+!!! warning "Ignore is Dangerous"
+
+    `ignore` is dangerous since it can result in error or data loss. It likely should never be used but could be useful as an "escape-hatch" or a way to workaround unexpected behavior.
 
 ### disable_restatement
 :   Set this to true to indicate that [data restatement](../plans.md#restatement-plans) is disabled for this model.

docs/guides/incremental_time.md

@@ -171,7 +171,12 @@ The check is performed at plan time based on the model definition. SQLMesh may n
 
 A model's `on_destructive_change` [configuration setting](../reference/model_configuration.md#incremental-models) determines what happens when SQLMesh detects a destructive change.
 
-By default, SQLMesh will error so no data is lost. You can set `on_destructive_change` to `warn` or `allow` in the model's `MODEL` block to allow destructive changes.
+By default, SQLMesh will error so no data is lost. You can set `on_destructive_change` to `warn` or `allow` in the model's `MODEL` block to allow destructive changes. 
+`ignore` can be used to not perform the schema change and allow the table's definition to diverge from the model definition.
+
+!!! warning "Ignore is Dangerous"
+
+    `ignore` is dangerous since it can result in error or data loss. It likely should never be used but could be useful as an "escape-hatch" or a way to workaround unexpected behavior.
 
 This example configures a model to silently `allow` destructive changes:
 

sqlmesh/core/engine_adapter/base.py

@@ -971,6 +971,8 @@ def get_alter_expressions(
         self,
         current_table_name: TableName,
         target_table_name: TableName,
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[exp.Alter]:
         """
         Determines the alter statements needed to change the current table into the structure of the target table.
@@ -979,6 +981,7 @@ def get_alter_expressions(
             current_table_name,
             self.columns(current_table_name),
             self.columns(target_table_name),
+            ignore_destructive=ignore_destructive,
         )
 
     def alter_table(
@@ -1463,6 +1466,7 @@ def scd_type_2_by_time(
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         truncate: bool = False,
         is_restatement: bool = False,
+        extra_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         **kwargs: t.Any,
     ) -> None:
         self._scd_type_2(
@@ -1480,6 +1484,7 @@ def scd_type_2_by_time(
             column_descriptions=column_descriptions,
             truncate=truncate,
             is_restatement=is_restatement,
+            extra_columns_to_types=extra_columns_to_types,
             **kwargs,
         )
 
@@ -1499,6 +1504,7 @@ def scd_type_2_by_column(
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         truncate: bool = False,
         is_restatement: bool = False,
+        extra_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         **kwargs: t.Any,
     ) -> None:
         self._scd_type_2(
@@ -1516,6 +1522,7 @@ def scd_type_2_by_column(
             column_descriptions=column_descriptions,
             truncate=truncate,
             is_restatement=is_restatement,
+            extra_columns_to_types=extra_columns_to_types,
             **kwargs,
         )
 
@@ -1538,6 +1545,7 @@ def _scd_type_2(
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         truncate: bool = False,
         is_restatement: bool = False,
+        extra_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         **kwargs: t.Any,
     ) -> None:
         def remove_managed_columns(
@@ -1934,6 +1942,14 @@ def remove_managed_columns(
                 )
             )
 
+            if extra_columns_to_types:
+                columns_to_types = columns_to_types.update(extra_columns_to_types)
+                select_columns = table_columns + [
+                    exp.cast(exp.Null(), dtype, copy=False).as_(col, copy=False, quoted=True)
+                    for col, dtype in extra_columns_to_types.items()
+                ]
+                query = query.select(*select_columns)
+
             self.replace_query(
                 target_table,
                 self.ensure_nulls_for_unmatched_after_join(query),

sqlmesh/core/engine_adapter/mixins.py

@@ -357,9 +357,15 @@ def _parse_clustering_key(self, clustering_key: t.Optional[str]) -> t.List[exp.E
         return parsed_cluster_key.expressions or [parsed_cluster_key.this]
 
     def get_alter_expressions(
-        self, current_table_name: TableName, target_table_name: TableName
+        self,
+        current_table_name: TableName,
+        target_table_name: TableName,
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[exp.Alter]:
-        expressions = super().get_alter_expressions(current_table_name, target_table_name)
+        expressions = super().get_alter_expressions(
+            current_table_name, target_table_name, ignore_destructive=ignore_destructive
+        )
 
         # check for a change in clustering
         current_table = exp.to_table(current_table_name)

sqlmesh/core/engine_adapter/trino.py

@@ -268,6 +268,7 @@ def _scd_type_2(
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         truncate: bool = False,
         is_restatement: bool = False,
+        extra_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         **kwargs: t.Any,
     ) -> None:
         if columns_to_types and self.current_catalog_type == "delta_lake":
@@ -291,6 +292,7 @@ def _scd_type_2(
             column_descriptions,
             truncate,
             is_restatement,
+            extra_columns_to_types,
             **kwargs,
         )
 

sqlmesh/core/model/kind.py

@@ -188,6 +188,7 @@ class OnDestructiveChange(str, Enum):
     ERROR = "ERROR"
     WARN = "WARN"
     ALLOW = "ALLOW"
+    IGNORE = "IGNORE"
 
     @property
     def is_error(self) -> bool:
@@ -201,6 +202,10 @@ def is_warn(self) -> bool:
     def is_allow(self) -> bool:
         return self == OnDestructiveChange.ALLOW
 
+    @property
+    def is_ignore(self) -> bool:
+        return self == OnDestructiveChange.IGNORE
+
 
 def _on_destructive_change_validator(
     cls: t.Type, v: t.Union[OnDestructiveChange, str, exp.Identifier]

sqlmesh/core/plan/builder.py

@@ -552,6 +552,7 @@ def _check_destructive_changes(self, directly_modified: t.Set[SnapshotId]) -> No
                     new.name,
                     old_columns_to_types,
                     new_columns_to_types,
+                    ignore_destructive=new.model.on_destructive_change.is_ignore,
                 )
 
                 if has_drop_alteration(schema_diff):

sqlmesh/core/schema_diff.py

@@ -556,6 +556,8 @@ def _alter_operation(
         current_type: t.Union[str, exp.DataType],
         root_struct: exp.DataType,
         new_kwarg: exp.ColumnDef,
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[TableAlterOperation]:
         # We don't copy on purpose here because current_type may need to be mutated inside
         # _get_operations (struct.expressions.pop and struct.expressions.insert)
@@ -570,6 +572,7 @@ def _alter_operation(
                         current_type,
                         new_type,
                         root_struct,
+                        ignore_destructive=ignore_destructive,
                     )
 
             if new_type.this == current_type.this == exp.DataType.Type.ARRAY:
@@ -587,6 +590,7 @@ def _alter_operation(
                             current_array_type,
                             new_array_type,
                             root_struct,
+                            ignore_destructive=ignore_destructive,
                         )
         if self._is_coerceable_type(current_type, new_type):
             return []
@@ -607,6 +611,8 @@ def _alter_operation(
                     col_pos,
                 )
             ]
+        if ignore_destructive:
+            return []
         return self._drop_operation(columns, root_struct, pos, root_struct) + self._add_operation(
             columns, pos, new_kwarg, struct, root_struct
         )
@@ -617,11 +623,16 @@ def _resolve_alter_operations(
         current_struct: exp.DataType,
         new_struct: exp.DataType,
         root_struct: exp.DataType,
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[TableAlterOperation]:
         operations = []
         for current_pos, current_kwarg in enumerate(current_struct.expressions.copy()):
             _, new_kwarg = self._get_matching_kwarg(current_kwarg, new_struct, current_pos)
-            assert new_kwarg
+            if new_kwarg is None:
+                if ignore_destructive:
+                    continue
+                raise ValueError("Cannot alter a column that is being dropped")
             _, new_type = _get_name_and_type(new_kwarg)
             _, current_type = _get_name_and_type(current_kwarg)
             columns = parent_columns + [TableAlterColumn.from_struct_kwarg(current_kwarg)]
@@ -636,6 +647,7 @@ def _resolve_alter_operations(
                     current_type,
                     root_struct,
                     new_kwarg,
+                    ignore_destructive=ignore_destructive,
                 )
             )
         return operations
@@ -646,62 +658,69 @@ def _get_operations(
         current_struct: exp.DataType,
         new_struct: exp.DataType,
         root_struct: exp.DataType,
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[TableAlterOperation]:
         root_struct = root_struct or current_struct
         parent_columns = parent_columns or []
         operations = []
-        operations.extend(
-            self._resolve_drop_operation(parent_columns, current_struct, new_struct, root_struct)
-        )
+        if not ignore_destructive:
+            operations.extend(
+                self._resolve_drop_operation(
+                    parent_columns, current_struct, new_struct, root_struct
+                )
+            )
         operations.extend(
             self._resolve_add_operations(parent_columns, current_struct, new_struct, root_struct)
         )
         operations.extend(
-            self._resolve_alter_operations(parent_columns, current_struct, new_struct, root_struct)
+            self._resolve_alter_operations(
+                parent_columns,
+                current_struct,
+                new_struct,
+                root_struct,
+                ignore_destructive=ignore_destructive,
+            )
         )
         return operations
 
     def _from_structs(
-        self, current_struct: exp.DataType, new_struct: exp.DataType
+        self,
+        current_struct: exp.DataType,
+        new_struct: exp.DataType,
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[TableAlterOperation]:
-        return self._get_operations([], current_struct, new_struct, current_struct)
+        return self._get_operations(
+            [], current_struct, new_struct, current_struct, ignore_destructive=ignore_destructive
+        )
 
-    def compare_structs(
-        self, table_name: t.Union[str, exp.Table], current: exp.DataType, new: exp.DataType
+    def _compare_structs(
+        self,
+        table_name: t.Union[str, exp.Table],
+        current: exp.DataType,
+        new: exp.DataType,
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[exp.Alter]:
-        """
-        Compares two schemas represented as structs.
-
-        Args:
-            current: The current schema.
-            new: The new schema.
-
-        Returns:
-            The list of table alter operations.
-        """
         return [
             op.expression(table_name, self.array_element_selector)
-            for op in self._from_structs(current, new)
+            for op in self._from_structs(current, new, ignore_destructive=ignore_destructive)
         ]
 
     def compare_columns(
         self,
         table_name: TableName,
         current: t.Dict[str, exp.DataType],
         new: t.Dict[str, exp.DataType],
+        *,
+        ignore_destructive: bool = False,
     ) -> t.List[exp.Alter]:
-        """
-        Compares two schemas represented as dictionaries of column names and types.
-
-        Args:
-            current: The current schema.
-            new: The new schema.
-
-        Returns:
-            The list of schema deltas.
-        """
-        return self.compare_structs(
-            table_name, columns_to_types_to_struct(current), columns_to_types_to_struct(new)
+        return self._compare_structs(
+            table_name,
+            columns_to_types_to_struct(current),
+            columns_to_types_to_struct(new),
+            ignore_destructive=ignore_destructive,
         )