Repository: TobikoData/sqlmesh
Pull request #2796 (2024-06-20 02:24:46)

Feat: Support custom materialization strategies
This update introduces an ability for users to provide implementation for their custom materialization strategies.

The custom implementation is provided in Python and in its simplest form looks something like this:

```python
import typing as t

from sqlmesh import CustomMaterialization, Model

if t.TYPE_CHECKING:
    from sqlmesh import QueryOrDF


class MyCustomMaterialization(CustomMaterialization):
    NAME = "my_custom_materialization"

    def insert(
        self,
        table_name: str,
        query_or_df: QueryOrDF,
        model: Model,
        is_first_insert: bool,
        **kwargs: t.Any,
    ) -> None:
        config_value = model.custom_materialization_properties["config_key"]
        self.adapter.execute(...)
```

In order to use the custom materialization strategy a new model kind `CUSTOM` has been introduced:

```sql
MODEL (
  ...,
  kind CUSTOM (
    materialization 'my_custom_materialization',
    materialization_properties (
      config_key = 'config_value'
    )
  )
);
```

There are two ways to add a new materialization strategy to a project: by adding a Python module with the implementation to the `materializations/ `project folder or by installing a Python package that contains the implementation.

## `materializations/` folder

SQLMesh loads all `*.py` files in this folder and registers custom materializations at project load time. This is the simplest way to add a custom materialization. 

However, this approach doesn't work when using the multi-repo setup or when using SQLMesh with Airflow. In the latter case, the custom implementation must be made available in the Airflow cluster, which can be achieved by packaging the custom materializations into a Python package.

## Python package

Custom materializations can be packaged into a Python package and exposed via [setuptools entrypoints](https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/#using-package-metadata) mechanism. Once the package is installed, SQLMesh will automatically load custom materializations from the entrypoint list.

Entrypoints can be specified in the `pyproject.toml` file of the package:
```
[project.entry-points."sqlmesh.materializations"]
my_materialization = "my_package.my_materialization:MyCustomMaterialization"
```

Or in `setup.py`:
```python
setup(
    ...,
    entry_points={
        "sqlmesh.materializations": [
            "my_materialization = my_package.my_materialization:MyCustomMaterialization",
        ],
    },
)
```

Documentation updates and guides are going to be a part of a follow up PR(s).


Patch:
diff --git a/examples/airflow/Dockerfile.template b/examples/airflow/Dockerfile.template
index a95542ce..29fe3c68 100644
--- a/examples/airflow/Dockerfile.template
+++ b/examples/airflow/Dockerfile.template
@@ -65,7 +65,11 @@ RUN pip install apache-airflow-providers-databricks==6.4.0 \
 # Install Deps
 USER root
 ADD setup.py /opt/sqlmesh/setup.py
-RUN mkdir /opt/sqlmesh/sqlmesh && touch /opt/sqlmesh/sqlmesh/__init__.py
-RUN chown -R airflow /opt/sqlmesh
+RUN mkdir /opt/sqlmesh/sqlmesh && touch /opt/sqlmesh/sqlmesh/__init__.py && chown -R airflow /opt/sqlmesh
+
+ADD examples/custom_materializations /opt/custom_materializations
+RUN chown -R airflow /opt/custom_materializations
+
 USER airflow
 RUN cd /opt/sqlmesh && pip install -e .[dbt]
+RUN cd /opt/custom_materializations && pip install -e .
diff --git a/examples/custom_materializations/custom_materializations/__init__.py b/examples/custom_materializations/custom_materializations/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/examples/custom_materializations/custom_materializations/full.py b/examples/custom_materializations/custom_materializations/full.py
new file mode 100644
index 00000000..79aa5023
--- /dev/null
+++ b/examples/custom_materializations/custom_materializations/full.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import typing as t
+
+from sqlmesh import CustomMaterialization, Model
+
+if t.TYPE_CHECKING:
+    from sqlmesh import QueryOrDF
+
+
+class CustomFullMaterialization(CustomMaterialization):
+    NAME = "custom_full"
+
+    def insert(
+        self,
+        table_name: str,
+        query_or_df: QueryOrDF,
+        model: Model,
+        is_first_insert: bool,
+        **kwargs: t.Any,
+    ) -> None:
+        self._replace_query_for_model(model, table_name, query_or_df)
diff --git a/examples/custom_materializations/setup.py b/examples/custom_materializations/setup.py
new file mode 100644
index 00000000..f2ce8176
--- /dev/null
+++ b/examples/custom_materializations/setup.py
@@ -0,0 +1,14 @@
+from setuptools import find_packages, setup
+
+setup(
+    name="custom_materializations",
+    packages=find_packages(include=["custom_materializations"]),
+    entry_points={
+        "sqlmesh.materializations": [
+            "custom_full_materialization = custom_materializations.full:CustomFullMaterialization",
+        ],
+    },
+    install_requires=[
+        "sqlmesh",
+    ],
+)
diff --git a/examples/sushi/models/active_customers.sql b/examples/sushi/models/active_customers.sql
new file mode 100644
index 00000000..813f5e82
--- /dev/null
+++ b/examples/sushi/models/active_customers.sql
@@ -0,0 +1,19 @@
+MODEL (
+  name sushi.active_customers,
+  kind CUSTOM (
+    materialization 'custom_full'
+  ),
+  owner jen,
+  cron '@daily',
+  tags (pii, fact),
+  grain customer_id,
+  description 'Sushi active customer data',
+  column_descriptions (
+    customer_id = 'customer_id uniquely identifies customers'
+  )
+);
+
+
+SELECT customer_id, zip
+FROM sushi.customers
+WHERE status = 'active'
diff --git a/setup.py b/setup.py
index 051f8f1e..4f7f9449 100644
--- a/setup.py
+++ b/setup.py
@@ -1,8 +1,11 @@
 import os
+from pathlib import Path
 from os.path import exists
 
 from setuptools import find_packages, setup
 
+custom_materializations_path = Path("examples/custom_materializations").absolute()
+
 description = open("README.md").read() if exists("README.md") else ""
 
 setup(
@@ -92,6 +95,7 @@ setup(
             "types-pytz",
             "types-requests==2.28.8",
             "typing-extensions",
+            f"custom_materializations @ file://localhost/{custom_materializations_path}",
         ],
         "cicdtest": [
             "dbt-databricks",
diff --git a/sqlmesh/__init__.py b/sqlmesh/__init__.py
index ed33bbb3..bfa0094d 100644
--- a/sqlmesh/__init__.py
+++ b/sqlmesh/__init__.py
@@ -25,6 +25,9 @@ from sqlmesh.core.engine_adapter import EngineAdapter as EngineAdapter
 from sqlmesh.core.macros import SQL as SQL, macro as macro
 from sqlmesh.core.model import Model as Model, model as model
 from sqlmesh.core.snapshot import Snapshot as Snapshot
+from sqlmesh.core.snapshot.evaluator import (
+    CustomMaterialization as CustomMaterialization,
+)
 from sqlmesh.utils import (
     debug_mode_enabled as debug_mode_enabled,
     enable_debug_mode as enable_debug_mode,
@@ -36,6 +39,10 @@ except ImportError:
     pass
 
 
+if t.TYPE_CHECKING:
+    from sqlmesh.core.engine_adapter._typing import QueryOrDF  # noqa: F401
+
+
 class RuntimeEnv(str, Enum):
     """Enum defining what environment SQLMesh is running in."""
 
diff --git a/sqlmesh/core/_typing.py b/sqlmesh/core/_typing.py
index 1791baea..197d07bf 100644
--- a/sqlmesh/core/_typing.py
+++ b/sqlmesh/core/_typing.py
@@ -8,3 +8,4 @@ if t.TYPE_CHECKING:
     TableName = t.Union[str, exp.Table]
     SchemaName = t.Union[str, exp.Table]
     SessionProperties = t.Dict[str, t.Union[exp.Expression, str, int, float, bool]]
+    CustomMaterializationProperties = t.Dict[str, t.Union[exp.Expression, str, int, float, bool]]
diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py
index 6baa09a5..cbecb9f6 100644
--- a/sqlmesh/core/console.py
+++ b/sqlmesh/core/console.py
@@ -53,6 +53,7 @@ SNAPSHOT_CHANGE_CATEGORY_STR = {
     SnapshotChangeCategory.FORWARD_ONLY: "Forward-only",
     SnapshotChangeCategory.INDIRECT_BREAKING: "Indirect Breaking",
     SnapshotChangeCategory.INDIRECT_NON_BREAKING: "Indirect Non-breaking",
+    SnapshotChangeCategory.METADATA: "Metadata",
 }
 
 
diff --git a/sqlmesh/core/constants.py b/sqlmesh/core/constants.py
index b2008dd1..4cc897b4 100644
--- a/sqlmesh/core/constants.py
+++ b/sqlmesh/core/constants.py
@@ -43,6 +43,7 @@ AUDITS = "audits"
 CACHE = ".cache"
 EXTERNAL_MODELS = "external_models"
 MACROS = "macros"
+MATERIALIZATIONS = "materializations"
 METRICS = "metrics"
 MODELS = "models"
 SEEDS = "seeds"
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index de8e711d..2f121ccf 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -497,6 +497,7 @@ def _create_parser(parser_type: t.Type[exp.Expression], table_keys: t.List[str])
                         ModelKindName.SCD_TYPE_2,
                         ModelKindName.SCD_TYPE_2_BY_TIME,
                         ModelKindName.SCD_TYPE_2_BY_COLUMN,
+                        ModelKindName.CUSTOM,
                     ) and self._match(TokenType.L_PAREN, advance=False):
                         props = self._parse_wrapped_csv(functools.partial(_parse_props, self))
                     else:
@@ -1069,3 +1070,19 @@ def normalize_and_quote(
     yield query
     if quote:
         quote_identifiers(query, dialect=dialect)
+
+
+def interpret_expression(e: exp.Expression) -> exp.Expression | str | int | float | bool:
+    if e.is_int:
+        return int(e.this)
+    if e.is_number:
+        return float(e.this)
+    if isinstance(e, (exp.Literal, exp.Boolean)):
+        return e.this
+    return e
+
+
+def interpret_key_value_pairs(
+    e: exp.Tuple,
+) -> t.Dict[str, exp.Expression | str | int | float | bool]:
+    return {i.this.name: interpret_expression(i.expression) for i in e.expressions}
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index 7f41302e..2ef6b7ad 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -182,10 +182,18 @@ class EngineAdapter:
         batch_size = self.DEFAULT_BATCH_SIZE if batch_size is None else batch_size
         if isinstance(query_or_df, (exp.Query, exp.DerivedTable)):
             return [SourceQuery(query_factory=lambda: query_or_df)]  # type: ignore
+
         if not columns_to_types:
             raise SQLMeshError(
-                "It is expected that if a DF is passed in then columns_to_types is set"
+                "It is expected that if a DataFrame is passed in then columns_to_types is set"
+            )
+        if query_or_df.empty:
+            raise SQLMeshError(
+                "Cannot construct source query from an empty DataFrame. This error is commonly "
+                "related to Python models that produce no data. For such models, consider yielding "
+                "from an empty generator if the resulting set is empty, i.e. use `yield from ()`."
             )
+
         return self._df_to_source_queries(
             query_or_df, columns_to_types, batch_size, target_table=target_table
         )
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/loader.py b/sqlmesh/core/loader.py
index 24d6be36..5d75d5bc 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -108,6 +108,8 @@ class Loader(abc.ABC):
         self._path_mtimes.clear()
         self._dag = DAG()
 
+        self._load_materializations()
+
         config_mtimes: t.Dict[Path, t.List[float]] = defaultdict(list)
         for context_path, config in self._context.configs.items():
             for config_file in context_path.glob("config.*"):
@@ -177,6 +179,9 @@ class Loader(abc.ABC):
     ) -> UniqueKeyDict[str, Audit]:
         """Loads all audits."""
 
+    def _load_materializations(self) -> None:
+        """Loads custom materializations."""
+
     def _load_metrics(self) -> UniqueKeyDict[str, MetricMeta]:
         return UniqueKeyDict("metrics")
 
@@ -377,6 +382,15 @@ class SqlMeshLoader(Loader):
 
         return models
 
+    def _load_materializations(self) -> None:
+        """Loads custom materializations."""
+        for context_path, config in self._context.configs.items():
+            for path in self._glob_paths(
+                context_path / c.MATERIALIZATIONS, config=config, extension=".py"
+            ):
+                if os.path.getsize(path):
+                    import_python_file(path, context_path)
+
     def _load_audits(
         self, macros: MacroRegistry, jinja_macros: JinjaMacroRegistry
     ) -> UniqueKeyDict[str, Audit]:
diff --git a/sqlmesh/core/model/__init__.py b/sqlmesh/core/model/__init__.py
index 7860458d..baac47ae 100644
--- a/sqlmesh/core/model/__init__.py
+++ b/sqlmesh/core/model/__init__.py
@@ -16,6 +16,7 @@ from sqlmesh.core.model.definition import (
     load_sql_based_model as load_sql_based_model,
 )
 from sqlmesh.core.model.kind import (
+    CustomKind as CustomKind,
     EmbeddedKind as EmbeddedKind,
     ExternalKind as ExternalKind,
     FullKind as FullKind,
diff --git a/sqlmesh/core/model/common.py b/sqlmesh/core/model/common.py
index ebb6e643..f3bcfd5b 100644
--- a/sqlmesh/core/model/common.py
+++ b/sqlmesh/core/model/common.py
@@ -141,6 +141,7 @@ properties_validator = field_validator(
     "physical_properties_",
     "virtual_properties_",
     "session_properties_",
+    "materialization_properties_",
     mode="before",
     check_fields=False,
 )(parse_properties)
diff --git a/sqlmesh/core/model/kind.py b/sqlmesh/core/model/kind.py
index 753d7cc9..ee05b336 100644
--- a/sqlmesh/core/model/kind.py
+++ b/sqlmesh/core/model/kind.py
@@ -12,7 +12,7 @@ from sqlglot.optimizer.simplify import gen
 from sqlglot.time import format_time
 
 from sqlmesh.core import dialect as d
-from sqlmesh.core.model.common import parse_properties
+from sqlmesh.core.model.common import parse_properties, properties_validator
 from sqlmesh.core.model.seed import CsvSettings
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import (
@@ -27,6 +27,7 @@ from sqlmesh.utils.pydantic import (
     field_validator,
     field_validator_v1_args,
     get_dialect,
+    validate_string,
 )
 
 if sys.version_info >= (3, 9):
@@ -36,6 +37,8 @@ else:
 
 
 if t.TYPE_CHECKING:
+    from sqlmesh.core._typing import CustomMaterializationProperties
+
     MODEL_KIND = t.TypeVar("MODEL_KIND", bound="_ModelKind")
 
 
@@ -107,6 +110,10 @@ class ModelKindMixin:
     def is_scd_type_2_by_column(self) -> bool:
         return self.model_kind_name == ModelKindName.SCD_TYPE_2_BY_COLUMN
 
+    @property
+    def is_custom(self) -> bool:
+        return self.model_kind_name == ModelKindName.CUSTOM
+
     @property
     def is_symbolic(self) -> bool:
         """A symbolic model is one that doesn't execute at all."""
@@ -149,6 +156,7 @@ class ModelKindName(str, ModelKindMixin, Enum):
     EMBEDDED = "EMBEDDED"
     SEED = "SEED"
     EXTERNAL = "EXTERNAL"
+    CUSTOM = "CUSTOM"
 
     @property
     def model_kind_name(self) -> t.Optional[ModelKindName]:
@@ -601,6 +609,70 @@ class ExternalKind(_ModelKind):
     name: Literal[ModelKindName.EXTERNAL] = ModelKindName.EXTERNAL
 
 
+class CustomKind(_ModelKind):
+    name: Literal[ModelKindName.CUSTOM] = ModelKindName.CUSTOM
+    materialization: str
+    materialization_properties_: t.Optional[exp.Tuple] = Field(
+        default=None, alias="materialization_properties"
+    )
+    forward_only: SQLGlotBool = False
+    disable_restatement: SQLGlotBool = False
+
+    _properties_validator = properties_validator
+
+    @field_validator("materialization", mode="before")
+    @classmethod
+    def _validate_materialization(cls, v: t.Any) -> str:
+        from sqlmesh.core.snapshot.evaluator import get_custom_materialization_type
+
+        materialization = validate_string(v)
+        # The below call fails if a materialization with the given name doesn't exist.
+        get_custom_materialization_type(materialization)
+        return materialization
+
+    @property
+    def materialization_properties(self) -> CustomMaterializationProperties:
+        """A dictionary of materialization properties."""
+        if not self.materialization_properties_:
+            return {}
+        return d.interpret_key_value_pairs(self.materialization_properties_)
+
+    @property
+    def data_hash_values(self) -> t.List[t.Optional[str]]:
+        return [
+            *super().data_hash_values,
+            self.materialization,
+            gen(self.materialization_properties_) if self.materialization_properties_ else None,
+        ]
+
+    @property
+    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
+        return [
+            *super().metadata_hash_values,
+            str(self.forward_only),
+            str(self.disable_restatement),
+        ]
+
+    def to_expression(self, **kwargs: t.Any) -> d.ModelKind:
+        return super().to_expression(
+            expressions=[
+                exp.Property(
+                    this=exp.var("materialization"),
+                    value=exp.Literal.string(self.materialization),
+                ),
+                exp.Property(
+                    this=exp.var("materialization_properties"),
+                    value=self.materialization_properties_,
+                ),
+                exp.Property(this=exp.var("forward_only"), value=exp.convert(self.forward_only)),
+                exp.Property(
+                    this=exp.var("disable_restatement"),
+                    value=exp.convert(self.disable_restatement),
+                ),
+            ],
+        )
+
+
 ModelKind = Annotated[
     t.Union[
         EmbeddedKind,
@@ -614,6 +686,7 @@ ModelKind = Annotated[
         ViewKind,
         SCDType2ByTimeKind,
         SCDType2ByColumnKind,
+        CustomKind,
     ],
     Field(discriminator="name"),
 ]
@@ -631,6 +704,7 @@ MODEL_KIND_NAME_TO_TYPE: t.Dict[str, t.Type[ModelKind]] = {
     ModelKindName.SCD_TYPE_2: SCDType2ByTimeKind,
     ModelKindName.SCD_TYPE_2_BY_TIME: SCDType2ByTimeKind,
     ModelKindName.SCD_TYPE_2_BY_COLUMN: SCDType2ByColumnKind,
+    ModelKindName.CUSTOM: CustomKind,
 }
 
 
diff --git a/sqlmesh/core/model/meta.py b/sqlmesh/core/model/meta.py
index b888f4d0..6f635551 100644
--- a/sqlmesh/core/model/meta.py
+++ b/sqlmesh/core/model/meta.py
@@ -19,6 +19,7 @@ from sqlmesh.core.model.common import (
     properties_validator,
 )
 from sqlmesh.core.model.kind import (
+    CustomKind,
     IncrementalByUniqueKeyKind,
     ModelKind,
     OnDestructiveChange,
@@ -42,7 +43,7 @@ from sqlmesh.utils.pydantic import (
 )
 
 if t.TYPE_CHECKING:
-    from sqlmesh.core._typing import SessionProperties
+    from sqlmesh.core._typing import CustomMaterializationProperties, SessionProperties
 
 AuditReference = t.Tuple[str, t.Dict[str, exp.Expression]]
 
@@ -424,20 +425,13 @@ class ModelMeta(_Node):
         if not self.session_properties_:
             return {}
 
-        def _interpret_expr(
-            e: exp.Expression,
-        ) -> t.Union[exp.Expression | str | int | float | bool]:
-            if e.is_int:
-                return int(e.this)
-            if e.is_number:
-                return float(e.this)
-            if isinstance(e, (exp.Literal, exp.Boolean)):
-                return e.this
-            return e
-
-        return {
-            e.this.name: _interpret_expr(e.expression) for e in self.session_properties_.expressions
-        }
+        return d.interpret_key_value_pairs(self.session_properties_)
+
+    @property
+    def custom_materialization_properties(self) -> CustomMaterializationProperties:
+        if isinstance(self.kind, CustomKind):
+            return self.kind.materialization_properties
+        return {}
 
     @property
     def all_references(self) -> t.List[Reference]:
diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py
index ced9f394..9b45e68d 100644
--- a/sqlmesh/core/plan/builder.py
+++ b/sqlmesh/core/plan/builder.py
@@ -23,8 +23,8 @@ from sqlmesh.core.snapshot import (
     DeployabilityIndex,
     Snapshot,
     SnapshotChangeCategory,
-    categorize_change,
 )
+from sqlmesh.core.snapshot.categorizer import categorize_change
 from sqlmesh.core.snapshot.definition import Interval, SnapshotId, start_date
 from sqlmesh.utils import columns_to_types_all_known, random_id
 from sqlmesh.utils.dag import DAG
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],
diff --git a/sqlmesh/core/snapshot/__init__.py b/sqlmesh/core/snapshot/__init__.py
index b3c65739..a474bcf1 100644
--- a/sqlmesh/core/snapshot/__init__.py
+++ b/sqlmesh/core/snapshot/__init__.py
@@ -1,4 +1,3 @@
-from sqlmesh.core.snapshot.categorizer import categorize_change as categorize_change
 from sqlmesh.core.snapshot.definition import (
     DeployabilityIndex as DeployabilityIndex,
     Intervals as Intervals,
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 2ff93e23..0740f36a 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -14,7 +14,7 @@ from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 
 from sqlmesh.core import constants as c
 from sqlmesh.core.audit import BUILT_IN_AUDITS, Audit, ModelAudit, StandaloneAudit
-from sqlmesh.core.model import Model, ModelKindMixin, ModelKindName, ViewKind
+from sqlmesh.core.model import Model, ModelKindMixin, ModelKindName, ViewKind, CustomKind
 from sqlmesh.core.model.definition import _Model
 from sqlmesh.core.node import IntervalUnit, NodeType
 from sqlmesh.utils import sanitize_name
@@ -423,6 +423,8 @@ class SnapshotTableInfo(PydanticModel, SnapshotInfoMixin, frozen=True):
     # This can be removed from this model once Pydantic 1 support is dropped (must remain in `Snapshot` though)
     base_table_name_override: t.Optional[str] = None
 
+    custom_materialization: t.Optional[str] = None
+
     def __lt__(self, other: SnapshotTableInfo) -> bool:
         return self.name < other.name
 
@@ -1017,6 +1019,13 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
     def table_info(self) -> SnapshotTableInfo:
         """Helper method to get the SnapshotTableInfo from the Snapshot."""
         self._ensure_categorized()
+
+        custom_materialization = (
+            self.model.kind.materialization
+            if self.is_model and isinstance(self.model.kind, CustomKind)
+            else None
+        )
+
         return SnapshotTableInfo(
             physical_schema=self.physical_schema,
             name=self.name,
@@ -1028,6 +1037,7 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
             change_category=self.change_category,
             kind_name=self.model_kind_name,
             node_type=self.node_type,
+            custom_materialization=custom_materialization,
         )
 
     @property
@@ -1161,6 +1171,12 @@ class Snapshot(PydanticModel, SnapshotInfoMixin):
     def expiration_ts(self) -> int:
         return to_timestamp(self.ttl, relative_base=to_datetime(self.updated_ts))
 
+    @property
+    def custom_materialization(self) -> t.Optional[str]:
+        if self.is_custom:
+            return t.cast(CustomKind, self.model.kind).materialization
+        return None
+
     def _ensure_categorized(self) -> None:
         if not self.change_category:
             raise SQLMeshError(f"Snapshot {self.snapshot_id} has not been categorized yet.")
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index c3ad5f3d..70e1d29e 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -25,6 +25,7 @@ from __future__ import annotations
 import abc
 import logging
 import typing as t
+import sys
 from collections import defaultdict
 from contextlib import contextmanager
 from functools import reduce
@@ -52,7 +53,6 @@ from sqlmesh.core.schema_diff import has_drop_alteration
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
     Intervals,
-    QualifiedViewName,
     Snapshot,
     SnapshotChangeCategory,
     SnapshotId,
@@ -67,6 +67,11 @@ from sqlmesh.utils.concurrency import (
 from sqlmesh.utils.date import TimeLike, now
 from sqlmesh.utils.errors import AuditError, ConfigError, SQLMeshError
 
+if sys.version_info >= (3, 12):
+    from importlib import metadata
+else:
+    import importlib_metadata as metadata  # type: ignore
+
 if t.TYPE_CHECKING:
     from sqlmesh.core.engine_adapter._typing import DF, QueryOrDF
     from sqlmesh.core.environment import EnvironmentNamingInfo
@@ -490,14 +495,19 @@ class SnapshotEvaluator:
 
         evaluation_strategy = _evaluation_strategy(snapshot, self.adapter)
 
+        # https://github.com/TobikoData/sqlmesh/issues/2609
+        # If there are no existing intervals yet; only consider this a first insert for the first snapshot in the batch
+        is_first_insert = not _intervals(snapshot, deployability_index) and batch_index == 0
+
         def apply(query_or_df: QueryOrDF, index: int = 0) -> None:
             if index > 0:
                 evaluation_strategy.append(
-                    snapshot,
-                    table_name,
-                    query_or_df,
-                    snapshots,
-                    deployability_index,
+                    table_name=table_name,
+                    query_or_df=query_or_df,
+                    model=snapshot.model,
+                    snapshot=snapshot,
+                    snapshots=snapshots,
+                    deployability_index=deployability_index,
                     batch_index=batch_index,
                     start=start,
                     end=end,
@@ -506,11 +516,13 @@ class SnapshotEvaluator:
             else:
                 logger.info("Inserting batch (%s, %s) into %s'", start, end, table_name)
                 evaluation_strategy.insert(
-                    snapshot,
-                    table_name,
-                    query_or_df,
-                    snapshots,
-                    deployability_index,
+                    table_name=table_name,
+                    query_or_df=query_or_df,
+                    is_first_insert=is_first_insert,
+                    model=snapshot.model,
+                    snapshot=snapshot,
+                    snapshots=snapshots,
+                    deployability_index=deployability_index,
                     batch_index=batch_index,
                     start=start,
                     end=end,
@@ -626,7 +638,6 @@ class SnapshotEvaluator:
         parent_snapshots_by_name[snapshot.name] = snapshot
 
         deployability_index = deployability_index or DeployabilityIndex.all_deployable()
-        is_snapshot_deployable = deployability_index.is_deployable(snapshot)
 
         common_render_kwargs: t.Dict[str, t.Any] = dict(
             engine_adapter=self.adapter,
@@ -661,10 +672,10 @@ class SnapshotEvaluator:
                 logger.info(f"Cloning table '{source_table_name}' into '{target_table_name}'")
 
                 evaluation_strategy.create(
-                    snapshot,
-                    tmp_table_name,
-                    False,
-                    is_snapshot_deployable,
+                    table_name=tmp_table_name,
+                    model=snapshot.model,
+                    is_table_deployable=False,
+                    snapshot=snapshot,
                     table_mapping={snapshot.name: tmp_table_name},
                     **create_render_kwargs,
                 )
@@ -685,10 +696,10 @@ class SnapshotEvaluator:
                     table_deployability_flags.append(True)
                 for is_table_deployable in table_deployability_flags:
                     evaluation_strategy.create(
-                        snapshot,
-                        snapshot.table_name(is_deployable=is_table_deployable),
-                        is_table_deployable,
-                        is_snapshot_deployable,
+                        table_name=snapshot.table_name(is_deployable=is_table_deployable),
+                        model=snapshot.model,
+                        is_table_deployable=is_table_deployable,
+                        snapshot=snapshot,
                         **create_render_kwargs,
                     )
 
@@ -722,11 +733,11 @@ class SnapshotEvaluator:
         tmp_table_name = snapshot.table_name(is_deployable=False)
         target_table_name = snapshot.table_name()
         _evaluation_strategy(snapshot, self.adapter).migrate(
-            snapshot,
-            parent_snapshots_by_name,
-            target_table_name,
-            tmp_table_name,
-            allow_destructive_snapshots,
+            target_table_name=target_table_name,
+            source_table_name=tmp_table_name,
+            snapshot=snapshot,
+            snapshots=parent_snapshots_by_name,
+            allow_destructive_snapshots=allow_destructive_snapshots,
         )
 
     def _promote_snapshot(
@@ -738,8 +749,14 @@ class SnapshotEvaluator:
     ) -> None:
         if snapshot.is_model:
             table_name = snapshot.table_name(deployability_index.is_representative(snapshot))
+            view_name = snapshot.qualified_view_name.for_environment(
+                environment_naming_info, dialect=self.adapter.dialect
+            )
             _evaluation_strategy(snapshot, self.adapter).promote(
-                snapshot.qualified_view_name, environment_naming_info, table_name, snapshot
+                table_name=table_name,
+                view_name=view_name,
+                model=snapshot.model,
+                environment=environment_naming_info.name,
             )
 
         if on_complete is not None:
@@ -751,9 +768,10 @@ class SnapshotEvaluator:
         environment_naming_info: EnvironmentNamingInfo,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
     ) -> None:
-        _evaluation_strategy(snapshot, self.adapter).demote(
-            snapshot.qualified_view_name, environment_naming_info
+        view_name = snapshot.qualified_view_name.for_environment(
+            environment_naming_info, dialect=self.adapter.dialect
         )
+        _evaluation_strategy(snapshot, self.adapter).demote(view_name)
 
         if on_complete is not None:
             on_complete(snapshot)
@@ -880,6 +898,13 @@ def _evaluation_strategy(snapshot: SnapshotInfoLike, adapter: EngineAdapter) ->
         klass = ViewStrategy
     elif snapshot.is_scd_type_2:
         klass = SCDType2Strategy
+    elif snapshot.is_custom:
+        if snapshot.custom_materialization is None:
+            raise SQLMeshError(
+                f"Missing the name of a custom evaluation strategy in model '{snapshot.name}'."
+            )
+        klass = get_custom_materialization_type(snapshot.custom_materialization)
+        return klass(adapter)
     else:
         raise SQLMeshError(f"Unexpected snapshot: {snapshot}")
 
@@ -893,87 +918,75 @@ class EvaluationStrategy(abc.ABC):
     @abc.abstractmethod
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        """Inserts the given query or a DataFrame into the target table or replaces a view.
+        """Inserts the given query or a DataFrame into the target table or a view.
 
         Args:
-            snapshot: The target snapshot.
-            name: The name of the target table or view.
-            query_or_df: The query or DataFrame to insert.
-            snapshots: Parent snapshots.
-            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
-            batch_index: Snapshot position in the current batch
+            table_name: The name of the target table or view.
+            query_or_df: A query or a DataFrame to insert.
+            model: The target model.
+            is_first_insert: Whether this is the first insert for this version of a model. This value is set to True
+                if no data has been previously inserted into the target table, or when the entire history of the target model has
+                been restated. Note that in the latter case, the table might contain data from previous executions, and it is the
+                responsibility of a specific evaluation strategy to handle the truncation of the table if necessary.
         """
 
     @abc.abstractmethod
     def append(
         self,
-        snapshot: Snapshot,
         table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
         **kwargs: t.Any,
     ) -> None:
         """Appends the given query or a DataFrame to the existing table.
 
         Args:
-            snapshot: The target snapshot.
             table_name: The target table name.
-            query_or_df: The query or DataFrame to insert.
-            snapshots: Parent snapshots.
-            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
-            batch_index: Snapshot position in the current batch
+            query_or_df: A query or a DataFrame to insert.
+            model: The target model.
         """
 
     @abc.abstractmethod
     def create(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
+        model: Model,
         is_table_deployable: bool,
-        is_snapshot_deployable: bool,
         **render_kwargs: t.Any,
     ) -> None:
         """Creates the target table or view.
 
         Args:
-            snapshot: The target snapshot.
-            name: The name of a table or a view.
+            table_name: The name of a table or a view.
+            model: The target model.
             is_table_deployable: Whether the table that is being created is deployable.
-            is_snapshot_deployable: Whether the snapshot is considered deployable.
             render_kwargs: Additional kwargs for node rendering.
         """
 
     @abc.abstractmethod
     def migrate(
         self,
-        snapshot: Snapshot,
-        snapshots: t.Dict[str, Snapshot],
         target_table_name: str,
         source_table_name: str,
-        allow_destructive_snapshots: t.Set[str] = set(),
+        snapshot: Snapshot,
+        **kwargs: t.Any,
     ) -> None:
         """Migrates the target table schema so that it corresponds to the source table schema.
 
         Args:
-            snapshot: The target snapshot.
-            snapshots: Parent snapshots.
             target_table_name: The target table name.
             source_table_name: The source table name.
-            allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
+            snapshot: The target snapshot.
         """
 
     @abc.abstractmethod
-    def delete(self, name: str) -> None:
+    def delete(self, name: str, **kwargs: t.Any) -> None:
         """Deletes a target table or a view.
 
         Args:
@@ -983,31 +996,27 @@ class EvaluationStrategy(abc.ABC):
     @abc.abstractmethod
     def promote(
         self,
-        view_name: QualifiedViewName,
-        environment_naming_info: EnvironmentNamingInfo,
         table_name: str,
-        snapshot: Snapshot,
+        view_name: str,
+        model: Model,
+        environment: str,
+        **kwargs: t.Any,
     ) -> None:
         """Updates the target view to point to the target table.
 
         Args:
-            view_name: The name of the target view.
-            environment_naming_info: The naming information for the target environment
-            table_name: The name of the target table.
-            snapshot: The target snapshot. Not ucrrently used but can be used by others if needed.
+            table_name: The name of a table in the physical layer that is being promoted.
+            view_name: The name of the target view in the virtual layer.
+            model: The model that is being promoted.
+            environment: The name of the target environment.
         """
 
     @abc.abstractmethod
-    def demote(
-        self,
-        view_name: QualifiedViewName,
-        environment_naming_info: EnvironmentNamingInfo,
-    ) -> None:
-        """Deletes the target view.
+    def demote(self, view_name: str, **kwargs: t.Any) -> None:
+        """Deletes the target view in the virtual layer.
 
         Args:
-            view_name: The name of the target view.
-            environment_naming_info: Naming information for the target environment.
+            view_name: The name of the target view in the virtual layer.
         """
 
     def _replace_query_for_model(self, model: Model, name: str, query_or_df: QueryOrDF) -> None:
@@ -1035,145 +1044,118 @@ class EvaluationStrategy(abc.ABC):
 class SymbolicStrategy(EvaluationStrategy):
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
         pass
 
     def append(
         self,
-        snapshot: Snapshot,
         table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
         **kwargs: t.Any,
     ) -> None:
         pass
 
     def create(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
+        model: Model,
         is_table_deployable: bool,
-        is_snapshot_deployable: bool,
         **render_kwargs: t.Any,
     ) -> None:
         pass
 
     def migrate(
         self,
-        snapshot: Snapshot,
-        snapshots: t.Dict[str, Snapshot],
         target_table_name: str,
         source_table_name: str,
-        allow_destructive_snapshots: t.Set[str] = set(),
+        snapshot: Snapshot,
+        **kwarg: t.Any,
     ) -> None:
         pass
 
-    def delete(self, name: str) -> None:
+    def delete(self, name: str, **kwargs: t.Any) -> None:
         pass
 
     def promote(
         self,
-        view_name: QualifiedViewName,
-        environment_naming_info: EnvironmentNamingInfo,
         table_name: str,
-        snapshot: Snapshot,
+        view_name: str,
+        model: Model,
+        environment: str,
+        **kwargs: t.Any,
     ) -> None:
         pass
 
-    def demote(
-        self,
-        view_name: QualifiedViewName,
-        environment_naming_info: EnvironmentNamingInfo,
-    ) -> None:
+    def demote(self, view_name: str, **kwargs: t.Any) -> None:
         pass
 
 
 class EmbeddedStrategy(SymbolicStrategy):
     def promote(
         self,
-        view_name: QualifiedViewName,
-        environment_naming_info: EnvironmentNamingInfo,
         table_name: str,
-        snapshot: Snapshot,
+        view_name: str,
+        model: Model,
+        environment: str,
+        **kwargs: t.Any,
     ) -> None:
-        target_name = view_name.for_environment(
-            environment_naming_info, dialect=self.adapter.dialect
-        )
-        logger.info("Dropping view '%s' for non-materialized table", target_name)
-        self.adapter.drop_view(target_name, cascade=False)
+        logger.info("Dropping view '%s' for non-materialized table", view_name)
+        self.adapter.drop_view(view_name, cascade=False)
 
 
 class PromotableStrategy(EvaluationStrategy):
     def promote(
         self,
-        view_name: QualifiedViewName,
-        environment_naming_info: EnvironmentNamingInfo,
         table_name: str,
-        snapshot: Snapshot,
+        view_name: str,
+        model: Model,
+        environment: str,
+        **kwargs: t.Any,
     ) -> None:
-        target_name = view_name.for_environment(
-            environment_naming_info, dialect=self.adapter.dialect
-        )
-        is_prod = environment_naming_info.name.lower() == c.PROD
-        logger.info("Updating view '%s' to point at table '%s'", target_name, table_name)
+        is_prod = environment == c.PROD
+        logger.info("Updating view '%s' to point at table '%s'", view_name, table_name)
         self.adapter.create_view(
-            target_name,
+            view_name,
             exp.select("*").from_(table_name, dialect=self.adapter.dialect),
-            table_description=snapshot.model.description if is_prod else None,
-            column_descriptions=snapshot.model.column_descriptions if is_prod else None,
-            view_properties=snapshot.model.virtual_properties,
+            table_description=model.description if is_prod else None,
+            column_descriptions=model.column_descriptions if is_prod else None,
+            view_properties=model.virtual_properties,
         )
 
-    def demote(
-        self,
-        view_name: QualifiedViewName,
-        environment_naming_info: EnvironmentNamingInfo,
-    ) -> None:
-        target_name = view_name.for_environment(
-            environment_naming_info, dialect=self.adapter.dialect
-        )
-        logger.info("Dropping view '%s'", target_name)
-        self.adapter.drop_view(target_name, cascade=False)
+    def demote(self, view_name: str, **kwargs: t.Any) -> None:
+        logger.info("Dropping view '%s'", view_name)
+        self.adapter.drop_view(view_name, cascade=False)
 
 
 class MaterializableStrategy(PromotableStrategy):
     def append(
         self,
-        snapshot: Snapshot,
         table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
         self.adapter.insert_append(table_name, query_or_df, columns_to_types=model.columns_to_types)
 
     def create(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
+        model: Model,
         is_table_deployable: bool,
-        is_snapshot_deployable: bool,
         **render_kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
         ctas_query = model.ctas_query(**render_kwargs)
 
-        logger.info("Creating table '%s'", name)
+        logger.info("Creating table '%s'", table_name)
         if model.annotated:
             self.adapter.create_table(
-                name,
+                table_name,
                 columns_to_types=model.columns_to_types_or_raise,
                 storage_format=model.storage_format,
                 partitioned_by=model.partitioned_by,
@@ -1194,7 +1176,7 @@ class MaterializableStrategy(PromotableStrategy):
                 self.adapter.fetchall(ctas_query)
         else:
             self.adapter.ctas(
-                name,
+                table_name,
                 ctas_query,
                 model.columns_to_types,
                 storage_format=model.storage_format,
@@ -1208,36 +1190,34 @@ class MaterializableStrategy(PromotableStrategy):
 
     def migrate(
         self,
-        snapshot: Snapshot,
-        snapshots: t.Dict[str, Snapshot],
         target_table_name: str,
         source_table_name: str,
-        allow_destructive_snapshots: t.Set[str] = set(),
+        snapshot: Snapshot,
+        **kwargs: t.Any,
     ) -> None:
         logger.info(f"Altering table '{target_table_name}'")
         alter_expressions = self.adapter.get_alter_expressions(target_table_name, source_table_name)
-        _check_destructive_schema_change(snapshot, alter_expressions, allow_destructive_snapshots)
+        _check_destructive_schema_change(
+            snapshot, alter_expressions, kwargs["allow_destructive_snapshots"]
+        )
         self.adapter.alter_table(alter_expressions)
 
-    def delete(self, table_name: str) -> None:
-        self.adapter.drop_table(table_name)
-        logger.info("Dropped table '%s'", table_name)
+    def delete(self, name: str, **kwargs: t.Any) -> None:
+        self.adapter.drop_table(name)
+        logger.info("Dropped table '%s'", name)
 
 
 class IncrementalByPartitionStrategy(MaterializableStrategy):
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
         self.adapter.insert_overwrite_by_partition(
-            name,
+            table_name,
             query_or_df,
             partitioned_by=model.partitioned_by,
             columns_to_types=model.columns_to_types,
@@ -1247,18 +1227,15 @@ class IncrementalByPartitionStrategy(MaterializableStrategy):
 class IncrementalByTimeRangeStrategy(MaterializableStrategy):
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
         assert model.time_column
         self.adapter.insert_overwrite_by_time_partition(
-            name,
+            table_name,
             query_or_df,
             time_formatter=model.convert_to_time_column,
             time_column=model.time_column,
@@ -1270,23 +1247,17 @@ class IncrementalByTimeRangeStrategy(MaterializableStrategy):
 class IncrementalByUniqueKeyStrategy(MaterializableStrategy):
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
-        # https://github.com/TobikoData/sqlmesh/issues/2609
-        # If there are no existing intervals yet; only clear the table for the first snapshot in the batch
-        # Otherwise, each subsequent snapshot in the batch will keep clearing the table when it runs instead of appending to it
-        if not _intervals(snapshot, deployability_index) and batch_index == 0:
-            self._replace_query_for_model(model, name, query_or_df)
+        if is_first_insert:
+            self._replace_query_for_model(model, table_name, query_or_df)
         else:
             self.adapter.merge(
-                name,
+                table_name,
                 query_or_df,
                 columns_to_types=model.columns_to_types,
                 unique_key=model.unique_key,
@@ -1295,15 +1266,11 @@ class IncrementalByUniqueKeyStrategy(MaterializableStrategy):
 
     def append(
         self,
-        snapshot: Snapshot,
         table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: t.Optional[DeployabilityIndex],
-        batch_index: int,
+        model: Model,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
         self.adapter.merge(
             table_name,
             query_or_df,
@@ -1316,32 +1283,26 @@ class IncrementalByUniqueKeyStrategy(MaterializableStrategy):
 class IncrementalUnmanagedStrategy(MaterializableStrategy):
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
-        if not _intervals(snapshot, deployability_index):
-            self._replace_query_for_model(model, name, query_or_df)
+        if is_first_insert:
+            self._replace_query_for_model(model, table_name, query_or_df)
         elif isinstance(model.kind, IncrementalUnmanagedKind) and model.kind.insert_overwrite:
             self.adapter.insert_overwrite_by_partition(
-                name,
+                table_name,
                 query_or_df,
                 model.partitioned_by,
                 columns_to_types=model.columns_to_types,
             )
         else:
             self.append(
-                snapshot,
-                name,
+                table_name,
                 query_or_df,
-                snapshots,
-                deployability_index,
-                batch_index,
+                model,
                 **kwargs,
             )
 
@@ -1349,61 +1310,55 @@ class IncrementalUnmanagedStrategy(MaterializableStrategy):
 class FullRefreshStrategy(MaterializableStrategy):
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
-        self._replace_query_for_model(model, name, query_or_df)
+        self._replace_query_for_model(model, table_name, query_or_df)
 
 
 class SeedStrategy(MaterializableStrategy):
     def create(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
+        model: Model,
         is_table_deployable: bool,
-        is_snapshot_deployable: bool,
         **render_kwargs: t.Any,
     ) -> None:
-        model = t.cast(SeedModel, snapshot.model)
-        if not model.is_hydrated and self.adapter.table_exists(name):
+        model = t.cast(SeedModel, model)
+        if not model.is_hydrated and self.adapter.table_exists(table_name):
             # This likely means that the table was created and populated previously, but the evaluation stage
             # failed before the interval could be added for this model.
             logger.warning(
                 "Seed model '%s' is not hydrated, but the table '%s' exists. Skipping creation",
                 model.name,
-                name,
+                table_name,
             )
             return
 
-        super().create(snapshot, name, is_table_deployable, is_snapshot_deployable, **render_kwargs)
+        super().create(table_name, model, is_table_deployable, **render_kwargs)
         if is_table_deployable:
             # For seeds we insert data at the time of table creation.
             try:
                 for index, df in enumerate(model.render_seed()):
                     if index == 0:
-                        self._replace_query_for_model(model, name, df)
+                        self._replace_query_for_model(model, table_name, df)
                     else:
                         self.adapter.insert_append(
-                            name, df, columns_to_types=model.columns_to_types
+                            table_name, df, columns_to_types=model.columns_to_types
                         )
             except Exception:
-                self.adapter.drop_table(name)
+                self.adapter.drop_table(table_name)
                 raise
 
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
         # Data has already been inserted at the time of table creation.
@@ -1413,21 +1368,19 @@ class SeedStrategy(MaterializableStrategy):
 class SCDType2Strategy(MaterializableStrategy):
     def create(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
+        model: Model,
         is_table_deployable: bool,
-        is_snapshot_deployable: bool,
         **render_kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
         assert isinstance(model.kind, (SCDType2ByTimeKind, SCDType2ByColumnKind))
         if model.annotated:
-            logger.info("Creating table '%s'", name)
+            logger.info("Creating table '%s'", table_name)
             columns_to_types = model.columns_to_types_or_raise
             if isinstance(model.kind, SCDType2ByTimeKind):
                 columns_to_types[model.kind.updated_at_name.name] = model.kind.time_data_type
             self.adapter.create_table(
-                name,
+                table_name,
                 columns_to_types=columns_to_types,
                 storage_format=model.storage_format,
                 partitioned_by=model.partitioned_by,
@@ -1442,56 +1395,51 @@ class SCDType2Strategy(MaterializableStrategy):
             # `time_data_type`. If that isn't the case, then the user might get an error about not being able
             # to do comparisons across different data types
             super().create(
-                snapshot,
-                name,
+                table_name,
+                model,
                 is_table_deployable,
-                is_snapshot_deployable,
                 **render_kwargs,
             )
 
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
-        truncate = not _intervals(snapshot, deployability_index)
         if isinstance(model.kind, SCDType2ByTimeKind):
             self.adapter.scd_type_2_by_time(
-                target_table=name,
+                target_table=table_name,
                 source_table=query_or_df,
                 unique_key=model.unique_key,
                 valid_from_col=model.kind.valid_from_name,
                 valid_to_col=model.kind.valid_to_name,
+                execution_time=kwargs["execution_time"],
                 updated_at_col=model.kind.updated_at_name,
                 invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
                 updated_at_as_valid_from=model.kind.updated_at_as_valid_from,
                 columns_to_types=model.columns_to_types,
                 table_description=model.description,
                 column_descriptions=model.column_descriptions,
-                truncate=truncate,
-                **kwargs,
+                truncate=is_first_insert,
             )
         elif isinstance(model.kind, SCDType2ByColumnKind):
             self.adapter.scd_type_2_by_column(
-                target_table=name,
+                target_table=table_name,
                 source_table=query_or_df,
                 unique_key=model.unique_key,
                 valid_from_col=model.kind.valid_from_name,
                 valid_to_col=model.kind.valid_to_name,
+                execution_time=kwargs["execution_time"],
                 check_columns=model.kind.columns,
-                columns_to_types=model.columns_to_types,
                 invalidate_hard_deletes=model.kind.invalidate_hard_deletes,
                 execution_time_as_valid_from=model.kind.execution_time_as_valid_from,
+                columns_to_types=model.columns_to_types,
                 table_description=model.description,
                 column_descriptions=model.column_descriptions,
-                truncate=truncate,
-                **kwargs,
+                truncate=is_first_insert,
             )
         else:
             raise SQLMeshError(
@@ -1500,15 +1448,11 @@ class SCDType2Strategy(MaterializableStrategy):
 
     def append(
         self,
-        snapshot: Snapshot,
         table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
         if isinstance(model.kind, SCDType2ByTimeKind):
             self.adapter.scd_type_2_by_time(
                 target_table=table_name,
@@ -1548,16 +1492,17 @@ class SCDType2Strategy(MaterializableStrategy):
 class ViewStrategy(PromotableStrategy):
     def insert(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
+        is_first_insert: bool,
         **kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
-        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
+        deployability_index = (
+            kwargs.get("deployability_index") or DeployabilityIndex.all_deployable()
+        )
+        snapshot = kwargs["snapshot"]
+        snapshots = kwargs["snapshots"]
         if (
             (
                 (
@@ -1574,14 +1519,14 @@ class ViewStrategy(PromotableStrategy):
                 or self.adapter.HAS_VIEW_BINDING
             )
             and snapshot.intervals  # Re-create the view during the first evaluation.
-            and self.adapter.table_exists(name)
+            and self.adapter.table_exists(table_name)
         ):
-            logger.info("Skipping creation of the view '%s'", name)
+            logger.info("Skipping creation of the view '%s'", table_name)
             return
 
-        logger.info("Replacing view '%s'", name)
+        logger.info("Replacing view '%s'", table_name)
         self.adapter.create_view(
-            name,
+            table_name,
             query_or_df,
             model.columns_to_types,
             materialized=self._is_materialized_view(model),
@@ -1592,29 +1537,23 @@ class ViewStrategy(PromotableStrategy):
 
     def append(
         self,
-        snapshot: Snapshot,
         table_name: str,
         query_or_df: QueryOrDF,
-        snapshots: t.Dict[str, Snapshot],
-        deployability_index: DeployabilityIndex,
-        batch_index: int,
+        model: Model,
         **kwargs: t.Any,
     ) -> None:
         raise ConfigError(f"Cannot append to a view '{table_name}'.")
 
     def create(
         self,
-        snapshot: Snapshot,
-        name: str,
+        table_name: str,
+        model: Model,
         is_table_deployable: bool,
-        is_snapshot_deployable: bool,
         **render_kwargs: t.Any,
     ) -> None:
-        model = snapshot.model
-
-        logger.info("Creating view '%s'", name)
+        logger.info("Creating view '%s'", table_name)
         self.adapter.create_view(
-            name,
+            table_name,
             model.render_query_or_raise(**render_kwargs),
             materialized=self._is_materialized_view(model),
             view_properties=model.physical_properties,
@@ -1624,18 +1563,17 @@ class ViewStrategy(PromotableStrategy):
 
     def migrate(
         self,
-        snapshot: Snapshot,
-        snapshots: t.Dict[str, Snapshot],
         target_table_name: str,
         source_table_name: str,
-        allow_destructive_snapshots: t.Set[str] = set(),
+        snapshot: Snapshot,
+        **kwargs: t.Any,
     ) -> None:
         logger.info("Migrating view '%s'", target_table_name)
         model = snapshot.model
         self.adapter.create_view(
             target_table_name,
             model.render_query_or_raise(
-                execution_time=now(), snapshots=snapshots, engine_adapter=self.adapter
+                execution_time=now(), snapshots=kwargs["snapshots"], engine_adapter=self.adapter
             ),
             model.columns_to_types,
             materialized=self._is_materialized_view(model),
@@ -1644,7 +1582,7 @@ class ViewStrategy(PromotableStrategy):
             column_descriptions=model.column_descriptions,
         )
 
-    def delete(self, name: str) -> None:
+    def delete(self, name: str, **kwargs: t.Any) -> None:
         try:
             self.adapter.drop_view(name)
         except Exception:
@@ -1660,6 +1598,69 @@ class ViewStrategy(PromotableStrategy):
         return isinstance(model.kind, ViewKind) and model.kind.materialized
 
 
+class CustomMaterialization(MaterializableStrategy):
+    """Base class for custom materializations."""
+
+    def insert(
+        self,
+        table_name: str,
+        query_or_df: QueryOrDF,
+        model: Model,
+        is_first_insert: bool,
+        **kwargs: t.Any,
+    ) -> None:
+        """Inserts the given query or a DataFrame into the target table or a view.
+
+        Args:
+            table_name: The name of the target table or view.
+            query_or_df: A query or a DataFrame to insert.
+            model: The target model.
+            is_first_insert: Whether this is the first insert for this version of a model. This value is set to True
+                if no data has been previously inserted into the target table, or when the entire history of the target model has
+                been restated. Note that in the latter case, the table might contain data from previous executions, and it is the
+                responsibility of a specific evaluation strategy to handle the truncation of the table if necessary.
+        """
+        raise NotImplementedError(
+            "Custom materialization strategies must implement the 'insert' method."
+        )
+
+
+_custom_materialization_type_cache: t.Optional[t.Dict[str, t.Type[CustomMaterialization]]] = None
+
+
+def get_custom_materialization_type(name: str) -> t.Type[CustomMaterialization]:
+    global _custom_materialization_type_cache
+
+    strategy_key = name.lower()
+
+    if (
+        _custom_materialization_type_cache is None
+        or strategy_key not in _custom_materialization_type_cache
+    ):
+        strategy_types = list(CustomMaterialization.__subclasses__())
+
+        entry_points = metadata.entry_points(group="sqlmesh.materializations")
+        for entry_point in entry_points:
+            strategy_type = entry_point.load()
+            if not issubclass(strategy_type, CustomMaterialization):
+                raise SQLMeshError(
+                    f"Custom materialization entry point '{entry_point.name}' must be a subclass of CustomMaterialization."
+                )
+            strategy_types.append(strategy_type)
+
+        _custom_materialization_type_cache = {
+            getattr(strategy_type, "NAME", strategy_type.__name__).lower(): strategy_type
+            for strategy_type in strategy_types
+        }
+
+    if strategy_key not in _custom_materialization_type_cache:
+        raise ConfigError(f"Materialization strategy with name '{name}' was not found.")
+
+    strategy_type = _custom_materialization_type_cache[strategy_key]
+    logger.debug("Resolved custom materialization '%s' to '%s'", name, strategy_type)
+    return strategy_type
+
+
 def _intervals(snapshot: Snapshot, deployability_index: DeployabilityIndex) -> Intervals:
     return (
         snapshot.intervals
diff --git a/sqlmesh/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index fd3c1de6..1c7b8e30 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -1379,7 +1379,9 @@ def _environment_to_df(environment: Environment) -> pd.DataFrame:
         [
             {
                 "name": environment.name,
-                "snapshots": json.dumps([snapshot.dict() for snapshot in environment.snapshots]),
+                "snapshots": json.dumps(
+                    [snapshot.dict(mode="json") for snapshot in environment.snapshots]
+                ),
                 "start_at": time_like_to_str(environment.start_at),
                 "end_at": time_like_to_str(environment.end_at) if environment.end_at else None,
                 "plan_id": environment.plan_id,
@@ -1395,7 +1397,10 @@ def _environment_to_df(environment: Environment) -> pd.DataFrame:
                 "catalog_name_override": environment.catalog_name_override,
                 "previous_finalized_snapshots": (
                     json.dumps(
-                        [snapshot.dict() for snapshot in environment.previous_finalized_snapshots]
+                        [
+                            snapshot.dict(mode="json")
+                            for snapshot in environment.previous_finalized_snapshots
+                        ]
                     )
                     if environment.previous_finalized_snapshots is not None
                     else None
diff --git a/web/client/openapi.json b/web/client/openapi.json
index 65313d81..48d0c0f9 100644
--- a/web/client/openapi.json
+++ b/web/client/openapi.json
@@ -871,6 +871,7 @@
             "title": "Stack"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["timestamp", "message", "origin"],
         "title": "ApiExceptionPayload"
@@ -897,6 +898,7 @@
           },
           "batches": { "type": "integer", "title": "Batches" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name", "interval", "batches"],
         "title": "BackfillDetails"
@@ -931,6 +933,7 @@
             "title": "Interval"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name", "completed", "total", "start"],
         "title": "BackfillTask"
@@ -1059,6 +1062,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name", "diff"],
         "title": "ChangeDirect"
@@ -1079,6 +1083,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name"],
         "title": "ChangeDisplay"
@@ -1099,6 +1104,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "view_name"],
         "title": "ChangeIndirect"
@@ -1112,6 +1118,7 @@
             "title": "Description"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "type"],
         "title": "Column"
@@ -1133,6 +1140,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path"],
         "title": "Directory"
@@ -1150,6 +1158,11 @@
             "anyOf": [{ "type": "string" }, { "type": "null" }],
             "title": "Catalog Name Override"
           },
+          "normalize_name": {
+            "type": "boolean",
+            "title": "Normalize Name",
+            "default": true
+          },
           "snapshots": {
             "items": { "$ref": "#/components/schemas/SnapshotTableInfo" },
             "type": "array",
@@ -1244,6 +1257,7 @@
             "default": ""
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "Environments"
       },
@@ -1282,6 +1296,7 @@
           },
           "limit": { "type": "integer", "title": "Limit", "default": 1000 }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["model", "start", "end", "execution_time"],
         "title": "EvaluateInput"
@@ -1291,6 +1306,7 @@
           "sql": { "type": "string", "title": "Sql" },
           "limit": { "type": "integer", "title": "Limit", "default": 1000 }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["sql"],
         "title": "FetchdfInput"
@@ -1309,6 +1325,7 @@
             "title": "Content"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path"],
         "title": "File"
@@ -1358,6 +1375,7 @@
             "title": "Models"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["models"],
         "title": "LineageColumn"
@@ -1371,6 +1389,7 @@
             "default": false
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["version"],
         "title": "Meta"
@@ -1407,6 +1426,7 @@
           },
           "hash": { "type": "string", "title": "Hash" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": [
           "name",
@@ -1526,6 +1546,7 @@
             "title": "Annotated"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "ModelDetails"
       },
@@ -1534,6 +1555,7 @@
         "enum": [
           "INCREMENTAL_BY_TIME_RANGE",
           "INCREMENTAL_BY_UNIQUE_KEY",
+          "INCREMENTAL_BY_PARTITION",
           "INCREMENTAL_UNMANAGED",
           "FULL",
           "SCD_TYPE_2",
@@ -1573,6 +1595,7 @@
             "default": []
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "ModelsDiff"
       },
@@ -1656,6 +1679,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanApplyStageTracker"
       },
@@ -1701,6 +1725,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanCancelStageTracker"
       },
@@ -1729,6 +1754,7 @@
             "title": "End"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanDates"
       },
@@ -1778,6 +1804,7 @@
             "default": false
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanOptions"
       },
@@ -1835,6 +1862,7 @@
             ]
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanOverviewStageTracker"
       },
@@ -1857,6 +1885,7 @@
             "default": {}
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageBackfill"
       },
@@ -1874,6 +1903,7 @@
             "title": "Models"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageBackfills"
       },
@@ -1881,6 +1911,7 @@
         "properties": {
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageCancel"
       },
@@ -1914,6 +1945,7 @@
           },
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageChanges"
       },
@@ -1923,6 +1955,7 @@
           "total_tasks": { "type": "integer", "title": "Total Tasks" },
           "num_tasks": { "type": "integer", "title": "Num Tasks" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["total_tasks", "num_tasks"],
         "title": "PlanStageCreation"
@@ -1937,6 +1970,7 @@
             "title": "Target Environment"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["total_tasks", "num_tasks", "target_environment"],
         "title": "PlanStagePromote"
@@ -1945,6 +1979,7 @@
         "properties": {
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageRestate"
       },
@@ -1952,11 +1987,13 @@
         "properties": {
           "meta": { "$ref": "#/components/schemas/TrackableMeta" }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "PlanStageValidation"
       },
       "Query": {
         "properties": { "sql": { "type": "string", "title": "Sql" } },
+        "additionalProperties": false,
         "type": "object",
         "required": ["sql"],
         "title": "Query"
@@ -1967,6 +2004,7 @@
           "expression": { "type": "string", "title": "Expression" },
           "unique": { "type": "boolean", "title": "Unique" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "expression", "unique"],
         "title": "Reference"
@@ -2021,6 +2059,7 @@
             "title": "Dialect"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["model"],
         "title": "RenderInput"
@@ -2039,6 +2078,7 @@
           "target_count": { "type": "integer", "title": "Target Count" },
           "count_pct_change": { "type": "number", "title": "Count Pct Change" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": [
           "source",
@@ -2081,6 +2121,7 @@
             "title": "Modified"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": [
           "source",
@@ -2095,9 +2136,9 @@
       },
       "SnapshotChangeCategory": {
         "type": "integer",
-        "enum": [1, 2, 3, 4, 5],
+        "enum": [1, 2, 3, 4, 5, 6],
         "title": "SnapshotChangeCategory",
-        "description": "Values are ordered by decreasing severity and that ordering is required.\n\nBREAKING: The change requires that snapshot modified and downstream dependencies be rebuilt\nNON_BREAKING: The change requires that only the snapshot modified be rebuilt\nFORWARD_ONLY: The change requires no rebuilding\nINDIRECT_BREAKING: The change was caused indirectly and is breaking.\nINDIRECT_NON_BREAKING: The change was caused indirectly by a non-breaking change."
+        "description": "Values are ordered by decreasing severity and that ordering is required.\n\nBREAKING: The change requires that snapshot modified and downstream dependencies be rebuilt\nNON_BREAKING: The change requires that only the snapshot modified be rebuilt\nFORWARD_ONLY: The change requires no rebuilding\nINDIRECT_BREAKING: The change was caused indirectly and is breaking.\nINDIRECT_NON_BREAKING: The change was caused indirectly by a non-breaking change.\nMETADATA: The change was caused by a metadata update."
       },
       "SnapshotDataVersion": {
         "properties": {
@@ -2222,6 +2263,7 @@
             "title": "On"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["schema_diff", "row_diff", "on"],
         "title": "TableDiff"
@@ -2231,6 +2273,7 @@
           "name": { "type": "string", "title": "Name" },
           "path": { "type": "string", "format": "path", "title": "Path" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path"],
         "title": "TestCase"
@@ -2241,6 +2284,7 @@
           "path": { "type": "string", "format": "path", "title": "Path" },
           "tb": { "type": "string", "title": "Tb" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path", "tb"],
         "title": "TestErrorOrFailure"
@@ -2269,6 +2313,7 @@
             "title": "Successes"
           }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["tests_run", "failures", "errors", "skipped", "successes"],
         "title": "TestResult"
@@ -2279,6 +2324,7 @@
           "path": { "type": "string", "format": "path", "title": "Path" },
           "reason": { "type": "string", "title": "Reason" }
         },
+        "additionalProperties": false,
         "type": "object",
         "required": ["name", "path", "reason"],
         "title": "TestSkipped"
@@ -2296,6 +2342,7 @@
           },
           "done": { "type": "boolean", "title": "Done", "default": false }
         },
+        "additionalProperties": false,
         "type": "object",
         "title": "TrackableMeta"
       },
diff --git a/web/client/package-lock.json b/web/client/package-lock.json
index 49954465..7bd0f0c6 100644
--- a/web/client/package-lock.json
+++ b/web/client/package-lock.json
@@ -8,13 +8,14 @@
       "name": "tobiko",
       "version": "0.0.0",
       "dependencies": {
-        "@codemirror/commands": "^6.2.5",
-        "@codemirror/lang-python": "^6.1.3",
-        "@codemirror/lang-sql": "^6.5.4",
-        "@codemirror/language": "^6.9.0",
-        "@codemirror/legacy-modes": "^6.3.3",
-        "@codemirror/state": "^6.2.1",
-        "@codemirror/view": "^6.17.0",
+        "@codemirror/autocomplete": "^6.16.2",
+        "@codemirror/commands": "^6.6.0",
+        "@codemirror/lang-python": "^6.1.6",
+        "@codemirror/lang-sql": "^6.6.4",
+        "@codemirror/language": "^6.10.2",
+        "@codemirror/legacy-modes": "^6.4.0",
+        "@codemirror/state": "^6.4.1",
+        "@codemirror/view": "^6.28.1",
         "@headlessui/react": "^1.7.17",
         "@heroicons/react": "^2.0.18",
         "@radix-ui/react-context-menu": "^2.1.4",
@@ -27,6 +28,7 @@
         "@uiw/react-codemirror": "^4.21.12",
         "apache-arrow": "^13.0.0",
         "clsx": "^2.0.0",
+        "diff": "^5.2.0",
         "elkjs": "^0.8.2",
         "pluralize": "^8.0.0",
         "react": "^18.2.0",
@@ -44,6 +46,7 @@
         "@testing-library/jest-dom": "^6.1.2",
         "@testing-library/react": "^14.0.0",
         "@testing-library/user-event": "^14.4.3",
+        "@types/diff": "^5.2.1",
         "@types/pluralize": "^0.0.30",
         "@types/react": "^18.2.21",
         "@types/react-dom": "^18.2.7",
@@ -381,8 +384,9 @@
       }
     },
     "node_modules/@codemirror/autocomplete": {
-      "version": "6.11.1",
-      "license": "MIT",
+      "version": "6.16.2",
+      "resolved": "https://registry.npmjs.org/@codemirror/autocomplete/-/autocomplete-6.16.2.tgz",
+      "integrity": "sha512-MjfDrHy0gHKlPWsvSsikhO1+BOh+eBHNgfH1OXs1+DAf30IonQldgMM3kxLDTG9ktE7kDLaA1j/l7KMPA4KNfw==",
       "dependencies": {
         "@codemirror/language": "^6.0.0",
         "@codemirror/state": "^6.0.0",
@@ -397,41 +401,48 @@
       }
     },
     "node_modules/@codemirror/commands": {
-      "version": "6.3.2",
-      "license": "MIT",
+      "version": "6.6.0",
+      "resolved": "https://registry.npmjs.org/@codemirror/commands/-/commands-6.6.0.tgz",
+      "integrity": "sha512-qnY+b7j1UNcTS31Eenuc/5YJB6gQOzkUoNmJQc0rznwqSRpeaWWpjkWy2C/MPTcePpsKJEM26hXrOXl1+nceXg==",
       "dependencies": {
         "@codemirror/language": "^6.0.0",
-        "@codemirror/state": "^6.2.0",
-        "@codemirror/view": "^6.0.0",
+        "@codemirror/state": "^6.4.0",
+        "@codemirror/view": "^6.27.0",
         "@lezer/common": "^1.1.0"
       }
     },
     "node_modules/@codemirror/lang-python": {
-      "version": "6.1.3",
-      "license": "MIT",
+      "version": "6.1.6",
+      "resolved": "https://registry.npmjs.org/@codemirror/lang-python/-/lang-python-6.1.6.tgz",
+      "integrity": "sha512-ai+01WfZhWqM92UqjnvorkxosZ2aq2u28kHvr+N3gu012XqY2CThD67JPMHnGceRfXPDBmn1HnyqowdpF57bNg==",
       "dependencies": {
         "@codemirror/autocomplete": "^6.3.2",
         "@codemirror/language": "^6.8.0",
+        "@codemirror/state": "^6.0.0",
+        "@lezer/common": "^1.2.1",
         "@lezer/python": "^1.1.4"
       }
     },
     "node_modules/@codemirror/lang-sql": {
-      "version": "6.5.4",
-      "license": "MIT",
+      "version": "6.6.4",
+      "resolved": "https://registry.npmjs.org/@codemirror/lang-sql/-/lang-sql-6.6.4.tgz",
+      "integrity": "sha512-n+FVfKGut+frOvor9dU5pFUalcP614WBNQ9IT1kOUj1t6LFLjWHi2I9DdxXnJuxqFV9jTyYF79coDV3ilSJqCw==",
       "dependencies": {
         "@codemirror/autocomplete": "^6.0.0",
         "@codemirror/language": "^6.0.0",
         "@codemirror/state": "^6.0.0",
+        "@lezer/common": "^1.2.0",
         "@lezer/highlight": "^1.0.0",
         "@lezer/lr": "^1.0.0"
       }
     },
     "node_modules/@codemirror/language": {
-      "version": "6.9.3",
-      "license": "MIT",
+      "version": "6.10.2",
+      "resolved": "https://registry.npmjs.org/@codemirror/language/-/language-6.10.2.tgz",
+      "integrity": "sha512-kgbTYTo0Au6dCSc/TFy7fK3fpJmgHDv1sG1KNQKJXVi+xBTEeBPY/M30YXiU6mMXeH+YIDLsbrT4ZwNRdtF+SA==",
       "dependencies": {
         "@codemirror/state": "^6.0.0",
-        "@codemirror/view": "^6.0.0",
+        "@codemirror/view": "^6.23.0",
         "@lezer/common": "^1.1.0",
         "@lezer/highlight": "^1.0.0",
         "@lezer/lr": "^1.0.0",
@@ -439,8 +450,9 @@
       }
     },
     "node_modules/@codemirror/legacy-modes": {
-      "version": "6.3.3",
-      "license": "MIT",
+      "version": "6.4.0",
+      "resolved": "https://registry.npmjs.org/@codemirror/legacy-modes/-/legacy-modes-6.4.0.tgz",
+      "integrity": "sha512-5m/K+1A6gYR0e+h/dEde7LoGimMjRtWXZFg4Lo70cc8HzjSdHe3fLwjWMR0VRl5KFT1SxalSap7uMgPKF28wBA==",
       "dependencies": {
         "@codemirror/language": "^6.0.0"
       }
@@ -464,8 +476,9 @@
       }
     },
     "node_modules/@codemirror/state": {
-      "version": "6.3.3",
-      "license": "MIT"
+      "version": "6.4.1",
+      "resolved": "https://registry.npmjs.org/@codemirror/state/-/state-6.4.1.tgz",
+      "integrity": "sha512-QkEyUiLhsJoZkbumGZlswmAhA7CBU02Wrz7zvH4SrcifbsqwlXShVXg65f3v/ts57W3dqyamEriMhij1Z3Zz4A=="
     },
     "node_modules/@codemirror/theme-one-dark": {
       "version": "6.1.2",
@@ -478,10 +491,11 @@
       }
     },
     "node_modules/@codemirror/view": {
-      "version": "6.22.1",
-      "license": "MIT",
+      "version": "6.28.1",
+      "resolved": "https://registry.npmjs.org/@codemirror/view/-/view-6.28.1.tgz",
+      "integrity": "sha512-BUWr+zCJpMkA/u69HlJmR+YkV4yPpM81HeMkOMZuwFa8iM5uJdEPKAs1icIRZKkKmy0Ub1x9/G3PQLTXdpBxrQ==",
       "dependencies": {
-        "@codemirror/state": "^6.1.4",
+        "@codemirror/state": "^6.4.0",
         "style-mod": "^4.1.0",
         "w3c-keyname": "^2.2.4"
       }
@@ -766,8 +780,9 @@
       }
     },
     "node_modules/@lezer/common": {
-      "version": "1.1.2",
-      "license": "MIT"
+      "version": "1.2.1",
+      "resolved": "https://registry.npmjs.org/@lezer/common/-/common-1.2.1.tgz",
+      "integrity": "sha512-yemX0ZD2xS/73llMZIK6KplkjIjf2EvAHcinDi/TfJ9hS25G0388+ClHt6/3but0oOxinTcQHJLDXh6w1crzFQ=="
     },
     "node_modules/@lezer/highlight": {
       "version": "1.2.0",
@@ -2630,6 +2645,13 @@
         "@types/d3-selection": "*"
       }
     },
+    "node_modules/@types/diff": {
+      "version": "5.2.1",
+      "resolved": "https://registry.npmjs.org/@types/diff/-/diff-5.2.1.tgz",
+      "integrity": "sha512-uxpcuwWJGhe2AR1g8hD9F5OYGCqjqWnBUQFD8gMZsDbv8oPHzxJF6iMO6n8Tk0AdzlxoaaoQhOYlIg/PukVU8g==",
+      "dev": true,
+      "license": "MIT"
+    },
     "node_modules/@types/es-aggregate-error": {
       "version": "1.0.6",
       "dev": true,
@@ -3627,7 +3649,9 @@
       }
     },
     "node_modules/caniuse-lite": {
-      "version": "1.0.30001566",
+      "version": "1.0.30001636",
+      "resolved": "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001636.tgz",
+      "integrity": "sha512-bMg2vmr8XBsbL6Lr0UHXy/21m84FTxDLWn2FSqMd5PrlbMxwJlQnC2YWYxVgp66PZE+BBNF2jYQUBKCo1FDeZg==",
       "dev": true,
       "funding": [
         {
@@ -4126,6 +4150,14 @@
       "version": "1.2.2",
       "license": "Apache-2.0"
     },
+    "node_modules/diff": {
+      "version": "5.2.0",
+      "resolved": "https://registry.npmjs.org/diff/-/diff-5.2.0.tgz",
+      "integrity": "sha512-uIFDxqpRZGZ6ThOk84hEfqWoHx2devRFvpTZcTHur85vImfaxUbTW9Ryh4CpCuDnToOP1CEtXKIgytHBPVff5A==",
+      "engines": {
+        "node": ">=0.3.1"
+      }
+    },
     "node_modules/diff-sequences": {
       "version": "29.6.3",
       "dev": true,
diff --git a/web/client/package.json b/web/client/package.json
index d66ab1aa..b56ec9fc 100644
--- a/web/client/package.json
+++ b/web/client/package.json
@@ -16,13 +16,14 @@
     "generate:api": "orval --config ./orval.config.ts"
   },
   "dependencies": {
-    "@codemirror/commands": "^6.2.5",
-    "@codemirror/lang-python": "^6.1.3",
-    "@codemirror/lang-sql": "^6.5.4",
-    "@codemirror/language": "^6.9.0",
-    "@codemirror/legacy-modes": "^6.3.3",
-    "@codemirror/state": "^6.2.1",
-    "@codemirror/view": "^6.17.0",
+    "@codemirror/autocomplete": "^6.16.2",
+    "@codemirror/commands": "^6.6.0",
+    "@codemirror/lang-python": "^6.1.6",
+    "@codemirror/lang-sql": "^6.6.4",
+    "@codemirror/language": "^6.10.2",
+    "@codemirror/legacy-modes": "^6.4.0",
+    "@codemirror/state": "^6.4.1",
+    "@codemirror/view": "^6.28.1",
     "@headlessui/react": "^1.7.17",
     "@heroicons/react": "^2.0.18",
     "@radix-ui/react-context-menu": "^2.1.4",
@@ -35,6 +36,7 @@
     "@uiw/react-codemirror": "^4.21.12",
     "apache-arrow": "^13.0.0",
     "clsx": "^2.0.0",
+    "diff": "^5.2.0",
     "elkjs": "^0.8.2",
     "pluralize": "^8.0.0",
     "react": "^18.2.0",
@@ -52,6 +54,7 @@
     "@testing-library/jest-dom": "^6.1.2",
     "@testing-library/react": "^14.0.0",
     "@testing-library/user-event": "^14.4.3",
+    "@types/diff": "^5.2.1",
     "@types/pluralize": "^0.0.30",
     "@types/react": "^18.2.21",
     "@types/react-dom": "^18.2.7",
diff --git a/web/client/src/index.css b/web/client/src/index.css
index c6fd970c..6e0a5233 100644
--- a/web/client/src/index.css
+++ b/web/client/src/index.css
@@ -222,7 +222,7 @@ html[mode='dark'] {
   --color-logo-lighter: var(--color-neutral-100);
 
   --color-editor-text: var(--color-neutral-300);
-  --color-editor-active-line-text: var(--color-neutral-100);
+  --color-editor-active-line-text: var(--color-neutral-200);
 
   --color-graph-edge-secondary: var(--color-primary-500);
   --color-graph-edge-main: var(--color-primary-20);
@@ -250,7 +250,7 @@ html[mode='light'] {
   --color-logo-lighter: var(--color-neutral-300);
 
   --color-editor-text: var(--color-neutral-600);
-  --color-editor-active-line-text: var(--color-neutral-800);
+  --color-editor-active-line-text: var(--color-neutral-700);
 
   --color-graph-edge-secondary: var(--color-secondary-500);
   --color-graph-edge-main: var(--color-secondary-20);
diff --git a/web/client/src/library/components/editor/Editor.css b/web/client/src/library/components/editor/Editor.css
index e36aca16..3a0efa73 100644
--- a/web/client/src/library/components/editor/Editor.css
+++ b/web/client/src/library/components/editor/Editor.css
@@ -2,98 +2,106 @@
 .cm-editor .cm-content {
   background-color: var(--color-editor);
   color: var(--color-editor-text);
+  font-weight: 400;
+  font-size: 0.85rem;
 }
-
 .cm-editor .cm-gutters {
   background-color: var(--color-editor);
   border: none;
   padding-left: 0.25rem;
   padding-right: 0.5rem;
 }
-
-.cm-editor .cm-activeLineGutter {
-  background-color: var(--color-theme-lighter);
-  color: var(--color-editor-active-line-text);
-}
-
+.cm-editor .cm-activeLineGutter,
 .cm-editor .cm-activeLine {
-  background-color: var(--color-theme-lighter);
   color: var(--color-editor-active-line-text);
 }
-
 .cm-editor .cm-line::selection,
 .cm-editor .cm-line > *::selection {
   background-color: var(--color-brand-100) !important;
   color: var(--color-brand-700) !important;
 }
-
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete {
-  border: 1px solid var(--color-neutral-300) !important;
-  border-radius: 0.25rem;
+  margin-top: 0.25rem;
 }
-
-.cm-editor .cm-tooltip-autocomplete ul li[aria-selected='true'] {
-  background: var(--color-primary-10);
+.cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul {
+  border-radius: 0.35rem;
+  padding: 0.25rem 0.25rem;
   color: var(--color-text);
 }
-
+.cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul > li {
+  padding: 0.25rem 0.5rem;
+  cursor: pointer;
+  border-radius: 0.3rem;
+  border: 1px solid transparent;
+  font-weight: normal;
+  display: flex;
+  align-items: center;
+}
+.cm-editor .cm-tooltip-autocomplete > ul > li[aria-selected='true'] {
+  color: var(--color-text);
+}
+.cm-editor
+  .cm-tooltip.cm-tooltip-autocomplete
+  > ul
+  > li
+  .cm-completionMatchedText {
+  opacity: 1;
+  font-weight: bold;
+  text-decoration: none;
+}
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-column,
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-model {
-  margin: 0.125rem 1.25rem 0.125rem 0;
+  margin-right: 1.5rem;
 }
-
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon::before {
   display: inline-block;
   padding: 0.125rem 0.25rem;
   border-radius: 0.25rem;
 }
-
+.cm-editor .cm-tooltip-autocomplete .cm-completionIcon {
+  display: none;
+}
+.cm-editor .cm-tooltip-autocomplete .cm-completionIcon-column,
+.cm-editor .cm-tooltip-autocomplete .cm-completionIcon-model {
+  display: block;
+  opacity: 1;
+}
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-column::before {
   content: 'col';
-  color: var(--color-brand-600);
-  background: var(--color-brand-10);
+  font-weight: bold;
 }
-
 .cm-editor .cm-tooltip-autocomplete .cm-completionIcon-model::before {
   content: 'mdl';
-  color: var(--color-primary-500);
-  background: var(--color-primary-10);
+  font-weight: bold;
 }
-
 .cm-editor .cm-scroller {
   font-family: inherit;
 }
-
 .cm-editor .cm-scroller::-webkit-scrollbar,
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete ul::-webkit-scrollbar {
   height: var(--scrollbar-size);
   width: var(--scrollbar-size);
 }
-
 .cm-editor .cm-scroller::-webkit-scrollbar-track,
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete ul::-webkit-scrollbar-track {
   background: transparent;
 }
-
 .cm-editor .cm-scroller::-webkit-scrollbar-thumb,
 .cm-editor .cm-tooltip.cm-tooltip-autocomplete ul::-webkit-scrollbar-thumb {
-  background: var(--color-brand);
+  background: var(--color-neutral-40);
   border-radius: var(--scrollbar-radius);
 }
-
 .cm-editor .cm-tooltip {
   border: none !important;
   outline: none !important;
   background: var(--color-theme) !important;
 }
-
 .sqlmesh-expression {
   background: var(--color-secondary-10);
   box-shadow:
     inset 0 -1px 0 0 var(--color-secondary-500),
     inset 0 1px 0 0 var(--color-secondary-500);
 }
-
 .sqlmesh-model,
 .sqlmesh-model > span {
   display: inline-block;
@@ -106,11 +114,9 @@
   cursor: pointer;
   box-shadow: inset 0 -2px 0 0 var(--color-primary-300);
 }
-
 .sqlmesh-model.--is-action-mode:hover {
   opacity: 0.8;
 }
-
 .sqlmesh-model.--is-active-model,
 .sqlmesh-model.--is-active-model:hover,
 .sqlmesh-model.--is-active-model > span,
@@ -119,7 +125,6 @@
   cursor: default;
   opacity: 1;
 }
-
 .sqlmesh-model__column.--is-original,
 .sqlmesh-model__column.--is-original > span,
 .sqlmesh-model__column.--is-alias,
@@ -128,7 +133,6 @@
   color: var(--color-accent-300);
   font-weight: bold;
 }
-
 .sqlmesh-model__column.--is-active-model.--is-alias,
 .sqlmesh-model__column.--is-active-model.--is-alias > span {
   display: inline-block;
@@ -144,3 +148,83 @@
 .sqlmesh-model__column.--is-action-mode:hover {
   opacity: 0.8;
 }
+
+[mode='dark'] .cm-editor .cm-activeLineGutter,
+[mode='dark'] .cm-editor .cm-activeLine {
+  background-color: var(--color-neutral-30);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-model::before {
+  color: var(--color-secondary-300);
+  background: var(--color-secondary-20);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip.cm-tooltip-autocomplete
+  > ul
+  > li
+  .cm-completionMatchedText {
+  color: var(--color-accent-200);
+  background: var(--color-accent-50);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-column::before {
+  color: var(--color-brand-500);
+  background: var(--color-brand-10);
+}
+[mode='dark'] .cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul {
+  border: 1px solid var(--color-neutral-50) !important;
+  box-shadow: 0 2px 3px 0 var(--color-neutral-20);
+}
+[mode='dark']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  > ul
+  > li[aria-selected='true'] {
+  background: var(--color-neutral-30);
+  border: 1px solid var(--color-neutral-50) !important;
+}
+
+[mode='light'] .cm-editor .cm-activeLineGutter,
+[mode='light'] .cm-editor .cm-activeLine {
+  background-color: var(--color-neutral-10);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-model::before {
+  color: var(--color-secondary-500);
+  background: var(--color-secondary-10);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip.cm-tooltip-autocomplete
+  > ul
+  > li
+  .cm-completionMatchedText {
+  color: var(--color-accent-500);
+  background: var(--color-accent-5);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  .cm-completionIcon-column::before {
+  color: var(--color-brand-500);
+  background: var(--color-brand-20);
+}
+[mode='light'] .cm-editor .cm-tooltip.cm-tooltip-autocomplete > ul {
+  border: 1px solid var(--color-neutral-20) !important;
+  box-shadow: 0 2px 3px 0 var(--color-neutral-20);
+}
+[mode='light']
+  .cm-editor
+  .cm-tooltip-autocomplete
+  > ul
+  > li[aria-selected='true'] {
+  background: var(--color-neutral-5);
+  border: 1px solid var(--color-neutral-10) !important;
+}
diff --git a/web/client/src/library/components/editor/Editor.tsx b/web/client/src/library/components/editor/Editor.tsx
index 9a4a06fe..6e108d5f 100644
--- a/web/client/src/library/components/editor/Editor.tsx
+++ b/web/client/src/library/components/editor/Editor.tsx
@@ -95,8 +95,10 @@ function EditorMain({ tab }: { tab: EditorTab }): JSX.Element {
     return [
       ...defaultKeymapsEditorTab,
       {
-        key: 'Shift-Mod-Enter',
+        key: 'Mod-Enter',
+        win: 'Ctrl-Enter',
         preventDefault: true,
+        stopPropagation: true,
         run(view) {
           const sql = view.state.doc.toString()
 
diff --git a/web/client/src/library/components/editor/EditorCode.tsx b/web/client/src/library/components/editor/EditorCode.tsx
index fb437b0a..eb594649 100644
--- a/web/client/src/library/components/editor/EditorCode.tsx
+++ b/web/client/src/library/components/editor/EditorCode.tsx
@@ -1,9 +1,9 @@
 import { useCallback, useEffect, useMemo, useState } from 'react'
 import CodeMirror from '@uiw/react-codemirror'
-import { type KeyBinding, keymap } from '@codemirror/view'
-import { type Extension } from '@codemirror/state'
+import { type EditorView, type KeyBinding, keymap } from '@codemirror/view'
+import { type ChangeSpec, type Extension } from '@codemirror/state'
 import { useApiFileByPath, useMutationApiSaveFile } from '~/api'
-import { debounceSync, isNil } from '~/utils'
+import { debounceSync, isNil, isNotNil } from '~/utils'
 import { useStoreContext } from '~/context/context'
 import { useStoreEditor } from '~/context/editor'
 import {
@@ -31,7 +31,9 @@ import {
   SQLMeshDialect,
   SQLMeshDialectCleanUp,
 } from './extensions/SQLMeshDialect'
-import { indentMore } from '@codemirror/commands'
+import { indentMore, defaultKeymap, historyKeymap } from '@codemirror/commands'
+import { EMPTY_STRING } from '@components/search/help'
+import { diffChars } from 'diff'
 
 export { CodeEditorDefault, CodeEditorRemoteFile }
 
@@ -55,25 +57,15 @@ function CodeEditorDefault({
   const { mode } = useColorScheme()
 
   const models = useStoreContext(s => s.models)
-
   const engine = useStoreEditor(s => s.engine)
   const dialects = useStoreEditor(s => s.dialects)
 
-  const [value, setValue] = useState(content)
-
+  const [editorView, setEditorView] = useState<Optional<EditorView>>()
   const [dialectOptions, setDialectOptions] = useState<{
     types: string
     keywords: string
   }>()
 
-  const debouncedChange = useCallback(
-    debounceSync(function handleChange(value): void {
-      setValue(value)
-      onChange?.(value)
-    }, 500),
-    [],
-  )
-
   const handleEngineWorkerMessage = useCallback((e: MessageEvent): void => {
     if (e.data.topic === 'dialect') {
       setDialectOptions(e.data.payload)
@@ -96,26 +88,34 @@ function CodeEditorDefault({
     () => dialects.map(d => d.dialect_title),
     [dialects],
   )
-  const extensionKeymap = useMemo(
-    () =>
-      keymap.of(
-        [
-          ...(keymaps ?? []),
-          [
-            {
-              key: 'Tab',
-              preventDefault: true,
-              run(e: any) {
-                return isNil(completionStatus(e.state))
-                  ? indentMore(e)
-                  : acceptCompletion(e)
-              },
-            },
-          ],
-        ].flat(),
-      ),
-    [keymaps],
-  )
+  const extensionKeymap = useMemo(() => {
+    const keys = [
+      defaultKeymap,
+      historyKeymap,
+      ...(keymaps ?? []),
+      [
+        {
+          key: 'Tab',
+          preventDefault: true,
+          run(e: any) {
+            return isNil(completionStatus(e.state))
+              ? indentMore(e)
+              : acceptCompletion(e)
+          },
+        },
+      ],
+    ]
+      .flat()
+      .reduce<Record<string, KeyBinding>>((acc, binding) => {
+        if (isNotNil(binding.key)) {
+          acc[binding.key] = binding
+        }
+
+        return acc
+      }, {})
+
+    return keymap.of(Object.values(keys))
+  }, [keymaps])
   const allModels = useMemo(() => Array.from(models.values()), [models])
   const allModelsNames = useMemo(
     () =>
@@ -191,8 +191,49 @@ function CodeEditorDefault({
   }, [dialect])
 
   useEffect(() => {
-    setValue(content)
-  }, [content])
+    updateEditor(content)
+  }, [editorView, content, extensionsAll])
+
+  function updateEditor(newContent: string = ''): void {
+    if (isNil(editorView)) return
+
+    const state = editorView.state
+    const currentContent = state.doc.toString()
+
+    if (currentContent === newContent) return
+
+    const changes: ChangeSpec[] = []
+
+    if (currentContent === EMPTY_STRING) {
+      changes.push({ from: 0, to: 0, insert: newContent })
+    } else {
+      const items = diffChars(currentContent, newContent)
+
+      let from = 0
+
+      items.forEach(item => {
+        const count = item.count ?? 0
+
+        if (isNotNil(item.removed)) {
+          const to = from + count
+          const change = { from, to }
+
+          from = to
+
+          changes.push(change)
+        } else if (isNotNil(item.added)) {
+          changes.push({ from, to: from, insert: item.value })
+        } else {
+          from = from + count
+        }
+      })
+    }
+
+    editorView.dispatch({
+      changes,
+      scrollIntoView: true,
+    })
+  }
 
   return (
     <div className={clsx('flex w-full h-full', className)}>
@@ -200,14 +241,16 @@ function CodeEditorDefault({
         height="100%"
         width="100%"
         className={clsx('flex w-full h-full font-mono text-xs', className)}
-        value={value}
         indentWithTab={false}
         extensions={extensionsAll}
-        onChange={debouncedChange}
+        onChange={onChange}
         readOnly={isNil(onChange)}
+        onCreateEditor={setEditorView}
         basicSetup={{
           autocompletion: false,
+          defaultKeymap: false,
         }}
+        autoFocus
       />
     </div>
   )
diff --git a/web/client/src/library/components/editor/EditorFooter.tsx b/web/client/src/library/components/editor/EditorFooter.tsx
index 7a6f7125..44ef59ff 100644
--- a/web/client/src/library/components/editor/EditorFooter.tsx
+++ b/web/client/src/library/components/editor/EditorFooter.tsx
@@ -50,7 +50,7 @@ export default function EditorFooter({ tab }: { tab: EditorTab }): JSX.Element {
   }
 
   return (
-    <div className="flex w-full items-center px-2 min-h-[2rem]">
+    <div className="flex w-full items-center px-2 min-h-[2rem] overflow-hidden">
       {tab.file.isRemote && (
         <EditorIndicator
           className="mr-2"
diff --git a/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts b/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts
index 38928131..b6fddd30 100644
--- a/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts
+++ b/web/client/src/library/components/editor/extensions/SQLMeshDialect.ts
@@ -3,6 +3,7 @@ import {
   completeFromList,
   type CompletionContext,
   type Completion,
+  type CompletionResult,
 } from '@codemirror/autocomplete'
 import { keywordCompletionSource, SQLDialect } from '@codemirror/lang-sql'
 import { LanguageSupport } from '@codemirror/language'
@@ -139,28 +140,19 @@ export const SQLMeshDialect: ExtensionSQLMeshDialect = function SQLMeshDialect(
           suggestions.push(...allColumnsNames)
         }
 
-        const wordLastChar = getFirstChar(word.text)
-        const isUpperCase =
-          isStringEmptyOrNil(wordLastChar) &&
-          wordLastChar === wordLastChar.toUpperCase()
+        const wordLastChar = word.text.trim().slice(-1)
+        const completionResult = keywordCompletionSource(
+          lang,
+          wordLastChar === wordLastChar.toUpperCase(),
+        )(ctx) as CompletionResult
 
-        if (isUpperCase) {
-          suggestions = suggestions.map(suggestion => ({
-            ...suggestion,
-            label: suggestion.label.toUpperCase(),
-          }))
+        if (isNotNil(completionResult)) {
+          const options = completionResult.options ?? []
+          completionResult.options = options.concat(suggestions)
+          completionResult.map = () => null // Updates suggestions to switch between lower and upper case
         }
 
-        const source = isArrayEmpty(suggestions)
-          ? keywordCompletionSource(lang, isUpperCase)(ctx)
-          : {
-              from: word.from,
-              to: word.to,
-              options: suggestions,
-              validFor: /^\w*$/,
-            }
-
-        return source
+        return completionResult
       },
     }),
   ])
@@ -262,18 +254,7 @@ function matchWordWithSpacesAfter(
 }
 
 function maybeSuggestion(suggestions: Completion[], word: string): boolean {
-  return (
-    word.length > 1 &&
-    suggestions.some(suggestion =>
-      suggestion.label.toLowerCase().includes(word.toLowerCase()),
-    )
+  return suggestions.some(suggestion =>
+    suggestion.label.toLowerCase().includes(word.toLowerCase()),
   )
 }
-
-function getFirstChar(word: string): string {
-  for (const char of word) {
-    if (/[a-zA-Z]/.test(char)) return char
-  }
-
-  return ''
-}
diff --git a/web/client/src/library/components/editor/extensions/index.ts b/web/client/src/library/components/editor/extensions/index.ts
index a2b0a87b..7dd557a8 100644
--- a/web/client/src/library/components/editor/extensions/index.ts
+++ b/web/client/src/library/components/editor/extensions/index.ts
@@ -10,6 +10,7 @@ import {
 } from '@codemirror/view'
 import { type ModelSQLMeshModel } from '@models/sqlmesh-model'
 import SQLMeshModel from './SQLMeshModel'
+import { isNil } from '@utils/index'
 
 export { events, HoverTooltip, SQLMeshModel, SQLMeshExpression }
 
@@ -36,7 +37,7 @@ function HoverTooltip(models: Map<string, ModelSQLMeshModel>): Extension {
       const modelName = view.state.doc.sliceString(start, end)
       const model = models.get(modelName)
 
-      if (model == null) return null
+      if (isNil(model)) return null
 
       return {
         pos: start,
diff --git a/web/client/src/library/components/editor/hooks.ts b/web/client/src/library/components/editor/hooks.ts
index 0982bf40..ae4bb08f 100644
--- a/web/client/src/library/components/editor/hooks.ts
+++ b/web/client/src/library/components/editor/hooks.ts
@@ -106,7 +106,7 @@ function useSQLMeshModelExtensions(
     )
 
     function handleEventModelClick(event: MouseEvent): void {
-      if (event.ctrlKey) {
+      if (event.metaKey) {
         const model = findModel(event, models)
 
         if (isNil(model)) return
@@ -116,7 +116,7 @@ function useSQLMeshModelExtensions(
     }
 
     function handleEventlColumnClick(event: MouseEvent): void {
-      if (event.ctrlKey) {
+      if (event.metaKey) {
         if (isNil(model)) return
 
         const column = findColumn(event, model)
@@ -127,28 +127,24 @@ function useSQLMeshModelExtensions(
       }
     }
 
-    return isNil(model)
-      ? []
-      : ([
-          models.size > 0 && isActionMode && HoverTooltip(models),
-          events({
-            keydown: e => {
-              if (e.ctrlKey) {
-                setIsActionMode(true)
-              }
-            },
-            keyup: e => {
-              if (isFalse(e.ctrlKey)) {
-                setIsActionMode(false)
-              }
-            },
-          }),
-          isNotNil(handleModelClick) &&
-            events({ click: handleEventModelClick }),
-          isNotNil(handleModelColumn) &&
-            events({ click: handleEventlColumnClick }),
-          SQLMeshModel(models, columns, isActionMode, model),
-        ].filter(Boolean) as Extension[])
+    return [
+      models.size > 0 && isActionMode && HoverTooltip(models),
+      events({
+        keydown: e => {
+          if (e.metaKey) {
+            setIsActionMode(true)
+          }
+        },
+        keyup: e => {
+          if (isFalse(e.metaKey)) {
+            setIsActionMode(false)
+          }
+        },
+      }),
+      isNotNil(handleModelClick) && events({ click: handleEventModelClick }),
+      isNotNil(handleModelColumn) && events({ click: handleEventlColumnClick }),
+      SQLMeshModel(models, columns, isActionMode, model),
+    ].filter(Boolean) as Extension[]
   }, [handleModelClick, handleModelColumn, model, models, files, isActionMode])
 
   return extensions
diff --git a/web/client/src/library/pages/root/Navigation.tsx b/web/client/src/library/pages/root/Navigation.tsx
index 84116c3b..6fc0cf4b 100644
--- a/web/client/src/library/pages/root/Navigation.tsx
+++ b/web/client/src/library/pages/root/Navigation.tsx
@@ -16,6 +16,7 @@ export default function PageNavigation(): JSX.Element {
       className={clsx(
         'relative min-w-[10rem] px-2 min-h-8 max-h-8 w-full flex items-center',
         modules.showHistoryNavigation ? 'justify-between' : 'justify-end',
+        isFetchingModels && 'overflow-hidden',
       )}
     >
       {modules.showHistoryNavigation && <HistoryNavigation />}


Test patch:
diff --git a/tests/core/analytics/test_collector.py b/tests/core/analytics/test_collector.py
index 42beecde..7c39081c 100644
--- a/tests/core/analytics/test_collector.py
+++ b/tests/core/analytics/test_collector.py
@@ -188,7 +188,7 @@ def test_on_plan_apply(
                 {
                     "seq_num": 0,
                     "event_type": "PLAN_APPLY_START",
-                    "event": f'{{"plan_id": "{plan_id}", "engine_type": "bigquery", "state_sync_type": "mysql", "scheduler_type": "builtin", "is_dev": false, "skip_backfill": false, "no_gaps": false, "forward_only": false, "ensure_finalized_snapshots": false, "has_restatements": false, "directly_modified_count": 17, "indirectly_modified_count": 0, "environment_name_hash": "d6e4a9b6646c62fc48baa6dd6150d1f7"}}',
+                    "event": f'{{"plan_id": "{plan_id}", "engine_type": "bigquery", "state_sync_type": "mysql", "scheduler_type": "builtin", "is_dev": false, "skip_backfill": false, "no_gaps": false, "forward_only": false, "ensure_finalized_snapshots": false, "has_restatements": false, "directly_modified_count": 18, "indirectly_modified_count": 0, "environment_name_hash": "d6e4a9b6646c62fc48baa6dd6150d1f7"}}',
                     **common_fields,
                 }
             ),
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index 2cbbf4ce..89e2cee5 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -7,6 +7,8 @@ from unittest.mock import PropertyMock, call, patch
 
 import freezegun
 import pytest
+import pandas as pd
+from pathlib import Path
 from pytest_mock.plugin import MockerFixture
 from sqlglot import parse_one
 from sqlglot.errors import SchemaError
@@ -24,7 +26,7 @@ from sqlmesh.core.config import (
 from sqlmesh.core.context import Context
 from sqlmesh.core.dialect import parse, schema_
 from sqlmesh.core.environment import Environment
-from sqlmesh.core.model import load_sql_based_model
+from sqlmesh.core.model import load_sql_based_model, model
 from sqlmesh.core.model.kind import ModelKindName
 from sqlmesh.core.plan import BuiltInPlanEvaluator, PlanBuilder
 from sqlmesh.utils.date import (
@@ -34,7 +36,7 @@ from sqlmesh.utils.date import (
     to_timestamp,
     yesterday_ds,
 )
-from sqlmesh.utils.errors import ConfigError
+from sqlmesh.utils.errors import ConfigError, SQLMeshError
 from tests.utils.test_filesystem import create_temp_file
 
 
@@ -285,6 +287,32 @@ def test_plan_execution_time():
     )
 
 
+def test_python_model_empty_df_raises(sushi_context, capsys):
+    @model(
+        "memory.sushi.test_model",
+        columns={"col": "int"},
+        kind=dict(name=ModelKindName.FULL),
+    )
+    def entrypoint(context, **kwargs):
+        yield pd.DataFrame({"col": []})
+
+    test_model = model.get_registry()["memory.sushi.test_model"].model(
+        module_path=Path("."), path=Path(".")
+    )
+    sushi_context.upsert_model(test_model)
+
+    capsys.readouterr()
+    with pytest.raises(SQLMeshError):
+        sushi_context.plan(no_prompts=True, auto_apply=True)
+
+    assert (
+        "Cannot construct source query from an empty \nDataFrame. This error "
+        "is commonly related to Python models that produce no data.\nFor such "
+        "models, consider yielding from an empty generator if the resulting set "
+        "\nis empty, i.e. use `yield from ()`"
+    ) in capsys.readouterr().out
+
+
 def test_env_and_default_schema_normalization(mocker: MockerFixture):
     from sqlglot.dialects import DuckDB
     from sqlglot.dialects.dialect import NormalizationStrategy
@@ -524,7 +552,7 @@ def test_janitor(sushi_context, mocker: MockerFixture) -> None:
     )
     # Assert that the views are dropped for each snapshot just once and make sure that the name used is the
     # view name with the environment as a suffix
-    assert adapter_mock.drop_view.call_count == 12
+    assert adapter_mock.drop_view.call_count == 13
     adapter_mock.drop_view.assert_has_calls(
         [
             call(
diff --git a/tests/core/test_integration.py b/tests/core/test_integration.py
index 14290b76..04b12c1c 100644
--- a/tests/core/test_integration.py
+++ b/tests/core/test_integration.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import typing as t
 from collections import Counter
 from datetime import timedelta
@@ -9,6 +11,7 @@ from pytest_mock.plugin import MockerFixture
 from sqlglot import exp
 from sqlglot.expressions import DataType
 
+from sqlmesh import CustomMaterialization
 from sqlmesh.core import constants as c
 from sqlmesh.core import dialect as d
 from sqlmesh.core.config import AutoCategorizationMode
@@ -19,6 +22,7 @@ from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.model import (
     IncrementalByTimeRangeKind,
     IncrementalByUniqueKeyKind,
+    Model,
     ModelKind,
     ModelKindName,
     SqlModel,
@@ -37,6 +41,10 @@ from sqlmesh.core.snapshot import (
 from sqlmesh.utils.date import TimeLike, now, to_date, to_datetime, to_timestamp
 from tests.conftest import DuckDBMetadata, SushiDataValidator
 
+
+if t.TYPE_CHECKING:
+    from sqlmesh import QueryOrDF
+
 pytestmark = pytest.mark.slow
 
 
@@ -350,13 +358,16 @@ def test_full_history_restatement_model_regular_plan_preview_enabled(
     context.upsert_model(model)
     snapshot = context.get_snapshot(model, raise_if_missing=True)
     customers_snapshot = context.get_snapshot("sushi.customers", raise_if_missing=True)
+    active_customers_snapshot = context.get_snapshot(
+        "sushi.active_customers", raise_if_missing=True
+    )
     waiter_as_customer_snapshot = context.get_snapshot(
         "sushi.waiter_as_customer_by_day", raise_if_missing=True
     )
 
     plan = context.plan("dev", no_prompts=True, skip_tests=True, enable_preview=True)
 
-    assert len(plan.new_snapshots) == 3
+    assert len(plan.new_snapshots) == 4
     assert (
         plan.context_diff.snapshots[snapshot.snapshot_id].change_category
         == SnapshotChangeCategory.FORWARD_ONLY
@@ -365,6 +376,10 @@ def test_full_history_restatement_model_regular_plan_preview_enabled(
         plan.context_diff.snapshots[customers_snapshot.snapshot_id].change_category
         == SnapshotChangeCategory.FORWARD_ONLY
     )
+    assert (
+        plan.context_diff.snapshots[active_customers_snapshot.snapshot_id].change_category
+        == SnapshotChangeCategory.FORWARD_ONLY
+    )
     assert (
         plan.context_diff.snapshots[waiter_as_customer_snapshot.snapshot_id].change_category
         == SnapshotChangeCategory.FORWARD_ONLY
@@ -372,6 +387,12 @@ def test_full_history_restatement_model_regular_plan_preview_enabled(
 
     assert plan.start == to_date("2023-01-07")
     assert plan.missing_intervals == [
+        SnapshotIntervals(
+            snapshot_id=active_customers_snapshot.snapshot_id,
+            intervals=[
+                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
+            ],
+        ),
         SnapshotIntervals(
             snapshot_id=customers_snapshot.snapshot_id,
             intervals=[
@@ -1250,6 +1271,41 @@ def test_incremental_by_partition(init_and_plan_context: t.Callable):
     ]
 
 
+@freeze_time("2023-01-08 15:00:00")
+def test_custom_materialization(init_and_plan_context: t.Callable):
+    context, _ = init_and_plan_context("examples/sushi")
+
+    custom_insert_called = False
+
+    class CustomFullMaterialization(CustomMaterialization):
+        NAME = "test_custom_full"
+
+        def insert(
+            self,
+            table_name: str,
+            query_or_df: QueryOrDF,
+            model: Model,
+            is_first_insert: bool,
+            **kwargs: t.Any,
+        ) -> None:
+            nonlocal custom_insert_called
+            custom_insert_called = True
+
+            self._replace_query_for_model(model, table_name, query_or_df)
+
+    model = context.get_model("sushi.top_waiters")
+    kwargs = {
+        **model.dict(),
+        # Make a breaking change.
+        "kind": dict(name="CUSTOM", materialization="test_custom_full"),
+    }
+    context.upsert_model(SqlModel.parse_obj(kwargs))
+
+    context.plan(auto_apply=True, no_prompts=True)
+
+    assert custom_insert_called
+
+
 @pytest.mark.parametrize(
     "context_fixture",
     ["sushi_context", "sushi_dbt_context", "sushi_test_dbt_context", "sushi_no_default_catalog"],
@@ -1877,7 +1933,7 @@ def test_environment_suffix_target_table(init_and_plan_context: t.Callable):
     assert set(metadata.schemas) - starting_schemas == {"raw"}
     prod_views = {x for x in metadata.qualified_views if x.db in environments_schemas}
     # Make sure that all models are present
-    assert len(prod_views) == 12
+    assert len(prod_views) == 13
     apply_to_environment(context, "dev")
     # Make sure no new schemas are created
     assert set(metadata.schemas) - starting_schemas == {"raw"}
@@ -1935,9 +1991,9 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
         user_default_tables,
         non_default_tables,
     ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
-    assert len(prod_views) == 12
+    assert len(prod_views) == 13
     assert len(dev_views) == 0
-    assert len(user_default_tables) == 22
+    assert len(user_default_tables) == 24
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
         {
@@ -1954,9 +2010,9 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
         user_default_tables,
         non_default_tables,
     ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
-    assert len(prod_views) == 12
-    assert len(dev_views) == 12
-    assert len(user_default_tables) == 22
+    assert len(prod_views) == 13
+    assert len(dev_views) == 13
+    assert len(user_default_tables) == 24
     assert len(non_default_tables) == 0
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
@@ -1974,9 +2030,9 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
         user_default_tables,
         non_default_tables,
     ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
-    assert len(prod_views) == 12
-    assert len(dev_views) == 24
-    assert len(user_default_tables) == 22
+    assert len(prod_views) == 13
+    assert len(dev_views) == 26
+    assert len(user_default_tables) == 24
     assert len(non_default_tables) == 0
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
@@ -1995,9 +2051,9 @@ def test_environment_catalog_mapping(init_and_plan_context: t.Callable):
         user_default_tables,
         non_default_tables,
     ) = get_default_catalog_and_non_tables(metadata, context.default_catalog)
-    assert len(prod_views) == 12
-    assert len(dev_views) == 12
-    assert len(user_default_tables) == 22
+    assert len(prod_views) == 13
+    assert len(dev_views) == 13
+    assert len(user_default_tables) == 24
     assert len(non_default_tables) == 0
     assert state_metadata.schemas == ["sqlmesh"]
     assert {x.sql() for x in state_metadata.qualified_tables}.issuperset(
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 3baf9a30..97f8aa97 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -24,6 +24,7 @@ from sqlmesh.core.context import Context, ExecutionContext
 from sqlmesh.core.dialect import parse
 from sqlmesh.core.macros import MacroEvaluator, macro
 from sqlmesh.core.model import (
+    CustomKind,
     PythonModel,
     FullKind,
     IncrementalByTimeRangeKind,
@@ -4672,3 +4673,50 @@ def my_model(context, **kwargs):
     context = Context(paths=tmp_path, config=config)
     assert context.get_model(expected_name).name == expected_name
     assert isinstance(context.get_model(expected_name), PythonModel)
+
+
+def test_custom_kind():
+    from sqlmesh import CustomMaterialization
+
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind CUSTOM (
+                materialization 'MyTestStrategy',
+                forward_only true,
+                disable_restatement true,
+                materialization_properties (
+                  'key_a' = 'value_a',
+                  key_b = 2,
+                  'key_c' = true,
+                  'key_d' = 1.23,
+                )
+            )
+        );
+
+        SELECT a, b
+        """
+    )
+
+    with pytest.raises(
+        ConfigError, match=r"Materialization strategy with name 'MyTestStrategy' was not found.*"
+    ):
+        load_sql_based_model(expressions)
+
+    class MyTestStrategy(CustomMaterialization):
+        pass
+
+    model = load_sql_based_model(expressions)
+    assert model.kind.is_custom
+
+    kind = t.cast(CustomKind, model.kind)
+    assert kind.disable_restatement
+    assert kind.forward_only
+    assert kind.materialization == "MyTestStrategy"
+    assert kind.materialization_properties == {
+        "key_a": "value_a",
+        "key_b": 2,
+        "key_c": True,
+        "key_d": 1.23,
+    }
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)
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 15a23474..85faa75f 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -37,12 +37,13 @@ from sqlmesh.core.snapshot import (
     Snapshot,
     SnapshotChangeCategory,
     SnapshotFingerprint,
-    categorize_change,
+    SnapshotTableInfo,
     earliest_start_date,
     fingerprint_from_node,
     has_paused_forward_only,
     missing_intervals,
 )
+from sqlmesh.core.snapshot.categorizer import categorize_change
 from sqlmesh.core.snapshot.definition import display_name
 from sqlmesh.utils import AttributeDict
 from sqlmesh.utils.date import to_date, to_datetime, to_timestamp
@@ -1815,3 +1816,35 @@ def test_external_model_audits(sushi_context):
     assert snapshot.evaluatable
     assert len(snapshot.model.audits) == 2
     assert snapshot.intervals
+
+
+def test_custom_model_kind(make_snapshot):
+    from sqlmesh import CustomMaterialization
+
+    class MyCustomStrategy(CustomMaterialization):
+        pass
+
+    snapshot = make_snapshot(
+        SqlModel(
+            name="test_model_name",
+            kind=dict(
+                name="CUSTOM",
+                materialization="MyCustomStrategy",
+                materialization_properties=parse_one("('test_key' = 'test_value')"),
+            ),
+            owner="owner",
+            dialect="",
+            cron="@hourly",
+            start="2023-01-29",
+            query=parse_one("SELECT id, @end_ds as ds FROM name"),
+        )
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    assert snapshot.custom_materialization == "MyCustomStrategy"
+
+    table_info = snapshot.table_info
+    assert table_info.custom_materialization == "MyCustomStrategy"
+
+    parsed_table_info = SnapshotTableInfo.parse_raw(table_info.json())
+    assert parsed_table_info.custom_materialization == "MyCustomStrategy"
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py
index 8fa9fc1a..286ffa21 100644
--- a/tests/core/test_snapshot_evaluator.py
+++ b/tests/core/test_snapshot_evaluator.py
@@ -1,3 +1,4 @@
+from __future__ import annotations
 import typing as t
 from unittest.mock import call, patch
 
@@ -21,6 +22,7 @@ from sqlmesh.core.engine_adapter.shared import (
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.macros import RuntimeStage, macro
 from sqlmesh.core.model import (
+    Model,
     FullKind,
     IncrementalByTimeRangeKind,
     IncrementalUnmanagedKind,
@@ -41,12 +43,17 @@ from sqlmesh.core.snapshot import (
     SnapshotEvaluator,
     SnapshotTableCleanupTask,
 )
+from sqlmesh.core.snapshot.evaluator import CustomMaterialization
 from sqlmesh.utils.concurrency import NodeExecutionFailedError
 from sqlmesh.utils.date import to_timestamp
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.metaprogramming import Executable
 
 
+if t.TYPE_CHECKING:
+    from sqlmesh.core.engine_adapter._typing import QueryOrDF
+
+
 @pytest.fixture
 def snapshot(duck_conn, make_snapshot) -> Snapshot:
     duck_conn.execute("CREATE VIEW tbl AS SELECT 1 AS a")
@@ -1442,10 +1449,8 @@ def test_insert_into_scd_type_2_by_time(
         unique_key=[exp.to_column("id", quoted=True)],
         valid_from_col=exp.column("valid_from", quoted=True),
         valid_to_col=exp.column("valid_to", quoted=True),
-        updated_at_col=exp.column("updated_at", quoted=True),
-        start="2020-01-01",
-        end="2020-01-02",
         execution_time="2020-01-02",
+        updated_at_col=exp.column("updated_at", quoted=True),
         invalidate_hard_deletes=False,
         table_description=None,
         column_descriptions={},
@@ -1610,8 +1615,6 @@ def test_insert_into_scd_type_2_by_column(
         check_columns=exp.Star(),
         valid_from_col=exp.column("valid_from", quoted=True),
         valid_to_col=exp.column("valid_to", quoted=True),
-        start="2020-01-01",
-        end="2020-01-02",
         execution_time="2020-01-02",
         execution_time_as_valid_from=False,
         invalidate_hard_deletes=False,
@@ -2079,3 +2082,58 @@ def test_evaluate_incremental_by_partition(mocker: MockerFixture, make_snapshot,
         ],
         columns_to_types=model.columns_to_types,
     )
+
+
+def test_custom_materialization_strategy(adapter_mock, make_snapshot):
+    custom_insert_called = False
+
+    class TestCustomMaterializationStrategy(CustomMaterialization):
+        NAME = "custom_materialization_test"
+
+        def insert(
+            self,
+            table_name: str,
+            query_or_df: QueryOrDF,
+            model: Model,
+            is_first_insert: bool,
+            **kwargs: t.Any,
+        ) -> None:
+            nonlocal custom_insert_called
+            custom_insert_called = True
+
+            assert model.custom_materialization_properties == {"test_property": "test_value"}
+
+            assert isinstance(query_or_df, exp.Query)
+            assert query_or_df.sql() == 'SELECT * FROM "tbl" AS "tbl"'
+
+    evaluator = SnapshotEvaluator(adapter_mock)
+    model = load_sql_based_model(
+        parse(  # type: ignore
+            """
+            MODEL (
+                name test_schema.test_model,
+                kind CUSTOM (
+                    materialization 'custom_materialization_test',
+                    materialization_properties (
+                        'test_property' = 'test_value'
+                    )
+                )
+            );
+
+            SELECT * FROM tbl;
+            """
+        )
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    evaluator.evaluate(
+        snapshot,
+        start="2020-01-01",
+        end="2020-01-02",
+        execution_time="2020-01-02",
+        snapshots={},
+    )
+
+    assert custom_insert_called
diff --git a/tests/dbt/test_integration.py b/tests/dbt/test_integration.py
index fc3a10bd..a7838897 100644
--- a/tests/dbt/test_integration.py
+++ b/tests/dbt/test_integration.py
@@ -192,11 +192,12 @@ test_config = config"""
         values: t.List[t.Tuple[int, str, str]],
     ):
         df = create_df(values, self.source_schema)
-        adapter.replace_query(
-            "sushi.raw_marketing",
-            df,
-            columns_to_types=columns_to_types_from_df(df),
-        )
+        columns_to_types = columns_to_types_from_df(df)
+
+        if values:
+            adapter.replace_query("sushi.raw_marketing", df, columns_to_types=columns_to_types)
+        else:
+            adapter.create_table("sushi.raw_marketing", columns_to_types=columns_to_types)
 
     def _normalize_dbt_dataframe(
         self,
diff --git a/tests/integrations/github/cicd/test_integration.py b/tests/integrations/github/cicd/test_integration.py
index b37f7349..e5d318d2 100644
--- a/tests/integrations/github/cicd/test_integration.py
+++ b/tests/integrations/github/cicd/test_integration.py
@@ -1573,6 +1573,7 @@ def test_overlapping_changes_models(
 ```
 
 **Indirectly Modified:**
+- `sushi.active_customers`
 - `sushi.waiter_as_customer_by_day`
 
 """
@@ -1612,7 +1613,7 @@ def test_overlapping_changes_models(
 """
     )
 
-    assert len(get_environment_objects(controller, "hello_world_2")) == 3
+    assert len(get_environment_objects(controller, "hello_world_2")) == 4
     assert "new_col" in get_columns(controller, "hello_world_2", "customers")
 
     assert mock_pull_request.merge.called
diff --git a/tests/integrations/jupyter/test_magics.py b/tests/integrations/jupyter/test_magics.py
index 2db6f8da..b27c933a 100644
--- a/tests/integrations/jupyter/test_magics.py
+++ b/tests/integrations/jupyter/test_magics.py
@@ -291,7 +291,7 @@ def test_plan(
 
     # TODO: Should this be going to stdout? This is printing the status updates for when each batch finishes for
     # the models and how long it took
-    assert len(output.stdout.strip().split("\n")) == 21
+    assert len(output.stdout.strip().split("\n")) == 22
     assert not output.stderr
     assert len(output.outputs) == 5
     text_output = convert_all_html_output_to_text(output)
@@ -530,7 +530,7 @@ def test_info(notebook, sushi_context, convert_all_html_output_to_text, get_all_
     assert not output.stderr
     assert len(output.outputs) == 3
     assert convert_all_html_output_to_text(output) == [
-        "Models: 16",
+        "Models: 17",
         "Macros: 5",
         "Data warehouse connection succeeded",
     ]
@@ -545,7 +545,7 @@ def test_info(notebook, sushi_context, convert_all_html_output_to_text, get_all_
                         "span",
                         # "color: #008000; text-decoration-color: #008000"
                         {"style": f"{NEUTRAL_STYLE}; font-weight: bold"},
-                        "16",
+                        "17",
                         autoescape=False,
                     )
                 ),