Fix!: mark vars referenced in metadata macros as metadata

sqlmesh/core/audit/definition.py

@@ -438,13 +438,15 @@ def load_audit(
 
     extra_kwargs: t.Dict[str, t.Any] = {}
     if is_standalone:
-        jinja_macro_refrences, used_variables = extract_macro_references_and_variables(
+        jinja_macro_refrences, referenced_variables = extract_macro_references_and_variables(
             *(gen(s) for s in statements),
             gen(query),
         )
         jinja_macros = (jinja_macros or JinjaMacroRegistry()).trim(jinja_macro_refrences)
         for jinja_macro in jinja_macros.root_macros.values():
-            used_variables.update(extract_macro_references_and_variables(jinja_macro.definition)[1])
+            referenced_variables.update(
+                extract_macro_references_and_variables(jinja_macro.definition)[1]
+            )
 
         extra_kwargs["jinja_macros"] = jinja_macros
         extra_kwargs["python_env"] = make_python_env(
@@ -453,7 +455,7 @@ def load_audit(
             module_path,
             macros or macro.get_registry(),
             variables=variables,
-            used_variables=used_variables,
+            referenced_variables=referenced_variables,
         )
         extra_kwargs["default_catalog"] = default_catalog
         if project is not None:

sqlmesh/core/constants.py

@@ -80,7 +80,9 @@
 DEFAULT_SCHEMA = "default"
 
 SQLMESH_VARS = "__sqlmesh__vars__"
+SQLMESH_VARS_METADATA = "__sqlmesh__vars__metadata__"
 SQLMESH_BLUEPRINT_VARS = "__sqlmesh__blueprint__vars__"
+SQLMESH_BLUEPRINT_VARS_METADATA = "__sqlmesh__blueprint__vars__metadata__"
 
 VAR = "var"
 BLUEPRINT_VAR = "blueprint_var"

sqlmesh/core/macros.py

@@ -210,7 +210,12 @@ def __init__(
                 self.macros[normalize_macro_name(k)] = self.env[k]
             elif v.is_value:
                 value = self.env[k]
-                if k in (c.SQLMESH_VARS, c.SQLMESH_BLUEPRINT_VARS):
+                if k in (
+                    c.SQLMESH_VARS,
+                    c.SQLMESH_VARS_METADATA,
+                    c.SQLMESH_BLUEPRINT_VARS,
+                    c.SQLMESH_BLUEPRINT_VARS_METADATA,
+                ):
                     value = {
                         var_name: (
                             self.parse_one(var_value.sql)
@@ -557,17 +562,25 @@ def views(self) -> t.List[str]:
 
     def var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
         """Returns the value of the specified variable, or the default value if it doesn't exist."""
-        return (self.locals.get(c.SQLMESH_VARS) or {}).get(var_name.lower(), default)
+        return {
+            **(self.locals.get(c.SQLMESH_VARS) or {}),
+            **(self.locals.get(c.SQLMESH_VARS_METADATA) or {}),
+        }.get(var_name.lower(), default)
 
     def blueprint_var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
         """Returns the value of the specified blueprint variable, or the default value if it doesn't exist."""
-        return (self.locals.get(c.SQLMESH_BLUEPRINT_VARS) or {}).get(var_name.lower(), default)
+        return {
+            **(self.locals.get(c.SQLMESH_BLUEPRINT_VARS) or {}),
+            **(self.locals.get(c.SQLMESH_BLUEPRINT_VARS_METADATA) or {}),
+        }.get(var_name.lower(), default)
 
     @property
     def variables(self) -> t.Dict[str, t.Any]:
         return {
             **self.locals.get(c.SQLMESH_VARS, {}),
+            **self.locals.get(c.SQLMESH_VARS_METADATA, {}),
             **self.locals.get(c.SQLMESH_BLUEPRINT_VARS, {}),
+            **self.locals.get(c.SQLMESH_BLUEPRINT_VARS_METADATA, {}),
         }
 
     def _coerce(self, expr: exp.Expression, typ: t.Any, strict: bool = False) -> t.Any: