Fix!: parse runtime-rendered fields, extract python env deps from merge_filter (#4905)

Author: georgesittas

sqlmesh/core/model/common.py

@@ -440,6 +440,19 @@ def _executable_to_str(k: str, v: Executable) -> str:
     return [_executable_to_str(k, v) for k, v in sort_python_env(python_env)]
 
 
+def parse_strings_with_macro_refs(value: t.Any, dialect: DialectType) -> t.Any:
+    if isinstance(value, str) and "@" in value:
+        return exp.maybe_parse(value, dialect=dialect)
+
+    if isinstance(value, dict):
+        for k, v in dict(value).items():
+            value[k] = parse_strings_with_macro_refs(v, dialect)
+    elif isinstance(value, list):
+        value = [parse_strings_with_macro_refs(v, dialect) for v in value]
+
+    return value
+
+
 expression_validator: t.Callable = field_validator(
     "query",
     "expressions_",

sqlmesh/core/model/definition.py

@@ -27,6 +27,7 @@
     expression_validator,
     make_python_env,
     parse_dependencies,
+    parse_strings_with_macro_refs,
     single_value_or_tuple,
     sorted_python_env_payloads,
     validate_extra_and_required_fields,
@@ -72,13 +73,14 @@
 
 logger = logging.getLogger(__name__)
 
+
+UNRENDERABLE_MODEL_FIELDS = {"cron", "description"}
+
 PROPERTIES = {"physical_properties", "session_properties", "virtual_properties"}
 
 RUNTIME_RENDERED_MODEL_FIELDS = {
     "audits",
     "signals",
-    "description",
-    "cron",
     "merge_filter",
 } | PROPERTIES
 
@@ -2469,6 +2471,9 @@ def _create_model(
         if isinstance(property_values, exp.Tuple):
             statements.extend(property_values.expressions)
 
+    if isinstance(getattr(kwargs.get("kind"), "merge_filter", None), exp.Expression):
+        statements.append(kwargs["kind"].merge_filter)
+
     jinja_macro_references, used_variables = extract_macro_references_and_variables(
         *(gen(e if isinstance(e, exp.Expression) else e[0]) for e in statements)
     )
@@ -2751,21 +2756,24 @@ def render_field_value(value: t.Any) -> t.Any:
 
     for field_name, field_info in ModelMeta.all_field_infos().items():
         field = field_info.alias or field_name
+        field_value = fields.get(field)
 
-        if field in RUNTIME_RENDERED_MODEL_FIELDS:
+        # We don't want to parse python model cron="@..." kwargs (e.g. @daily) into MacroVar
+        if field == "cron" or field_value is None:
             continue
 
-        field_value = fields.get(field)
-        if field_value is None:
+        if field in RUNTIME_RENDERED_MODEL_FIELDS:
+            fields[field] = parse_strings_with_macro_refs(field_value, dialect)
             continue
 
         if isinstance(field_value, dict):
             rendered_dict = {}
             for key, value in field_value.items():
                 if key in RUNTIME_RENDERED_MODEL_FIELDS:
-                    rendered_dict[key] = value
+                    rendered_dict[key] = parse_strings_with_macro_refs(value, dialect)
                 elif (rendered := render_field_value(value)) is not None:
                     rendered_dict[key] = rendered
+
             if rendered_dict:
                 fields[field] = rendered_dict
             else:

tests/core/test_model.py

@@ -6436,7 +6436,7 @@ def end_date_macro(evaluator: MacroEvaluator, var: bool):
         owner="@IF(@gateway = 'dev', @{dev_owner}, @{prod_owner})",
         stamp="@{stamp}",
         tags=["@{tag1}", "@{tag2}"],
-        description="Model desc @{test_}",
+        description="'Model desc @{test_}'",
     )
     def model_with_macros(evaluator, **kwargs):
         return exp.select(
@@ -6480,26 +6480,109 @@ def model_with_macros(evaluator, **kwargs):
     assert query.sql() == """SELECT 'test_value' AS "a" """.strip()
 
 
+def test_unrendered_macros_sql_model(mocker: MockerFixture) -> None:
+    model = load_sql_based_model(
+        parse(
+            """
+            MODEL (
+              name db.employees,
+              kind INCREMENTAL_BY_UNIQUE_KEY (
+                unique_key @{key},
+                merge_filter source.id > 0 and target.updated_at < @end_ds and source.updated_at > @start_ds and @merge_filter_var
+              ),
+              cron '@daily',
+              allow_partials @IF(@gateway = 'dev', True, False),
+              physical_properties (
+                location1 = @'s3://bucket/prefix/@{schema_name}/@{table_name}',
+                location2 = @IF(@gateway = 'dev', @'hdfs://@{catalog_name}/@{schema_name}/dev/@{table_name}', @'s3://prod/@{table_name}'),
+                foo = @physical_var
+              ),
+              virtual_properties (
+                creatable_type = @{create_type},
+                bar = @virtual_var,
+              ),
+              session_properties (
+                'spark.executor.cores' = @IF(@gateway = 'dev', 1, 2),
+                'spark.executor.memory' = '1G',
+                baz = @session_var
+              ),
+            );
+
+            SELECT * FROM src;
+        """
+        ),
+        variables={
+            "gateway": "dev",
+            "key": "a",  # Not included in python_env because kind is rendered at load time
+            "create_type": "'SECURE'",
+            "merge_filter_var": True,
+            "physical_var": "bla",
+            "virtual_var": "blb",
+            "session_var": "blc",
+        },
+    )
+
+    assert model.python_env[c.SQLMESH_VARS] == Executable.value(
+        {
+            "gateway": "dev",
+            "create_type": "'SECURE'",
+            "merge_filter_var": True,
+            "physical_var": "bla",
+            "virtual_var": "blb",
+            "session_var": "blc",
+        }
+    )
+
+    assert "location1" in model.physical_properties
+    assert "location2" in model.physical_properties
+
+    # The properties will stay unrendered at load time
+    assert model.session_properties == {
+        "spark.executor.cores": exp.maybe_parse("@IF(@gateway = 'dev', 1, 2)"),
+        "spark.executor.memory": "1G",
+        "baz": exp.maybe_parse("@session_var"),
+    }
+    assert model.virtual_properties["creatable_type"] == exp.maybe_parse("@{create_type}")
+
+    assert (
+        model.physical_properties["location1"].sql()
+        == "@'s3://bucket/prefix/@{schema_name}/@{table_name}'"
+    )
+    assert (
+        model.physical_properties["location2"].sql()
+        == "@IF(@gateway = 'dev', @'hdfs://@{catalog_name}/@{schema_name}/dev/@{table_name}', @'s3://prod/@{table_name}')"
+    )
+
+    # merge_filter will stay unrendered as well
+    assert model.unique_key[0] == exp.column("a", quoted=True)
+    assert (
+        t.cast(exp.Expression, model.merge_filter).sql()
+        == '"__merge_source__"."id" > 0 AND "__merge_target__"."updated_at" < @end_ds AND "__merge_source__"."updated_at" > @start_ds AND @merge_filter_var'
+    )
+
+
 def test_unrendered_macros_python_model(mocker: MockerFixture) -> None:
     @model(
         "test_unrendered_macros_python_model_@{bar}",
         is_sql=True,
         kind=dict(
             name=ModelKindName.INCREMENTAL_BY_UNIQUE_KEY,
             unique_key="@{key}",
-            merge_filter="source.id > 0 and target.updated_at < @end_ds and source.updated_at > @start_ds",
+            merge_filter="source.id > 0 and target.updated_at < @end_ds and source.updated_at > @start_ds and @merge_filter_var",
         ),
         cron="@daily",
         columns={"a": "string"},
         allow_partials="@IF(@gateway = 'dev', True, False)",
         physical_properties=dict(
             location1="@'s3://bucket/prefix/@{schema_name}/@{table_name}'",
             location2="@IF(@gateway = 'dev', @'hdfs://@{catalog_name}/@{schema_name}/dev/@{table_name}', @'s3://prod/@{table_name}')",
+            foo="@physical_var",
         ),
-        virtual_properties={"creatable_type": "@{create_type}"},
+        virtual_properties={"creatable_type": "@{create_type}", "bar": "@virtual_var"},
         session_properties={
             "spark.executor.cores": "@IF(@gateway = 'dev', 1, 2)",
             "spark.executor.memory": "1G",
+            "baz": "@session_var",
         },
     )
     def model_with_macros(evaluator, **kwargs):
@@ -6517,12 +6600,24 @@ def model_with_macros(evaluator, **kwargs):
             "gateway": "dev",
             "key": "a",
             "create_type": "'SECURE'",
+            "merge_filter_var": True,
+            "physical_var": "bla",
+            "virtual_var": "blb",
+            "session_var": "blc",
         },
     )
 
     assert python_sql_model.name == "test_unrendered_macros_python_model_suffix"
     assert python_sql_model.python_env[c.SQLMESH_VARS] == Executable.value(
-        {"test_var_a": "test_value"}
+        {
+            "test_var_a": "test_value",
+            "gateway": "dev",
+            "create_type": "'SECURE'",
+            "merge_filter_var": True,
+            "physical_var": "bla",
+            "virtual_var": "blb",
+            "session_var": "blc",
+        }
     )
     assert python_sql_model.enabled
 
@@ -6536,25 +6631,28 @@ def model_with_macros(evaluator, **kwargs):
 
     # The properties will stay unrendered at load time
     assert python_sql_model.session_properties == {
-        "spark.executor.cores": "@IF(@gateway = 'dev', 1, 2)",
+        "spark.executor.cores": exp.maybe_parse("@IF(@gateway = 'dev', 1, 2)"),
         "spark.executor.memory": "1G",
+        "baz": exp.maybe_parse("@session_var"),
     }
-    assert python_sql_model.virtual_properties["creatable_type"] == exp.convert("@{create_type}")
+    assert python_sql_model.virtual_properties["creatable_type"] == exp.maybe_parse(
+        "@{create_type}"
+    )
 
     assert (
-        python_sql_model.physical_properties["location1"].text("this")
+        python_sql_model.physical_properties["location1"].sql()
         == "@'s3://bucket/prefix/@{schema_name}/@{table_name}'"
     )
     assert (
-        python_sql_model.physical_properties["location2"].text("this")
+        python_sql_model.physical_properties["location2"].sql()
         == "@IF(@gateway = 'dev', @'hdfs://@{catalog_name}/@{schema_name}/dev/@{table_name}', @'s3://prod/@{table_name}')"
     )
 
     # merge_filter will stay unrendered as well
     assert python_sql_model.unique_key[0] == exp.column("a", quoted=True)
     assert (
         python_sql_model.merge_filter.sql()
-        == '"source"."id" > 0 AND "target"."updated_at" < @end_ds AND "source"."updated_at" > @start_ds'
+        == '"__merge_source__"."id" > 0 AND "__merge_target__"."updated_at" < @end_ds AND "__merge_source__"."updated_at" > @start_ds AND @merge_filter_var'
     )