- Fixed an obscure bug where the wrong results would be

fetched when joining/joinedloading across a many-to-many
relationship to a single-table-inheriting
subclass with a specific discriminator value, due to "secondary"
rows that would come back.  The "secondary" and right-side
tables are now inner joined inside of parenthesis for all
ORM joins on many-to-many relationships so that the left->right
join can accurately filtered.
[ticket:2369]

7 files changed, 375 insertions, 97 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 1c27d7bf6..9d4bbfea1 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -7,6 +7,19 @@
     :version: 0.9.0
 
     .. change::
+        :tags: bug, orm
+        :tickets: 2369
+
+        Fixed an obscure bug where the wrong results would be
+        fetched when joining/joinedloading across a many-to-many
+        relationship to a single-table-inheriting
+        subclass with a specific discriminator value, due to "secondary"
+        rows that would come back.  The "secondary" and right-side
+        tables are now inner joined inside of parenthesis for all
+        ORM joins on many-to-many relationships so that the left->right
+        join can accurately filtered.
+
+    .. change::
         :tags: bug, mssql
         :tickets: 2747
 
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index 7ac3ac96a..fb6471ac6 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -918,8 +918,13 @@ class _ORMJoin(expression.Join):
                             of_type=right_info.mapper)
 
             if sj is not None:
-                left = sql.join(left, secondary, pj, isouter)
-                onclause = sj
+                if isouter:
+                    # note this is an inner join from secondary->right
+                    right = sql.join(secondary, right, sj)
+                    onclause = pj
+                else:
+                    left = sql.join(left, secondary, pj, isouter)
+                    onclause = sj
             else:
                 onclause = pj
             self._target_adapter = target_adapter
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 4422705cd..bf3f3397e 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -213,6 +213,8 @@ def surface_selectables(clause):
         yield elem
         if isinstance(elem, expression.Join):
             stack.extend((elem.left, elem.right))
+        elif isinstance(elem, expression.FromGrouping):
+            stack.append(elem.element)
 
 def selectables_overlap(left, right):
     """Return True if left/right have some overlapping selectable"""
diff --git a/test/orm/inheritance/test_relationship.py b/test/orm/inheritance/test_relationship.py
index f30a37941..20c6de284 100644
--- a/test/orm/inheritance/test_relationship.py
+++ b/test/orm/inheritance/test_relationship.py
@@ -448,6 +448,7 @@ class M2MFilterTest(fixtures.MappedTest):
             [Organization(name='org1')])
 
 class SelfReferentialM2MTest(fixtures.MappedTest, AssertsCompiledSQL):
+    __dialect__ = "default"
 
     @classmethod
     def define_tables(cls, metadata):
@@ -550,8 +551,7 @@ class SelfReferentialM2MTest(fixtures.MappedTest, AssertsCompiledSQL):
             "(parent AS parent_1 JOIN child1 AS child1_1 ON parent_1.id = child1_1.id) "
             "ON parent_1.id = secondary_2.right_id WHERE "
             "parent_1.id = secondary_1.right_id AND :param_1 = "
-            "secondary_1.left_id",
-            dialect=default.DefaultDialect()
+            "secondary_1.left_id"
         )
 
     def test_eager_join(self):
@@ -571,14 +571,14 @@ class SelfReferentialM2MTest(fixtures.MappedTest, AssertsCompiledSQL):
             "child2_1.id AS child2_1_id, parent_1.id AS "
             "parent_1_id, parent_1.cls AS parent_1_cls FROM "
             "(SELECT child1.id AS child1_id, parent.id AS parent_id, "
-            "parent.cls AS parent_cls FROM parent JOIN child1 ON parent.id = "
-            "child1.id LIMIT :param_1) AS anon_1 LEFT OUTER JOIN secondary "
-            "AS secondary_1 ON anon_1.parent_id = secondary_1.right_id LEFT "
-            "OUTER JOIN (parent AS parent_1 JOIN child2 AS child2_1 ON "
-            "parent_1.id = child2_1.id) ON parent_1.id = "
-            "secondary_1.left_id",
-            {'param_1':1},
-            dialect=default.DefaultDialect())
+            "parent.cls AS parent_cls "
+            "FROM parent JOIN child1 ON parent.id = child1.id "
+            "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN "
+            "(secondary AS secondary_1 JOIN "
+            "(parent AS parent_1 JOIN child2 AS child2_1 "
+            "ON parent_1.id = child2_1.id) ON parent_1.id = secondary_1.left_id) "
+            "ON anon_1.parent_id = secondary_1.right_id",
+            {'param_1':1})
 
         # another way to check
         assert q.limit(1).with_labels().subquery().count().scalar() == 1
diff --git a/test/orm/inheritance/test_single.py b/test/orm/inheritance/test_single.py
index de6e55e95..434642ca1 100644
--- a/test/orm/inheritance/test_single.py
+++ b/test/orm/inheritance/test_single.py
@@ -4,7 +4,7 @@ from sqlalchemy.orm import *
 
 from sqlalchemy import testing
 from test.orm import _fixtures
-from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import fixtures, AssertsCompiledSQL
 from sqlalchemy.testing.schema import Table, Column
 
 
@@ -418,7 +418,8 @@ class RelationshipToSingleTest(testing.AssertsCompiledSQL, fixtures.MappedTest):
 
 
     def test_relationship_to_subclass(self):
-        JuniorEngineer, Company, companies, Manager, Employee, employees, Engineer = (self.classes.JuniorEngineer,
+        JuniorEngineer, Company, companies, Manager, \
+                    Employee, employees, Engineer = (self.classes.JuniorEngineer,
                                 self.classes.Company,
                                 self.tables.companies,
                                 self.classes.Manager,
@@ -511,6 +512,125 @@ class RelationshipToSingleTest(testing.AssertsCompiledSQL, fixtures.MappedTest):
             )
         go()
 
+
+class ManyToManyToSingleTest(fixtures.MappedTest, AssertsCompiledSQL):
+    __dialect__ = 'default'
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('parent', metadata,
+                Column('id', Integer, primary_key=True,
+                            test_needs_autoincrement=True)
+            )
+        Table('m2m', metadata,
+                Column('parent_id', Integer,
+                            ForeignKey('parent.id'), primary_key=True),
+                Column('child_id', Integer,
+                            ForeignKey('child.id'), primary_key=True),
+            )
+        Table('child', metadata,
+                Column('id', Integer, primary_key=True,
+                            test_needs_autoincrement=True),
+                Column('discriminator', String(20)),
+                Column('name', String(20))
+            )
+
+    @classmethod
+    def setup_classes(cls):
+        class Parent(cls.Comparable):
+            pass
+
+        class Child(cls.Comparable):
+            pass
+
+        class SubChild1(Child):
+            pass
+
+        class SubChild2(Child):
+            pass
+
+    @classmethod
+    def setup_mappers(cls):
+        mapper(cls.classes.Parent, cls.tables.parent, properties={
+                "s1": relationship(cls.classes.SubChild1,
+                        secondary=cls.tables.m2m,
+                        uselist=False),
+                "s2": relationship(cls.classes.SubChild2,
+                        secondary=cls.tables.m2m)
+            })
+        mapper(cls.classes.Child, cls.tables.child,
+                        polymorphic_on=cls.tables.child.c.discriminator)
+        mapper(cls.classes.SubChild1, inherits=cls.classes.Child,
+                    polymorphic_identity='sub1')
+        mapper(cls.classes.SubChild2, inherits=cls.classes.Child,
+                    polymorphic_identity='sub2')
+
+    @classmethod
+    def insert_data(cls):
+        Parent = cls.classes.Parent
+        SubChild1 = cls.classes.SubChild1
+        SubChild2 = cls.classes.SubChild2
+        s = Session()
+        s.add_all([
+                Parent(s1=SubChild1(name='sc1_1'),
+                    s2=[SubChild2(name="sc2_1"), SubChild2(name="sc2_2")]
+                ),
+        ])
+        s.commit()
+
+    def test_eager_join(self):
+        Parent = self.classes.Parent
+        SubChild1 = self.classes.SubChild1
+
+        s = Session()
+
+        p1 = s.query(Parent).options(joinedload(Parent.s1)).all()[0]
+        eq_(p1.__dict__['s1'], SubChild1(name='sc1_1'))
+
+    def test_manual_join(self):
+        Parent = self.classes.Parent
+        Child = self.classes.Child
+        SubChild1 = self.classes.SubChild1
+
+        s = Session()
+
+        p1, c1 = s.query(Parent, Child).outerjoin(Parent.s1).all()[0]
+        eq_(c1, SubChild1(name='sc1_1'))
+
+    def test_assert_join_sql(self):
+        Parent = self.classes.Parent
+        Child = self.classes.Child
+
+        s = Session()
+
+        self.assert_compile(
+            s.query(Parent, Child).outerjoin(Parent.s1),
+            "SELECT parent.id AS parent_id, child.id AS child_id, "
+            "child.discriminator AS child_discriminator, "
+            "child.name AS child_name "
+            "FROM parent LEFT OUTER JOIN (m2m AS m2m_1 "
+            "JOIN child ON child.id = m2m_1.child_id "
+            "AND child.discriminator IN (:discriminator_1)) "
+            "ON parent.id = m2m_1.parent_id"
+        )
+
+    def test_assert_joinedload_sql(self):
+        Parent = self.classes.Parent
+        Child = self.classes.Child
+
+        s = Session()
+
+        self.assert_compile(
+            s.query(Parent).options(joinedload(Parent.s1)),
+            "SELECT parent.id AS parent_id, child_1.id AS child_1_id, "
+            "child_1.discriminator AS child_1_discriminator, "
+            "child_1.name AS child_1_name "
+            "FROM parent LEFT OUTER JOIN "
+            "(m2m AS m2m_1 JOIN child AS child_1 "
+            "ON child_1.id = m2m_1.child_id AND child_1.discriminator "
+            "IN (:discriminator_1)) ON parent.id = m2m_1.parent_id"
+        )
+
 class SingleOnJoinedTest(fixtures.MappedTest):
     @classmethod
     def define_tables(cls, metadata):
@@ -536,7 +656,8 @@ class SingleOnJoinedTest(fixtures.MappedTest):
         class Manager(Employee):
             pass
 
-        mapper(Person, persons_table, polymorphic_on=persons_table.c.type, polymorphic_identity='person')
+        mapper(Person, persons_table, polymorphic_on=persons_table.c.type,
+                    polymorphic_identity='person')
         mapper(Employee, employees_table, inherits=Person,polymorphic_identity='engineer')
         mapper(Manager, inherits=Employee,polymorphic_identity='manager')
 
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py
index 52f45a2d4..e53ff6669 100644
--- a/test/orm/test_eager_relations.py
+++ b/test/orm/test_eager_relations.py
@@ -23,6 +23,7 @@ import datetime
 class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
     run_inserts = 'once'
     run_deletes = None
+    __dialect__ = 'default'
 
     def test_basic(self):
         users, Address, addresses, User = (self.tables.users,
@@ -787,7 +788,8 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
         """test that the subquery wrapping only occurs with
         limit/offset and m2m or o2m joins present."""
 
-        users, items, order_items, Order, Item, User, Address, orders, addresses = (self.tables.users,
+        users, items, order_items, Order, Item, User, \
+                    Address, orders, addresses = (self.tables.users,
                                 self.tables.items,
                                 self.tables.order_items,
                                 self.classes.Order,
@@ -802,7 +804,8 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             orders=relationship(Order, backref='user')
         ))
         mapper(Order, orders, properties=odict([
-            ('items', relationship(Item, secondary=order_items, backref='orders')),
+            ('items', relationship(Item, secondary=order_items,
+                        backref='orders')),
             ('address', relationship(Address))
             ]))
         mapper(Address, addresses)
@@ -812,63 +815,86 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
 
         self.assert_compile(
             sess.query(User).options(joinedload(User.orders)).limit(10),
-            "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS anon_1_users_name, "
-            "orders_1.id AS orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS "
-            "orders_1_address_id, orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen "
+            "SELECT anon_1.users_id AS anon_1_users_id, "
+            "anon_1.users_name AS anon_1_users_name, "
+            "orders_1.id AS orders_1_id, orders_1.user_id AS orders_1_user_id, "
+            "orders_1.address_id AS "
+            "orders_1_address_id, orders_1.description AS orders_1_description, "
+            "orders_1.isopen AS orders_1_isopen "
             "FROM (SELECT users.id AS users_id, users.name AS users_name "
             "FROM users "
-            "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN orders AS orders_1 ON anon_1.users_id = orders_1.user_id",
-            {'param_1':10},
-            use_default_dialect=True
+            "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN orders AS orders_1 "
+            "ON anon_1.users_id = orders_1.user_id",
+            {'param_1':10}
         )
 
         self.assert_compile(
             sess.query(Order).options(joinedload(Order.user)).limit(10),
-            "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, orders.address_id AS "
-            "orders_address_id, orders.description AS orders_description, orders.isopen AS orders_isopen, "
-            "users_1.id AS users_1_id, users_1.name AS users_1_name FROM orders LEFT OUTER JOIN users AS "
+            "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, "
+            "orders.address_id AS "
+            "orders_address_id, orders.description AS orders_description, "
+            "orders.isopen AS orders_isopen, "
+            "users_1.id AS users_1_id, users_1.name AS users_1_name "
+            "FROM orders LEFT OUTER JOIN users AS "
             "users_1 ON users_1.id = orders.user_id LIMIT :param_1",
-            {'param_1':10},
-            use_default_dialect=True
+            {'param_1':10}
         )
 
         self.assert_compile(
-            sess.query(Order).options(joinedload(Order.user, innerjoin=True)).limit(10),
-            "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, orders.address_id AS "
-            "orders_address_id, orders.description AS orders_description, orders.isopen AS orders_isopen, "
-            "users_1.id AS users_1_id, users_1.name AS users_1_name FROM orders JOIN users AS "
+            sess.query(Order).options(
+                        joinedload(Order.user, innerjoin=True)).limit(10),
+            "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, "
+            "orders.address_id AS "
+            "orders_address_id, orders.description AS orders_description, "
+            "orders.isopen AS orders_isopen, "
+            "users_1.id AS users_1_id, users_1.name AS users_1_name "
+            "FROM orders JOIN users AS "
             "users_1 ON users_1.id = orders.user_id LIMIT :param_1",
-            {'param_1':10},
-            use_default_dialect=True
+            {'param_1':10}
         )
 
         self.assert_compile(
-            sess.query(User).options(joinedload_all("orders.address")).limit(10),
-            "SELECT anon_1.users_id AS anon_1_users_id, anon_1.users_name AS anon_1_users_name, "
-            "addresses_1.id AS addresses_1_id, addresses_1.user_id AS addresses_1_user_id, "
-            "addresses_1.email_address AS addresses_1_email_address, orders_1.id AS orders_1_id, "
-            "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, "
-            "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen FROM "
-            "(SELECT users.id AS users_id, users.name AS users_name FROM users LIMIT :param_1) AS anon_1 "
-            "LEFT OUTER JOIN orders AS orders_1 ON anon_1.users_id = orders_1.user_id LEFT OUTER JOIN "
+            sess.query(User).options(
+                        joinedload_all("orders.address")).limit(10),
+            "SELECT anon_1.users_id AS anon_1_users_id, "
+            "anon_1.users_name AS anon_1_users_name, "
+            "addresses_1.id AS addresses_1_id, "
+            "addresses_1.user_id AS addresses_1_user_id, "
+            "addresses_1.email_address AS addresses_1_email_address, "
+            "orders_1.id AS orders_1_id, "
+            "orders_1.user_id AS orders_1_user_id, "
+            "orders_1.address_id AS orders_1_address_id, "
+            "orders_1.description AS orders_1_description, "
+            "orders_1.isopen AS orders_1_isopen FROM "
+            "(SELECT users.id AS users_id, users.name AS users_name "
+            "FROM users LIMIT :param_1) AS anon_1 "
+            "LEFT OUTER JOIN orders AS orders_1 "
+            "ON anon_1.users_id = orders_1.user_id LEFT OUTER JOIN "
             "addresses AS addresses_1 ON addresses_1.id = orders_1.address_id",
-            {'param_1':10},
-            use_default_dialect=True
+            {'param_1':10}
         )
 
         self.assert_compile(
-            sess.query(User).options(joinedload_all("orders.items"), joinedload("orders.address")),
-            "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS items_1_id, "
-            "items_1.description AS items_1_description, addresses_1.id AS addresses_1_id, "
-            "addresses_1.user_id AS addresses_1_user_id, addresses_1.email_address AS "
-            "addresses_1_email_address, orders_1.id AS orders_1_id, orders_1.user_id AS "
-            "orders_1_user_id, orders_1.address_id AS orders_1_address_id, orders_1.description "
-            "AS orders_1_description, orders_1.isopen AS orders_1_isopen FROM users LEFT OUTER JOIN "
-            "orders AS orders_1 ON users.id = orders_1.user_id LEFT OUTER JOIN order_items AS "
-            "order_items_1 ON orders_1.id = order_items_1.order_id LEFT OUTER JOIN items AS "
-            "items_1 ON items_1.id = order_items_1.item_id LEFT OUTER JOIN addresses AS "
-            "addresses_1 ON addresses_1.id = orders_1.address_id"
-            ,use_default_dialect=True
+            sess.query(User).options(joinedload_all("orders.items"),
+                                joinedload("orders.address")),
+            "SELECT users.id AS users_id, users.name AS users_name, "
+                        "items_1.id AS items_1_id, "
+            "items_1.description AS items_1_description, "
+                        "addresses_1.id AS addresses_1_id, "
+            "addresses_1.user_id AS addresses_1_user_id, "
+            "addresses_1.email_address AS "
+            "addresses_1_email_address, orders_1.id AS orders_1_id, "
+            "orders_1.user_id AS "
+            "orders_1_user_id, orders_1.address_id AS orders_1_address_id, "
+            "orders_1.description "
+            "AS orders_1_description, orders_1.isopen AS orders_1_isopen "
+            "FROM users LEFT OUTER JOIN orders AS orders_1 "
+            "ON users.id = orders_1.user_id "
+            "LEFT OUTER JOIN (order_items AS order_items_1 "
+            "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) "
+            "ON orders_1.id = order_items_1.order_id "
+            "LEFT OUTER JOIN addresses AS addresses_1 "
+            "ON addresses_1.id = orders_1.address_id"
         )
 
         self.assert_compile(
@@ -882,8 +908,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "FROM users "
             "LIMIT :param_1) AS anon_1 LEFT OUTER JOIN orders AS orders_1 ON anon_1.users_id = "
             "orders_1.user_id LEFT OUTER JOIN addresses AS addresses_1 ON addresses_1.id = orders_1.address_id",
-            {'param_1':10},
-            use_default_dialect=True
+            {'param_1':10}
         )
 
         self.assert_compile(
@@ -898,8 +923,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "FROM users "
             "LIMIT :param_1) AS anon_1 JOIN orders AS orders_1 ON anon_1.users_id = "
             "orders_1.user_id JOIN addresses AS addresses_1 ON addresses_1.id = orders_1.address_id",
-            {'param_1':10},
-            use_default_dialect=True
+            {'param_1':10}
         )
 
     def test_one_to_many_scalar(self):
@@ -1159,8 +1183,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
                 "SELECT users.id AS users_id, users.name AS users_name, "
                 "addresses_1.id AS addresses_1_id, addresses_1.user_id AS addresses_1_user_id, "
                 "addresses_1.email_address AS addresses_1_email_address FROM users JOIN "
-                "addresses AS addresses_1 ON users.id = addresses_1.user_id ORDER BY addresses_1.id"
-        , use_default_dialect=True)
+                "addresses AS addresses_1 ON users.id = addresses_1.user_id ORDER BY addresses_1.id")
 
     def test_inner_join_chaining_options(self):
         users, items, order_items, Order, Item, User, orders = (self.tables.users,
@@ -1191,8 +1214,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "orders_1.isopen AS orders_1_isopen FROM users JOIN orders AS orders_1 ON "
             "users.id = orders_1.user_id JOIN order_items AS order_items_1 ON orders_1.id = "
             "order_items_1.order_id JOIN items AS items_1 ON items_1.id = "
-            "order_items_1.item_id",
-            use_default_dialect=True
+            "order_items_1.item_id"
         )
 
         self.assert_compile(
@@ -1201,11 +1223,12 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "items_1_id, items_1.description AS items_1_description, orders_1.id AS "
             "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS "
             "orders_1_address_id, orders_1.description AS orders_1_description, "
-            "orders_1.isopen AS orders_1_isopen FROM users LEFT OUTER JOIN orders AS orders_1 ON "
-            "users.id = orders_1.user_id LEFT OUTER JOIN order_items AS order_items_1 ON orders_1.id = "
-            "order_items_1.order_id LEFT OUTER JOIN items AS items_1 ON items_1.id = "
-            "order_items_1.item_id",
-            use_default_dialect=True
+            "orders_1.isopen AS orders_1_isopen "
+            "FROM users LEFT OUTER JOIN orders AS orders_1 "
+            "ON users.id = orders_1.user_id "
+            "LEFT OUTER JOIN (order_items AS order_items_1 "
+            "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) "
+            "ON orders_1.id = order_items_1.order_id"
         )
 
         self.assert_compile(
@@ -1214,11 +1237,70 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "items_1_id, items_1.description AS items_1_description, orders_1.id AS "
             "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS "
             "orders_1_address_id, orders_1.description AS orders_1_description, "
-            "orders_1.isopen AS orders_1_isopen FROM users JOIN orders AS orders_1 ON "
-            "users.id = orders_1.user_id LEFT OUTER JOIN order_items AS order_items_1 ON orders_1.id = "
-            "order_items_1.order_id LEFT OUTER JOIN items AS items_1 ON items_1.id = "
-            "order_items_1.item_id",
-            use_default_dialect=True
+            "orders_1.isopen AS orders_1_isopen "
+            "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id "
+            "LEFT OUTER JOIN (order_items AS order_items_1 "
+            "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) "
+            "ON orders_1.id = order_items_1.order_id"
+
+        )
+
+    def test_catch_the_right_target(self):
+        # test eager join chaining to the "nested" join on the left,
+        # a new feature as of [ticket:2369]
+
+        users, Keyword, orders, items, order_items, Order, Item, \
+                User, keywords, item_keywords = (self.tables.users,
+                                self.classes.Keyword,
+                                self.tables.orders,
+                                self.tables.items,
+                                self.tables.order_items,
+                                self.classes.Order,
+                                self.classes.Item,
+                                self.classes.User,
+                                self.tables.keywords,
+                                self.tables.item_keywords)
+
+        mapper(User, users, properties={
+            'orders':relationship(Order, backref='user'), # o2m, m2o
+        })
+        mapper(Order, orders, properties={
+            'items':relationship(Item, secondary=order_items,
+                                        order_by=items.c.id),  #m2m
+        })
+        mapper(Item, items, properties={
+            'keywords':relationship(Keyword, secondary=item_keywords,
+                                        order_by=keywords.c.id) #m2m
+        })
+        mapper(Keyword, keywords)
+
+        sess = create_session()
+        q = sess.query(User).join(User.orders).join(Order.items).\
+                options(joinedload_all("orders.items.keywords"))
+
+        # here, the eager join for keywords can catch onto
+        # join(Order.items) or the nested (orders LEFT OUTER JOIN items),
+        # it should catch the latter
+        self.assert_compile(
+            q,
+            "SELECT users.id AS users_id, users.name AS users_name, "
+            "keywords_1.id AS keywords_1_id, keywords_1.name AS keywords_1_name, "
+            "items_1.id AS items_1_id, items_1.description AS items_1_description, "
+            "orders_1.id AS orders_1_id, orders_1.user_id AS orders_1_user_id, "
+            "orders_1.address_id AS orders_1_address_id, "
+            "orders_1.description AS orders_1_description, "
+            "orders_1.isopen AS orders_1_isopen "
+            "FROM users JOIN orders ON users.id = orders.user_id "
+            "JOIN order_items AS order_items_1 ON orders.id = order_items_1.order_id "
+            "JOIN items ON items.id = order_items_1.item_id "
+            "LEFT OUTER JOIN orders AS orders_1 ON users.id = orders_1.user_id "
+            "LEFT OUTER JOIN (order_items AS order_items_2 "
+            "JOIN items AS items_1 ON items_1.id = order_items_2.item_id) "
+            "ON orders_1.id = order_items_2.order_id "
+            "LEFT OUTER JOIN (item_keywords AS item_keywords_1 "
+            "JOIN keywords AS keywords_1 ON keywords_1.id = item_keywords_1.keyword_id) "
+            "ON items_1.id = item_keywords_1.item_id "
+            "ORDER BY items_1.id, keywords_1.id"
         )
 
     def test_inner_join_chaining_fixed(self):
@@ -1249,10 +1331,10 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "orders_1_id, orders_1.user_id AS orders_1_user_id, orders_1.address_id AS "
             "orders_1_address_id, orders_1.description AS orders_1_description, "
             "orders_1.isopen AS orders_1_isopen FROM users LEFT OUTER JOIN orders AS orders_1 ON "
-            "users.id = orders_1.user_id LEFT OUTER JOIN order_items AS order_items_1 ON orders_1.id = "
-            "order_items_1.order_id LEFT OUTER JOIN items AS items_1 ON items_1.id = "
-            "order_items_1.item_id",
-            use_default_dialect=True
+            "users.id = orders_1.user_id LEFT OUTER JOIN "
+            "(order_items AS order_items_1 JOIN items AS items_1 ON items_1.id = "
+            "order_items_1.item_id) ON orders_1.id = "
+            "order_items_1.order_id"
         )
 
         # joining just from Order, innerjoin=True can be respected
@@ -1264,8 +1346,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "AS items_1_id, items_1.description AS items_1_description FROM "
             "orders JOIN order_items AS order_items_1 ON orders.id = "
             "order_items_1.order_id JOIN items AS items_1 ON items_1.id = "
-            "order_items_1.item_id",
-            use_default_dialect=True
+            "order_items_1.item_id"
         )
 
 
@@ -1291,8 +1372,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "SELECT users.id AS users_id, users.name AS users_name, orders_1.id AS orders_1_id, "
             "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, "
             "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen "
-            "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id ORDER BY orders_1.id"
-        , use_default_dialect=True)
+            "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id ORDER BY orders_1.id")
 
         self.assert_compile(sess.query(User).options(joinedload_all(User.orders, Order.items, innerjoin=True)),
             "SELECT users.id AS users_id, users.name AS users_name, items_1.id AS items_1_id, "
@@ -1301,8 +1381,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen "
             "FROM users JOIN orders AS orders_1 ON users.id = orders_1.user_id JOIN order_items AS "
             "order_items_1 ON orders_1.id = order_items_1.order_id JOIN items AS items_1 ON "
-            "items_1.id = order_items_1.item_id ORDER BY orders_1.id, items_1.id"
-        , use_default_dialect=True)
+            "items_1.id = order_items_1.item_id ORDER BY orders_1.id, items_1.id")
 
         def go():
             eq_(
@@ -1330,8 +1409,7 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             "orders_address_id, orders.description AS orders_description, orders.isopen AS "
             "orders_isopen, users_1.id AS users_1_id, users_1.name AS users_1_name "
             "FROM orders JOIN users AS users_1 ON users_1.id = orders.user_id "
-            "WHERE orders.description = :description_1",
-            use_default_dialect=True
+            "WHERE orders.description = :description_1"
         )
 
 
@@ -2141,6 +2219,7 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
     run_setup_mappers = 'once'
     run_inserts = 'once'
     run_deletes = None
+    __dialect__ = 'default'
 
     @classmethod
     def setup_mappers(cls):
@@ -2249,7 +2328,7 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
 
 
 
-    def test_aliased_entity(self):
+    def test_aliased_entity_one(self):
         Item, Order, User, Address = (self.classes.Item,
                                 self.classes.Order,
                                 self.classes.User,
@@ -2272,6 +2351,17 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             )
         self.assert_sql_count(testing.db, go, 1)
 
+    def test_aliased_entity_two(self):
+        Item, Order, User, Address = (self.classes.Item,
+                                self.classes.Order,
+                                self.classes.User,
+                                self.classes.Address)
+
+        sess = create_session()
+
+        oalias = sa.orm.aliased(Order)
+
+
         # one FROM clause
         def go():
             eq_(
@@ -2287,19 +2377,35 @@ class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
             )
         self.assert_sql_count(testing.db, go, 1)
 
-        from sqlalchemy.engine.default import DefaultDialect
+
+    def test_aliased_entity_three(self):
+        Item, Order, User, Address = (self.classes.Item,
+                                self.classes.Order,
+                                self.classes.User,
+                                self.classes.Address)
+
+        sess = create_session()
+
+        oalias = sa.orm.aliased(Order)
+
 
         # improper setup: oalias in the columns clause but join to usual
         # orders alias.  this should create two FROM clauses even though the
         # query has a from_clause set up via the join
-        self.assert_compile(sess.query(User, oalias).join(User.orders).options(joinedload(oalias.items)).with_labels().statement,
-        "SELECT users.id AS users_id, users.name AS users_name, orders_1.id AS orders_1_id, "\
-        "orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, "\
-        "orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen, items_1.id AS items_1_id, "\
-        "items_1.description AS items_1_description FROM users JOIN orders ON users.id = orders.user_id, "\
-        "orders AS orders_1 LEFT OUTER JOIN order_items AS order_items_1 ON orders_1.id = order_items_1.order_id "\
-        "LEFT OUTER JOIN items AS items_1 ON items_1.id = order_items_1.item_id ORDER BY items_1.id",
-        dialect=DefaultDialect()
+        self.assert_compile(
+                sess.query(User, oalias).join(User.orders).
+                    options(joinedload(oalias.items)).with_labels().statement,
+        "SELECT users.id AS users_id, users.name AS users_name, "
+        "orders_1.id AS orders_1_id, "
+        "orders_1.user_id AS orders_1_user_id, "
+        "orders_1.address_id AS orders_1_address_id, "
+        "orders_1.description AS orders_1_description, "
+        "orders_1.isopen AS orders_1_isopen, items_1.id AS items_1_id, "
+        "items_1.description AS items_1_description FROM users "
+        "JOIN orders ON users.id = orders.user_id, "
+        "orders AS orders_1 LEFT OUTER JOIN (order_items AS order_items_1 "
+        "JOIN items AS items_1 ON items_1.id = order_items_1.item_id) "
+        "ON orders_1.id = order_items_1.order_id ORDER BY items_1.id"
         )
 
 
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index 1a972d965..2403f4aae 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -1929,7 +1929,8 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
 
 
     def test_more_joins(self):
-        users, Keyword, orders, items, order_items, Order, Item, User, keywords, item_keywords = (self.tables.users,
+        users, Keyword, orders, items, order_items, Order, Item, \
+                User, keywords, item_keywords = (self.tables.users,
                                 self.classes.Keyword,
                                 self.tables.orders,
                                 self.tables.items,
@@ -1972,6 +1973,36 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
             User(name='jack',id=7)
         ])
 
+    def test_very_nested_joins_with_joinedload(self):
+        users, Keyword, orders, items, order_items, Order, Item, \
+                User, keywords, item_keywords = (self.tables.users,
+                                self.classes.Keyword,
+                                self.tables.orders,
+                                self.tables.items,
+                                self.tables.order_items,
+                                self.classes.Order,
+                                self.classes.Item,
+                                self.classes.User,
+                                self.tables.keywords,
+                                self.tables.item_keywords)
+
+        mapper(User, users, properties={
+            'orders':relationship(Order, backref='user'), # o2m, m2o
+        })
+        mapper(Order, orders, properties={
+            'items':relationship(Item, secondary=order_items,
+                                        order_by=items.c.id),  #m2m
+        })
+        mapper(Item, items, properties={
+            'keywords':relationship(Keyword, secondary=item_keywords,
+                                        order_by=keywords.c.id) #m2m
+        })
+        mapper(Keyword, keywords)
+
+        sess = create_session()
+
+        sel = users.select(users.c.id.in_([7, 8]))
+
         def go():
             eq_(
                 sess.query(User).select_entity_from(sel).