Merge "track item schema names to identify name collisions w/ default schema" into main

4 files changed, 57 insertions, 2 deletions

diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 7841ce88a..74469b035 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -500,7 +500,7 @@ class CompileState:
 
     """
 
-    __slots__ = ("statement",)
+    __slots__ = ("statement", "_ambiguous_table_name_map")
 
     plugins = {}
 
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index cb10811c6..af39f0672 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1466,6 +1466,7 @@ class SQLCompiler(Compiled):
         add_to_result_map=None,
         include_table=True,
         result_map_targets=(),
+        ambiguous_table_name_map=None,
         **kwargs,
     ):
         name = orig_name = column.name
@@ -1502,6 +1503,14 @@ class SQLCompiler(Compiled):
             else:
                 schema_prefix = ""
             tablename = table.name
+
+            if (
+                not effective_schema
+                and ambiguous_table_name_map
+                and tablename in ambiguous_table_name_map
+            ):
+                tablename = ambiguous_table_name_map[tablename]
+
             if isinstance(tablename, elements._truncated_label):
                 tablename = self._truncated_identifier("alias", tablename)
 
@@ -3252,6 +3261,10 @@ class SQLCompiler(Compiled):
         compile_state = select_stmt._compile_state_factory(
             select_stmt, self, **kwargs
         )
+        kwargs[
+            "ambiguous_table_name_map"
+        ] = compile_state._ambiguous_table_name_map
+
         select_stmt = compile_state.statement
 
         toplevel = not self.stack
@@ -3732,6 +3745,7 @@ class SQLCompiler(Compiled):
         fromhints=None,
         use_schema=True,
         from_linter=None,
+        ambiguous_table_name_map=None,
         **kwargs,
     ):
         if from_linter:
@@ -3748,6 +3762,20 @@ class SQLCompiler(Compiled):
                 )
             else:
                 ret = self.preparer.quote(table.name)
+
+                if (
+                    not effective_schema
+                    and ambiguous_table_name_map
+                    and table.name in ambiguous_table_name_map
+                ):
+                    anon_name = self._truncated_identifier(
+                        "alias", ambiguous_table_name_map[table.name]
+                    )
+
+                    ret = ret + self.get_render_as_alias_suffix(
+                        self.preparer.format_alias(None, anon_name)
+                    )
+
             if fromhints and table in fromhints:
                 ret = self.format_from_hint_text(
                     ret, table, fromhints[table], iscrud
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 65f345fb3..43979b4ae 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -246,6 +246,7 @@ class ClauseElement(
     is_clause_element = True
     is_selectable = False
 
+    _is_table = False
     _is_textual = False
     _is_from_clause = False
     _is_returns_rows = False
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 00e20e3fb..e1bbcffec 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -2287,6 +2287,8 @@ class TableClause(roles.DMLTableRole, Immutable, FromClause):
 
     named_with_column = True
 
+    _is_table = True
+
     implicit_returning = False
     """:class:`_expression.TableClause`
     doesn't support having a primary key or column
@@ -3660,6 +3662,8 @@ class SelectState(util.MemoizedSlots, CompileState):
         return go
 
     def _get_froms(self, statement):
+        self._ambiguous_table_name_map = ambiguous_table_name_map = {}
+
         return self._normalize_froms(
             itertools.chain(
                 itertools.chain.from_iterable(
@@ -3677,10 +3681,16 @@ class SelectState(util.MemoizedSlots, CompileState):
                 self.from_clauses,
             ),
             check_statement=statement,
+            ambiguous_table_name_map=ambiguous_table_name_map,
         )
 
     @classmethod
-    def _normalize_froms(cls, iterable_of_froms, check_statement=None):
+    def _normalize_froms(
+        cls,
+        iterable_of_froms,
+        check_statement=None,
+        ambiguous_table_name_map=None,
+    ):
         """given an iterable of things to select FROM, reduce them to what
         would actually render in the FROM clause of a SELECT.
 
@@ -3693,6 +3703,7 @@ class SelectState(util.MemoizedSlots, CompileState):
         froms = []
 
         for item in iterable_of_froms:
+
             if item._is_subquery and item.element is check_statement:
                 raise exc.InvalidRequestError(
                     "select() construct refers to itself as a FROM"
@@ -3713,6 +3724,21 @@ class SelectState(util.MemoizedSlots, CompileState):
                 # using a list to maintain ordering
                 froms = [f for f in froms if f not in toremove]
 
+            if ambiguous_table_name_map is not None:
+                ambiguous_table_name_map.update(
+                    (
+                        fr.name,
+                        _anonymous_label.safe_construct(
+                            hash(fr.name), fr.name
+                        ),
+                    )
+                    for item in froms
+                    for fr in item._from_objects
+                    if fr._is_table
+                    and fr.schema
+                    and fr.name not in ambiguous_table_name_map
+                )
+
         return froms
 
     def _get_display_froms(