Repository: TobikoData/sqlmesh
Pull request #4801 (2025-06-24 19:54:27)

Fix: Propagation of snapshots with missing intervals into the physical layer update stage
This can sometimes cause a misleading output in the plan explainer. The `snapshots_with_missing_intervals` should only reflect snapshots selected for backfill.


Patch:
diff --git a/examples/sushi/models/orders.py b/examples/sushi/models/orders.py
index aa0e0455..8d8718a3 100644
--- a/examples/sushi/models/orders.py
+++ b/examples/sushi/models/orders.py
@@ -36,6 +36,7 @@ WAITERS = list(range(0, 10))
         "end_ts": "int",
         "event_date": "date",
     },
+    signals=[("test_signal", {"arg": 1})],
 )
 def execute(
     context: ExecutionContext,
diff --git a/sqlmesh/core/engine_adapter/mssql.py b/sqlmesh/core/engine_adapter/mssql.py
index 88b3f51e..a00e11e0 100644
--- a/sqlmesh/core/engine_adapter/mssql.py
+++ b/sqlmesh/core/engine_adapter/mssql.py
@@ -90,18 +90,18 @@ class MSSQLEngineAdapter(
 
         sql = (
             exp.select(
-                "column_name",
-                "data_type",
-                "character_maximum_length",
-                "numeric_precision",
-                "numeric_scale",
+                "COLUMN_NAME",
+                "DATA_TYPE",
+                "CHARACTER_MAXIMUM_LENGTH",
+                "NUMERIC_PRECISION",
+                "NUMERIC_SCALE",
             )
-            .from_("information_schema.columns")
-            .where(f"table_name = '{table.name}'")
+            .from_("INFORMATION_SCHEMA.COLUMNS")
+            .where(f"TABLE_NAME = '{table.name}'")
         )
         database_name = table.db
         if database_name:
-            sql = sql.where(f"table_schema = '{database_name}'")
+            sql = sql.where(f"TABLE_SCHEMA = '{database_name}'")
 
         columns_raw = self.fetchall(sql, quote_identifiers=True)
 
@@ -145,12 +145,12 @@ class MSSQLEngineAdapter(
 
         sql = (
             exp.select("1")
-            .from_("information_schema.tables")
-            .where(f"table_name = '{table.alias_or_name}'")
+            .from_("INFORMATION_SCHEMA.TABLES")
+            .where(f"TABLE_NAME = '{table.alias_or_name}'")
         )
         database_name = table.db
         if database_name:
-            sql = sql.where(f"table_schema = '{database_name}'")
+            sql = sql.where(f"TABLE_SCHEMA = '{database_name}'")
 
         result = self.fetchone(sql, quote_identifiers=True)
 
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index 7f90c0de..41bb8a0a 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -672,6 +672,7 @@ class SqlMeshLoader(Loader):
                             default_catalog=self.context.default_catalog,
                             infer_names=self.config.model_naming.infer_names,
                             audit_definitions=audits,
+                            signal_definitions=signals,
                             default_catalog_per_gateway=self.context.default_catalog_per_gateway,
                         ):
                             if model.enabled:
diff --git a/sqlmesh/core/model/decorator.py b/sqlmesh/core/model/decorator.py
index 0151e9ec..3b78efc6 100644
--- a/sqlmesh/core/model/decorator.py
+++ b/sqlmesh/core/model/decorator.py
@@ -9,6 +9,7 @@ from sqlglot import exp
 from sqlglot.dialects.dialect import DialectType
 
 from sqlmesh.core.macros import MacroRegistry
+from sqlmesh.core.signal import SignalRegistry
 from sqlmesh.utils.jinja import JinjaMacroRegistry
 from sqlmesh.core import constants as c
 from sqlmesh.core.dialect import MacroFunc, parse_one
@@ -48,23 +49,24 @@ class model(registry_decorator):
         self.kwargs = kwargs
 
         # Make sure that argument values are expressions in order to pass validation in ModelMeta.
-        calls = self.kwargs.pop("audits", [])
-        self.kwargs["audits"] = [
-            (
-                (call, {})
-                if isinstance(call, str)
-                else (
-                    call[0],
-                    {
-                        arg_key: exp.convert(
-                            tuple(arg_value) if isinstance(arg_value, list) else arg_value
-                        )
-                        for arg_key, arg_value in call[1].items()
-                    },
+        for function_call_attribute in ("audits", "signals"):
+            calls = self.kwargs.pop(function_call_attribute, [])
+            self.kwargs[function_call_attribute] = [
+                (
+                    (call, {})
+                    if isinstance(call, str)
+                    else (
+                        call[0],
+                        {
+                            arg_key: exp.convert(
+                                tuple(arg_value) if isinstance(arg_value, list) else arg_value
+                            )
+                            for arg_key, arg_value in call[1].items()
+                        },
+                    )
                 )
-            )
-            for call in calls
-        ]
+                for call in calls
+            ]
 
         if "default_catalog" in kwargs:
             raise ConfigError("`default_catalog` cannot be set on a per-model basis.")
@@ -142,6 +144,7 @@ class model(registry_decorator):
         defaults: t.Optional[t.Dict[str, t.Any]] = None,
         macros: t.Optional[MacroRegistry] = None,
         jinja_macros: t.Optional[JinjaMacroRegistry] = None,
+        signal_definitions: t.Optional[SignalRegistry] = None,
         audit_definitions: t.Optional[t.Dict[str, ModelAudit]] = None,
         dialect: t.Optional[str] = None,
         time_column_format: str = c.DEFAULT_TIME_COLUMN_FORMAT,
@@ -223,6 +226,7 @@ class model(registry_decorator):
             "macros": macros,
             "jinja_macros": jinja_macros,
             "audit_definitions": audit_definitions,
+            "signal_definitions": signal_definitions,
             "blueprint_variables": blueprint_variables,
             **rendered_fields,
         }
diff --git a/sqlmesh/core/plan/stages.py b/sqlmesh/core/plan/stages.py
index b50edfbb..863e5338 100644
--- a/sqlmesh/core/plan/stages.py
+++ b/sqlmesh/core/plan/stages.py
@@ -431,7 +431,11 @@ class PlanStagesBuilder:
         return PhysicalLayerUpdateStage(
             snapshots=self._get_snapshots_to_create(plan, snapshots),
             all_snapshots=snapshots,
-            snapshots_with_missing_intervals={s.snapshot_id for s in snapshots_to_intervals},
+            snapshots_with_missing_intervals={
+                s.snapshot_id
+                for s in snapshots_to_intervals
+                if plan.is_selected_for_backfill(s.name)
+            },
             deployability_index=deployability_index,
         )
 


Test patch:
diff --git a/tests/core/engine_adapter/test_mssql.py b/tests/core/engine_adapter/test_mssql.py
index beeaa59c..22f37174 100644
--- a/tests/core/engine_adapter/test_mssql.py
+++ b/tests/core/engine_adapter/test_mssql.py
@@ -79,7 +79,7 @@ def test_columns(adapter: MSSQLEngineAdapter):
     }
 
     adapter.cursor.execute.assert_called_once_with(
-        """SELECT [column_name], [data_type], [character_maximum_length], [numeric_precision], [numeric_scale] FROM [information_schema].[columns] WHERE [table_name] = 'table' AND [table_schema] = 'db';"""
+        """SELECT [COLUMN_NAME], [DATA_TYPE], [CHARACTER_MAXIMUM_LENGTH], [NUMERIC_PRECISION], [NUMERIC_SCALE] FROM [INFORMATION_SCHEMA].[COLUMNS] WHERE [TABLE_NAME] = 'table' AND [TABLE_SCHEMA] = 'db';"""
     )
 
 
@@ -149,8 +149,8 @@ def test_table_exists(make_mocked_engine_adapter: t.Callable):
     resp = adapter.table_exists("db.table")
     adapter.cursor.execute.assert_called_once_with(
         """SELECT 1 """
-        """FROM [information_schema].[tables] """
-        """WHERE [table_name] = 'table' AND [table_schema] = 'db';"""
+        """FROM [INFORMATION_SCHEMA].[TABLES] """
+        """WHERE [TABLE_NAME] = 'table' AND [TABLE_SCHEMA] = 'db';"""
     )
     assert resp
     adapter.cursor.fetchone.return_value = None
@@ -506,7 +506,7 @@ def test_replace_query(make_mocked_engine_adapter: t.Callable):
     adapter.replace_query("test_table", parse_one("SELECT a FROM tbl"), {"a": "int"})
 
     assert to_sql_calls(adapter) == [
-        """SELECT 1 FROM [information_schema].[tables] WHERE [table_name] = 'test_table';""",
+        """SELECT 1 FROM [INFORMATION_SCHEMA].[TABLES] WHERE [TABLE_NAME] = 'test_table';""",
         "TRUNCATE TABLE [test_table];",
         "INSERT INTO [test_table] ([a]) SELECT [a] FROM [tbl];",
     ]
diff --git a/tests/core/test_context.py b/tests/core/test_context.py
index e08f5346..213c4cec 100644
--- a/tests/core/test_context.py
+++ b/tests/core/test_context.py
@@ -2122,7 +2122,7 @@ def test_check_intervals(sushi_context, mocker):
     intervals = sushi_context.check_intervals(environment=None, no_signals=False, select_models=[])
 
     min_intervals = 19
-    assert spy.call_count == 1
+    assert spy.call_count == 2
     assert len(intervals) >= min_intervals
 
     for i in intervals.values():
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 7c65f258..a1f90344 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -5303,6 +5303,30 @@ def test_signals():
     )
 
 
+def test_load_python_model_with_signals():
+    @signal()
+    def always_true(batch):
+        return True
+
+    @model(
+        name="model_with_signal",
+        kind="full",
+        columns={'"COL"': "int"},
+        signals=[("always_true", {})],
+    )
+    def model_with_signal(context, **kwargs):
+        return pd.DataFrame([{"COL": 1}])
+
+    models = model.get_registry()["model_with_signal"].models(
+        get_variables=lambda _: {},
+        path=Path("."),
+        module_path=Path("."),
+        signal_definitions=signal.get_registry(),
+    )
+    assert len(models) == 1
+    assert models[0].signals == [("always_true", {})]
+
+
 def test_null_column_type():
     expressions = d.parse(
         """
diff --git a/tests/core/test_plan_stages.py b/tests/core/test_plan_stages.py
index c8989f9e..b806b95b 100644
--- a/tests/core/test_plan_stages.py
+++ b/tests/core/test_plan_stages.py
@@ -141,6 +141,10 @@ def test_build_plan_stages_basic(
         snapshot_a.snapshot_id,
         snapshot_b.snapshot_id,
     }
+    assert {s.snapshot_id for s in physical_stage.snapshots_with_missing_intervals} == {
+        snapshot_a.snapshot_id,
+        snapshot_b.snapshot_id,
+    }
     assert physical_stage.deployability_index == DeployabilityIndex.all_deployable()
 
     # Verify BackfillStage
@@ -357,6 +361,7 @@ def test_build_plan_stages_select_models(
     assert len(physical_stage.snapshots) == 1
     assert {s.snapshot_id for s in physical_stage.snapshots} == {snapshot_a.snapshot_id}
     assert physical_stage.deployability_index == DeployabilityIndex.all_deployable()
+    assert physical_stage.snapshots_with_missing_intervals == {snapshot_a.snapshot_id}
 
     # Verify BackfillStage
     backfill_stage = stages[2]