Repository: TobikoData/sqlmesh
Pull request #2768 (2024-06-13 15:48:42)

Feat: make SchemaDiffer precision aware
Some data types can include user-specified precision parameters, such as decimal precision `DECIMAL(10,1)` or character length `VARCHAR(100)`. 

Currently, the SchemaDiffer outputs a DROP/ADD for all  precision changes, even though precision increases can be done with an ALTER. This PR makes the SchemaDiffer aware of precision such that it outputs ALTER statements when precision is increased.

Precision is implemented differently across engines and types:
- Some types have one parameter `VARCHAR(1)`, and others have two `DECIMAL(10,1)`.
- Sometimes a parameter is optional and has a default value (e.g., `VARCHAR` = `VARCHAR(256)`, `DECIMAL(10)` = `DECIMAL(10,0)`). The SchemaDiffer will automatically provide these default values where possible.
- Some engines/types allow explicitly making a type its maximum precision (e.g., `VARCHAR(max)`).  SchemaDiffer substitutes the corresponding numeric precision.
- Sometimes a parameter is optional and the default is "unlimited" precision (e.g., `VARCHAR` allows strings of any length up to system limits). SchemaDiffer will allow ALTERs from any explicit precision to an unlimited type.
- 
Default values for all engines and types are documented in [this Notion doc](https://www.notion.so/SchemaDiffer-type-docs-4464135715254cc597d638e4550c20dd?pvs=4).

## Complexities

Type aliases:
- Some engines provide multiple aliases for the same underlying type. For example, DuckDB has a single "unlimited" length character type and maps all of `VARCHAR`, `VARCHAR(1)`, `CHAR`, and others to that underlying type.
- In some engines, the aliases behave differently depending on whether a parameter is provided. For example, in Postgres `BPCHAR(1)` == `CHAR(1)` but `BPCHAR` != `CHAR`. 
- Therefore, direct comparison of parameter values only occurs WITHIN a SQLGlot `exp.DataType.Type`. If two type aliases are parsed to the same underlying SQLGlot type their parameters will be compared, if not SchemaDiffer will return DROP/ADD.


Patch:
diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py
index 00ff6973..d039f88e 100644
--- a/sqlmesh/core/engine_adapter/bigquery.py
+++ b/sqlmesh/core/engine_adapter/bigquery.py
@@ -73,6 +73,20 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
             },
         },
         support_coercing_compatible_types=True,
+        parameterized_type_defaults={
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(38, 9), (0,)],
+            exp.DataType.build("BIGDECIMAL", dialect=DIALECT).this: [(76.76, 38), (0,)],
+        },
+        types_with_unlimited_length={
+            # parameterized `STRING(n)` can ALTER to unparameterized `STRING`
+            exp.DataType.build("STRING", dialect=DIALECT).this: {
+                exp.DataType.build("STRING", dialect=DIALECT).this,
+            },
+            # parameterized `BYTES(n)` can ALTER to unparameterized `BYTES`
+            exp.DataType.build("BYTES", dialect=DIALECT).this: {
+                exp.DataType.build("BYTES", dialect=DIALECT).this,
+            },
+        },
     )
 
     @property
diff --git a/sqlmesh/core/engine_adapter/databricks.py b/sqlmesh/core/engine_adapter/databricks.py
index 368b6e4a..8e7b52c6 100644
--- a/sqlmesh/core/engine_adapter/databricks.py
+++ b/sqlmesh/core/engine_adapter/databricks.py
@@ -38,6 +38,9 @@ class DatabricksEngineAdapter(SparkEngineAdapter):
         support_positional_add=True,
         support_nested_operations=True,
         array_element_selector="element",
+        parameterized_type_defaults={
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(10, 0), (0,)],
+        },
     )
     CATALOG_SUPPORT = CatalogSupport.FULL_SUPPORT
     SUPPORTS_ROW_LEVEL_OP = True
diff --git a/sqlmesh/core/engine_adapter/duckdb.py b/sqlmesh/core/engine_adapter/duckdb.py
index 744c25d3..aa7c0e0d 100644
--- a/sqlmesh/core/engine_adapter/duckdb.py
+++ b/sqlmesh/core/engine_adapter/duckdb.py
@@ -18,6 +18,7 @@ from sqlmesh.core.engine_adapter.shared import (
     set_catalog,
 )
 from sqlmesh.utils import major_minor
+from sqlmesh.core.schema_diff import SchemaDiffer
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
@@ -29,6 +30,11 @@ class DuckDBEngineAdapter(LogicalMergeMixin, GetCurrentCatalogFromFunctionMixin)
     DIALECT = "duckdb"
     SUPPORTS_TRANSACTIONS = False
     CATALOG_SUPPORT = CatalogSupport.FULL_SUPPORT
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(18, 3), (0,)],
+        },
+    )
 
     # TODO: remove once we stop supporting DuckDB 0.9
     COMMENT_CREATION_TABLE, COMMENT_CREATION_VIEW = (
diff --git a/sqlmesh/core/engine_adapter/mssql.py b/sqlmesh/core/engine_adapter/mssql.py
index e9aa4423..a632f01d 100644
--- a/sqlmesh/core/engine_adapter/mssql.py
+++ b/sqlmesh/core/engine_adapter/mssql.py
@@ -26,6 +26,7 @@ from sqlmesh.core.engine_adapter.shared import (
     SourceQuery,
     set_catalog,
 )
+from sqlmesh.core.schema_diff import SchemaDiffer
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
@@ -48,6 +49,25 @@ class MSSQLEngineAdapter(
     COMMENT_CREATION_TABLE = CommentCreationTable.UNSUPPORTED
     COMMENT_CREATION_VIEW = CommentCreationView.UNSUPPORTED
     SUPPORTS_REPLACE_TABLE = False
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(18, 0), (0,)],
+            exp.DataType.build("BINARY", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("VARBINARY", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("CHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("VARCHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("NCHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("NVARCHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("TIME", dialect=DIALECT).this: [(7,)],
+            exp.DataType.build("DATETIME2", dialect=DIALECT).this: [(7,)],
+            exp.DataType.build("DATETIMEOFFSET", dialect=DIALECT).this: [(7,)],
+        },
+        max_parameter_length={
+            exp.DataType.build("VARBINARY", dialect=DIALECT).this: 2147483647,  # 2 GB
+            exp.DataType.build("VARCHAR", dialect=DIALECT).this: 2147483647,
+            exp.DataType.build("NVARCHAR", dialect=DIALECT).this: 2147483647,
+        },
+    )
 
     def columns(
         self,
diff --git a/sqlmesh/core/engine_adapter/mysql.py b/sqlmesh/core/engine_adapter/mysql.py
index 6b37145c..fd0538c8 100644
--- a/sqlmesh/core/engine_adapter/mysql.py
+++ b/sqlmesh/core/engine_adapter/mysql.py
@@ -18,6 +18,7 @@ from sqlmesh.core.engine_adapter.shared import (
     DataObjectType,
     set_catalog,
 )
+from sqlmesh.core.schema_diff import SchemaDiffer
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
@@ -39,6 +40,19 @@ class MySQLEngineAdapter(
     MAX_TABLE_COMMENT_LENGTH = 2048
     MAX_COLUMN_COMMENT_LENGTH = 1024
     SUPPORTS_REPLACE_TABLE = False
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            exp.DataType.build("BIT", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("BINARY", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(10, 0), (0,)],
+            exp.DataType.build("CHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("NCHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("TEXT", dialect=DIALECT).this: [(65535,)],
+            exp.DataType.build("TIME", dialect=DIALECT).this: [(0,)],
+            exp.DataType.build("DATETIME", dialect=DIALECT).this: [(0,)],
+            exp.DataType.build("TIMESTAMP", dialect=DIALECT).this: [(0,)],
+        },
+    )
 
     def get_current_catalog(self) -> t.Optional[str]:
         """Returns the catalog name of the current connection."""
diff --git a/sqlmesh/core/engine_adapter/postgres.py b/sqlmesh/core/engine_adapter/postgres.py
index 1e8a36ff..17805994 100644
--- a/sqlmesh/core/engine_adapter/postgres.py
+++ b/sqlmesh/core/engine_adapter/postgres.py
@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import logging
 import typing as t
-
 from sqlglot import exp
 
 from sqlmesh.core.engine_adapter.base_postgres import BasePostgresEngineAdapter
@@ -11,6 +10,7 @@ from sqlmesh.core.engine_adapter.mixins import (
     PandasNativeFetchDFSupportMixin,
 )
 from sqlmesh.core.engine_adapter.shared import set_catalog
+from sqlmesh.core.schema_diff import SchemaDiffer
 
 if t.TYPE_CHECKING:
     from sqlmesh.core.engine_adapter._typing import DF
@@ -29,6 +29,34 @@ class PostgresEngineAdapter(
     HAS_VIEW_BINDING = True
     CURRENT_CATALOG_EXPRESSION = exp.column("current_catalog")
     SUPPORTS_REPLACE_TABLE = False
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            # DECIMAL without precision is "up to 131072 digits before the decimal point; up to 16383 digits after the decimal point"
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(131072 + 16383, 16383), (0,)],
+            exp.DataType.build("CHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("TIME", dialect=DIALECT).this: [(6,)],
+            exp.DataType.build("TIMESTAMP", dialect=DIALECT).this: [(6,)],
+        },
+        types_with_unlimited_length={
+            # all can ALTER to `TEXT`
+            exp.DataType.build("TEXT", dialect=DIALECT).this: {
+                exp.DataType.build("VARCHAR", dialect=DIALECT).this,
+                exp.DataType.build("CHAR", dialect=DIALECT).this,
+                exp.DataType.build("BPCHAR", dialect=DIALECT).this,
+            },
+            # all can ALTER to unparameterized `VARCHAR`
+            exp.DataType.build("VARCHAR", dialect=DIALECT).this: {
+                exp.DataType.build("VARCHAR", dialect=DIALECT).this,
+                exp.DataType.build("CHAR", dialect=DIALECT).this,
+                exp.DataType.build("BPCHAR", dialect=DIALECT).this,
+                exp.DataType.build("TEXT", dialect=DIALECT).this,
+            },
+            # parameterized `BPCHAR(n)` can ALTER to unparameterized `BPCHAR`
+            exp.DataType.build("BPCHAR", dialect=DIALECT).this: {
+                exp.DataType.build("BPCHAR", dialect=DIALECT).this
+            },
+        },
+    )
 
     def _fetch_native_df(
         self, query: t.Union[exp.Expression, str], quote_identifiers: bool = False
diff --git a/sqlmesh/core/engine_adapter/redshift.py b/sqlmesh/core/engine_adapter/redshift.py
index 91c804c3..f54ca01d 100644
--- a/sqlmesh/core/engine_adapter/redshift.py
+++ b/sqlmesh/core/engine_adapter/redshift.py
@@ -20,6 +20,7 @@ from sqlmesh.core.engine_adapter.shared import (
     SourceQuery,
     set_catalog,
 )
+from sqlmesh.core.schema_diff import SchemaDiffer
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
@@ -39,6 +40,20 @@ class RedshiftEngineAdapter(
     # Redshift doesn't support comments for VIEWs WITH NO SCHEMA BINDING (which we always use)
     COMMENT_CREATION_VIEW = CommentCreationView.UNSUPPORTED
     SUPPORTS_REPLACE_TABLE = False
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            exp.DataType.build("VARBYTE", dialect=DIALECT).this: [(64000,)],
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(18, 0), (0,)],
+            exp.DataType.build("CHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("VARCHAR", dialect=DIALECT).this: [(256,)],
+            exp.DataType.build("NCHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("NVARCHAR", dialect=DIALECT).this: [(256,)],
+        },
+        max_parameter_length={
+            exp.DataType.build("CHAR", dialect=DIALECT).this: 4096,
+            exp.DataType.build("VARCHAR", dialect=DIALECT).this: 65535,
+        },
+    )
 
     def _columns_query(self, table: exp.Table) -> exp.Select:
         sql = (
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index e628dfb8..7f925d62 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -18,6 +18,7 @@ from sqlmesh.core.engine_adapter.shared import (
     SourceQuery,
     set_catalog,
 )
+from sqlmesh.core.schema_diff import SchemaDiffer
 from sqlmesh.utils import optional_import
 from sqlmesh.utils.errors import SQLMeshError
 
@@ -42,6 +43,21 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin):
     SUPPORTS_CLONING = True
     CATALOG_SUPPORT = CatalogSupport.FULL_SUPPORT
     CURRENT_CATALOG_EXPRESSION = exp.func("current_database")
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            exp.DataType.build("BINARY", dialect=DIALECT).this: [(8388608,)],
+            exp.DataType.build("VARBINARY", dialect=DIALECT).this: [(8388608,)],
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(38, 0), (0,)],
+            exp.DataType.build("CHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("NCHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("VARCHAR", dialect=DIALECT).this: [(16777216,)],
+            exp.DataType.build("TIME", dialect=DIALECT).this: [(9,)],
+            exp.DataType.build("TIMESTAMP", dialect=DIALECT).this: [(9,)],
+            exp.DataType.build("TIMESTAMP_LTZ", dialect=DIALECT).this: [(9,)],
+            exp.DataType.build("TIMESTAMP_NTZ", dialect=DIALECT).this: [(9,)],
+            exp.DataType.build("TIMESTAMP_TZ", dialect=DIALECT).this: [(9,)],
+        },
+    )
 
     @contextlib.contextmanager
     def session(self, properties: SessionProperties) -> t.Iterator[None]:
diff --git a/sqlmesh/core/engine_adapter/spark.py b/sqlmesh/core/engine_adapter/spark.py
index 2da8b2bf..16fba976 100644
--- a/sqlmesh/core/engine_adapter/spark.py
+++ b/sqlmesh/core/engine_adapter/spark.py
@@ -22,6 +22,7 @@ from sqlmesh.core.engine_adapter.shared import (
     SourceQuery,
     set_catalog,
 )
+from sqlmesh.core.schema_diff import SchemaDiffer
 from sqlmesh.utils import classproperty
 from sqlmesh.utils.errors import SQLMeshError
 
@@ -58,6 +59,12 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
 
     WAP_PREFIX = "wap_"
     BRANCH_PREFIX = "branch_"
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            # default decimal precision varies across backends
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(), (0,)],
+        },
+    )
 
     @property
     def connection(self) -> SparkSessionConnection:
diff --git a/sqlmesh/core/engine_adapter/trino.py b/sqlmesh/core/engine_adapter/trino.py
index fbeea496..47619a34 100644
--- a/sqlmesh/core/engine_adapter/trino.py
+++ b/sqlmesh/core/engine_adapter/trino.py
@@ -24,6 +24,7 @@ from sqlmesh.core.engine_adapter.shared import (
     SourceQuery,
     set_catalog,
 )
+from sqlmesh.core.schema_diff import SchemaDiffer
 from sqlmesh.utils.date import TimeLike
 
 if t.TYPE_CHECKING:
@@ -53,6 +54,14 @@ class TrinoEngineAdapter(
     SUPPORTS_REPLACE_TABLE = False
     DEFAULT_CATALOG_TYPE = "hive"
     QUOTE_IDENTIFIERS_IN_VIEWS = False
+    SCHEMA_DIFFER = SchemaDiffer(
+        parameterized_type_defaults={
+            # default decimal precision varies across backends
+            exp.DataType.build("DECIMAL", dialect=DIALECT).this: [(), (0,)],
+            exp.DataType.build("CHAR", dialect=DIALECT).this: [(1,)],
+            exp.DataType.build("TIMESTAMP", dialect=DIALECT).this: [(3,)],
+        },
+    )
 
     @property
     def connection(self) -> TrinoConnection:
diff --git a/sqlmesh/core/schema_diff.py b/sqlmesh/core/schema_diff.py
index b11bd37c..56566acb 100644
--- a/sqlmesh/core/schema_diff.py
+++ b/sqlmesh/core/schema_diff.py
@@ -4,7 +4,6 @@ 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
 
@@ -294,8 +293,29 @@ class SchemaDiffer(PydanticModel):
     change. As a result historical data is lost.
 
     Potential future improvements:
-    1. Support precision changes on columns like VARCHAR and DECIMAL. Each engine has different rules on what is allowed
-    2. Support column moves. Databricks Delta supports moves and would allow exact matches.
+    1. Support column moves. Databricks Delta supports moves and would allow exact matches.
+
+    Args:
+        support_positional_add: Whether the engine for which the diff is being computed supports adding columns in a
+            specific position in the set of existing columns.
+        support_nested_operations: Whether the engine for which the diff is being computed supports modifications to
+            nested data types like STRUCTs and ARRAYs.
+        compatible_types: Types that are compatible and automatically coerced in actions like UNION ALL. Dict key is data
+            type, and value is the set of types that are compatible with it.
+        support_coercing_compatible_types: Whether or not the engine for which the diff is being computed supports direct
+            coercion of compatible types.
+        parameterized_type_defaults: Default values for parameterized data types. Dict key is a sqlglot exp.DataType.Type,
+            but in the engine adapter specification we build it from the dialect string instead of specifying it directly.
+            Example: `exp.DataType.build("STRING", dialect=DIALECT).this` instead of the underlying `exp.DataType.Type.TEXT`
+            to which it parses. We do that because parameter default replacement will silently break if we specify type
+            directly and SQLGlot changes the dialect's mapping of type string to exp.DataType.Type. Dict value is default
+            values in a list, where the list index contains the remaining defaults given the number of parameter values
+            provided by the user. Example: if user provides 0 parameters "DECIMAL", we return index 0 values for the two
+            omitted parameters `(38, 9)` -> "DECIMAL(38,9)". Example: if user provides 1 parameters "DECIMAL(10)", we return
+            index 1 value for the one omitted parameters `(0,)` -> "DECIMAL(10,0)".
+        max_parameter_length: Numeric parameter values corresponding to "max". Example: `VARCHAR(max)` -> `VARCHAR(65535)`.
+        types_with_unlimited_length: Data types that accept values of any length up to system limits. Any explicitly
+            parameterized type can ALTER to its unlimited length version, along with different types in some engines.
     """
 
     support_positional_add: bool = False
@@ -303,6 +323,11 @@ class SchemaDiffer(PydanticModel):
     array_element_selector: str = ""
     compatible_types: t.Dict[exp.DataType, t.Set[exp.DataType]] = {}
     support_coercing_compatible_types: bool = False
+    parameterized_type_defaults: t.Dict[
+        exp.DataType.Type, t.List[t.Tuple[t.Union[int, float], ...]]
+    ] = {}
+    max_parameter_length: t.Dict[exp.DataType.Type, t.Union[int, float]] = {}
+    types_with_unlimited_length: t.Dict[exp.DataType.Type, t.Set[exp.DataType.Type]] = {}
 
     _coerceable_types: t.Dict[exp.DataType, t.Set[exp.DataType]] = {}
 
@@ -319,10 +344,16 @@ class SchemaDiffer(PydanticModel):
         return self._coerceable_types
 
     def _is_compatible_type(self, current_type: exp.DataType, new_type: exp.DataType) -> bool:
-        if current_type == new_type:
+        # types are identical or both types are parameterized and new has higher precision
+        # - default parameter values are automatically provided if not present
+        if current_type == new_type or self._is_precision_increase(current_type, new_type):
             return True
+        # types are un-parameterized and compatible
         if current_type in self.compatible_types:
             return new_type in self.compatible_types[current_type]
+        # new type is un-parameterized and has unlimited length, current type is compatible
+        if not new_type.expressions and new_type.this in self.types_with_unlimited_length:
+            return current_type.this in self.types_with_unlimited_length[new_type.this]
         return False
 
     def _is_coerceable_type(self, current_type: exp.DataType, new_type: exp.DataType) -> bool:
@@ -337,6 +368,46 @@ class SchemaDiffer(PydanticModel):
             return is_coerceable
         return False
 
+    def _is_precision_increase(self, current_type: exp.DataType, new_type: exp.DataType) -> bool:
+        if current_type.this == new_type.this and not current_type.is_type(
+            *exp.DataType.NESTED_TYPES
+        ):
+            current_params = self._get_type_parameters(current_type)
+            new_params = self._get_type_parameters(new_type)
+
+            if len(current_params) != len(new_params):
+                return False
+
+            return all(new >= current for current, new in zip(current_params, new_params))
+        return False
+
+    def _get_type_parameters(self, type: exp.DataType) -> t.List[t.Union[int, float]]:
+        def _str_to_number(string: str, allows_max_param: bool) -> t.Union[int, float]:
+            try:
+                return int(string)
+            except ValueError:
+                try:
+                    return float(string)
+                except ValueError:
+                    if allows_max_param and string.upper() == "MAX":
+                        return self.max_parameter_length[type.this]
+                    raise ValueError(f"Could not convert '{string}' to a number")
+
+        # extract existing parameters
+        params = [
+            _str_to_number(param.this.this, type.this in self.max_parameter_length)
+            for param in type.expressions
+        ]
+
+        # maybe get default parameter values
+        param_defaults: t.Tuple[t.Union[int, float], ...] = ()
+        if type.this in self.parameterized_type_defaults:
+            param_defaults_list = self.parameterized_type_defaults[type.this]
+            if len(params) < len(param_defaults_list):
+                param_defaults = param_defaults_list[len(params)]
+
+        return [*params, *param_defaults]
+
     def _get_matching_kwarg(
         self,
         current_kwarg: t.Union[str, exp.ColumnDef],


Test patch:
diff --git a/tests/core/test_schema_diff.py b/tests/core/test_schema_diff.py
index a5ebfa25..f5581c58 100644
--- a/tests/core/test_schema_diff.py
+++ b/tests/core/test_schema_diff.py
@@ -710,14 +710,31 @@ def test_schema_diff_calculate_type_transitions():
             ],
             {},
         ),
-        # Precision VARCHAR is a no-op with no changes
+        # ############
+        # Precision Tests
+        # ############
+        # Identical precision is a no-op with no changes
         (
             "STRUCT<id INT, address VARCHAR(120)>",
             "STRUCT<id INT, address VARCHAR(120)>",
             [],
             {},
         ),
-        # Change the precision bits of a VARCHAR
+        # Increase the precision of a type is ALTER
+        (
+            "STRUCT<id INT, address VARCHAR(120)>",
+            "STRUCT<id INT, address VARCHAR(121)>",
+            [
+                TableAlterOperation.alter_type(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR(121)",
+                    current_type="VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR(121)>",
+                )
+            ],
+            {},
+        ),
+        # Decrease the precision of a type is DROP/ADD
         (
             "STRUCT<id INT, address VARCHAR(120)>",
             "STRUCT<id INT, address VARCHAR(100)>",
@@ -739,6 +756,216 @@ def test_schema_diff_calculate_type_transitions():
                 support_nested_operations=True,
             ),
         ),
+        # Type with precision to same type with no precision and no default is DROP/ADD
+        (
+            "STRUCT<id INT, address VARCHAR(120)>",
+            "STRUCT<id INT, address VARCHAR>",
+            [
+                TableAlterOperation.drop(
+                    TableAlterColumn.primitive("address"),
+                    "STRUCT<id INT>",
+                    "VARCHAR(120)",
+                ),
+                TableAlterOperation.add(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR>",
+                    position=TableAlterColumnPosition.last("id"),
+                ),
+            ],
+            dict(
+                support_positional_add=True,
+            ),
+        ),
+        # Type with no precision and no default to same type with precision is DROP/ADD
+        (
+            "STRUCT<id INT, address VARCHAR>",
+            "STRUCT<id INT, address VARCHAR(120)>",
+            [
+                TableAlterOperation.drop(
+                    TableAlterColumn.primitive("address"),
+                    "STRUCT<id INT>",
+                    "VARCHAR",
+                ),
+                TableAlterOperation.add(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR(120)>",
+                    position=TableAlterColumnPosition.last("id"),
+                ),
+            ],
+            dict(
+                support_positional_add=True,
+            ),
+        ),
+        # Increase precision of a type from a default is ALTER
+        (
+            "STRUCT<id INT, address VARCHAR>",  # default of 1 --> VARCHAR(1)
+            "STRUCT<id INT, address VARCHAR(2)>",
+            [
+                TableAlterOperation.alter_type(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR(2)",
+                    current_type="VARCHAR",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR(2)>",
+                )
+            ],
+            dict(
+                parameterized_type_defaults={
+                    exp.DataType.build("VARCHAR").this: [(1,)],
+                },
+            ),
+        ),
+        # Decrease precision of a type to a default is DROP/ADD
+        (
+            "STRUCT<id INT, address VARCHAR(120)>",
+            "STRUCT<id INT, address VARCHAR>",  # default of 1 --> VARCHAR(1)
+            [
+                TableAlterOperation.drop(
+                    TableAlterColumn.primitive("address"),
+                    "STRUCT<id INT>",
+                    "VARCHAR(120)",
+                ),
+                TableAlterOperation.add(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR>",
+                    position=TableAlterColumnPosition.last("id"),
+                ),
+            ],
+            dict(
+                parameterized_type_defaults={
+                    exp.DataType.build("VARCHAR").this: [(1,)],
+                },
+                support_positional_add=True,
+            ),
+        ),
+        # Change the precision of a type to exact "max" value is ALTER
+        (
+            "STRUCT<id INT, address VARCHAR(120)>",
+            "STRUCT<id INT, address VARCHAR(max)>",
+            [
+                TableAlterOperation.alter_type(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR(max)",
+                    current_type="VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR(max)>",
+                )
+            ],
+            dict(
+                max_parameter_length={
+                    exp.DataType.build("VARCHAR").this: 120,
+                },
+            ),
+        ),
+        # Increase the precision of a type to larger "max" is ALTER
+        (
+            "STRUCT<id INT, address VARCHAR(120)>",
+            "STRUCT<id INT, address VARCHAR(max)>",
+            [
+                TableAlterOperation.alter_type(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR(max)",
+                    current_type="VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR(max)>",
+                )
+            ],
+            dict(
+                max_parameter_length={
+                    exp.DataType.build("VARCHAR").this: 121,
+                },
+            ),
+        ),
+        # Decrease the precision of a type from "max" to smaller value is DROP/ADD
+        (
+            "STRUCT<id INT, address VARCHAR(max)>",
+            "STRUCT<id INT, address VARCHAR(120)>",
+            [
+                TableAlterOperation.drop(
+                    TableAlterColumn.primitive("address"),
+                    "STRUCT<id INT>",
+                    "VARCHAR(max)",
+                ),
+                TableAlterOperation.add(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR(120)>",
+                    position=TableAlterColumnPosition.last("id"),
+                ),
+            ],
+            dict(
+                support_positional_add=True,
+                max_parameter_length={
+                    exp.DataType.build("VARCHAR").this: 121,
+                },
+            ),
+        ),
+        # Increase the precision of a type to no-precision unlimited is ALTER
+        (
+            "STRUCT<id INT, address VARCHAR(120)>",
+            "STRUCT<id INT, address VARCHAR>",
+            [
+                TableAlterOperation.alter_type(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR",
+                    current_type="VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR>",
+                )
+            ],
+            dict(
+                types_with_unlimited_length={
+                    exp.DataType.build("VARCHAR").this: {
+                        exp.DataType.build("VARCHAR").this,
+                    },
+                }
+            ),
+        ),
+        # Decrease the precision of a type from no-precision unlimited to any precision is DROP/ADD
+        (
+            "STRUCT<id INT, address VARCHAR>",
+            "STRUCT<id INT, address VARCHAR(120)>",
+            [
+                TableAlterOperation.drop(
+                    TableAlterColumn.primitive("address"),
+                    "STRUCT<id INT>",
+                    "VARCHAR",
+                ),
+                TableAlterOperation.add(
+                    TableAlterColumn.primitive("address"),
+                    "VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address VARCHAR(120)>",
+                    position=TableAlterColumnPosition.last("id"),
+                ),
+            ],
+            dict(
+                support_positional_add=True,
+                types_with_unlimited_length={
+                    exp.DataType.build("VARCHAR").this: {
+                        exp.DataType.build("VARCHAR").this,
+                    },
+                },
+            ),
+        ),
+        # Increase the precision of one type to a different no-precision unlimited type is ALTER
+        (
+            "STRUCT<id INT, address VARCHAR(120)>",
+            "STRUCT<id INT, address TEXT>",
+            [
+                TableAlterOperation.alter_type(
+                    TableAlterColumn.primitive("address"),
+                    "TEXT",
+                    current_type="VARCHAR(120)",
+                    expected_table_struct="STRUCT<id INT, address TEXT>",
+                )
+            ],
+            dict(
+                types_with_unlimited_length={
+                    exp.DataType.build("TEXT").this: {
+                        exp.DataType.build("VARCHAR").this,
+                    },
+                }
+            ),
+        ),
         # ############
         # Coerce Tests
         # ############
@@ -787,9 +1014,6 @@ def test_struct_diff(
     expected_diff: t.List[TableAlterOperation],
     config: t.Dict[str, t.Any],
 ):
-    config = {
-        **config,
-    }
     resolver = SchemaDiffer(**config)
     operations = resolver._from_structs(
         exp.DataType.build(current_struct), exp.DataType.build(new_struct)