Fix(optimizer): constrain UDTF scope boundary (#3226)

* Fix(optimizer): constrain UDTF scope boundary

* Replace end-to-end test with a minimal projection pushdown test

sqlglot/optimizer/scope.py

@@ -800,10 +800,9 @@ def walk_in_scope(expression, bfs=True, prune=None):
         if (
             isinstance(node, exp.CTE)
             or (
-                _is_derived_table(node)
-                and isinstance(node.parent, (exp.From, exp.Join, exp.Subquery))
+                isinstance(node.parent, (exp.From, exp.Join, exp.Subquery))
+                and (_is_derived_table(node) or isinstance(node, exp.UDTF))
             )
-            or isinstance(node, exp.UDTF)
             or isinstance(node, exp.UNWRAPPED_QUERIES)
         ):
             crossed_scope_boundary = True

tests/fixtures/optimizer/pushdown_projections.sql

@@ -79,6 +79,9 @@ WITH y AS (SELECT MAX(1) AS _ FROM x AS x) SELECT 1 AS "1" FROM y AS y;
 WITH y AS (SELECT a FROM x GROUP BY a) SELECT 1 FROM y;
 WITH y AS (SELECT 1 AS _ FROM x AS x GROUP BY x.a) SELECT 1 AS "1" FROM y AS y;
 
+WITH cte AS (SELECT col FROM t) SELECT IF(1 IN UNNEST(col), 1, 0) AS col FROM cte;
+WITH cte AS (SELECT t.col AS col FROM t AS t) SELECT CASE WHEN 1 IN (SELECT UNNEST(cte.col)) THEN 1 ELSE 0 END AS col FROM cte AS cte;
+
 --------------------------------------
 -- Unknown Star Expansion
 --------------------------------------

tests/test_optimizer.py

@@ -505,6 +505,9 @@ def test_file_schema(self):
         )
 
     def test_scope(self):
+        ast = parse_one("SELECT IF(a IN UNNEST(b), 1, 0) AS c FROM t", dialect="bigquery")
+        self.assertEqual(build_scope(ast).columns, [exp.column("a"), exp.column("b")])
+
         many_unions = parse_one(" UNION ALL ".join(["SELECT x FROM t"] * 10000))
         scopes_using_traverse = list(build_scope(many_unions).traverse())
         scopes_using_traverse_scope = traverse_scope(many_unions)