PR feedback

Author: georgesittas

sqlmesh/core/model/common.py

@@ -74,7 +74,7 @@ def make_python_env(
                 # If this macro has been seen before as a non-metadata macro, prioritize that
                 used_macros[name] = (
                     macros[name],
-                    (used_macros.get(name) or (None, is_metadata))[1],
+                    used_macros.get(name, (None, is_metadata))[1],
                 )
                 if name == c.VAR:
                     args = macro_func_or_var.this.expressions
@@ -92,7 +92,7 @@ def make_python_env(
                     # If this macro has been seen before as a non-metadata macro, prioritize that
                     used_macros[name] = (
                         macros[name],
-                        (used_macros.get(name) or (None, is_metadata))[1],
+                        used_macros.get(name, (None, is_metadata))[1],
                     )
                 elif name in variables:
                     used_variables.add(name)
@@ -114,14 +114,13 @@ def make_python_env(
         if isinstance(used_macro, Executable):
             python_env[name] = used_macro
         elif not hasattr(used_macro, c.SQLMESH_BUILTIN) and name not in python_env:
-            used_macro_func = used_macro.func
-            previous_is_metadata = getattr(used_macro_func, c.SQLMESH_METADATA, None)
-
-            if is_metadata:
-                setattr(used_macro_func, c.SQLMESH_METADATA, is_metadata)
-
-            build_env(used_macro.func, env=env, name=name, path=module_path)
-            setattr(used_macro_func, c.SQLMESH_METADATA, previous_is_metadata)
+            build_env(
+                used_macro.func,
+                env=env,
+                name=name,
+                path=module_path,
+                is_metadata_obj=is_metadata,
+            )
 
     python_env.update(serialize_env(env, path=module_path))
     return _add_variables_to_python_env(

sqlmesh/migrations/v0078_warn_if_non_migratable_python_env_and_make_audits_and_signals_macros_metadata.py

@@ -34,10 +34,10 @@ def migrate(state_sync, **kwargs):  # type: ignore
     warning = (
         "SQLMesh detected that it may not be able to fully migrate the state database. This should not impact "
         "the migration process, but may result in unexpected changes being reported by the next `sqlmesh plan` "
-        "command. Please run the `sqlmesh diff` (https://sqlmesh.readthedocs.io/en/stable/reference/cli/?h=cli#diff) "
-        "command after the migration has completed, before making any new changes. If any unexpected changes are reported, "
-        "consider running a forward-only plan (https://sqlmesh.readthedocs.io/en/stable/concepts/plans/#forward-only-plans) "
-        "to avoid unnecessary backfills.\n"
+        "command. Please run `sqlmesh diff prod` after the migration has completed, before making any new "
+        "changes. If any unexpected changes are reported, consider running a forward-only plan to apply these "
+        "changes and avoid unnecessary backfills: sqlmesh plan prod --forward-only. "
+        "See https://sqlmesh.readthedocs.io/en/stable/concepts/plans/#forward-only-plans for more details.\n"
     )
 
     for (snapshot,) in engine_adapter.fetchall(
@@ -50,7 +50,6 @@ def migrate(state_sync, **kwargs):  # type: ignore
         if node.get("source_type") == "audit":
             continue
 
-        name = node["name"]
         python_env = node.get("python_env") or {}
 
         has_metadata = False
@@ -69,29 +68,33 @@ def migrate(state_sync, **kwargs):  # type: ignore
         dialect = node.get("dialect")
         metadata_hash_statements = []
 
-        if on_virtual_update := node.get("on_virtual_update"):
-            metadata_hash_statements.extend(parse_expression(on_virtual_update, dialect))
-
-        for _, audit_args in func_call_validator(node.get("audits") or []):
-            metadata_hash_statements.extend(audit_args.values())
-
-        for signal_name, signal_args in func_call_validator(
-            node.get("signals") or [], is_signal=True
-        ):
-            metadata_hash_statements.extend(signal_args.values())
-
-        if audit_definitions := node.get("audit_definitions"):
-            audit_queries = [
-                parse_expression(audit["query"], audit["dialect"])
-                for audit in audit_definitions.values()
-            ]
-            metadata_hash_statements.extend(audit_queries)
-
-        for macro_name in extract_used_macros(metadata_hash_statements):
-            serialized_macro = python_env.get(macro_name)
-            if isinstance(serialized_macro, dict) and not serialized_macro.get("is_metadata"):
-                get_console().log_warning(warning)
-                return
+        # We use try-except here as a conservative measure to avoid any unexpected exceptions
+        try:
+            if on_virtual_update := node.get("on_virtual_update"):
+                metadata_hash_statements.extend(parse_expression(on_virtual_update, dialect))
+
+            for _, audit_args in func_call_validator(node.get("audits") or []):
+                metadata_hash_statements.extend(audit_args.values())
+
+            for signal_name, signal_args in func_call_validator(
+                node.get("signals") or [], is_signal=True
+            ):
+                metadata_hash_statements.extend(signal_args.values())
+
+            if audit_definitions := node.get("audit_definitions"):
+                audit_queries = [
+                    parse_expression(audit["query"], audit["dialect"])
+                    for audit in audit_definitions.values()
+                ]
+                metadata_hash_statements.extend(audit_queries)
+
+            for macro_name in extract_used_macros(metadata_hash_statements):
+                serialized_macro = python_env.get(macro_name)
+                if isinstance(serialized_macro, dict) and not serialized_macro.get("is_metadata"):
+                    get_console().log_warning(warning)
+                    return
+        except Exception:
+            pass
 
 
 def extract_used_macros(expressions):

sqlmesh/utils/metaprogramming.py

@@ -269,6 +269,7 @@ def build_env(
     env: t.Dict[str, t.Tuple[t.Any, t.Optional[bool]]],
     name: str,
     path: Path,
+    is_metadata_obj: t.Optional[bool] = None,
 ) -> None:
     """Fills in env dictionary with all globals needed to execute the object.
 
@@ -279,6 +280,7 @@ def build_env(
         env: Dictionary to store the env.
         name: Name of the object in the env.
         path: The module path to serialize. Other modules will not be walked and treated as imports.
+        is_metadata_obj: An optional flag that determins whether the input object is metadata-only.
     """
     # We don't rely on `env` to keep track of visited objects, because it's populated in post-order
     visited: t.Set[str] = set()
@@ -356,7 +358,7 @@ def walk(obj: t.Any, name: str, is_metadata: t.Optional[bool] = None) -> None:
         env[name] = (obj, is_metadata)
 
     # The "metadata only" annotation of the object is transitive
-    walk(obj, name, getattr(obj, c.SQLMESH_METADATA, None))
+    walk(obj, name, is_metadata_obj or getattr(obj, c.SQLMESH_METADATA, None))
 
 
 @dataclass