fix: skip limit on recursive ctes (#2278)

Author: tobymao

sqlmesh/core/model/definition.py

@@ -425,12 +425,17 @@ def ctas_query(self, **render_kwarg: t.Any) -> exp.Query:
         query = self.render_query_or_raise(**render_kwarg).copy()
 
         for select_or_union in query.find_all(exp.Select, exp.Union):
+            cte = select_or_union.find_ancestor(exp.With, exp.Select, exp.Subquery)
+            skip_limit = isinstance(cte, exp.With) and cte.recursive
+
             if isinstance(select_or_union, exp.Select) and select_or_union.args.get("from"):
                 select_or_union.where(exp.false(), copy=False)
-                if not isinstance(select_or_union.parent, exp.Union):
+                if not skip_limit and not isinstance(select_or_union.parent, exp.Union):
                     select_or_union.limit(0, copy=False)
-            elif isinstance(select_or_union, exp.Union) and not isinstance(
-                select_or_union.parent, exp.Union
+            elif (
+                not skip_limit
+                and isinstance(select_or_union, exp.Union)
+                and not isinstance(select_or_union.parent, exp.Union)
             ):
                 select_or_union.set("limit", exp.Limit(expression=exp.Literal.number(0)))
 

tests/core/test_model.py

@@ -1811,6 +1811,24 @@ def test_model_ctas_query():
         == 'SELECT 1 AS "a" FROM "t" AS "t" WHERE FALSE UNION ALL SELECT 2 AS "a" FROM "t" AS "t" WHERE FALSE ORDER BY 1 LIMIT 0'
     )
 
+    expressions = d.parse(
+        """
+        MODEL (name `a-b-c.table`, kind FULL, dialect bigquery);
+        WITH RECURSIVE a AS (
+            SELECT * FROM x
+        ), b AS (
+            SELECT * FROM a UNION ALL SELECT * FROM a
+        )
+        SELECT * FROM b
+
+    """
+    )
+
+    assert (
+        load_sql_based_model(expressions, dialect="bigquery").ctas_query().sql()
+        == 'WITH RECURSIVE "a" AS (SELECT * FROM "x" AS "x" WHERE FALSE), "b" AS (SELECT * FROM "a" AS "a" WHERE FALSE UNION ALL SELECT * FROM "a" AS "a" WHERE FALSE) SELECT * FROM "b" AS "b" WHERE FALSE LIMIT 0'
+    )
+
 
 def test_is_breaking_change():
     model = create_external_model("a", columns={"a": "int", "limit": "int"})