Run SelectState from obj normalize ahead of calcing ORM joins

Fixed regression where the full combination of joined inheritance, global
with_polymorphic, self-referential relationship and joined loading would
fail to be able to produce a query with the scope of lazy loads and object
refresh operations that also attempted to render the joined loader.

Fixes: #6495
Change-Id: If74a744c237069e3a89617498096c18b9b6e5dde

3 files changed, 49 insertions, 23 deletions

diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index ea84805b4..baad28835 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -591,9 +591,15 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         self.create_eager_joins = []
         self._fallback_from_clauses = []
 
-        self.from_clauses = [
+        # normalize the FROM clauses early by themselves, as this makes
+        # it an easier job when we need to assemble a JOIN onto these,
+        # for select.join() as well as joinedload().   As of 1.4 there are now
+        # potentially more complex sets of FROM objects here as the use
+        # of lambda statements for lazyload, load_on_pk etc. uses more
+        # cloning of the select() construct.  See #6495
+        self.from_clauses = self._normalize_froms(
             info.selectable for info in select_statement._from_obj
-        ]
+        )
 
         # this is a fairly arbitrary break into a second method,
         # so it might be nicer to break up create_for_statement()
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 2c37ecaa0..dca45730c 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -2226,6 +2226,7 @@ class JoinedLoader(AbstractRelationshipLoader):
             and not should_nest_selectable
             and compile_state.from_clauses
         ):
+
             indexes = sql_util.find_left_clause_that_matches_given(
                 compile_state.from_clauses, query_entity.selectable
             )
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 7007bb430..997c3588e 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -4209,38 +4209,57 @@ class SelectState(util.MemoizedSlots, CompileState):
         return go
 
     def _get_froms(self, statement):
+        return self._normalize_froms(
+            itertools.chain(
+                itertools.chain.from_iterable(
+                    [
+                        element._from_objects
+                        for element in statement._raw_columns
+                    ]
+                ),
+                itertools.chain.from_iterable(
+                    [
+                        element._from_objects
+                        for element in statement._where_criteria
+                    ]
+                ),
+                self.from_clauses,
+            ),
+            check_statement=statement,
+        )
+
+    def _normalize_froms(self, iterable_of_froms, check_statement=None):
+        """given an iterable of things to select FROM, reduce them to what
+        would actually render in the FROM clause of a SELECT.
+
+        This does the job of checking for JOINs, tables, etc. that are in fact
+        overlapping due to cloning, adaption, present in overlapping joins,
+        etc.
+
+        """
         seen = set()
         froms = []
 
-        for item in itertools.chain(
-            itertools.chain.from_iterable(
-                [element._from_objects for element in statement._raw_columns]
-            ),
-            itertools.chain.from_iterable(
-                [
-                    element._from_objects
-                    for element in statement._where_criteria
-                ]
-            ),
-            self.from_clauses,
-        ):
-            if item._is_subquery and item.element is statement:
+        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"
                 )
+
             if not seen.intersection(item._cloned_set):
                 froms.append(item)
                 seen.update(item._cloned_set)
 
-        toremove = set(
-            itertools.chain.from_iterable(
-                [_expand_cloned(f._hide_froms) for f in froms]
+        if froms:
+            toremove = set(
+                itertools.chain.from_iterable(
+                    [_expand_cloned(f._hide_froms) for f in froms]
+                )
             )
-        )
-        if toremove:
-            # filter out to FROM clauses not in the list,
-            # using a list to maintain ordering
-            froms = [f for f in froms if f not in toremove]
+            if toremove:
+                # filter out to FROM clauses not in the list,
+                # using a list to maintain ordering
+                froms = [f for f in froms if f not in toremove]
 
         return froms