Fixed bug in polymorphic SQL generation where multiple joined-inheritance

entities against the same base class joined to each other as well
would not track columns on the base table independently of each other if
the string of joins were more than two entities long.  Also in 0.8.2.
[ticket:2759]

4 files changed, 119 insertions, 6 deletions

diff --git a/doc/build/changelog/changelog_08.rst b/doc/build/changelog/changelog_08.rst
index 8e81ff1b9..ffb2a24d4 100644
--- a/doc/build/changelog/changelog_08.rst
+++ b/doc/build/changelog/changelog_08.rst
@@ -7,6 +7,15 @@
     :version: 0.8.2
 
     .. change::
+        :tags: bug, orm
+        :tickets: 2759
+
+        Fixed bug in polymorphic SQL generation where multiple joined-inheritance
+        entities against the same base class joined to each other as well
+        would not track columns on the base table independently of each other if
+        the string of joins were more than two entities long.
+
+    .. change::
         :tags: bug, engine
         :pullreq: 6
 
diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 1ec648474..288642817 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -7,6 +7,15 @@
     :version: 0.9.0
 
     .. change::
+        :tags: bug, orm
+        :tickets: 2759
+
+        Fixed bug in polymorphic SQL generation where multiple joined-inheritance
+        entities against the same base class joined to each other as well
+        would not track columns on the base table independently of each other if
+        the string of joins were more than two entities long.  Also in 0.8.2.
+
+    .. change::
         :tags: bug, engine
         :pullreq: 6
 
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 54f5d7393..bd3ee8f72 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -203,17 +203,22 @@ class Query(object):
                 self._polymorphic_adapters.pop(m.local_table, None)
 
     def _adapt_polymorphic_element(self, element):
+        if "parententity" in element._annotations:
+            search = element._annotations['parententity']
+            alias = self._polymorphic_adapters.get(search, None)
+            if alias:
+                return alias.adapt_clause(element)
+
         if isinstance(element, expression.FromClause):
             search = element
         elif hasattr(element, 'table'):
             search = element.table
         else:
-            search = None
+            return None
 
-        if search is not None:
-            alias = self._polymorphic_adapters.get(search, None)
-            if alias:
-                return alias.adapt_clause(element)
+        alias = self._polymorphic_adapters.get(search, None)
+        if alias:
+            return alias.adapt_clause(element)
 
     def _adapt_col_list(self, cols):
         return [
diff --git a/test/orm/inheritance/test_relationship.py b/test/orm/inheritance/test_relationship.py
index 20c6de284..ecb4bf407 100644
--- a/test/orm/inheritance/test_relationship.py
+++ b/test/orm/inheritance/test_relationship.py
@@ -8,7 +8,7 @@ from sqlalchemy.engine import default
 from sqlalchemy.testing import AssertsCompiledSQL, fixtures
 from sqlalchemy import testing
 from sqlalchemy.testing.schema import Table, Column
-from sqlalchemy.testing import assert_raises, eq_
+from sqlalchemy.testing import assert_raises, eq_, is_
 
 class Company(fixtures.ComparableEntity):
     pass
@@ -1354,3 +1354,93 @@ class SubClassToSubClassMultiTest(AssertsCompiledSQL, fixtures.MappedTest):
             "JOIN ep2 ON anon_1.base2_id = ep2.base2_id"
         )
 
+class MultipleAdaptUsesEntityOverTableTest(AssertsCompiledSQL, fixtures.MappedTest):
+    __dialect__ = 'default'
+    run_create_tables = None
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('a', metadata,
+                Column('id', Integer, primary_key=True),
+                Column('name', String)
+        )
+        Table('b', metadata,
+                Column('id', Integer, ForeignKey('a.id'), primary_key=True)
+        )
+        Table('c', metadata,
+                Column('id', Integer, ForeignKey('a.id'), primary_key=True),
+                Column('bid', Integer, ForeignKey('b.id'))
+        )
+        Table('d', metadata,
+                Column('id', Integer, ForeignKey('a.id'), primary_key=True),
+                Column('cid', Integer, ForeignKey('c.id'))
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class A(cls.Comparable):
+            pass
+        class B(A):
+            pass
+        class C(A):
+            pass
+        class D(A):
+            pass
+
+    @classmethod
+    def setup_mappers(cls):
+        A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D
+        a, b, c, d = cls.tables.a, cls.tables.b, cls.tables.c, cls.tables.d
+        mapper(A, a)
+        mapper(B, b, inherits=A)
+        mapper(C, c, inherits=A)
+        mapper(D, d, inherits=A)
+
+    def _two_join_fixture(self):
+        A, B, C, D = self.classes.A, self.classes.B, self.classes.C, self.classes.D
+        s = Session()
+        return s.query(B.name, C.name, D.name).select_from(B).\
+                        join(C, C.bid == B.id).\
+                        join(D, D.cid == C.id)
+
+    def test_two_joins_adaption(self):
+        a, b, c, d = self.tables.a, self.tables.b, self.tables.c, self.tables.d
+        q = self._two_join_fixture()
+
+        btoc = q._from_obj[0].left
+
+        ac_adapted = btoc.right.element.left
+        c_adapted = btoc.right.element.right
+
+        is_(ac_adapted.element, a)
+        is_(c_adapted.element, c)
+
+        ctod = q._from_obj[0].right
+        ad_adapted = ctod.left
+        d_adapted = ctod.right
+        is_(ad_adapted.element, a)
+        is_(d_adapted.element, d)
+
+        bname, cname, dname = q._entities
+
+        b_name_adapted = bname._resolve_expr_against_query_aliases(
+                                        q, bname.column, None)
+        c_name_adapted = cname._resolve_expr_against_query_aliases(
+                                        q, cname.column, None)
+        d_name_adapted = dname._resolve_expr_against_query_aliases(
+                                        q, dname.column, None)
+
+        assert bool(b_name_adapted == a.c.name)
+        assert bool(c_name_adapted == ac_adapted.c.name)
+        assert bool(d_name_adapted == ad_adapted.c.name)
+
+    def test_two_joins_sql(self):
+        q = self._two_join_fixture()
+        self.assert_compile(q,
+            "SELECT a.name AS a_name, a_1.name AS a_1_name, "
+            "a_2.name AS a_2_name "
+            "FROM a JOIN b ON a.id = b.id JOIN "
+            "(a AS a_1 JOIN c AS c_1 ON a_1.id = c_1.id) ON c_1.bid = b.id "
+            "JOIN (a AS a_2 JOIN d AS d_1 ON a_2.id = d_1.id) "
+            "ON d_1.cid = c_1.id"
+        )
\ No newline at end of file