Rework _apply_joins(), _prep_for_joins() totally

The approach here is still error prone
and hard to follow.  Reorganize the whole
thing to take a pretty blunt approach to
the structure of to_join().  Also fix some never-called
code (!) in _prep_for_joins() and ensure we re-use
an aliased object.

Fixes: #3774
Change-Id: Ie6319415ae7213b4a33eac2ab70803ad2880d340

3 files changed, 209 insertions, 45 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index 1e2c87fdf..a443fa197 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -20,12 +20,13 @@
 
     .. change::
         :tags: bug, orm
-        :tickets: 3773
+        :tickets: 3773, 3774
         :versions: 1.1.0
 
         Fixed bug in subquery eager loading where a subqueryload
         of an "of_type()" object linked to a second subqueryload of a plain
-        mapped class would fail to link the joins correctly.
+        mapped class, or a longer chain of several "of_type()" attributes,
+        would fail to link the joins correctly.
 
     .. change::
         :tags: bug, sql
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 1d0058ca2..0b22b8486 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -915,28 +915,21 @@ class SubqueryLoader(AbstractRelationshipLoader):
 
         # determine the immediate parent class we are joining from,
         # which needs to be aliased.
-        if len(to_join) > 1:
-            info = inspect(to_join[-1][0])
 
         if len(to_join) < 2:
             # in the case of a one level eager load, this is the
             # leftmost "left_alias".
             parent_alias = left_alias
-        elif info.mapper.isa(self.parent):
-            # In the case of multiple levels, retrieve
-            # it from subq_path[-2]. This is the same as self.parent
-            # in the vast majority of cases, and [ticket:2014]
-            # illustrates a case where sub_path[-2] is a subclass
-            # of self.parent
-            parent_alias = orm_util.AliasedClass(
-                to_join[-1][0],
-                use_mapper_path=True)
         else:
-            # if of_type() were used leading to this relationship,
-            # self.parent is more specific than subq_path[-2]
-            parent_alias = orm_util.AliasedClass(
-                self.parent,
-                use_mapper_path=True)
+            info = inspect(to_join[-1][0])
+            if info.is_aliased_class:
+                parent_alias = info.entity
+            else:
+                # alias a plain mapper as we may be
+                # joining multiple times
+                parent_alias = orm_util.AliasedClass(
+                    info.entity,
+                    use_mapper_path=True)
 
         local_cols = self.parent_property.local_columns
 
@@ -949,35 +942,46 @@ class SubqueryLoader(AbstractRelationshipLoader):
     def _apply_joins(
             self, q, to_join, left_alias, parent_alias,
             effective_entity):
-        for i, (mapper, key) in enumerate(to_join):
-
-            # we need to use query.join() as opposed to
-            # orm.join() here because of the
-            # rich behavior it brings when dealing with
-            # "with_polymorphic" mappers.  "aliased"
-            # and "from_joinpoint" take care of most of
-            # the chaining and aliasing for us.
-
-            first = i == 0
-            middle = i < len(to_join) - 1
-            second_to_last = i == len(to_join) - 2
-            last = i == len(to_join) - 1
-
-            if first:
-                attr = getattr(left_alias, key)
-                if last and effective_entity is not self.mapper:
-                    attr = attr.of_type(effective_entity)
-            else:
-                if last:
-                    attr = getattr(parent_alias, key).\
-                        of_type(effective_entity)
+
+        ltj = len(to_join)
+        if ltj == 1:
+            to_join = [
+                getattr(left_alias, to_join[0][1]).of_type(effective_entity)
+            ]
+        elif ltj == 2:
+            to_join = [
+                getattr(left_alias, to_join[0][1]).of_type(parent_alias),
+                getattr(parent_alias, to_join[-1][1]).of_type(effective_entity)
+            ]
+        elif ltj > 2:
+            middle = [
+                (
+                    orm_util.AliasedClass(item[0])
+                    if not inspect(item[0]).is_aliased_class
+                    else item[0].entity,
+                    item[1]
+                ) for item in to_join[1:-1]
+            ]
+            inner = []
+
+            while middle:
+                item = middle.pop(0)
+                attr = getattr(item[0], item[1])
+                if middle:
+                    attr = attr.of_type(middle[0][0])
                 else:
-                    attr = getattr(mapper.entity, key)
+                    attr = attr.of_type(parent_alias)
 
-            if second_to_last:
-                q = q.join(parent_alias, attr, from_joinpoint=True)
-            else:
-                q = q.join(attr, aliased=middle, from_joinpoint=True)
+                inner.append(attr)
+
+            to_join = [
+                getattr(left_alias, to_join[0][1]).of_type(inner[0].parent)
+            ] + inner + [
+                getattr(parent_alias, to_join[-1][1]).of_type(effective_entity)
+            ]
+
+        for attr in to_join:
+            q = q.join(attr, from_joinpoint=True)
         return q
 
     def _setup_options(self, q, subq_path, orig_query, effective_entity):
diff --git a/test/orm/test_of_type.py b/test/orm/test_of_type.py
index 44ea347e1..0b6ef1649 100644
--- a/test/orm/test_of_type.py
+++ b/test/orm/test_of_type.py
@@ -13,6 +13,7 @@ from .inheritance._poly_fixtures import Company, Person, Engineer, Manager, Boss
     Machine, Paperwork, _PolymorphicFixtureBase, _Polymorphic,\
     _PolymorphicPolymorphic, _PolymorphicUnions, _PolymorphicJoins,\
     _PolymorphicAliasedJoins
+from sqlalchemy.testing.assertsql import AllOf, CompiledSQL
 
 
 class _PolymorphicTestBase(object):
@@ -719,3 +720,161 @@ class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeM
                 "AS anon_1 ON data_container.id = anon_1.job_container_id"
         )
 
+
+class SubclassRelationshipTest2(
+        testing.AssertsCompiledSQL, fixtures.DeclarativeMappedTest):
+
+    run_setup_classes = 'once'
+    run_setup_mappers = 'once'
+    run_inserts = 'once'
+    run_deletes = None
+    __dialect__ = 'default'
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class A(Base):
+            __tablename__ = 't_a'
+
+            id = Column(Integer, primary_key=True,
+                        test_needs_autoincrement=True)
+
+        class B(Base):
+            __tablename__ = 't_b'
+
+            type = Column(String(2))
+            __mapper_args__ = {
+                'polymorphic_identity': 'b',
+                'polymorphic_on': type
+            }
+
+            id = Column(Integer, primary_key=True,
+                        test_needs_autoincrement=True)
+
+            # Relationship to A
+            a_id = Column(Integer, ForeignKey('t_a.id'))
+            a = relationship('A', backref='bs')
+
+        class B2(B):
+            __tablename__ = 't_b2'
+
+            __mapper_args__ = {
+                'polymorphic_identity': 'b2',
+            }
+
+            id = Column(Integer, ForeignKey('t_b.id'), primary_key=True)
+
+        class C(Base):
+            __tablename__ = 't_c'
+
+            type = Column(String(2))
+            __mapper_args__ = {
+                'polymorphic_identity': 'c',
+                'polymorphic_on': type
+            }
+
+            id = Column(Integer, primary_key=True,
+                        test_needs_autoincrement=True)
+
+            # Relationship to B
+            b_id = Column(Integer, ForeignKey('t_b.id'))
+            b = relationship('B', backref='cs')
+
+        class C2(C):
+            __tablename__ = 't_c2'
+
+            __mapper_args__ = {
+                'polymorphic_identity': 'c2',
+            }
+
+            id = Column(Integer, ForeignKey('t_c.id'), primary_key=True)
+
+        class D(Base):
+            __tablename__ = 't_d'
+
+            id = Column(Integer, primary_key=True,
+                        test_needs_autoincrement=True)
+
+            # Relationship to B
+            c_id = Column(Integer, ForeignKey('t_c.id'))
+            c = relationship('C', backref='ds')
+
+    @classmethod
+    def insert_data(cls):
+        s = Session(testing.db)
+
+        s.add_all(cls._fixture())
+        s.commit()
+
+    @classmethod
+    def _fixture(cls):
+        A, B, B2, C, C2, D = cls.classes('A', 'B', 'B2', 'C', 'C2', 'D')
+
+        return [
+            A(bs=[B2(cs=[C2(ds=[D()])])]),
+            A(bs=[B2(cs=[C2(ds=[D()])])]),
+        ]
+
+    def test_all_subq_query(self):
+        A, B, B2, C, C2, D = self.classes('A', 'B', 'B2', 'C', 'C2', 'D')
+
+        session = Session(testing.db)
+
+        b_b2 = with_polymorphic(B, [B2], flat=True)
+        c_c2 = with_polymorphic(C, [C2], flat=True)
+
+        q = session.query(
+            A
+        ).options(
+            subqueryload(
+                A.bs.of_type(b_b2)
+            ).subqueryload(
+                b_b2.cs.of_type(c_c2)
+            ).subqueryload(
+                c_c2.ds
+            )
+        )
+
+        self.assert_sql_execution(
+            testing.db,
+            q.all,
+            CompiledSQL(
+                "SELECT t_a.id AS t_a_id FROM t_a",
+                {}
+            ),
+            CompiledSQL(
+                "SELECT t_b_1.type AS t_b_1_type, t_b_1.id AS t_b_1_id, "
+                "t_b_1.a_id AS t_b_1_a_id, t_b2_1.id AS t_b2_1_id, "
+                "anon_1.t_a_id AS anon_1_t_a_id FROM "
+                "(SELECT t_a.id AS t_a_id FROM t_a) AS anon_1 "
+                "JOIN (t_b AS t_b_1 LEFT OUTER JOIN t_b2 AS t_b2_1 "
+                "ON t_b_1.id = t_b2_1.id) ON anon_1.t_a_id = t_b_1.a_id "
+                "ORDER BY anon_1.t_a_id",
+                {}
+            ),
+            CompiledSQL(
+                "SELECT t_c_1.type AS t_c_1_type, t_c_1.id AS t_c_1_id, "
+                "t_c_1.b_id AS t_c_1_b_id, t_c2_1.id AS t_c2_1_id, "
+                "t_b_1.id AS t_b_1_id FROM (SELECT t_a.id AS t_a_id FROM t_a) "
+                "AS anon_1 JOIN (t_b AS t_b_1 LEFT OUTER JOIN t_b2 AS t_b2_1 "
+                "ON t_b_1.id = t_b2_1.id) ON anon_1.t_a_id = t_b_1.a_id "
+                "JOIN (t_c AS t_c_1 LEFT OUTER JOIN t_c2 AS t_c2_1 ON "
+                "t_c_1.id = t_c2_1.id) ON t_b_1.id = t_c_1.b_id "
+                "ORDER BY t_b_1.id",
+                {}
+            ),
+            CompiledSQL(
+                "SELECT t_d.id AS t_d_id, t_d.c_id AS t_d_c_id, "
+                "t_c_1.id AS t_c_1_id "
+                "FROM (SELECT t_a.id AS t_a_id FROM t_a) AS anon_1 "
+                "JOIN (t_b AS t_b_1 LEFT OUTER JOIN t_b2 AS t_b2_1 "
+                "ON t_b_1.id = t_b2_1.id) "
+                "ON anon_1.t_a_id = t_b_1.a_id "
+                "JOIN (t_c AS t_c_1 LEFT OUTER JOIN t_c2 AS t_c2_1 "
+                "ON t_c_1.id = t_c2_1.id) "
+                "ON t_b_1.id = t_c_1.b_id "
+                "JOIN t_d ON t_c_1.id = t_d.c_id ORDER BY t_c_1.id",
+                {}
+            )
+        )