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:
 

examples/custom_materializations/custom_materializations/custom_kind.py

@@ -24,8 +24,9 @@ def insert(
         query_or_df: QueryOrDF,
         model: Model,
         is_first_insert: bool,
+        render_kwargs: t.Dict[str, t.Any],
         **kwargs: t.Any,
     ) -> None:
         assert type(model.kind).__name__ == "ExtendedCustomKind"
 
-        self._replace_query_for_model(model, table_name, query_or_df)
+        self._replace_query_for_model(model, table_name, query_or_df, render_kwargs)

examples/custom_materializations/custom_materializations/full.py

@@ -17,6 +17,7 @@ def insert(
         query_or_df: QueryOrDF,
         model: Model,
         is_first_insert: bool,
+        render_kwargs: t.Dict[str, t.Any],
         **kwargs: t.Any,
     ) -> None:
-        self._replace_query_for_model(model, table_name, query_or_df)
+        self._replace_query_for_model(model, table_name, query_or_df, render_kwargs)

sqlmesh/core/dialect.py

@@ -22,6 +22,7 @@
 from sqlglot.schema import MappingSchema
 from sqlglot.tokens import Token
 
+from sqlmesh import EngineAdapter
 from sqlmesh.core.constants import MAX_MODEL_DEFINITION_SIZE
 from sqlmesh.utils.errors import SQLMeshError, ConfigError
 from sqlmesh.utils.pandas import columns_to_types_from_df
@@ -1134,31 +1135,45 @@ def select_from_values_for_batch_range(
     batch_start: int,
     batch_end: int,
     alias: str = "t",
+    source_columns: t.Optional[t.Set[str]] = None,
 ) -> exp.Select:
-    casted_columns = [
-        exp.alias_(exp.cast(exp.column(column), to=kind), column, copy=False)
-        for column, kind in columns_to_types.items()
-    ]
+    source_columns = source_columns or set(columns_to_types)
+    source_columns_to_types = EngineAdapter.get_source_columns_to_types(
+        columns_to_types, source_columns
+    )
 
     if not values:
         # Ensures we don't generate an empty VALUES clause & forces a zero-row output
         where = exp.false()
-        expressions = [tuple(exp.cast(exp.null(), to=kind) for kind in columns_to_types.values())]
+        expressions = [
+            tuple(exp.cast(exp.null(), to=kind) for kind in source_columns_to_types.values())
+        ]
     else:
         where = None
         expressions = [
-            tuple(transform_values(v, columns_to_types)) for v in values[batch_start:batch_end]
+            tuple(transform_values(v, source_columns_to_types))
+            for v in values[batch_start:batch_end]
         ]
 
-    values_exp = exp.values(expressions, alias=alias, columns=columns_to_types)
+    values_exp = exp.values(expressions, alias=alias, columns=source_columns_to_types)
     if values:
         # BigQuery crashes on `SELECT CAST(x AS TIMESTAMP) FROM UNNEST([NULL]) AS x`, but not
         # on `SELECT CAST(x AS TIMESTAMP) FROM UNNEST([CAST(NULL AS TIMESTAMP)]) AS x`. This
         # ensures nulls under the `Values` expression are cast to avoid similar issues.
-        for value, kind in zip(values_exp.expressions[0].expressions, columns_to_types.values()):
+        for value, kind in zip(
+            values_exp.expressions[0].expressions, source_columns_to_types.values()
+        ):
             if isinstance(value, exp.Null):
                 value.replace(exp.cast(value, to=kind))
 
+    casted_columns = [
+        exp.alias_(
+            exp.cast(exp.column(column) if column in source_columns else exp.Null(), to=kind),
+            column,
+            copy=False,
+        )
+        for column, kind in columns_to_types.items()
+    ]
     return exp.select(*casted_columns).from_(values_exp, copy=False).where(where, copy=False)
 
 

sqlmesh/core/engine_adapter/_typing.py

@@ -13,7 +13,7 @@
 
     snowpark = optional_import("snowflake.snowpark")
 
-    Query = t.Union[exp.Query, exp.DerivedTable]
+    Query = exp.Query
     PySparkSession = t.Union[pyspark.sql.SparkSession, pyspark.sql.connect.dataframe.SparkSession]
     PySparkDataFrame = t.Union[pyspark.sql.DataFrame, pyspark.sql.connect.dataframe.DataFrame]
 

sqlmesh/core/engine_adapter/athena.py

@@ -434,6 +434,7 @@ def replace_query(
         columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
+        source_columns: t.Optional[t.Set[str]] = None,
         **kwargs: t.Any,
     ) -> None:
         table = exp.to_table(table_name)
@@ -447,6 +448,7 @@ def replace_query(
             columns_to_types=columns_to_types,
             table_description=table_description,
             column_descriptions=column_descriptions,
+            source_columns=source_columns,
             **kwargs,
         )