Fix: do not apply model_defaults statements to external models

docs/reference/model_configuration.md

@@ -136,7 +136,7 @@ You can also use the `@model_kind_name` variable to fine-tune control over `phys
     )
     ```
 
-You can aso define `pre_statements`, `post_statements` and `on_virtual_update` statements at the project level that will be applied to all models. These default statements are merged with any model-specific statements, with default statements executing first, followed by model-specific statements.
+You can aso define `pre_statements`, `post_statements` and `on_virtual_update` statements at the project level that will be applied to all models except [external models](../concepts/models/external_models.md). These default statements are merged with any model-specific statements, with default statements executing first, followed by model-specific statements.
 
 === "YAML"
 

sqlmesh/core/model/definition.py

@@ -2583,6 +2583,12 @@ def _create_model(
     use_original_sql: bool = False,
     **kwargs: t.Any,
 ) -> Model:
+    """Creates a model of the given class, applying the project's model defaults.
+
+    Defaults that don't apply to the model class are dropped. External models, for example, don't
+    inherit the default gateway or the default pre-statements, post-statements and on-virtual-update
+    statements.
+    """
     validate_extra_and_required_fields(
         klass,
         {"name", *kwargs} - {"grain", "table_properties"},
@@ -2614,6 +2620,10 @@ def _create_model(
         # external_models.yaml, so it must remain explicit rather than inheriting the
         # gateway used to execute managed models in the project.
         defaults.pop("gateway", None)
+        # External models are symbolic, so their statements are never executed. They are also
+        # built without the project's macros, so default statements could not be rendered.
+        for statement_field in ("pre_statements", "post_statements", "on_virtual_update"):
+            defaults.pop(statement_field, None)
     if not issubclass(klass, SqlModel):
         defaults.pop("optimize_query", None)
 

sqlmesh/migrations/v0103_exclude_default_statements_from_external_models.py

@@ -0,0 +1,10 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Exclude model_defaults statements from the fingerprint of external models."""
+
+
+def migrate_schemas(engine_adapter, schema, **kwargs):  # type: ignore
+    pass
+
+
+def migrate_rows(engine_adapter, schema, **kwargs):  # type: ignore
+    pass

tests/core/test_context.py

@@ -4427,6 +4427,72 @@ def test_model_defaults_statements_with_on_virtual_update(tmp_path: Path):
     assert model.on_virtual_update[1].sql() == "SELECT 'Model-specific update' AS message"
 
 
+@pytest.mark.parametrize("statement_type", ["pre_statements", "post_statements"])
+def test_model_defaults_statements_not_applied_to_external_models(
+    tmp_path: Path, statement_type: str
+):
+    """Changing a default statement that calls a project macro doesn't break the plan.
+
+    External models used to inherit the statement without the project's macros, so the second
+    plan failed when it rendered their copy to compare it.
+    """
+    (tmp_path / "macros").mkdir()
+    (tmp_path / "macros" / "tag.py").write_text(
+        """
+from sqlmesh import macro
+
+
+@macro()
+def tag_query(evaluator, version):
+    return None
+"""
+    )
+    (tmp_path / "external_models.yaml").write_text(
+        """
+- name: '"db"."ext_db"."ext_table"'
+  columns:
+    id: int
+"""
+    )
+    (tmp_path / "models").mkdir()
+    (tmp_path / "models" / "test_model.sql").write_text(
+        """
+MODEL (name test.test_model, kind FULL);
+
+SELECT id FROM db.ext_db.ext_table
+"""
+    )
+
+    def write_config(version: str) -> None:
+        (tmp_path / "config.yaml").write_text(
+            f"""
+gateways:
+    local:
+        connection:
+            type: duckdb
+            database: {tmp_path / "db.duckdb"}
+default_gateway: local
+model_defaults:
+    dialect: duckdb
+    {statement_type}:
+        - "@tag_query('{version}')"
+"""
+        )
+
+    write_config("v1")
+    ctx = Context(paths=[tmp_path])
+    ctx.engine_adapter.execute("CREATE SCHEMA ext_db")
+    ctx.engine_adapter.execute("CREATE TABLE ext_db.ext_table AS SELECT 1 AS id")
+    ctx.plan(auto_apply=True, no_prompts=True)
+
+    write_config("v2")
+    ctx = Context(paths=[tmp_path])
+    plan = ctx.plan(no_prompts=True)
+    assert [s.name for s in plan.directly_modified] == ['"db"."test"."test_model"']
+    assert not ctx.get_model("db.ext_db.ext_table").pre_statements
+    assert not ctx.get_model("db.ext_db.ext_table").post_statements
+
+
 def test_uppercase_gateway_external_models(tmp_path):
     # Create a temporary SQLMesh project with uppercase gateway name
     config_py = tmp_path / "config.py"

tests/core/test_model.py

@@ -13552,6 +13552,34 @@ def test_external_model_does_not_inherit_model_defaults_gateway() -> None:
     assert explicit_external_model.gateway == "source_gateway"
 
 
+def test_external_model_does_not_inherit_model_defaults_statements() -> None:
+    """External models drop the default statements but keep explicit ones; SQL models inherit them."""
+    defaults = ModelDefaultsConfig(
+        dialect="duckdb",
+        pre_statements=["SET a = 1"],
+        post_statements=["ANALYZE"],
+        on_virtual_update=["SELECT 1"],
+    ).dict()
+
+    external_model = create_external_model("source_schema.source", defaults=defaults)
+    assert not external_model.pre_statements
+    assert not external_model.post_statements
+    assert not external_model.on_virtual_update
+
+    # Statements passed explicitly are kept.
+    explicit_model = create_external_model(
+        "source_schema.explicit_source",
+        defaults=defaults,
+        pre_statements=[d.parse_one("SET b = 2")],
+    )
+    assert [s.sql() for s in explicit_model.pre_statements] == ["SET b = 2"]
+
+    sql_model = create_sql_model("a.b", d.parse_one("SELECT 1 AS x"), defaults=defaults)
+    assert [s.sql() for s in sql_model.pre_statements] == ["SET a = 1"]
+    assert [s.sql() for s in sql_model.post_statements] == ["ANALYZE"]
+    assert [s.sql() for s in sql_model.on_virtual_update] == ["SELECT 1"]
+
+
 def test_no_gateway_uses_global_default_catalog():
     """
     Control test: when a model does NOT specify a gateway, the global