Repository: tobymao/sqlglot
Pull request: Fix!(optimizer): preserve parenthesized joins, tables
Fixes #1898

Problem statement:
Optimizer transforms nested joins incorrectly
Hey @GeorgeSittas, thanks a lot for the fix on the nested joins. That will work on my part, as i only need to be able to parse and use the scope class. However i did a check on the optimizer as well and the SQL seems to be transformed into SQL that is functionally different from the original.

As you can see in the output the query is beeing transformed into regular joins and the table b and c arent nested anymore.

The output:
```
SELECT
  "a"."id" AS "id",
  "b"."test" AS "test",
  "c"."test2" AS "test2"
FROM "a" AS "a"
LEFT JOIN "b" AS "b"
  ON "b"."id" = "a"."id"
JOIN "c" AS "c"
  ON "c"."id" = "b"."id"
```

The code i used:
```
import sqlglot
from sqlglot import optimizer
expr = sqlglot.parse_one("""
SELECT
 a.id
,b.test
,c.test2
FROM a
LEFT JOIN (
    b
    INNER JOIN c
        ON c.id = b.id
)
    ON b.id = a.id
""", read="tsql")

schema={
    "testcat": {
        "testdb": {
            "a": {
                "id": "INT",
            },
            "b": {
                "id": "INT",
                "test": "STRING",
            },
            "c": {
                "id": "INT",
                "test2": "STRING",
            },
        }
    }
}

expr = optimizer.optimize(
    expression=expr,
    schema=schema,
)

postgres = sqlglot.Dialect.get_or_raise("postgres")()
print(postgres.generate(expr, pretty=True))
```

Hints from issue discussion:
Hey @10bas10, happy to know your issue was resolved.

> However i did a check on the optimizer as well and the SQL seems to be transformed into SQL that is functionally different from the original.

So after discussing this, we agreed on flattening the `JOIN` sequence in the optimizer because preserving the parentheses wasn't a trivial thing to do. This shouldn't actually be "functionally different" (as in "different semantics"), because AFAIK specifying an explicit `JOIN` order is simply an optimization and the parentheses don't change the returned row set. This is behaviour is now documented in the docstring of `qualify_tables`.

```
    Note:
        This rule effectively enforces a left-to-right join order, since all joins
        are unnested. This means that the optimizer doesn't necessarily preserve the
        original join order, e.g. when parentheses are used to specify it explicitly.
```

Note that this is consistent with SQLGlot's optimizer in general: it doesn't make any guarantees about the JOIN order. Even without my recent PR, you could get different `JOIN` orders for some optimized queries.
I'm gonna go ahead to close this one, but happy to continue the discussion either here or in Slack.
Hey @GeorgeSittas,

It actually is functionally different. Let me demonstrate a case where the results in differerent output, it is actually the same case as i provided in this bug.

In the following query, where nested joins are applied, the inner join is applied in a different context level. The inner join is applied to the table b.

As you can see this query gives all records that are in table a even with the joins not supplying any matches:
![image](https://github.com/tobymao/sqlglot/assets/81879308/3aca1be2-1f12-4b67-9d96-6ce5736fcfc9)

```
;WITH a AS (
	SELECT 1 AS id
),
b AS (
	SELECT 2 AS id, 'b' as test
),
c AS (
	SELECT 2 AS id, 'c' as test2
)
SELECT
 a.id
,b.test
,c.test2
FROM a
LEFT JOIN (
    b
    INNER JOIN c
        ON c.id = b.id
)
    ON b.id = a.id
```

In the query down below, actually the query beeing outputted by the optimizer, the join is indeed flattened. However this gives a different end result as the inner join is actually performed in a different context. This query doesnt give any output as you can see:
![image](https://github.com/tobymao/sqlglot/assets/81879308/b484ad69-819a-4d99-bf8d-5f98cd87b6a6)


```
;WITH a AS (
	SELECT 1 AS id
),
b AS (
	SELECT 2 AS id, 'b' as test
),
c AS (
	SELECT 2 AS id, 'c' as test2
)
SELECT
 a.id
,b.test
,c.test2
FROM a
LEFT JOIN b
	ON b.id = a.id
INNER JOIN c
	ON c.id = b.id
```
I see, thanks for the explanation -- it seems like this is indeed incorrect then. I'll try to resolve this.

Commit message: Remove selects, named_selects from Scope

Patch to review:
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index 242e66ce65..1fb8765486 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -878,11 +878,11 @@ def alias_column_names(self) -> t.List[str]:
         return [c.name for c in table_alias.args.get("columns") or []]
 
     @property
-    def selects(self):
+    def selects(self) -> t.List[Expression]:
         return self.this.selects if isinstance(self.this, Subqueryable) else []
 
     @property
-    def named_selects(self):
+    def named_selects(self) -> t.List[str]:
         return [select.output_name for select in self.selects]
 
 
@@ -959,7 +959,7 @@ def except_(
 
 class UDTF(DerivedTable, Unionable):
     @property
-    def selects(self):
+    def selects(self) -> t.List[Expression]:
         alias = self.args.get("alias")
         return alias.columns if alias else []
 
@@ -2194,11 +2194,11 @@ def ctes(self):
         return with_.expressions
 
     @property
-    def selects(self):
+    def selects(self) -> t.List[Expression]:
         raise NotImplementedError("Subqueryable objects must implement `selects`")
 
     @property
-    def named_selects(self):
+    def named_selects(self) -> t.List[str]:
         raise NotImplementedError("Subqueryable objects must implement `named_selects`")
 
     def with_(
@@ -2282,7 +2282,6 @@ class Table(Expression):
         "pivots": False,
         "hints": False,
         "system_time": False,
-        "wrapped": False,
     }
 
     @property
@@ -2299,6 +2298,14 @@ def db(self) -> str:
     def catalog(self) -> str:
         return self.text("catalog")
 
+    @property
+    def selects(self) -> t.List[Expression]:
+        return []
+
+    @property
+    def named_selects(self) -> t.List[str]:
+        return []
+
     @property
     def parts(self) -> t.List[Identifier]:
         """Return the parts of a table in order catalog, db, table."""
@@ -2390,7 +2397,7 @@ def select(
         return this
 
     @property
-    def named_selects(self):
+    def named_selects(self) -> t.List[str]:
         return self.this.unnest().named_selects
 
     @property
@@ -2398,7 +2405,7 @@ def is_star(self) -> bool:
         return self.this.is_star or self.expression.is_star
 
     @property
-    def selects(self):
+    def selects(self) -> t.List[Expression]:
         return self.this.unnest().selects
 
     @property
diff --git a/sqlglot/generator.py b/sqlglot/generator.py
index 4ac988f371..ed65c94b51 100644
--- a/sqlglot/generator.py
+++ b/sqlglot/generator.py
@@ -1215,8 +1215,7 @@ def table_sql(self, expression: exp.Table, sep: str = " AS ") -> str:
         system_time = expression.args.get("system_time")
         system_time = f" {self.sql(expression, 'system_time')}" if system_time else ""
 
-        sql = f"{table}{system_time}{alias}{hints}{pivots}{joins}{laterals}"
-        return f"({sql})" if expression.args.get("wrapped") else sql
+        return f"{table}{system_time}{alias}{hints}{pivots}{joins}{laterals}"
 
     def tablesample_sql(
         self, expression: exp.TableSample, seed_prefix: str = "SEED", sep=" AS "
diff --git a/sqlglot/lineage.py b/sqlglot/lineage.py
index 04a807322a..9f5ae9a757 100644
--- a/sqlglot/lineage.py
+++ b/sqlglot/lineage.py
@@ -104,7 +104,7 @@ def to_node(
         # Find the specific select clause that is the source of the column we want.
         # This can either be a specific, named select or a generic `*` clause.
         select = next(
-            (select for select in scope.selects if select.alias_or_name == column_name),
+            (select for select in scope.expression.selects if select.alias_or_name == column_name),
             exp.Star() if scope.expression.is_star else None,
         )
 
diff --git a/sqlglot/optimizer/eliminate_joins.py b/sqlglot/optimizer/eliminate_joins.py
index cd8ba3b3c0..3134e65986 100644
--- a/sqlglot/optimizer/eliminate_joins.py
+++ b/sqlglot/optimizer/eliminate_joins.py
@@ -85,7 +85,7 @@ def _unique_outputs(scope):
         grouped_outputs = set()
 
         unique_outputs = set()
-        for select in scope.selects:
+        for select in scope.expression.selects:
             output = select.unalias()
             if output in grouped_expressions:
                 grouped_outputs.add(output)
@@ -105,7 +105,7 @@ def _unique_outputs(scope):
 
 def _has_single_output_row(scope):
     return isinstance(scope.expression, exp.Select) and (
-        all(isinstance(e.unalias(), exp.AggFunc) for e in scope.selects)
+        all(isinstance(e.unalias(), exp.AggFunc) for e in scope.expression.selects)
         or _is_limit_1(scope)
         or not scope.expression.args.get("from")
     )
diff --git a/sqlglot/optimizer/eliminate_subqueries.py b/sqlglot/optimizer/eliminate_subqueries.py
index 84f50e9e25..5ae1fa0474 100644
--- a/sqlglot/optimizer/eliminate_subqueries.py
+++ b/sqlglot/optimizer/eliminate_subqueries.py
@@ -113,7 +113,7 @@ def _eliminate_union(scope, existing_ctes, taken):
     taken[alias] = scope
 
     # Try to maintain the selections
-    expressions = scope.selects
+    expressions = scope.expression.selects
     selects = [
         exp.alias_(exp.column(e.alias_or_name, table=alias), alias=e.alias_or_name, copy=False)
         for e in expressions
diff --git a/sqlglot/optimizer/isolate_table_selects.py b/sqlglot/optimizer/isolate_table_selects.py
index 79e3ed55bc..a6524b8f98 100644
--- a/sqlglot/optimizer/isolate_table_selects.py
+++ b/sqlglot/optimizer/isolate_table_selects.py
@@ -12,7 +12,12 @@ def isolate_table_selects(expression, schema=None):
             continue
 
         for _, source in scope.selected_sources.values():
-            if not isinstance(source, exp.Table) or not schema.column_names(source):
+            if (
+                not isinstance(source, exp.Table)
+                or not schema.column_names(source)
+                or isinstance(source.parent, exp.Subquery)
+                or isinstance(source.parent.parent, exp.Table)
+            ):
                 continue
 
             if not source.alias:
diff --git a/sqlglot/optimizer/merge_subqueries.py b/sqlglot/optimizer/merge_subqueries.py
index e156d5e4fc..6609c77ff1 100644
--- a/sqlglot/optimizer/merge_subqueries.py
+++ b/sqlglot/optimizer/merge_subqueries.py
@@ -97,8 +97,11 @@ def merge_derived_tables(expression, leave_tables_isolated=False):
         for subquery in outer_scope.derived_tables:
             from_or_join = subquery.find_ancestor(exp.From, exp.Join)
             alias = subquery.alias_or_name
-            inner_scope = outer_scope.sources[alias]
-            if _mergeable(outer_scope, inner_scope, leave_tables_isolated, from_or_join):
+            inner_scope = outer_scope.sources.get(alias)
+
+            if inner_scope and _mergeable(
+                outer_scope, inner_scope, leave_tables_isolated, from_or_join
+            ):
                 _rename_inner_sources(outer_scope, inner_scope, alias)
                 _merge_from(outer_scope, inner_scope, subquery, alias)
                 _merge_expressions(outer_scope, inner_scope, alias)
@@ -107,6 +110,7 @@ def merge_derived_tables(expression, leave_tables_isolated=False):
                 _merge_order(outer_scope, inner_scope)
                 _merge_hints(outer_scope, inner_scope)
                 outer_scope.clear_cache()
+
     return expression
 
 
@@ -166,7 +170,7 @@ def _outer_select_joins_on_inner_select_join():
         if not inner_from:
             return False
         inner_from_table = inner_from.alias_or_name
-        inner_projections = {s.alias_or_name: s for s in inner_scope.selects}
+        inner_projections = {s.alias_or_name: s for s in inner_scope.expression.selects}
         return any(
             col.table != inner_from_table
             for selection in selections
diff --git a/sqlglot/optimizer/optimize_joins.py b/sqlglot/optimizer/optimize_joins.py
index d51276f021..7b3b2b1ee1 100644
--- a/sqlglot/optimizer/optimize_joins.py
+++ b/sqlglot/optimizer/optimize_joins.py
@@ -59,7 +59,7 @@ def reorder_joins(expression):
         dag = {name: other_table_names(join) for name, join in joins.items()}
         parent.set(
             "joins",
-            [joins[name] for name in tsort(dag) if name != from_.alias_or_name],
+            [joins[name] for name in tsort(dag) if name != from_.alias_or_name and name in joins],
         )
     return expression
 
diff --git a/sqlglot/optimizer/pushdown_predicates.py b/sqlglot/optimizer/pushdown_predicates.py
index fb1662ddca..58b988dd39 100644
--- a/sqlglot/optimizer/pushdown_predicates.py
+++ b/sqlglot/optimizer/pushdown_predicates.py
@@ -42,7 +42,10 @@ def pushdown_predicates(expression):
             # so we limit the selected sources to only itself
             for join in select.args.get("joins") or []:
                 name = join.alias_or_name
-                pushdown(join.args.get("on"), {name: scope.selected_sources[name]}, scope_ref_count)
+                if name in scope.selected_sources:
+                    pushdown(
+                        join.args.get("on"), {name: scope.selected_sources[name]}, scope_ref_count
+                    )
 
     return expression
 
diff --git a/sqlglot/optimizer/pushdown_projections.py b/sqlglot/optimizer/pushdown_projections.py
index be3ddb25a8..97e8ff682f 100644
--- a/sqlglot/optimizer/pushdown_projections.py
+++ b/sqlglot/optimizer/pushdown_projections.py
@@ -48,12 +48,12 @@ def pushdown_projections(expression, schema=None, remove_unused_selections=True)
             left, right = scope.union_scopes
             referenced_columns[left] = parent_selections
 
-            if any(select.is_star for select in right.selects):
+            if any(select.is_star for select in right.expression.selects):
                 referenced_columns[right] = parent_selections
-            elif not any(select.is_star for select in left.selects):
+            elif not any(select.is_star for select in left.expression.selects):
                 referenced_columns[right] = [
-                    right.selects[i].alias_or_name
-                    for i, select in enumerate(left.selects)
+                    right.expression.selects[i].alias_or_name
+                    for i, select in enumerate(left.expression.selects)
                     if SELECT_ALL in parent_selections or select.alias_or_name in parent_selections
                 ]
 
@@ -90,7 +90,7 @@ def _remove_unused_selections(scope, parent_selections, schema):
     removed = False
     star = False
 
-    for selection in scope.selects:
+    for selection in scope.expression.selects:
         name = selection.alias_or_name
 
         if SELECT_ALL in parent_selections or name in parent_selections or name in order_refs:
diff --git a/sqlglot/optimizer/qualify_columns.py b/sqlglot/optimizer/qualify_columns.py
index 435585cfdd..83c91bdd1a 100644
--- a/sqlglot/optimizer/qualify_columns.py
+++ b/sqlglot/optimizer/qualify_columns.py
@@ -198,7 +198,7 @@ def replace_columns(
                 else:
                     column.replace(alias_expr.copy())
 
-    for i, projection in enumerate(scope.selects):
+    for i, projection in enumerate(scope.expression.selects):
         replace_columns(projection)
 
         if isinstance(projection, exp.Alias):
@@ -239,7 +239,7 @@ def _expand_order_by(scope: Scope, resolver: Resolver):
         ordered.set("this", new_expression)
 
     if scope.expression.args.get("group"):
-        selects = {s.this: exp.column(s.alias_or_name) for s in scope.selects}
+        selects = {s.this: exp.column(s.alias_or_name) for s in scope.expression.selects}
 
         for ordered in ordereds:
             ordered = ordered.this
@@ -270,7 +270,7 @@ def _expand_positional_references(scope: Scope, expressions: t.Iterable[E]) -> t
 
 def _select_by_pos(scope: Scope, node: exp.Literal) -> exp.Alias:
     try:
-        return scope.selects[int(node.this) - 1].assert_is(exp.Alias)
+        return scope.expression.selects[int(node.this) - 1].assert_is(exp.Alias)
     except IndexError:
         raise OptimizeError(f"Unknown output column: {node.name}")
 
@@ -347,7 +347,7 @@ def _expand_stars(
         if not pivot_output_columns:
             pivot_output_columns = [col.alias_or_name for col in pivot.expressions]
 
-    for expression in scope.selects:
+    for expression in scope.expression.selects:
         if isinstance(expression, exp.Star):
             tables = list(scope.selected_sources)
             _add_except_columns(expression, tables, except_columns)
@@ -446,7 +446,7 @@ def _qualify_outputs(scope: Scope):
     new_selections = []
 
     for i, (selection, aliased_column) in enumerate(
-        itertools.zip_longest(scope.selects, scope.outer_column_list)
+        itertools.zip_longest(scope.expression.selects, scope.outer_column_list)
     ):
         if isinstance(selection, exp.Subquery):
             if not selection.output_name:
diff --git a/sqlglot/optimizer/qualify_tables.py b/sqlglot/optimizer/qualify_tables.py
index af8c716458..7533780451 100644
--- a/sqlglot/optimizer/qualify_tables.py
+++ b/sqlglot/optimizer/qualify_tables.py
@@ -15,26 +15,13 @@ def qualify_tables(
     schema: t.Optional[Schema] = None,
 ) -> E:
     """
-    Rewrite sqlglot AST to have fully qualified, unnested tables.
+    Rewrite sqlglot AST to have fully qualified tables.
 
     Examples:
         >>> import sqlglot
         >>> expression = sqlglot.parse_one("SELECT 1 FROM tbl")
         >>> qualify_tables(expression, db="db").sql()
         'SELECT 1 FROM db.tbl AS tbl'
-        >>>
-        >>> expression = sqlglot.parse_one("SELECT * FROM (tbl)")
-        >>> qualify_tables(expression).sql()
-        'SELECT * FROM tbl AS tbl'
-        >>>
-        >>> expression = sqlglot.parse_one("SELECT * FROM (tbl1 JOIN tbl2 ON id1 = id2)")
-        >>> qualify_tables(expression).sql()
-        'SELECT * FROM tbl1 AS tbl1 JOIN tbl2 AS tbl2 ON id1 = id2'
-
-    Note:
-        This rule effectively enforces a left-to-right join order, since all joins
-        are unnested. This means that the optimizer doesn't necessarily preserve the
-        original join order, e.g. when parentheses are used to specify it explicitly.
 
     Args:
         expression: Expression to qualify
@@ -49,6 +36,15 @@ def qualify_tables(
 
     for scope in traverse_scope(expression):
         for derived_table in itertools.chain(scope.ctes, scope.derived_tables):
+            # Expand join construct, for example
+            # SELECT * FROM (t1 JOIN t2) AS t -> SELECT * FROM (SELECT * FROM t1 JOIN t2) AS t
+            if isinstance(derived_table, exp.Subquery):
+                unnested = derived_table.unnest()
+                if isinstance(unnested, exp.Table):
+                    joins = unnested.args.pop("joins", None)
+                    derived_table.this.replace(exp.select("*").from_(unnested.copy(), copy=False))
+                    derived_table.this.set("joins", joins)
+
             if not derived_table.args.get("alias"):
                 alias_ = next_alias_name()
                 derived_table.set("alias", exp.TableAlias(this=exp.to_identifier(alias_)))
@@ -66,19 +62,9 @@ def qualify_tables(
                     if not source.args.get("catalog"):
                         source.set("catalog", exp.to_identifier(catalog))
 
-                # Unnest joins attached in tables by appending them to the closest query
-                for join in source.args.get("joins") or []:
-                    scope.expression.append("joins", join)
-
-                source.set("joins", None)
-                source.set("wrapped", None)
-
                 if not source.alias:
-                    source = source.replace(
-                        alias(
-                            source, name or source.name or next_alias_name(), copy=True, table=True
-                        )
-                    )
+                    # Mutates source by attaching an alias to it
+                    alias(source, name or source.name or next_alias_name(), copy=False, table=True)
 
                 pivots = source.args.get("pivots")
                 if pivots and not pivots[0].alias:
diff --git a/sqlglot/optimizer/scope.py b/sqlglot/optimizer/scope.py
index 7dcfb37e88..b2b4230ab3 100644
--- a/sqlglot/optimizer/scope.py
+++ b/sqlglot/optimizer/scope.py
@@ -122,7 +122,11 @@ def _collect(self):
                 self._udtfs.append(node)
             elif isinstance(node, exp.CTE):
                 self._ctes.append(node)
-            elif isinstance(node, exp.Subquery) and isinstance(parent, (exp.From, exp.Join)):
+            elif (
+                isinstance(node, exp.Subquery)
+                and isinstance(parent, (exp.From, exp.Join))
+                and _is_subquery_scope(node)
+            ):
                 self._derived_tables.append(node)
             elif isinstance(node, exp.Subqueryable):
                 self._subqueries.append(node)
@@ -274,6 +278,7 @@ def columns(self):
                     not ancestor
                     or column.table
                     or isinstance(ancestor, exp.Select)
+                    or (isinstance(ancestor, exp.Table) and not isinstance(ancestor.this, exp.Func))
                     or (
                         isinstance(ancestor, exp.Order)
                         and (
@@ -340,23 +345,6 @@ def cte_sources(self):
             if isinstance(scope, Scope) and scope.is_cte
         }
 
-    @property
-    def selects(self):
-        """
-        Select expressions of this scope.
-
-        For example, for the following expression:
-            SELECT 1 as a, 2 as b FROM x
-
-        The outputs are the "1 as a" and "2 as b" expressions.
-
-        Returns:
-            list[exp.Expression]: expressions
-        """
-        if isinstance(self.expression, exp.Union):
-            return self.expression.unnest().selects
-        return self.expression.selects
-
     @property
     def external_columns(self):
         """
@@ -548,6 +536,8 @@ def _traverse_scope(scope):
         yield from _traverse_union(scope)
     elif isinstance(scope.expression, exp.Subquery):
         yield from _traverse_subqueries(scope)
+    elif isinstance(scope.expression, exp.Table):
+        yield from _traverse_tables(scope)
     elif isinstance(scope.expression, exp.UDTF):
         pass
     else:
@@ -620,6 +610,15 @@ def _traverse_ctes(scope):
     scope.sources.update(sources)
 
 
+def _is_subquery_scope(expression: exp.Subquery) -> bool:
+    """
+    We represent (tbl1 JOIN tbl2) as a Subquery, but it's not really a new scope.
+    If an alias is present, it shadows all names under the Subquery, so that's an
+    exception to this rule.
+    """
+    return bool(not isinstance(expression.unnest(), exp.Table) or expression.alias)
+
+
 def _traverse_tables(scope):
     sources = {}
 
@@ -629,9 +628,8 @@ def _traverse_tables(scope):
     if from_:
         expressions.append(from_.this)
 
-    for expression in (scope.expression, *scope.find_all(exp.Table)):
-        for join in expression.args.get("joins") or []:
-            expressions.append(join.this)
+    for join in scope.expression.args.get("joins") or []:
+        expressions.append(join.this)
 
     if isinstance(scope.expression, exp.Table):
         expressions.append(scope.expression)
@@ -655,6 +653,8 @@ def _traverse_tables(scope):
                 sources[find_new_name(sources, table_name)] = expression
             else:
                 sources[source_name] = expression
+
+            expressions.extend(join.this for join in expression.args.get("joins") or [])
             continue
 
         if not isinstance(expression, exp.DerivedTable):
@@ -664,10 +664,15 @@ def _traverse_tables(scope):
             lateral_sources = sources
             scope_type = ScopeType.UDTF
             scopes = scope.udtf_scopes
-        else:
+        elif _is_subquery_scope(expression):
             lateral_sources = None
             scope_type = ScopeType.DERIVED_TABLE
             scopes = scope.derived_table_scopes
+        else:
+            # Makes sure we check for possible sources in nested table constructs
+            expressions.append(expression.this)
+            expressions.extend(join.this for join in expression.args.get("joins") or [])
+            continue
 
         for child_scope in _traverse_scope(
             scope.branch(
@@ -728,7 +733,11 @@ def walk_in_scope(expression, bfs=True):
             continue
         if (
             isinstance(node, exp.CTE)
-            or (isinstance(node, exp.Subquery) and isinstance(parent, (exp.From, exp.Join)))
+            or (
+                isinstance(node, exp.Subquery)
+                and isinstance(parent, (exp.From, exp.Join))
+                and _is_subquery_scope(node)
+            )
             or isinstance(node, exp.UDTF)
             or isinstance(node, exp.Subqueryable)
         ):
diff --git a/sqlglot/parser.py b/sqlglot/parser.py
index c7f4050858..b0d2acda42 100644
--- a/sqlglot/parser.py
+++ b/sqlglot/parser.py
@@ -1969,31 +1969,9 @@ def _parse_select(
 
             self._match_r_paren()
 
-            alias = None
-
-            # Ensure "wrapped" tables are not parsed as Subqueries. The exception to this is when there's
-            # an alias that can be applied to the parentheses, because that would shadow all wrapped table
-            # names, and so we want to parse it as a Subquery to represent the inner scope appropriately.
-            # Additionally, we want the node under the Subquery to be an actual query, so we will replace
-            # the table reference with a star query that selects from it.
-            if isinstance(this, exp.Table):
-                alias = self._parse_table_alias()
-                if not alias:
-                    this.set("wrapped", True)
-                    return this
-
-                this.set("wrapped", None)
-                joins = this.args.pop("joins", None)
-                this = this.replace(exp.select("*").from_(this.copy(), copy=False))
-                this.set("joins", joins)
-
-            subquery = self._parse_subquery(this, parse_alias=parse_subquery_alias and not alias)
-            if subquery and alias:
-                subquery.set("alias", alias)
-
             # We return early here so that the UNION isn't attached to the subquery by the
             # following call to _parse_set_operations, but instead becomes the parent node
-            return subquery
+            return self._parse_subquery(this, parse_alias=parse_subquery_alias)
         elif self._match(TokenType.VALUES):
             this = self.expression(
                 exp.Values,
diff --git a/tests/fixtures/identity.sql b/tests/fixtures/identity.sql
index 63631c4dd0..093536e2a5 100644
--- a/tests/fixtures/identity.sql
+++ b/tests/fixtures/identity.sql
@@ -423,6 +423,7 @@ SELECT 1 INTERSECT SELECT 2
 SELECT 1 INTERSECT SELECT 2
 SELECT 1 AS delete, 2 AS alter
 SELECT * FROM (x)
+SELECT * FROM ((x))
 SELECT * FROM ((SELECT 1))
 SELECT * FROM (x CROSS JOIN foo LATERAL VIEW EXPLODE(y))
 SELECT * FROM (SELECT 1) AS x
@@ -724,6 +725,11 @@ SELECT * FROM project.dataset.INFORMATION_SCHEMA.TABLES
 SELECT * FROM (table1 AS t1 LEFT JOIN table2 AS t2 ON 1 = 1)
 SELECT * FROM (tbl1 LEFT JOIN tbl2 ON 1 = 1)
 SELECT * FROM (tbl1, tbl2 JOIN tbl3 ON TRUE)
+SELECT * FROM (tbl)
+SELECT * FROM (((tbl)))
+SELECT * FROM (tbl1 CROSS JOIN tbl2)
+SELECT * FROM (tbl1 CROSS JOIN tbl2) AS t
+SELECT * FROM (tbl AS tbl) AS t
 SELECT * FROM (tbl1 JOIN (tbl2 CROSS JOIN tbl3) ON bla = foo)
 SELECT * FROM (tbl1, LATERAL (SELECT * FROM bla) AS tbl)
 SELECT CAST(x AS INT) /* comment */ FROM foo
diff --git a/tests/fixtures/optimizer/optimizer.sql b/tests/fixtures/optimizer/optimizer.sql
index 38e64d7d17..2a922d7beb 100644
--- a/tests/fixtures/optimizer/optimizer.sql
+++ b/tests/fixtures/optimizer/optimizer.sql
@@ -701,3 +701,90 @@ SELECT
     "x"."a" * "x"."b" - "x"."b"
   ) AS "f"
 FROM "x" AS "x";
+
+# title: wrapped table without alias
+# execute: false
+SELECT * FROM (tbl);
+SELECT
+  *
+FROM (
+  "tbl" AS "tbl"
+);
+
+# title: wrapped table with alias
+# execute: false
+SELECT * FROM (tbl AS tbl);
+SELECT
+  *
+FROM (
+  "tbl" AS "tbl"
+);
+
+# title: wrapped join of tables without alias
+SELECT a, c FROM (x LEFT JOIN y ON a = c);
+SELECT
+  "x"."a" AS "a",
+  "y"."c" AS "c"
+FROM (
+  "x" AS "x"
+    LEFT JOIN "y" AS "y"
+      ON "x"."a" = "y"."c"
+);
+
+# title: wrapped join of tables with alias
+# execute: false
+SELECT a, c FROM (x LEFT JOIN y ON a = c) AS t;
+SELECT
+  "x"."a" AS "a",
+  "y"."c" AS "c"
+FROM "x" AS "x"
+LEFT JOIN "y" AS "y"
+  ON "x"."a" = "y"."c";
+
+# title: chained wrapped joins without aliases
+# execute: false
+SELECT * FROM ((a CROSS JOIN ((b CROSS JOIN c) CROSS JOIN (d CROSS JOIN e))));
+SELECT
+  *
+FROM (
+  (
+    "a" AS "a"
+      CROSS JOIN (
+        (
+          "b" AS "b"
+            CROSS JOIN "c" AS "c"
+        )
+        CROSS JOIN (
+          "d" AS "d"
+            CROSS JOIN "e" AS "e"
+        )
+      )
+  )
+);
+
+# title: chained wrapped joins with aliases
+# execute: false
+SELECT * FROM ((a AS foo CROSS JOIN b AS bar) CROSS JOIN c AS baz);
+SELECT
+  *
+FROM (
+  (
+    "a" AS "foo"
+      CROSS JOIN "b" AS "bar"
+  )
+  CROSS JOIN "c" AS "baz"
+);
+
+# title: table joined with joined construct
+SELECT x.a, y.b, z.c FROM x LEFT JOIN (y INNER JOIN z ON y.c = z.c) ON x.b = y.b;
+SELECT
+  "x"."a" AS "a",
+  "y"."b" AS "b",
+  "z"."c" AS "c"
+FROM "x" AS "x"
+LEFT JOIN (
+  "y" AS "y"
+    JOIN "z" AS "z"
+      ON "y"."c" = "z"."c"
+)
+  ON "x"."b" = "y"."b";
diff --git a/tests/fixtures/optimizer/qualify_tables.sql b/tests/fixtures/optimizer/qualify_tables.sql
index d8ce2b0e63..cd749a0a35 100644
--- a/tests/fixtures/optimizer/qualify_tables.sql
+++ b/tests/fixtures/optimizer/qualify_tables.sql
@@ -26,45 +26,49 @@ SELECT (SELECT y.c FROM c.db.y AS y) FROM c.db.x AS x;
 SELECT * FROM x PIVOT (SUM(a) FOR b IN ('a', 'b'));
 SELECT * FROM c.db.x AS x PIVOT(SUM(a) FOR b IN ('a', 'b')) AS _q_0;
 
------------------------------------------------------------
---- Unnest wrapped tables / joins, expand join constructs
------------------------------------------------------------
-
 # title: wrapped table without alias
 SELECT * FROM (tbl);
-SELECT * FROM c.db.tbl AS tbl;
+SELECT * FROM (c.db.tbl AS tbl);
 
 # title: wrapped table with alias
 SELECT * FROM (tbl AS tbl);
-SELECT * FROM c.db.tbl AS tbl;
+SELECT * FROM (c.db.tbl AS tbl);
 
-# title: wrapped table with alias and multiple redundant parentheses
+# title: wrapped table with alias using multiple (redundant) parentheses
 SELECT * FROM ((((tbl AS tbl))));
-SELECT * FROM c.db.tbl AS tbl;
+SELECT * FROM ((((c.db.tbl AS tbl))));
+
+# title: wrapped join of tables without alias
+SELECT * FROM (t1 CROSS JOIN t2);
+SELECT * FROM (c.db.t1 AS t1 CROSS JOIN c.db.t2 AS t2);
+
+# title: wrapped join of tables with alias, expansion of join construct
+SELECT * FROM (t1 CROSS JOIN t2) AS t;
+SELECT * FROM (SELECT * FROM c.db.t1 AS t1 CROSS JOIN c.db.t2 AS t2) AS t;
 
 # title: chained wrapped joins without aliases (1)
 SELECT * FROM ((a CROSS JOIN b) CROSS JOIN c);
-SELECT * FROM c.db.a AS a CROSS JOIN c.db.b AS b CROSS JOIN c.db.c AS c;
+SELECT * FROM ((c.db.a AS a CROSS JOIN c.db.b AS b) CROSS JOIN c.db.c AS c);
 
 # title: chained wrapped joins without aliases (2)
 SELECT * FROM (a CROSS JOIN (b CROSS JOIN c));
-SELECT * FROM c.db.a AS a CROSS JOIN c.db.b AS b CROSS JOIN c.db.c AS c;
+SELECT * FROM (c.db.a AS a CROSS JOIN (c.db.b AS b CROSS JOIN c.db.c AS c));
 
 # title: chained wrapped joins without aliases (3)
 SELECT * FROM ((a CROSS JOIN ((b CROSS JOIN c) CROSS JOIN d)));
-SELECT * FROM c.db.a AS a CROSS JOIN c.db.b AS b CROSS JOIN c.db.c AS c CROSS JOIN c.db.d AS d;
+SELECT * FROM ((c.db.a AS a CROSS JOIN ((c.db.b AS b CROSS JOIN c.db.c AS c) CROSS JOIN c.db.d AS d)));
 
 # title: chained wrapped joins without aliases (4)
 SELECT * FROM ((a CROSS JOIN ((b CROSS JOIN c) CROSS JOIN (d CROSS JOIN e))));
-SELECT * FROM c.db.a AS a CROSS JOIN c.db.b AS b CROSS JOIN c.db.c AS c CROSS JOIN c.db.d AS d CROSS JOIN c.db.e AS e;
+SELECT * FROM ((c.db.a AS a CROSS JOIN ((c.db.b AS b CROSS JOIN c.db.c AS c) CROSS JOIN (c.db.d AS d CROSS JOIN c.db.e AS e))));
 
 # title: chained wrapped joins with aliases
 SELECT * FROM ((a AS foo CROSS JOIN b AS bar) CROSS JOIN c AS baz);
-SELECT * FROM c.db.a AS foo CROSS JOIN c.db.b AS bar CROSS JOIN c.db.c AS baz;
+SELECT * FROM ((c.db.a AS foo CROSS JOIN c.db.b AS bar) CROSS JOIN c.db.c AS baz);
 
 # title: wrapped join with subquery without alias
 SELECT * FROM (tbl1 CROSS JOIN (SELECT * FROM tbl2) AS t1);
-SELECT * FROM c.db.tbl1 AS tbl1 CROSS JOIN (SELECT * FROM c.db.tbl2 AS tbl2) AS t1;
+SELECT * FROM (c.db.tbl1 AS tbl1 CROSS JOIN (SELECT * FROM c.db.tbl2 AS tbl2) AS t1);
 
 # title: wrapped join with subquery with alias, parentheses can't be omitted because of alias
 SELECT * FROM (tbl1 CROSS JOIN (SELECT * FROM tbl2) AS t1) AS t2;
@@ -72,32 +76,28 @@ SELECT * FROM (SELECT * FROM c.db.tbl1 AS tbl1 CROSS JOIN (SELECT * FROM c.db.tb
 
 # title: join construct as the right operand of a left join
 SELECT * FROM a LEFT JOIN (b INNER JOIN c ON c.id = b.id) ON b.id = a.id;
-SELECT * FROM c.db.a AS a LEFT JOIN c.db.b AS b ON b.id = a.id INNER JOIN c.db.c AS c ON c.id = b.id;
+SELECT * FROM c.db.a AS a LEFT JOIN (c.db.b AS b INNER JOIN c.db.c AS c ON c.id = b.id) ON b.id = a.id;
 
-# title: nested joins converted to canonical form
+# title: nested joins
 SELECT * FROM a LEFT JOIN b INNER JOIN c ON c.id = b.id ON b.id = a.id;
-SELECT * FROM c.db.a AS a LEFT JOIN c.db.b AS b ON b.id = a.id INNER JOIN c.db.c AS c ON c.id = b.id;
+SELECT * FROM c.db.a AS a LEFT JOIN c.db.b AS b INNER JOIN c.db.c AS c ON c.id = b.id ON b.id = a.id;
 
 # title: parentheses can't be omitted because alias shadows inner table names
 SELECT t.a FROM (tbl AS tbl) AS t;
 SELECT t.a FROM (SELECT * FROM c.db.tbl AS tbl) AS t;
 
-# title: outermost set of parentheses can't be omitted due to shadowing (1)
-SELECT * FROM ((tbl AS tbl)) AS _q_0;
-SELECT * FROM (SELECT * FROM c.db.tbl AS tbl) AS _q_0;
-
-# title: outermost set of parentheses can't be omitted due to shadowing (2)
+# title: wrapped aliased table with outer alias
 SELECT * FROM ((((tbl AS tbl)))) AS _q_0;
 SELECT * FROM (SELECT * FROM c.db.tbl AS tbl) AS _q_0;
 
-# title: join construct with three tables in canonical form
+# title: join construct with three tables
 SELECT * FROM (tbl1 AS tbl1 JOIN tbl2 AS tbl2 ON id1 = id2 JOIN tbl3 AS tbl3 ON id1 = id3) AS _q_0;
 SELECT * FROM (SELECT * FROM c.db.tbl1 AS tbl1 JOIN c.db.tbl2 AS tbl2 ON id1 = id2 JOIN c.db.tbl3 AS tbl3 ON id1 = id3) AS _q_0;
 
-# title: join construct with three tables in canonical form and redundant set of parentheses
+# title: join construct with three tables and redundant set of parentheses
 SELECT * FROM ((tbl1 AS tbl1 JOIN tbl2 AS tbl2 ON id1 = id2 JOIN tbl3 AS tbl3 ON id1 = id3)) AS _q_0;
 SELECT * FROM (SELECT * FROM c.db.tbl1 AS tbl1 JOIN c.db.tbl2 AS tbl2 ON id1 = id2 JOIN c.db.tbl3 AS tbl3 ON id1 = id3) AS _q_0;
 
-# title: nested join construct in canonical form
+# title: join construct within join construct
 SELECT * FROM (tbl1 AS tbl1 JOIN (tbl2 AS tbl2 JOIN tbl3 AS tbl3 ON id2 = id3) AS _q_0 ON id1 = id3) AS _q_1;
 SELECT * FROM (SELECT * FROM c.db.tbl1 AS tbl1 JOIN (SELECT * FROM c.db.tbl2 AS tbl2 JOIN c.db.tbl3 AS tbl3 ON id2 = id3) AS _q_0 ON id1 = id3) AS _q_1;
diff --git a/tests/test_parser.py b/tests/test_parser.py
index 891dcef3c6..2fa6a09604 100644
--- a/tests/test_parser.py
+++ b/tests/test_parser.py
@@ -580,58 +580,3 @@ def test_parse_properties(self):
 
     def test_parse_floats(self):
         self.assertTrue(parse_one("1. ").is_number)
-
-    def test_parse_wrapped_tables(self):
-        expr = parse_one("select * from (table)")
-        self.assertIsInstance(expr.args["from"].this, exp.Table)
-        self.assertTrue(expr.args["from"].this.args["wrapped"])
-
-        expr = parse_one("select * from (((table)))")
-        self.assertIsInstance(expr.args["from"].this, exp.Table)
-        self.assertTrue(expr.args["from"].this.args["wrapped"])
-
-        self.assertEqual(expr.sql(), "SELECT * FROM (table)")
-
-        expr = parse_one("select * from (tbl1 join tbl2)")
-        self.assertIsInstance(expr.args["from"].this, exp.Table)
-        self.assertTrue(expr.args["from"].this.args["wrapped"])
-        self.assertEqual(len(expr.args["from"].this.args["joins"]), 1)
-
-        expr = parse_one("select * from (tbl1 join tbl2) t")
-        self.assertIsInstance(expr.args["from"].this, exp.Subquery)
-        self.assertIsInstance(expr.args["from"].this.this, exp.Select)
-        self.assertEqual(expr.sql(), "SELECT * FROM (SELECT * FROM tbl1, tbl2) AS t")
-
-        expr = parse_one("select * from (tbl as tbl) t")
-        self.assertEqual(expr.sql(), "SELECT * FROM (SELECT * FROM tbl AS tbl) AS t")
-
-        expr = parse_one("select * from ((a cross join b) cross join c)")
-        self.assertIsInstance(expr.args["from"].this, exp.Table)
-        self.assertTrue(expr.args["from"].this.args["wrapped"])
-        self.assertEqual(len(expr.args["from"].this.args["joins"]), 2)
-        self.assertEqual(expr.sql(), "SELECT * FROM (a CROSS JOIN b CROSS JOIN c)")
-
-        expr = parse_one("select * from ((a cross join b) cross join c) t")
-        self.assertIsInstance(expr.args["from"].this, exp.Subquery)
-        self.assertEqual(len(expr.args["from"].this.this.args["joins"]), 2)
-        self.assertEqual(
-            expr.sql(), "SELECT * FROM (SELECT * FROM a CROSS JOIN b CROSS JOIN c) AS t"
-        )
-
-        expr = parse_one("select * from (a cross join (b cross join c))")
-        self.assertIsInstance(expr.args["from"].this, exp.Table)
-        self.assertTrue(expr.args["from"].this.args["wrapped"])
-        self.assertEqual(len(expr.args["from"].this.args["joins"]), 1)
-        self.assertIsInstance(expr.args["from"].this.args["joins"][0].this, exp.Table)
-        self.assertTrue(expr.args["from"].this.args["joins"][0].this.args["wrapped"])
-        self.assertEqual(expr.sql(), "SELECT * FROM (a CROSS JOIN (b CROSS JOIN c))")
-
-        expr = parse_one("select * from ((a cross join ((b cross join c) cross join d)))")
-        self.assertEqual(expr.sql(), "SELECT * FROM (a CROSS JOIN (b CROSS JOIN c CROSS JOIN d))")
-
-        expr = parse_one(
-            "select * from ((a cross join ((b cross join c) cross join (d cross join e))))"
-        )
-        self.assertEqual(
-            expr.sql(), "SELECT * FROM (a CROSS JOIN (b CROSS JOIN c CROSS JOIN (d CROSS JOIN e)))"
-        )


Code review comments:
diff_hunk: @@ -97,8 +97,11 @@ def merge_derived_tables(expression, leave_tables_isolated=False):
         for subquery in outer_scope.derived_tables:
             from_or_join = subquery.find_ancestor(exp.From, exp.Join)
             alias = subquery.alias_or_name
-            inner_scope = outer_scope.sources[alias]
-            if _mergeable(outer_scope, inner_scope, leave_tables_isolated, from_or_join):
+            inner_scope = outer_scope.sources.get(alias)
+
+            if inner_scope and _mergeable(
+                outer_scope, inner_scope, leave_tables_isolated, from_or_join
+            ):
path: sqlglot/optimizer/merge_subqueries.py
text: @author:
The invariant "every source has an alias" does not hold anymore, since you can have queries like:

```sql
SELECT * FROM (t1 JOIN t2)
```

There's a single scope here, since there's no alias attached to the nested source `(t1 JOIN t2)`.

@author:
Update on this: these changes were not needed after all because I "accidentally" solved the problem by not treating `(t1 join t2)` as a derived table, and so we _should_ always have an alias here theoretically.