fix: adding subquery udtf projections should be non-breaking (TobikoD…

…ata#2754)

sqlmesh/core/model/definition.py

@@ -1094,9 +1094,17 @@ def is_breaking_change(self, previous: Model) -> t.Optional[bool]:
         for edit in edits:
             if isinstance(edit, Insert):
                 expr = edit.expression
-                if _is_udtf(expr) or (
-                    not _is_projection(expr) and expr.parent not in inserted_expressions
-                ):
+                if _is_udtf(expr):
+                    # projection subqueries do not change cardinality, engines don't allow these to return
+                    # more than one row of data
+                    parent = expr.find_ancestor(exp.Subquery)
+
+                    if not parent:
+                        return None
+
+                    expr = parent
+
+                if not _is_projection(expr) and expr.parent not in inserted_expressions:
                     return None
             elif not isinstance(edit, Keep):
                 return None

tests/core/test_snapshot.py

@@ -1041,6 +1041,28 @@ def test_categorize_change_sql(make_snapshot):
         is SnapshotChangeCategory.NON_BREAKING
     )
 
+    # A UDTF subquery has been added.
+    assert (
+        categorize_change(
+            new=make_snapshot(
+                SqlModel(name="a", query=parse_one("select 1, ds, (select x from unnest(a) x)"))
+            ),
+            old=old_snapshot,
+            config=config,
+        )
+        is SnapshotChangeCategory.NON_BREAKING
+    )
+    assert (
+        categorize_change(
+            new=make_snapshot(
+                SqlModel(name="a", query=parse_one("select 1, ds, (select x from posexplode(a) x)"))
+            ),
+            old=old_snapshot,
+            config=config,
+        )
+        is SnapshotChangeCategory.NON_BREAKING
+    )
+
 
 def test_categorize_change_seed(make_snapshot, tmp_path):
     config = CategorizerConfig(seed=AutoCategorizationMode.SEMI)