fix(bigquery): star referencing outer-scope from correlated subquery

sqlglot/optimizer/qualify_columns.py

@@ -615,15 +615,31 @@ def _convert_columns_to_dots(scope: Scope, resolver: Resolver) -> None:
             )
         ):
             root, *parts = column.parts
-
-            if isinstance(root, exp.Identifier) and root.name in scope.selected_sources:
-                # The struct is already qualified, but we still need to change the AST
-                column_table = root
-                root, *parts = parts
-                was_qualified = True
-            else:
-                column_table = resolver.get_table(root.name)
-                was_qualified = False
+            was_qualified = False
+
+            # Unlike columns, correlated stars can't be deferred to the outer scopes, since they
+            # must be expanded in this one, so they're resolved against those scopes as well
+            resolvers: t.Iterable[Resolver] = (
+                itertools.chain((resolver,), resolver.outer_resolvers())
+                if isinstance(column.this, exp.Star)
+                else (resolver,)
+            )
+            for source_resolver in resolvers:
+                selected_sources = source_resolver.scope.selected_sources
+                if column.table in selected_sources:
+                    # The star is over a table, so it's expanded as is
+                    column_table = None
+                    break
+                if isinstance(root, exp.Identifier) and root.name in selected_sources:
+                    # The struct is already qualified, but we still need to change the AST
+                    column_table = root
+                    root, *parts = parts
+                    was_qualified = True
+                    break
+
+                column_table = source_resolver.get_table(root.name)
+                if column_table:
+                    break
 
             if column_table:
                 converted = True
@@ -1011,7 +1027,12 @@ def _expand_stars(
                     if annotated_ahead:
                         annotator.uncache(expression)
 
-                    new_selections.extend(struct_fields)
+                    star = expression.expression
+                    excluded = {e.name for e in star.args.get("except_") or []}
+                    replaced = {e.alias: e for e in star.args.get("replace") or []}
+                    new_selections.extend(
+                        replaced.get(f.alias) or f for f in struct_fields if f.alias not in excluded
+                    )
                     continue
 
         if not tables:
@@ -1020,6 +1041,7 @@ def _expand_stars(
 
         for table in tables:
             source = scope.sources.get(table)
+            source_resolver = resolver
             pivots: list[exp.Pivot] | None = None
             source_table = table
 
@@ -1038,10 +1060,18 @@ def _expand_stars(
                     source_table = parent.alias_or_name
                     source = scope.sources.get(source_table)
 
+                if source is None:
+                    # Correlated stars, e.g. (SELECT AS STRUCT x.* EXCEPT (a)), expand an outer source
+                    for outer_resolver in resolver.outer_resolvers():
+                        source = outer_resolver.scope.sources.get(table)
+                        if source:
+                            source_resolver = outer_resolver
+                            break
+
                 if source is None:
                     raise OptimizeError(f"Unknown table: {table}")
 
-            columns = resolver.get_source_columns(source_table, only_visible=True)
+            columns = source_resolver.get_source_columns(source_table, only_visible=True)
             columns = columns or scope.outer_columns
 
             if pseudocolumns and dialect.EXCLUDES_PSEUDOCOLUMNS_FROM_STAR:
@@ -1140,6 +1170,8 @@ def _expand_stars(
             annotator.uncache(scope_expression, deep=False)
 
         scope_expression.set("expressions", new_selections)
+        # The parent scope sees this scope's columns, so they must reflect the expansions
+        scope.clear_cache()
 
 
 def _output_identifier(selection: exp.Expr | None) -> exp.Identifier | None:

sqlglot/optimizer/resolver.py

@@ -84,6 +84,13 @@ def get_table(self, column: str | exp.Column) -> exp.Identifier | None:
 
         return exp.to_identifier(table_name)
 
+    def outer_resolvers(self) -> t.Iterator[Resolver]:
+        """Resolvers for the outer scopes a correlated subquery can reference, innermost first."""
+        scope = self.scope
+        while scope.can_be_correlated and scope.parent:
+            scope = scope.parent
+            yield Resolver(scope, self.schema, self._infer_schema)
+
     @property
     def all_columns(self) -> set[str]:
         """All available columns of all sources in this scope"""