Look up adapter info for previous left side in chained query.join()

Fixed issue where chaining multiple join elements inside of
:meth:`.Query.join` might not correctly adapt to the previous left-hand
side, when chaining joined inheritance classes that share the same base
class.

Change-Id: I4b846430b7362912dbebf50599ec15a1eb978fd4
Fixes: #3505

3 files changed, 120 insertions, 0 deletions

diff --git a/doc/build/changelog/unreleased_12/3505.rst b/doc/build/changelog/unreleased_12/3505.rst
new file mode 100644
index 000000000..b2990596c
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/3505.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 3505
+
+    Fixed issue where chaining multiple join elements inside of
+    :meth:`.Query.join` might not correctly adapt to the previous left-hand
+    side, when chaining joined inheritance classes that share the same base
+    class.
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 74d3079af..98747c680 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -2244,6 +2244,13 @@ class Query(object):
 
                 left_entity = onclause._parententity
 
+                alias = self._polymorphic_adapters.get(left_entity, None)
+                # could be None or could be ColumnAdapter also
+                if isinstance(alias, ORMAdapter) and \
+                        alias.mapper.isa(left_entity):
+                    left_entity = alias.aliased_class
+                    onclause = getattr(left_entity, onclause.key)
+
                 prop = onclause.property
                 if not isinstance(onclause, attributes.QueryableAttribute):
                     onclause = prop
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 5b24b0c8d..6af8ce725 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -2677,3 +2677,108 @@ class SelfReferentialM2MTest(fixtures.MappedTest):
         eq_(sess.query(Node).select_from(join(Node, n1, 'children'))
             .filter(n1.data.in_(['n3', 'n7'])).order_by(Node.id).all(),
             [Node(data='n1'), Node(data='n2')])
+
+
+class AliasFromCorrectLeftTest(
+        fixtures.DeclarativeMappedTest, AssertsCompiledSQL):
+    run_create_tables = None
+    __dialect__ = 'default'
+
+    @classmethod
+    def setup_classes(cls):
+        Base = cls.DeclarativeBasic
+
+        class Object(Base):
+            __tablename__ = 'object'
+
+            type = Column(String(30))
+            __mapper_args__ = {
+                'polymorphic_identity': 'object',
+                'polymorphic_on': type
+            }
+
+            id = Column(Integer, primary_key=True)
+            name = Column(String(256))
+
+        class A(Object):
+            __tablename__ = 'a'
+
+            __mapper_args__ = {'polymorphic_identity': 'a'}
+
+            id = Column(Integer, ForeignKey('object.id'), primary_key=True)
+
+            b_list = relationship(
+                'B',
+                secondary='a_b_association',
+                backref='a_list'
+            )
+
+        class B(Object):
+            __tablename__ = 'b'
+
+            __mapper_args__ = {'polymorphic_identity': 'b'}
+
+            id = Column(Integer, ForeignKey('object.id'), primary_key=True)
+
+        class ABAssociation(Base):
+            __tablename__ = 'a_b_association'
+
+            a_id = Column(Integer, ForeignKey('a.id'), primary_key=True)
+            b_id = Column(Integer, ForeignKey('b.id'), primary_key=True)
+
+        class X(Base):
+            __tablename__ = 'x'
+
+            id = Column(Integer, primary_key=True)
+            name = Column(String(30))
+
+            obj_id = Column(Integer, ForeignKey('object.id'))
+            obj = relationship('Object', backref='x_list')
+
+    def test_join_prop_to_string(self):
+        A, B, X = self.classes("A", "B", "X")
+
+        s = Session()
+
+        q = s.query(B).\
+            join(B.a_list, 'x_list').filter(X.name == 'x1')
+
+        self.assert_compile(
+            q,
+            "SELECT object.type AS object_type, b.id AS b_id, "
+            "object.id AS object_id, object.name AS object_name "
+            "FROM object JOIN b ON object.id = b.id "
+            "JOIN a_b_association AS a_b_association_1 "
+            "ON b.id = a_b_association_1.b_id "
+            "JOIN ("
+            "object AS object_1 "
+            "JOIN a AS a_1 ON object_1.id = a_1.id"
+            ") ON a_1.id = a_b_association_1.a_id "
+            "JOIN x ON object_1.id = x.obj_id WHERE x.name = :name_1"
+        )
+
+    def test_join_prop_to_prop(self):
+        A, B, X = self.classes("A", "B", "X")
+
+        s = Session()
+
+        # B -> A, but both are Object.  So when we say A.x_list, make sure
+        # we pick the correct right side
+        q = s.query(B).\
+            join(B.a_list, A.x_list).filter(X.name == 'x1')
+
+        self.assert_compile(
+            q,
+            "SELECT object.type AS object_type, b.id AS b_id, "
+            "object.id AS object_id, object.name AS object_name "
+            "FROM object JOIN b ON object.id = b.id "
+            "JOIN a_b_association AS a_b_association_1 "
+            "ON b.id = a_b_association_1.b_id "
+            "JOIN ("
+            "object AS object_1 "
+            "JOIN a AS a_1 ON object_1.id = a_1.id"
+            ") ON a_1.id = a_b_association_1.a_id "
+            "JOIN x ON object_1.id = x.obj_id WHERE x.name = :name_1"
+        )
+
+