feat: support bigquery no op coerce types (#1566)

Author: eakmanrq

sqlmesh/core/engine_adapter/bigquery.py

@@ -54,14 +54,20 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
             exp.DataType.build("INT64", dialect=DIALECT): {
                 exp.DataType.build("NUMERIC", dialect=DIALECT),
                 exp.DataType.build("FLOAT64", dialect=DIALECT),
+                exp.DataType.build("BIGNUMERIC", dialect=DIALECT),
             },
             exp.DataType.build("NUMERIC", dialect=DIALECT): {
                 exp.DataType.build("FLOAT64", dialect=DIALECT),
+                exp.DataType.build("BIGNUMERIC", dialect=DIALECT),
+            },
+            exp.DataType.build("BIGNUMERIC", dialect=DIALECT): {
+                exp.DataType.build("FLOAT64", dialect=DIALECT),
             },
             exp.DataType.build("DATE", dialect=DIALECT): {
                 exp.DataType.build("DATETIME", dialect=DIALECT),
             },
         },
+        support_coercing_compatible_types=True,
     )
 
     @property

sqlmesh/core/schema_diff.py

@@ -1,13 +1,20 @@
 from __future__ import annotations
 
+import logging
 import typing as t
+from collections import defaultdict
 from enum import Enum, auto
 
 from sqlglot import exp
 from sqlglot.helper import ensure_list, seq_get
 
 from sqlmesh.utils.pydantic import PydanticModel
 
+if t.TYPE_CHECKING:
+    from sqlmesh.core._typing import TableName
+
+logger = logging.getLogger(__name__)
+
 
 class TableAlterOperationType(Enum):
     ADD = auto()
@@ -292,6 +299,9 @@ class SchemaDiffer(PydanticModel):
     support_nested_operations: bool = False
     array_element_selector: str = ""
     compatible_types: t.Dict[exp.DataType, t.Set[exp.DataType]] = {}
+    support_coercing_compatible_types: bool = False
+
+    _coerceable_types: t.Dict[exp.DataType, t.Set[exp.DataType]] = {}
 
     @classmethod
     def _dict_to_struct(cls, value: t.Dict[str, exp.DataType]) -> exp.DataType:
@@ -303,13 +313,37 @@ def _dict_to_struct(cls, value: t.Dict[str, exp.DataType]) -> exp.DataType:
             nested=True,
         )
 
+    @property
+    def coerceable_types(self) -> t.Dict[exp.DataType, t.Set[exp.DataType]]:
+        if not self._coerceable_types:
+            if not self.support_coercing_compatible_types or not self.compatible_types:
+                return {}
+            coerceable_types = defaultdict(set)
+            for source_type, target_types in self.compatible_types.items():
+                for target_type in target_types:
+                    coerceable_types[target_type].add(source_type)
+            self._coerceable_types = coerceable_types
+        return self._coerceable_types
+
     def _is_compatible_type(self, current_type: exp.DataType, new_type: exp.DataType) -> bool:
         if current_type == new_type:
             return True
         if current_type in self.compatible_types:
             return new_type in self.compatible_types[current_type]
         return False
 
+    def _is_coerceable_type(self, current_type: exp.DataType, new_type: exp.DataType) -> bool:
+        if not self.support_coercing_compatible_types:
+            return False
+        if current_type in self.coerceable_types:
+            is_coerceable = new_type in self.coerceable_types[current_type]
+            if is_coerceable:
+                logger.warning(
+                    f"Coercing type {current_type} to {new_type} which means an alter will not be performed and therefore the resulting table structure will not match what is in the query.\nUpdate your model to cast the value to {current_type} type in order to remove this warning.",
+                )
+            return is_coerceable
+        return False
+
     def _get_matching_kwarg(
         self,
         current_kwarg: t.Union[str, exp.ColumnDef],
@@ -438,7 +472,9 @@ def _alter_operation(
                         new_array_type,
                         root_struct,
                     )
-        if self._is_compatible_type(current_type, new_type):
+        if self._is_coerceable_type(current_type, new_type):
+            return []
+        elif self._is_compatible_type(current_type, new_type):
             struct.expressions.pop(pos)
             struct.expressions.insert(pos, new_kwarg)
             col_pos = (
@@ -535,7 +571,7 @@ def compare_structs(
 
     def compare_columns(
         self,
-        table_name: t.Union[str, exp.Table],
+        table_name: TableName,
         current: t.Dict[str, exp.DataType],
         new: t.Dict[str, exp.DataType],
     ) -> t.List[exp.AlterTable]:

tests/core/test_schema_diff.py

@@ -688,6 +688,46 @@ def test_schema_diff_calculate_type_transitions():
                 support_nested_operations=True,
             ),
         ),
+        # ############
+        # Coerce Tests
+        # ############
+        # Single coercion results in no operation
+        (
+            "STRUCT<id INT, name STRING, revenue FLOAT>",
+            "STRUCT<id INT, name STRING, revenue INT>",
+            [],
+            dict(
+                support_positional_add=True,
+                support_nested_operations=True,
+                support_coercing_compatible_types=True,
+                compatible_types={
+                    exp.DataType.build("INT"): {exp.DataType.build("FLOAT")},
+                },
+            ),
+        ),
+        # Coercion with an alter results in a single alter
+        (
+            "STRUCT<id INT, name STRING, revenue FLOAT, total INT>",
+            "STRUCT<id INT, name STRING, revenue INT, total FLOAT>",
+            [
+                TableAlterOperation.alter_type(
+                    TableAlterColumn.primitive("total"),
+                    "FLOAT",
+                    current_type="INT",
+                    # Note that the resulting table struct will not match what we defined as the desired
+                    # result since it could be coerced
+                    expected_table_struct="STRUCT<id INT, name STRING, revenue FLOAT, total FLOAT>",
+                )
+            ],
+            dict(
+                support_positional_add=False,
+                support_nested_operations=True,
+                support_coercing_compatible_types=True,
+                compatible_types={
+                    exp.DataType.build("INT"): {exp.DataType.build("FLOAT")},
+                },
+            ),
+        ),
     ],
 )
 def test_struct_diff(