Merge branch 'main' into add_fabric_warehouse

Author: fresioAS

README.md

@@ -145,8 +145,7 @@ python -m venv .venv
 source .venv/bin/activate
 pip install 'sqlmesh[lsp]' # install the sqlmesh package with extensions to work with VSCode
 source .venv/bin/activate # reactivate the venv to ensure you're using the right installation
-sqlmesh init duckdb # get started right away with a local duckdb instance
-sqlmesh plan # see the plan for the changes you're making
+sqlmesh init # follow the prompts to get started (choose DuckDB)
 ```
 
 </details>
@@ -163,13 +162,12 @@ python -m venv .venv
 .\.venv\Scripts\Activate.ps1
 pip install 'sqlmesh[lsp]' # install the sqlmesh package with extensions to work with VSCode
 .\.venv\Scripts\Activate.ps1 # reactivate the venv to ensure you're using the right installation
-sqlmesh init duckdb # get started right away with a local duckdb instance
-sqlmesh plan # see the plan for the changes you're making
+sqlmesh init # follow the prompts to get started (choose DuckDB)
 ```
 </details>
 
 
-Follow the [quickstart guide](https://sqlmesh.readthedocs.io/en/stable/quickstart/cli/#1-create-the-sqlmesh-project) to learn how to use SQLMesh. You already have a head start!
+Follow the [quickstart guide](https://sqlmesh.readthedocs.io/en/stable/quickstart/cli/) to learn how to use SQLMesh. You already have a head start!
 
 Follow the [crash course](https://sqlmesh.readthedocs.io/en/stable/examples/sqlmesh_cli_crash_course/) to learn the core movesets and use the easy to reference cheat sheet. 
 

docs/readme/architecture_diagram.png

@@ -36,6 +36,7 @@
         "end_ts": "int",
         "event_date": "date",
     },
+    signals=[("test_signal", {"arg": 1})],
 )
 def execute(
     context: ExecutionContext,

examples/sushi/models/orders.py

@@ -90,18 +90,18 @@ def columns(
 
         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 @@ def table_exists(self, table_name: TableName) -> bool:
 
         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)
 

sqlmesh/core/engine_adapter/mssql.py

@@ -672,6 +672,7 @@ def _load_python_models(
                             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:

sqlmesh/core/loader.py

@@ -9,6 +9,7 @@
 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 @@ def __init__(self, name: t.Optional[str] = None, is_sql: bool = False, **kwargs:
         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 @@ def model(
         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 @@ def model(
             "macros": macros,
             "jinja_macros": jinja_macros,
             "audit_definitions": audit_definitions,
+            "signal_definitions": signal_definitions,
             "blueprint_variables": blueprint_variables,
             **rendered_fields,
         }

sqlmesh/core/model/decorator.py

@@ -431,7 +431,11 @@ def _get_physical_layer_update_stage(
         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,
         )
 

sqlmesh/core/plan/stages.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];",
     ]

tests/core/engine_adapter/test_mssql.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():

tests/core/test_context.py

@@ -5303,6 +5303,30 @@ def my_signal(batch):
     )
 
 
+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(
         """

tests/core/test_model.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]