Repository: TobikoData/sqlmesh
Pull request #2794 (2024-06-19 17:09:16)

Fix: truncate backfill models and adjust threshold
- Adjust `INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD` to 10.
- Limit the output for backfill, as well as added and removed models.


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/config/scheduler.py b/sqlmesh/core/config/scheduler.py
index 1278d2ed..093bdd4d 100644
--- a/sqlmesh/core/config/scheduler.py
+++ b/sqlmesh/core/config/scheduler.py
@@ -218,6 +218,7 @@ class AirflowSchedulerConfig(_BaseAirflowSchedulerConfig, BaseConfig):
     airflow_url: str = "http://localhost:8080/"
     username: str = "airflow"
     password: str = "airflow"
+    token: t.Optional[str] = None
     dag_run_poll_interval_secs: int = 10
     dag_creation_poll_interval_secs: int = 30
     dag_creation_max_retry_attempts: int = 10
@@ -236,8 +237,10 @@ class AirflowSchedulerConfig(_BaseAirflowSchedulerConfig, BaseConfig):
 
     def get_client(self, console: t.Optional[Console] = None) -> AirflowClient:
         session = Session()
-        session.headers.update({"Content-Type": "application/json"})
-        session.auth = (self.username, self.password)
+        if self.token is None:
+            session.auth = (self.username, self.password)
+        else:
+            session.headers.update({"Authorization": f"Bearer {self.token}"})
 
         return AirflowClient(
             session=session,
diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py
index 6baa09a5..e7d11749 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",
 }
 
 
@@ -60,7 +61,7 @@ class Console(abc.ABC):
     """Abstract base class for defining classes used for displaying information to the user and also interact
     with them when their input is needed."""
 
-    INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD = 20
+    INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD = 10
 
     @abc.abstractmethod
     def start_plan_evaluation(self, plan: Plan) -> None:
@@ -701,20 +702,20 @@ class TerminalConsole(Console):
         tree = Tree(f"[bold]{header}:")
         if added_snapshot_ids:
             added_tree = Tree("[bold][added]Added:")
-            for s_id in added_snapshot_ids:
+            for s_id in sorted(added_snapshot_ids):
                 snapshot = context_diff.snapshots[s_id]
                 added_tree.add(
                     f"[added]{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}"
                 )
-            tree.add(added_tree)
+            tree.add(self._limit_model_names(added_tree, self.verbose))
         if removed_snapshot_ids:
             removed_tree = Tree("[bold][removed]Removed:")
-            for s_id in removed_snapshot_ids:
+            for s_id in sorted(removed_snapshot_ids):
                 snapshot_table_info = context_diff.removed_snapshots[s_id]
                 removed_tree.add(
                     f"[removed]{snapshot_table_info.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}"
                 )
-            tree.add(removed_tree)
+            tree.add(self._limit_model_names(removed_tree, self.verbose))
         if modified_snapshot_ids:
             direct = Tree("[bold][direct]Directly Modified:")
             indirect = Tree("[bold][indirect]Indirectly Modified:")
@@ -862,6 +863,8 @@ class TerminalConsole(Console):
             backfill.add(
                 f"{snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)}: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}"
             )
+        if backfill:
+            backfill = self._limit_model_names(backfill, self.verbose)
         self._print(backfill)
 
     def _prompt_effective_from(
@@ -1509,10 +1512,17 @@ class MarkdownConsole(CaptureTerminalConsole):
         added_snapshot_models = {s for s in added_snapshots if s.is_model}
         if added_snapshot_models:
             self._print("\n**Added Models:**")
-            for snapshot in sorted(added_snapshot_models):
-                self._print(
-                    f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
-                )
+            added_models = sorted(added_snapshot_models)
+            list_length = len(added_models)
+            if not self.verbose and list_length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD:
+                self._print(added_models[0])
+                self._print(f"- `.... {list_length-2} more ....`\n")
+                self._print(added_models[-1])
+            else:
+                for snapshot in added_models:
+                    self._print(
+                        f"- `{snapshot.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
+                    )
 
         added_snapshot_audits = {s for s in added_snapshots if s.is_audit}
         if added_snapshot_audits:
@@ -1530,10 +1540,17 @@ class MarkdownConsole(CaptureTerminalConsole):
         removed_model_snapshot_table_infos = {s for s in removed_snapshot_table_infos if s.is_model}
         if removed_model_snapshot_table_infos:
             self._print("\n**Removed Models:**")
-            for snapshot_table_info in sorted(removed_model_snapshot_table_infos):
-                self._print(
-                    f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
-                )
+            removed_models = sorted(removed_model_snapshot_table_infos)
+            list_length = len(removed_models)
+            if not self.verbose and list_length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD:
+                self._print(removed_models[0])
+                self._print(f"- `.... {list_length-2} more ....`\n")
+                self._print(removed_models[-1])
+            else:
+                for snapshot_table_info in removed_models:
+                    self._print(
+                        f"- `{snapshot_table_info.display_name(environment_naming_info, default_catalog, dialect=self.dialect)}`"
+                    )
 
         removed_audit_snapshot_table_infos = {s for s in removed_snapshot_table_infos if s.is_audit}
         if removed_audit_snapshot_table_infos:
@@ -1605,6 +1622,7 @@ class MarkdownConsole(CaptureTerminalConsole):
         if not missing_intervals:
             return
         self._print("\n**Models needing backfill (missing dates):**")
+        snapshots = []
         for missing in missing_intervals:
             snapshot = plan.context_diff.snapshots[missing.snapshot_id]
             if not snapshot.is_model:
@@ -1614,10 +1632,19 @@ class MarkdownConsole(CaptureTerminalConsole):
             if not plan.deployability_index.is_deployable(snapshot):
                 preview_modifier = " (**preview**)"
 
-            self._print(
+            snapshots.append(
                 f"* `{snapshot.display_name(plan.environment_naming_info, default_catalog, dialect=self.dialect)}`: {missing.format_intervals(snapshot.node.interval_unit)}{preview_modifier}"
             )
 
+        length = len(snapshots)
+        if not self.verbose and length > self.INDIRECTLY_MODIFIED_DISPLAY_THRESHOLD:
+            self._print(snapshots[0])
+            self._print(f"- `.... {length-2} more ....`\n")
+            self._print(snapshots[-1])
+        else:
+            for snap in snapshots:
+                self._print(snap)
+
     def _show_categorized_snapshots(self, plan: Plan, default_catalog: t.Optional[str]) -> None:
         context_diff = plan.context_diff
         for snapshot in plan.categorized:
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/sqlmesh/schedulers/airflow/client.py b/sqlmesh/schedulers/airflow/client.py
index 3bafb42b..f2d1a657 100644
--- a/sqlmesh/schedulers/airflow/client.py
+++ b/sqlmesh/schedulers/airflow/client.py
@@ -230,6 +230,7 @@ class AirflowClient(BaseAirflowClient):
         response = self._session.post(
             urljoin(self._airflow_url, PLANS_PATH),
             data=request.json(),
+            headers={"Content-Type": "application/json"},
         )
         raise_for_status(response)
 
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,
                     )
                 ),
diff --git a/tests/schedulers/airflow/test_client.py b/tests/schedulers/airflow/test_client.py
index 02506dad..9dd7cd88 100644
--- a/tests/schedulers/airflow/test_client.py
+++ b/tests/schedulers/airflow/test_client.py
@@ -77,6 +77,7 @@ def test_apply_plan(mocker: MockerFixture, snapshot: Snapshot):
     args, data = apply_plan_mock.call_args_list[0]
 
     assert args[0] == "http://localhost:8080/sqlmesh/api/v1/plans"
+    assert data["headers"] == {"Content-Type": "application/json"}
     assert json.loads(data["data"]) == {
         "new_snapshots": [
             {