Feat: expose model fqn in the macro evaluator (#4510)

Author: georgesittas

sqlmesh/core/macros.py

@@ -173,6 +173,7 @@ def __init__(
         default_catalog: t.Optional[str] = None,
         path: Path = Path(),
         environment_naming_info: t.Optional[EnvironmentNamingInfo] = None,
+        model_fqn: t.Optional[str] = None,
     ):
         self.dialect = dialect
         self.generator = MacroDialect().generator()
@@ -198,6 +199,7 @@ def __init__(
         self._snapshots = snapshots if snapshots is not None else {}
         self._path = path
         self._environment_naming_info = environment_naming_info
+        self._model_fqn = model_fqn
 
         prepare_env(self.python_env, self.env)
         for k, v in self.python_env.items():
@@ -476,6 +478,12 @@ def this_model(self) -> str:
             raise SQLMeshError("Model name is not available in the macro evaluator.")
         return this_model.sql(dialect=self.dialect, identify=True, comments=False)
 
+    @property
+    def this_model_fqn(self) -> str:
+        if self._model_fqn is None:
+            raise SQLMeshError("Model name is not available in the macro evaluator.")
+        return self._model_fqn
+
     @property
     def engine_adapter(self) -> EngineAdapter:
         engine_adapter = self.locals.get("engine_adapter")

sqlmesh/core/renderer.py

@@ -169,6 +169,7 @@ def _resolve_table(table: str | exp.Table) -> str:
             default_catalog=self._default_catalog,
             path=self._path,
             environment_naming_info=environment_naming_info,
+            model_fqn=self._model_fqn,
         )
 
         start_time, end_time = (

tests/core/test_model.py

@@ -9993,6 +9993,7 @@ def test_extract_schema_in_post_statement(tmp_path: Path) -> None:
         SELECT c FROM x;
         ON_VIRTUAL_UPDATE_BEGIN;
         @check_schema('y');
+        @check_self_schema();
         ON_VIRTUAL_UPDATE_END;
         """
     )
@@ -10007,6 +10008,11 @@ def test_extract_schema_in_post_statement(tmp_path: Path) -> None:
 def check_schema(evaluator, model_name: str):
     if evaluator.runtime_stage != 'loading':
         assert evaluator.columns_to_types(model_name) == {"c": exp.DataType.build("INT")}
+
+@macro()
+def check_self_schema(evaluator):
+    if evaluator.runtime_stage != 'loading':
+        assert evaluator.columns_to_types(evaluator.this_model_fqn) == {"c": exp.DataType.build("INT")}
 """)
 
     context = Context(paths=tmp_path, config=config)