Fix: Create empty physical tables for fully annotated self-referential models

sqlmesh/core/snapshot/evaluator.py

@@ -746,11 +746,15 @@ def _evaluate_snapshot(
             adapter.execute(model.render_pre_statements(**render_statements_kwargs))
 
             if not target_table_exists or (model.is_seed and not snapshot.intervals):
-                columns_to_types_provided = (
+                # Only create the empty table if the columns were provided explicitly by the user
+                should_create_empty_table = (
                     model.kind.is_materialized
                     and model.columns_to_types_
                     and columns_to_types_all_known(model.columns_to_types_)
                 )
+                if not should_create_empty_table:
+                    # Or if the model is self-referential and its query is fully annotated with types
+                    should_create_empty_table = model.depends_on_self and model.annotated
                 if self._can_clone(snapshot, deployability_index):
                     self._clone_snapshot_in_dev(
                         snapshot=snapshot,
@@ -763,7 +767,7 @@ def _evaluate_snapshot(
                     )
                     runtime_stage = RuntimeStage.EVALUATING
                     target_table_exists = True
-                elif columns_to_types_provided or model.is_seed or model.kind.is_scd_type_2:
+                elif should_create_empty_table or model.is_seed or model.kind.is_scd_type_2:
                     self._execute_create(
                         snapshot=snapshot,
                         table_name=target_table_name,

tests/core/test_integration.py

@@ -980,6 +980,32 @@ def test_new_forward_only_model(init_and_plan_context: t.Callable):
     assert context.engine_adapter.table_exists(snapshot.table_name(is_deployable=False))
 
 
+@time_machine.travel("2023-01-08 00:00:00 UTC")
+def test_annotated_self_referential_model(init_and_plan_context: t.Callable):
+    context, _ = init_and_plan_context("examples/sushi")
+
+    # Projections are fully annotated in the query but columns were not specified explicitly
+    expressions = d.parse(
+        f"""
+        MODEL (
+            name memory.sushi.test_self_ref,
+            kind FULL,
+            start '2023-01-01',
+        );
+
+        SELECT 1::INT AS one FROM memory.sushi.test_self_ref;
+        """
+    )
+    model = load_sql_based_model(expressions)
+    assert model.depends_on_self
+    context.upsert_model(model)
+
+    context.plan("prod", skip_tests=True, no_prompts=True, auto_apply=True)
+
+    df = context.fetchdf("SELECT one FROM memory.sushi.test_self_ref")
+    assert len(df) == 0
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_plan_set_choice_is_reflected_in_missing_intervals(init_and_plan_context: t.Callable):
     context, _ = init_and_plan_context("examples/sushi")