fix(macros): extract the audit subquery table from its FROM clause

Signed-off-by: sravankumarkunadi <sravankumarkunadi@users.noreply.github.com>

Author: sravankumarkunadi

sqlmesh/core/macros.py

@@ -1379,15 +1379,17 @@ def resolve_template(
     """
     Generates either a String literal or an exp.Table representing a physical table location, based on rendering the provided template String literal.
 
-    Note: It relies on the @this_model variable being available in the evaluation context (@this_model resolves to an exp.Table object
-    representing the current physical table).
+    Note: It relies on the @this_model variable being available in the evaluation context. @this_model usually resolves to an
+    exp.Table object representing the current physical table, but in an audit on a model with a time column it resolves to a
+    subquery that selects from that table and filters it down to the audited time range. In that case the placeholders below
+    are resolved against the physical table the subquery selects from.
     Therefore, the @resolve_template macro must be used at creation or evaluation time and not at load time.
 
     Args:
         template: Template string literal. Can contain the following placeholders:
-            @{catalog_name} -> replaced with the catalog of the exp.Table returned from @this_model
-            @{schema_name} -> replaced with the schema of the exp.Table returned from @this_model
-            @{table_name} -> replaced with the name of the exp.Table returned from @this_model
+            @{catalog_name} -> replaced with the catalog of the physical table @this_model refers to
+            @{schema_name} -> replaced with the schema of the physical table @this_model refers to
+            @{table_name} -> replaced with the name of the physical table @this_model refers to
         mode: What to return.
             'literal' -> return an exp.Literal string
             'table' -> return an exp.Table
@@ -1400,13 +1402,24 @@ def resolve_template(
         >>> evaluator.locals.update({"this_model": exp.to_table("test_catalog.sqlmesh__test.test__test_model__2517971505")})
         >>> evaluator.transform(parse_one(sql)).sql()
         "'s3://data-bucket/prod/test_catalog/sqlmesh__test/test__test_model__2517971505'"
+
+        The same template resolves to the same location when @this_model is the time-filtered
+        subquery that audits on models with a time column receive:
+
+        >>> table = exp.to_table("test_catalog.sqlmesh__test.test__test_model__2517971505")
+        >>> subquery = exp.select("*").from_(table).where(exp.column("ds").eq("2020-01-01")).subquery()
+        >>> evaluator.locals.update({"this_model": subquery})
+        >>> evaluator.transform(parse_one(sql)).sql()
+        "'s3://data-bucket/prod/test_catalog/sqlmesh__test/test__test_model__2517971505'"
     """
     if "this_model" in evaluator.locals:
         this_model_expr = evaluator.locals["this_model"]
         if isinstance(this_model_expr, exp.Subquery):
             # Audits on models with a time column render @this_model as a subquery that filters the
-            # physical table on the audited time range, so extract the table it selects from
-            this_model_expr = this_model_expr.find(exp.Table) or this_model_expr
+            # physical table on the audited time range, so resolve against the table it selects from
+            from_ = this_model_expr.unnest().args.get("from_")
+            if from_ is not None and isinstance(from_.this, exp.Table):
+                this_model_expr = from_.this
 
         this_model = exp.to_table(this_model_expr, dialect=evaluator.dialect)
         template_str: str = template.this

tests/core/test_macros.py

@@ -1160,6 +1160,26 @@ def test_resolve_template_subquery():
         == 'SELECT * FROM "test_catalog"."sqlmesh__test"."test__test_model__2517971505$partitions"'
     )
 
+    # the table is taken from the subquery's FROM clause, so other tables referenced elsewhere
+    # in it (here, in a WHERE subquery) can't be picked up instead
+    evaluator.locals.update(
+        {
+            "this_model": exp.select("*")
+            .from_(exp.to_table("test_catalog.sqlmesh__test.test__test_model__2517971505"))
+            .where(
+                exp.column("ds").isin(
+                    query=exp.select("ds").from_(exp.to_table("other_catalog.other_schema.other"))
+                )
+            )
+            .subquery()
+        }
+    )
+
+    assert (
+        evaluator.transform(parsed_sql).sql(identify=True)
+        == 'SELECT * FROM "test_catalog"."sqlmesh__test"."test__test_model__2517971505$partitions"'
+    )
+
 
 def test_macro_with_spaces():
     evaluator = MacroEvaluator()