fix(optimizer): qualify columns with circular dependency (#6873)

* fix(optimizer): qualify columns with circular dependency

* fix style

* add clear cache for column index

Author: geooo109

sqlglot/optimizer/qualify_columns.py

@@ -610,6 +610,13 @@ def _qualify_columns(
             # column_table can be a '' because bigquery unnest has no table alias
             column_table = resolver.get_table(column)
 
+            if (
+                column_table
+                and isinstance(source := scope.sources.get(column_table.name), Scope)
+                and id(column) in source.column_index
+            ):
+                continue
+
             if column_table:
                 column.set("table", column_table)
             elif (

sqlglot/optimizer/scope.py

@@ -103,6 +103,7 @@ def clear_cache(self):
         self._pivots = None
         self._references = None
         self._semi_anti_join_tables = None
+        self._column_index = None
 
     def branch(
         self, expression, scope_type, sources=None, cte_sources=None, lateral_sources=None, **kwargs
@@ -131,6 +132,7 @@ def _collect(self):
         self._stars = []
         self._join_hints = []
         self._semi_anti_join_tables = set()
+        self._column_index = set()
 
         for node in self.walk(bfs=False):
             if node is self.expression:
@@ -139,6 +141,8 @@ def _collect(self):
             if isinstance(node, exp.Dot) and node.is_star:
                 self._stars.append(node)
             elif isinstance(node, exp.Column) and not isinstance(node, exp.Pseudocolumn):
+                self._column_index.add(id(node))
+
                 if isinstance(node.this, exp.Star):
                     self._stars.append(node)
                 else:
@@ -259,6 +263,14 @@ def stars(self) -> t.List[exp.Column | exp.Dot]:
         self._ensure_collected()
         return self._stars
 
+    @property
+    def column_index(self) -> t.Set[int]:
+        """
+        Set of column object IDs that belong to this scope's expression.
+        """
+        self._ensure_collected()
+        return self._column_index
+
     @property
     def columns(self):
         """

tests/test_optimizer.py

@@ -574,6 +574,30 @@ def test_qualify_columns(self, logger):
         )
         self.check_file("qualify_columns_ddl", qualify_columns, schema=self.schema)
 
+        self.assertEqual(
+            optimizer.qualify.qualify(
+                parse_one(
+                    """
+                    SELECT
+                    (
+                        SELECT
+                        col_st.value
+                        FROM UNNEST(col_st) AS col_st
+                    ) AS vcol1
+                    FROM t AS b
+                    """,
+                    read="bigquery",
+                ),
+                schema={
+                    "t": {
+                        "col_st": "ARRAY<STRUCT<key STRING, value INT>>",
+                    }
+                },
+                dialect="bigquery",
+            ).sql(dialect="bigquery"),
+            "SELECT (SELECT `col_st`.`value` AS `value` FROM UNNEST(`b`.`col_st`) AS `col_st`) AS `vcol1` FROM `t` AS `b`",
+        )
+
     def test_validate_columns(self):
         with self.assertRaisesRegex(
             OptimizeError, "Column 'foo' could not be resolved. Line: 1, Col: 10"