Merge branch 'ticket_2587'

Conflicts:
	test/profiles.txt
	test/sql/test_selectable.py

20 files changed, 723 insertions, 299 deletions

diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 1ca8f4e64..c7e09b164 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -592,6 +592,7 @@ class SQLiteDialect(default.DefaultDialect):
     supports_empty_insert = False
     supports_cast = True
     supports_multivalues_insert = True
+    supports_right_nested_joins = False
 
     default_paramstyle = 'qmark'
     execution_ctx_cls = SQLiteExecutionContext
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 91869ab75..2ad7002c4 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -49,6 +49,8 @@ class DefaultDialect(interfaces.Dialect):
     postfetch_lastrowid = True
     implicit_returning = False
 
+    supports_right_nested_joins = True
+
     supports_native_enum = False
     supports_native_boolean = False
 
@@ -106,6 +108,7 @@ class DefaultDialect(interfaces.Dialect):
     def __init__(self, convert_unicode=False,
                  encoding='utf-8', paramstyle=None, dbapi=None,
                  implicit_returning=None,
+                 supports_right_nested_joins=None,
                  case_sensitive=True,
                  label_length=None, **kwargs):
 
@@ -130,6 +133,8 @@ class DefaultDialect(interfaces.Dialect):
         self.positional = self.paramstyle in ('qmark', 'format', 'numeric')
         self.identifier_preparer = self.preparer(self)
         self.type_compiler = self.type_compiler(self)
+        if supports_right_nested_joins is not None:
+            self.supports_right_nested_joins = supports_right_nested_joins
 
         self.case_sensitive = case_sensitive
 
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index beae7aba0..39ed8d8bf 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -447,6 +447,8 @@ class Query(object):
                         statement
         if self._params:
             stmt = stmt.params(self._params)
+
+
         # TODO: there's no tests covering effects of
         # the annotation not being there
         return stmt._annotate({'no_replacement_traverse': True})
@@ -1795,6 +1797,7 @@ class Query(object):
                                 right_entity, onclause,
                                 outerjoin, create_aliases, prop)
 
+
     def _join_left_to_right(self, left, right,
                             onclause, outerjoin, create_aliases, prop):
         """append a JOIN to the query's from clause."""
@@ -1814,10 +1817,21 @@ class Query(object):
                         "are the same entity" %
                         (left, right))
 
+        l_info = inspect(left)
+        r_info = inspect(right)
+
+        overlap = not create_aliases and \
+                        sql_util.selectables_overlap(l_info.selectable,
+                            r_info.selectable)
+        if overlap and l_info.selectable is r_info.selectable:
+            raise sa_exc.InvalidRequestError(
+                    "Can't join table/selectable '%s' to itself" %
+                        l_info.selectable)
+
         right, onclause = self._prepare_right_side(
-                                            right, onclause,
+                                r_info, right, onclause,
                                             create_aliases,
-                                            prop)
+                                            prop, overlap)
 
         # if joining on a MapperProperty path,
         # track the path to prevent redundant joins
@@ -1829,10 +1843,11 @@ class Query(object):
         else:
             self._joinpoint = {'_joinpoint_entity': right}
 
-        self._join_to_left(left, right, onclause, outerjoin)
+        self._join_to_left(l_info, left, right, onclause, outerjoin)
 
-    def _prepare_right_side(self, right, onclause, create_aliases, prop):
-        info = inspect(right)
+    def _prepare_right_side(self, r_info, right, onclause, create_aliases,
+                                    prop, overlap):
+        info = r_info
 
         right_mapper, right_selectable, right_is_aliased = \
             getattr(info, 'mapper', None), \
@@ -1862,19 +1877,23 @@ class Query(object):
                     (right_selectable.description,
                     right_mapper.mapped_table.description))
 
-            if not isinstance(right_selectable, expression.Alias):
+            if isinstance(right_selectable, expression.SelectBase):
+                # TODO: this isn't even covered now!
                 right_selectable = right_selectable.alias()
+                need_adapter = True
 
             right = aliased(right_mapper, right_selectable)
-            need_adapter = True
 
         aliased_entity = right_mapper and \
                             not right_is_aliased and \
                             (
-                                right_mapper.with_polymorphic or
                                 isinstance(
-                                    right_mapper.mapped_table,
-                                    expression.Join)
+                                    right_mapper._with_polymorphic_selectable,
+                                    expression.Alias)
+                                or
+                                overlap # test for overlap:
+                                        # orm/inheritance/relationships.py
+                                        # SelfReferentialM2MTest
                             )
 
         if not need_adapter and (create_aliases or aliased_entity):
@@ -1910,8 +1929,8 @@ class Query(object):
 
         return right, onclause
 
-    def _join_to_left(self, left, right, onclause, outerjoin):
-        info = inspect(left)
+    def _join_to_left(self, l_info, left, right, onclause, outerjoin):
+        info = l_info
         left_mapper = getattr(info, 'mapper', None)
         left_selectable = info.selectable
 
@@ -1946,7 +1965,6 @@ class Query(object):
             clause = left_selectable
 
         assert clause is not None
-
         try:
             clause = orm_join(clause, right, onclause, isouter=outerjoin)
         except sa_exc.ArgumentError as ae:
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index 95fa28613..33377d3ec 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -17,7 +17,7 @@ from .. import sql, util, exc as sa_exc, schema
 from ..sql.util import (
     ClauseAdapter,
     join_condition, _shallow_annotate, visit_binary_product,
-    _deep_deannotate, find_tables
+    _deep_deannotate, find_tables, selectables_overlap
     )
 from ..sql import operators, expression, visitors
 from .interfaces import MANYTOMANY, MANYTOONE, ONETOMANY
@@ -404,11 +404,7 @@ class JoinCondition(object):
     def _tables_overlap(self):
         """Return True if parent/child tables have some overlap."""
 
-        return  bool(
-            set(find_tables(self.parent_selectable)).intersection(
-                find_tables(self.child_selectable)
-            )
-        )
+        return selectables_overlap(self.parent_selectable, self.child_selectable)
 
     def _annotate_remote(self):
         """Annotate the primaryjoin and secondaryjoin
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index bd8228f2c..c21e7eace 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -493,6 +493,7 @@ class AliasedClass(object):
     """
     def __init__(self, cls, alias=None,
                             name=None,
+                            flat=True,
                             adapt_on_names=False,
                             #  TODO: None for default here?
                             with_polymorphic_mappers=(),
@@ -501,7 +502,7 @@ class AliasedClass(object):
                             use_mapper_path=False):
         mapper = _class_to_mapper(cls)
         if alias is None:
-            alias = mapper._with_polymorphic_selectable.alias(name=name)
+            alias = mapper._with_polymorphic_selectable.alias(name=name, flat=flat)
         self._aliased_insp = AliasedInsp(
             self,
             mapper,
@@ -837,7 +838,7 @@ def with_polymorphic(base, classes, selectable=False,
                     _with_polymorphic_args(classes, selectable,
                                 innerjoin=innerjoin)
     if aliased:
-        selectable = selectable.alias()
+        selectable = selectable.alias(flat=True)
     return AliasedClass(base,
                 selectable,
                 with_polymorphic_mappers=mappers,
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 73b094053..dd2a6e08c 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1113,23 +1113,115 @@ class SQLCompiler(engine.Compiled):
     def get_crud_hint_text(self, table, text):
         return None
 
+    def _transform_select_for_nested_joins(self, select):
+        """Rewrite any "a JOIN (b JOIN c)" expression as
+        "a JOIN (select * from b JOIN c) AS anon", to support
+        databases that can't parse a parenthesized join correctly
+        (i.e. sqlite the main one).
+
+        """
+        cloned = {}
+        column_translate = [{}]
+
+        # TODO: should we be using isinstance() for this,
+        # as this whole system won't work for custom Join/Select
+        # subclasses where compilation routines
+        # call down to compiler.visit_join(), compiler.visit_select()
+        join_name = sql.Join.__visit_name__
+        select_name = sql.Select.__visit_name__
+
+        def visit(element, **kw):
+            if element in column_translate[-1]:
+                return column_translate[-1][element]
+
+            elif element in cloned:
+                return cloned[element]
+
+            newelem = cloned[element] = element._clone()
+
+            if newelem.__visit_name__ is join_name and \
+                isinstance(newelem.right, sql.FromGrouping):
+
+                newelem._reset_exported()
+                newelem.left = visit(newelem.left, **kw)
+
+                right = visit(newelem.right, **kw)
+
+                selectable = sql.select(
+                                    [right.element],
+                                    use_labels=True).alias()
+
+                for c in selectable.c:
+                    c._label = c._key_label = c.name
+                translate_dict = dict(
+                        zip(right.element.c, selectable.c)
+                    )
+                translate_dict[right.element.left] = selectable
+                translate_dict[right.element.right] = selectable
+
+                # propagate translations that we've gained
+                # from nested visit(newelem.right) outwards
+                # to the enclosing select here.  this happens
+                # only when we have more than one level of right
+                # join nesting, i.e. "a JOIN (b JOIN (c JOIN d))"
+                for k, v in list(column_translate[-1].items()):
+                    if v in translate_dict:
+                        # remarkably, no current ORM tests (May 2013)
+                        # hit this condition, only test_join_rewriting
+                        # does.
+                        column_translate[-1][k] = translate_dict[v]
+
+                column_translate[-1].update(translate_dict)
+
+                newelem.right = selectable
+                newelem.onclause = visit(newelem.onclause, **kw)
+            elif newelem.__visit_name__ is select_name:
+                column_translate.append({})
+                newelem._copy_internals(clone=visit, **kw)
+                del column_translate[-1]
+            else:
+                newelem._copy_internals(clone=visit, **kw)
+
+            return newelem
+
+        return visit(select)
+
+    def _transform_result_map_for_nested_joins(self, select, transformed_select):
+        inner_col = dict((c._key_label, c) for
+                        c in transformed_select.inner_columns)
+        d = dict(
+                    (inner_col[c._key_label], c)
+                    for c in select.inner_columns
+                )
+        for key, (name, objs, typ) in list(self.result_map.items()):
+            objs = tuple([d.get(col, col) for col in objs])
+            self.result_map[key] = (name, objs, typ)
+
     def visit_select(self, select, asfrom=False, parens=True,
                             iswrapper=False, fromhints=None,
                             compound_index=0,
                             force_result_map=False,
-                            positional_names=None, **kwargs):
-        entry = self.stack and self.stack[-1] or {}
-
-        existingfroms = entry.get('from', None)
-
-        froms = select._get_display_froms(existingfroms, asfrom=asfrom)
-
-        correlate_froms = set(sql._from_objects(*froms))
+                            positional_names=None,
+                            nested_join_translation=False, **kwargs):
+
+        needs_nested_translation = \
+                            select.use_labels and \
+                            not nested_join_translation and \
+                            not self.stack and \
+                            not self.dialect.supports_right_nested_joins
+
+        if needs_nested_translation:
+            transformed_select = self._transform_select_for_nested_joins(select)
+            text = self.visit_select(
+                            transformed_select, asfrom=asfrom, parens=parens,
+                            iswrapper=iswrapper, fromhints=fromhints,
+                            compound_index=compound_index,
+                            force_result_map=force_result_map,
+                            positional_names=positional_names,
+                            nested_join_translation=True, **kwargs
+                        )
 
-        # TODO: might want to propagate existing froms for
-        # select(select(select)) where innermost select should correlate
-        # to outermost if existingfroms: correlate_froms =
-        # correlate_froms.union(existingfroms)
+        entry = self.stack and self.stack[-1] or {}
 
         populate_result_map = force_result_map or (
                                         compound_index == 0 and (
@@ -1138,6 +1230,19 @@ class SQLCompiler(engine.Compiled):
                                         )
                                     )
 
+        if needs_nested_translation:
+            if populate_result_map:
+                self._transform_result_map_for_nested_joins(
+                                                select, transformed_select)
+            return text
+
+        existingfroms = entry.get('from', None)
+
+        froms = select._get_display_froms(existingfroms, asfrom=asfrom)
+
+        correlate_froms = set(sql._from_objects(*froms))
+
+
         self.stack.append({'from': correlate_froms,
                             'iswrapper': iswrapper})
 
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index 6dc134d98..f0c6134e5 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -795,7 +795,7 @@ def intersect_all(*selects, **kwargs):
     return CompoundSelect(CompoundSelect.INTERSECT_ALL, *selects, **kwargs)
 
 
-def alias(selectable, name=None):
+def alias(selectable, name=None, flat=False):
     """Return an :class:`.Alias` object.
 
     An :class:`.Alias` represents any :class:`.FromClause`
@@ -2636,7 +2636,7 @@ class FromClause(Selectable):
 
         return Join(self, right, onclause, True)
 
-    def alias(self, name=None):
+    def alias(self, name=None, flat=False):
         """return an alias of this :class:`.FromClause`.
 
         This is shorthand for calling::
@@ -3980,7 +3980,7 @@ class Join(FromClause):
     def bind(self):
         return self.left.bind or self.right.bind
 
-    def alias(self, name=None):
+    def alias(self, name=None, flat=False):
         """return an alias of this :class:`.Join`.
 
         Used against a :class:`.Join` object,
@@ -4008,7 +4008,17 @@ class Join(FromClause):
         aliases.
 
         """
-        return self.select(use_labels=True, correlate=False).alias(name)
+        if flat:
+            assert name is None, "Can't send name argument with flat"
+            left_a, right_a = self.left.alias(flat=True), \
+                                self.right.alias(flat=True)
+            adapter = sqlutil.ClauseAdapter(left_a).\
+                        chain(sqlutil.ClauseAdapter(right_a))
+
+            return left_a.join(right_a,
+                        adapter.traverse(self.onclause), isouter=self.isouter)
+        else:
+            return self.select(use_labels=True, correlate=False).alias(name)
 
     @property
     def _hide_froms(self):
@@ -4138,7 +4148,7 @@ class CTE(Alias):
         self._restates = _restates
         super(CTE, self).__init__(selectable, name=name)
 
-    def alias(self, name=None):
+    def alias(self, name=None, flat=False):
         return CTE(
             self.original,
             name=name,
@@ -4221,10 +4231,10 @@ class FromGrouping(FromClause):
 
     @property
     def foreign_keys(self):
-        # this could be
-        # self.element.foreign_keys
-        # see SelectableTest.test_join_condition
-        return set()
+        return self.element.foreign_keys
+
+    def is_derived_from(self, element):
+        return self.element.is_derived_from(element)
 
     @property
     def _hide_froms(self):
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 91740dc16..6f4d27e1b 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -200,15 +200,28 @@ def clause_is_present(clause, search):
 
     """
 
-    stack = [search]
-    while stack:
-        elem = stack.pop()
+    for elem in surface_selectables(search):
         if clause == elem:  # use == here so that Annotated's compare
             return True
-        elif isinstance(elem, expression.Join):
+    else:
+        return False
+
+def surface_selectables(clause):
+    stack = [clause]
+    while stack:
+        elem = stack.pop()
+        yield elem
+        if isinstance(elem, expression.Join):
             stack.extend((elem.left, elem.right))
-    return False
 
+def selectables_overlap(left, right):
+    """Return True if left/right have some overlapping selectable"""
+
+    return bool(
+                set(surface_selectables(left)).intersection(
+                        surface_selectables(right)
+                    )
+            )
 
 def bind_values(clause):
     """Return an ordered list of "bound" values in the given clause.
@@ -797,8 +810,11 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
     def __init__(self, selectable, equivalents=None,
                         include=None, exclude=None,
                         include_fn=None, exclude_fn=None,
-                        adapt_on_names=False):
+                        adapt_on_names=False,
+                        traverse_options=None):
         self.__traverse_options__ = {'stop_on': [selectable]}
+        if traverse_options:
+            self.__traverse_options__.update(traverse_options)
         self.selectable = selectable
         if include:
             assert not include_fn
@@ -829,10 +845,11 @@ class ClauseAdapter(visitors.ReplacingCloningVisitor):
             newcol = self.selectable.c.get(col.name)
         return newcol
 
+    magic_flag = False
     def replace(self, col):
-        if isinstance(col, expression.FromClause) and \
+        if not self.magic_flag and isinstance(col, expression.FromClause) and \
             self.selectable.is_derived_from(col):
-                return self.selectable
+            return self.selectable
         elif not isinstance(col, expression.ColumnElement):
             return None
         elif self.include_fn and not self.include_fn(col):
diff --git a/lib/sqlalchemy/sql/visitors.py b/lib/sqlalchemy/sql/visitors.py
index 62f46ab64..c5a45ffd4 100644
--- a/lib/sqlalchemy/sql/visitors.py
+++ b/lib/sqlalchemy/sql/visitors.py
@@ -30,6 +30,7 @@ import operator
 __all__ = ['VisitableType', 'Visitable', 'ClauseVisitor',
     'CloningVisitor', 'ReplacingCloningVisitor', 'iterate',
     'iterate_depthfirst', 'traverse_using', 'traverse',
+    'traverse_depthfirst',
     'cloned_traverse', 'replacement_traverse']
 
 
@@ -255,7 +256,11 @@ def cloned_traverse(obj, opts, visitors):
     """clone the given expression structure, allowing
     modifications by visitors."""
 
-    cloned = util.column_dict()
+
+    if "cloned" in opts:
+        cloned = opts['cloned']
+    else:
+        cloned = util.column_dict()
     stop_on = util.column_set(opts.get('stop_on', []))
 
     def clone(elem):
@@ -281,10 +286,12 @@ def replacement_traverse(obj, opts, replace):
 
     cloned = util.column_dict()
     stop_on = util.column_set([id(x) for x in opts.get('stop_on', [])])
+    unconditional = opts.get('unconditional', False)
 
     def clone(elem, **kw):
         if id(elem) in stop_on or \
-            'no_replacement_traverse' in elem._annotations:
+            (not unconditional
+                    and 'no_replacement_traverse' in elem._annotations):
             return elem
         else:
             newelem = replace(elem)
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index c04153961..96a8bc023 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -184,15 +184,20 @@ class AssertsCompiledSQL(object):
                         allow_dialect_select=False):
         if use_default_dialect:
             dialect = default.DefaultDialect()
-        elif dialect == None and not allow_dialect_select:
-            dialect = getattr(self, '__dialect__', None)
-            if dialect == 'default':
-                dialect = default.DefaultDialect()
-            elif dialect is None:
+        elif allow_dialect_select:
+            dialect = None
+        else:
+            if dialect is None:
+                dialect = getattr(self, '__dialect__', None)
+
+            if dialect is None:
                 dialect = config.db.dialect
+            elif dialect == 'default':
+                dialect = default.DefaultDialect()
             elif isinstance(dialect, util.string_types):
                 dialect = create_engine("%s://" % dialect).dialect
 
+
         kw = {}
         if params is not None:
             kw['column_keys'] = list(params)
diff --git a/test/orm/inheritance/test_polymorphic_rel.py b/test/orm/inheritance/test_polymorphic_rel.py
index 8c1f22114..eecfb036b 100644
--- a/test/orm/inheritance/test_polymorphic_rel.py
+++ b/test/orm/inheritance/test_polymorphic_rel.py
@@ -37,6 +37,7 @@ class _PolymorphicTestBase(object):
         e1, e2, e3, b1, m1 = \
             cls.e1, cls.e2, cls.e3, cls.b1, cls.m1
 
+
     def test_loads_at_once(self):
         """
         Test that all objects load from the full query, when
@@ -672,7 +673,17 @@ class _PolymorphicTestBase(object):
               expected)
         self.assert_sql_count(testing.db, go, 1)
 
+    def test_subqueryload_on_subclass(self):
         sess = create_session()
+        expected = [
+            Engineer(
+                name="dilbert",
+                engineer_name="dilbert",
+                primary_language="java",
+                status="regular engineer",
+                machines=[
+                    Machine(name="IBM ThinkPad"),
+                    Machine(name="IPhone")])]
         def go():
             # test load People with subqueryload to engineers + machines
             eq_(sess.query(Person)
@@ -726,6 +737,14 @@ class _PolymorphicTestBase(object):
                 .join(Engineer.machines).all(),
             [c1, c2])
 
+    def test_join_to_subclass_six_point_five(self):
+        sess = create_session()
+        eq_(sess.query(Company)
+                .join(people.join(engineers), 'employees')
+                .join(Engineer.machines)
+                .filter(Engineer.name == 'dilbert').all(),
+            [c1])
+
     def test_join_to_subclass_seven(self):
         sess = create_session()
         eq_(sess.query(Company)
@@ -898,7 +917,8 @@ class _PolymorphicTestBase(object):
                 .filter(Paperwork.description.like('%#%')).all(),
             [c1, c2])
 
-    def test_explicit_polymorphic_join(self):
+
+    def test_explicit_polymorphic_join_one(self):
         sess = create_session()
 
         # join from Company to Engineer; join condition formulated by
@@ -910,6 +930,9 @@ class _PolymorphicTestBase(object):
                 .filter(Engineer.engineer_name == 'vlad').one(),
             c2)
 
+    def test_explicit_polymorphic_join_two(self):
+        sess = create_session()
+
         # same, using explicit join condition.  Query.join() must
         # adapt the on clause here to match the subquery wrapped around
         # "people join engineers".
diff --git a/test/orm/inheritance/test_relationship.py b/test/orm/inheritance/test_relationship.py
index 809884f52..3f1eb849f 100644
--- a/test/orm/inheritance/test_relationship.py
+++ b/test/orm/inheritance/test_relationship.py
@@ -546,11 +546,10 @@ class SelfReferentialM2MTest(fixtures.MappedTest, AssertsCompiledSQL):
             "SELECT child2.id AS child2_id, parent.id AS parent_id, "
             "parent.cls AS parent_cls FROM secondary AS secondary_1, "
             "parent JOIN child2 ON parent.id = child2.id JOIN secondary AS "
-            "secondary_2 ON parent.id = secondary_2.left_id JOIN (SELECT "
-            "parent.id AS parent_id, parent.cls AS parent_cls, child1.id AS "
-            "child1_id FROM parent JOIN child1 ON parent.id = child1.id) AS "
-            "anon_1 ON anon_1.parent_id = secondary_2.right_id WHERE "
-            "anon_1.parent_id = secondary_1.right_id AND :param_1 = "
+            "secondary_2 ON parent.id = secondary_2.left_id JOIN "
+            "(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()
         )
@@ -569,15 +568,14 @@ class SelfReferentialM2MTest(fixtures.MappedTest, AssertsCompiledSQL):
         self.assert_compile(q.limit(1).with_labels().statement,
             "SELECT anon_1.child1_id AS anon_1_child1_id, anon_1.parent_id "
             "AS anon_1_parent_id, anon_1.parent_cls AS anon_1_parent_cls, "
-            "anon_2.child2_id AS anon_2_child2_id, anon_2.parent_id AS "
-            "anon_2_parent_id, anon_2.parent_cls AS anon_2_parent_cls FROM "
+            "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 (SELECT parent.id AS parent_id, parent.cls AS "
-            "parent_cls, child2.id AS child2_id FROM parent JOIN child2 ON "
-            "parent.id = child2.id) AS anon_2 ON anon_2.parent_id = "
+            "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())
@@ -1224,18 +1222,13 @@ class SubClassToSubClassMultiTest(AssertsCompiledSQL, fixtures.MappedTest):
                 join(Sub2.ep1).
                 join(Sub2.ep2),
             "SELECT parent.id AS parent_id, parent.data AS parent_data "
-            "FROM parent JOIN (SELECT base1.id AS base1_id, "
-            "base1.data AS base1_data, sub1.id AS sub1_id, "
-            "sub1.parent_id AS sub1_parent_id, sub1.subdata AS sub1_subdata "
-            "FROM base1 JOIN sub1 ON base1.id = sub1.id) AS anon_1 "
-            "ON parent.id = anon_1.sub1_parent_id JOIN "
-            "(SELECT base2.id AS base2_id, base2.base1_id AS base2_base1_id, "
-            "base2.data AS base2_data, sub2.id AS sub2_id, "
-            "sub2.subdata AS sub2_subdata FROM base2 JOIN sub2 "
-            "ON base2.id = sub2.id) AS anon_2 "
-            "ON anon_1.base1_id = anon_2.base2_base1_id "
-            "JOIN ep1 ON anon_2.base2_id = ep1.base2_id "
-            "JOIN ep2 ON anon_2.base2_id = ep2.base2_id"
+            "FROM parent JOIN (base1 JOIN sub1 ON base1.id = sub1.id) "
+            "ON parent.id = sub1.parent_id JOIN "
+            "(base2 JOIN sub2 "
+            "ON base2.id = sub2.id) "
+            "ON base1.id = base2.base1_id "
+            "JOIN ep1 ON base2.id = ep1.base2_id "
+            "JOIN ep2 ON base2.id = ep2.base2_id"
         )
 
     def test_two(self):
@@ -1248,16 +1241,11 @@ class SubClassToSubClassMultiTest(AssertsCompiledSQL, fixtures.MappedTest):
             s.query(Parent).join(Parent.sub1).
                 join(s2a, Sub1.sub2),
             "SELECT parent.id AS parent_id, parent.data AS parent_data "
-            "FROM parent JOIN (SELECT base1.id AS base1_id, "
-            "base1.data AS base1_data, sub1.id AS sub1_id, "
-            "sub1.parent_id AS sub1_parent_id, sub1.subdata AS sub1_subdata "
-            "FROM base1 JOIN sub1 ON base1.id = sub1.id) AS anon_1 "
-            "ON parent.id = anon_1.sub1_parent_id JOIN "
-            "(SELECT base2.id AS base2_id, base2.base1_id AS base2_base1_id, "
-            "base2.data AS base2_data, sub2.id AS sub2_id, "
-            "sub2.subdata AS sub2_subdata FROM base2 JOIN sub2 "
-            "ON base2.id = sub2.id) AS anon_2 "
-            "ON anon_1.base1_id = anon_2.base2_base1_id"
+            "FROM parent JOIN (base1 JOIN sub1 ON base1.id = sub1.id) "
+            "ON parent.id = sub1.parent_id JOIN "
+            "(base2 AS base2_1 JOIN sub2 AS sub2_1 "
+            "ON base2_1.id = sub2_1.id) "
+            "ON base1.id = base2_1.base1_id"
         )
 
     def test_three(self):
@@ -1269,13 +1257,11 @@ class SubClassToSubClassMultiTest(AssertsCompiledSQL, fixtures.MappedTest):
                 join(Sub2.ep1).\
                 join(Sub2.ep2),
             "SELECT base1.id AS base1_id, base1.data AS base1_data "
-            "FROM base1 JOIN (SELECT base2.id AS base2_id, base2.base1_id "
-            "AS base2_base1_id, base2.data AS base2_data, sub2.id AS sub2_id, "
-            "sub2.subdata AS sub2_subdata FROM base2 JOIN sub2 "
-            "ON base2.id = sub2.id) AS anon_1 ON base1.id = "
-            "anon_1.base2_base1_id "
-            "JOIN ep1 ON anon_1.base2_id = ep1.base2_id "
-            "JOIN ep2 ON anon_1.base2_id = ep2.base2_id"
+            "FROM base1 JOIN (base2 JOIN sub2 "
+            "ON base2.id = sub2.id) ON base1.id = "
+            "base2.base1_id "
+            "JOIN ep1 ON base2.id = ep1.base2_id "
+            "JOIN ep2 ON base2.id = ep2.base2_id"
         )
 
     def test_four(self):
@@ -1308,11 +1294,8 @@ class SubClassToSubClassMultiTest(AssertsCompiledSQL, fixtures.MappedTest):
             "sub2.subdata AS sub2_subdata "
             "FROM base2 JOIN sub2 ON base2.id = sub2.id "
             "JOIN "
-            "(SELECT base1.id AS base1_id, base1.data AS base1_data, "
-                "sub1.id AS sub1_id, sub1.parent_id AS sub1_parent_id, "
-                "sub1.subdata AS sub1_subdata "
-                "FROM base1 JOIN sub1 ON base1.id = sub1.id) AS anon_1 "
-            "ON anon_1.sub1_id = base2.base1_id "
+            "(base1 JOIN sub1 ON base1.id = sub1.id) "
+            "ON sub1.id = base2.base1_id "
             "JOIN ep1 ON base2.id = ep1.base2_id "
             "JOIN ep2 ON base2.id = ep2.base2_id"
         )
@@ -1352,28 +1335,22 @@ class SubClassToSubClassMultiTest(AssertsCompiledSQL, fixtures.MappedTest):
                 join(Sub2.ep2),
             "SELECT anon_1.parent_id AS anon_1_parent_id, "
             "anon_1.parent_data AS anon_1_parent_data, "
-            "anon_1.anon_2_sub2_id AS anon_1_anon_2_sub2_id, "
-            "anon_1.anon_2_base2_id AS anon_1_anon_2_base2_id, "
-            "anon_1.anon_2_base2_base1_id AS anon_1_anon_2_base2_base1_id, "
-            "anon_1.anon_2_base2_data AS anon_1_anon_2_base2_data, "
-            "anon_1.anon_2_sub2_subdata AS anon_1_anon_2_sub2_subdata "
+            "anon_1.sub2_id AS anon_1_sub2_id, "
+            "anon_1.base2_id AS anon_1_base2_id, "
+            "anon_1.base2_base1_id AS anon_1_base2_base1_id, "
+            "anon_1.base2_data AS anon_1_base2_data, "
+            "anon_1.sub2_subdata AS anon_1_sub2_subdata "
             "FROM (SELECT parent.id AS parent_id, parent.data AS parent_data, "
-            "anon_2.sub2_id AS anon_2_sub2_id, "
-            "anon_2.base2_id AS anon_2_base2_id, "
-            "anon_2.base2_base1_id AS anon_2_base2_base1_id, "
-            "anon_2.base2_data AS anon_2_base2_data, "
-            "anon_2.sub2_subdata AS anon_2_sub2_subdata "
-            "FROM parent JOIN (SELECT base1.id AS base1_id, "
-            "base1.data AS base1_data, sub1.id AS sub1_id, "
-            "sub1.parent_id AS sub1_parent_id, sub1.subdata AS sub1_subdata "
-            "FROM base1 JOIN sub1 ON base1.id = sub1.id) AS anon_3 "
-            "ON parent.id = anon_3.sub1_parent_id JOIN "
-            "(SELECT base2.id AS base2_id, base2.base1_id AS base2_base1_id, "
-            "base2.data AS base2_data, sub2.id AS sub2_id, "
+            "sub2.id AS sub2_id, "
+            "base2.id AS base2_id, "
+            "base2.base1_id AS base2_base1_id, "
+            "base2.data AS base2_data, "
             "sub2.subdata AS sub2_subdata "
-            "FROM base2 JOIN sub2 ON base2.id = sub2.id) AS anon_2 "
-            "ON anon_3.base1_id = anon_2.base2_base1_id) AS anon_1 "
-            "JOIN ep1 ON anon_1.anon_2_base2_id = ep1.base2_id "
-            "JOIN ep2 ON anon_1.anon_2_base2_id = ep2.base2_id"
+            "FROM parent JOIN (base1 JOIN sub1 ON base1.id = sub1.id) "
+            "ON parent.id = sub1.parent_id JOIN "
+            "(base2 JOIN sub2 ON base2.id = sub2.id) "
+            "ON base1.id = base2.base1_id) AS anon_1 "
+            "JOIN ep1 ON anon_1.base2_id = ep1.base2_id "
+            "JOIN ep2 ON anon_1.base2_id = ep2.base2_id"
         )
 
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py
index 54a106c5a..52f45a2d4 100644
--- a/test/orm/test_eager_relations.py
+++ b/test/orm/test_eager_relations.py
@@ -2630,16 +2630,15 @@ class CyclicalInheritingEagerTestTwo(fixtures.DeclarativeMappedTest,
         self.assert_compile(
             s.query(Director).options(joinedload('*')),
             "SELECT director.id AS director_id, persistent.id AS persistent_id, "
-            "director.name AS director_name, anon_1.movie_id AS anon_1_movie_id, "
-            "anon_1.persistent_id AS anon_1_persistent_id, "
-            "anon_1.movie_director_id AS anon_1_movie_director_id, "
-            "anon_1.movie_title AS anon_1_movie_title "
+            "director.name AS director_name, movie_1.id AS movie_1_id, "
+            "persistent_1.id AS persistent_1_id, "
+            "movie_1.director_id AS movie_1_director_id, "
+            "movie_1.title AS movie_1_title "
             "FROM persistent JOIN director ON persistent.id = director.id "
             "LEFT OUTER JOIN "
-            "(SELECT persistent.id AS persistent_id, movie.id AS movie_id, "
-                "movie.director_id AS movie_director_id, movie.title AS movie_title "
-                "FROM persistent JOIN movie ON persistent.id = movie.id) AS anon_1 "
-            "ON director.id = anon_1.movie_director_id"
+            "(persistent AS persistent_1 JOIN movie AS movie_1 "
+                "ON persistent_1.id = movie_1.id) "
+            "ON director.id = movie_1.director_id"
         )
 
     def test_integrate(self):
@@ -2702,15 +2701,12 @@ class CyclicalInheritingEagerTestThree(fixtures.DeclarativeMappedTest,
             sess.query(PersistentObject).options(joinedload(Director.other, join_depth=1)),
             "SELECT persistent.id AS persistent_id, director.id AS director_id, "
             "director.other_id AS director_other_id, "
-            "director.name AS director_name, anon_1.persistent_id AS "
-            "anon_1_persistent_id, anon_1.director_id AS anon_1_director_id, "
-            "anon_1.director_other_id AS anon_1_director_other_id, "
-            "anon_1.director_name AS anon_1_director_name "
+            "director.name AS director_name, persistent_1.id AS "
+            "persistent_1_id, director_1.id AS director_1_id, "
+            "director_1.other_id AS director_1_other_id, "
+            "director_1.name AS director_1_name "
             "FROM persistent LEFT OUTER JOIN director ON director.id = persistent.id "
-            "LEFT OUTER JOIN (SELECT persistent.id AS persistent_id, "
-                "director.id AS director_id, director.other_id AS director_other_id, "
-                "director.name AS director_name "
-                "FROM persistent LEFT OUTER JOIN director ON "
-                "director.id = persistent.id) "
-            "AS anon_1 ON director.other_id = anon_1.persistent_id"
+            "LEFT OUTER JOIN (persistent AS persistent_1 LEFT OUTER JOIN director AS director_1 ON "
+                "director_1.id = persistent_1.id) "
+            "ON director.other_id = persistent_1.id"
         )
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 2dac59150..cb9412e1d 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -203,15 +203,11 @@ class InheritedJoinTest(fixtures.MappedTest, AssertsCompiledSQL):
 
         self.assert_compile(
             sess.query(Company).join(Company.employees.of_type(Engineer)),
-            "SELECT companies.company_id AS companies_company_id, companies.name AS companies_name "
-            "FROM companies JOIN (SELECT people.person_id AS people_person_id, "
-            "people.company_id AS people_company_id, people.name AS people_name, "
-            "people.type AS people_type, engineers.person_id AS "
-            "engineers_person_id, engineers.status AS engineers_status, "
-            "engineers.engineer_name AS engineers_engineer_name, "
-            "engineers.primary_language AS engineers_primary_language "
-            "FROM people JOIN engineers ON people.person_id = engineers.person_id) AS "
-            "anon_1 ON companies.company_id = anon_1.people_company_id"
+            "SELECT companies.company_id AS companies_company_id, "
+            "companies.name AS companies_name "
+            "FROM companies JOIN "
+            "(people JOIN engineers ON people.person_id = engineers.person_id) "
+            "ON companies.company_id = people.company_id"
             , use_default_dialect = True
         )
 
@@ -259,7 +255,7 @@ class InheritedJoinTest(fixtures.MappedTest, AssertsCompiledSQL):
             , use_default_dialect=True
         )
 
-    def test_explicit_polymorphic_join(self):
+    def test_explicit_polymorphic_join_one(self):
         Company, Engineer = self.classes.Company, self.classes.Engineer
 
         sess = create_session()
@@ -268,35 +264,28 @@ class InheritedJoinTest(fixtures.MappedTest, AssertsCompiledSQL):
             sess.query(Company).join(Engineer).filter(Engineer.engineer_name=='vlad'),
             "SELECT companies.company_id AS companies_company_id, companies.name AS "
             "companies_name "
-            "FROM companies JOIN (SELECT people.person_id AS people_person_id, "
-            "people.company_id AS "
-            "people_company_id, people.name AS people_name, people.type AS people_type,"
-            " engineers.person_id AS "
-            "engineers_person_id, engineers.status AS engineers_status, "
-            "engineers.engineer_name AS engineers_engineer_name, "
-            "engineers.primary_language AS engineers_primary_language "
-            "FROM people JOIN engineers ON people.person_id = engineers.person_id) "
-            "AS anon_1 ON "
-            "companies.company_id = anon_1.people_company_id "
-            "WHERE anon_1.engineers_engineer_name = :engineer_name_1"
+            "FROM companies JOIN (people JOIN engineers "
+                "ON people.person_id = engineers.person_id) "
+            "ON "
+            "companies.company_id = people.company_id "
+            "WHERE engineers.engineer_name = :engineer_name_1"
             , use_default_dialect=True
         )
+
+    def test_explicit_polymorphic_join_two(self):
+        Company, Engineer = self.classes.Company, self.classes.Engineer
+
+        sess = create_session()
         self.assert_compile(
             sess.query(Company).join(Engineer, Company.company_id==Engineer.company_id).
                     filter(Engineer.engineer_name=='vlad'),
             "SELECT companies.company_id AS companies_company_id, companies.name "
             "AS companies_name "
-            "FROM companies JOIN (SELECT people.person_id AS people_person_id, "
-            "people.company_id AS "
-            "people_company_id, people.name AS people_name, people.type AS "
-            "people_type, engineers.person_id AS "
-            "engineers_person_id, engineers.status AS engineers_status, "
-            "engineers.engineer_name AS engineers_engineer_name, "
-            "engineers.primary_language AS engineers_primary_language "
-            "FROM people JOIN engineers ON people.person_id = engineers.person_id) AS "
-            "anon_1 ON "
-            "companies.company_id = anon_1.people_company_id "
-            "WHERE anon_1.engineers_engineer_name = :engineer_name_1"
+            "FROM companies JOIN "
+            "(people JOIN engineers ON people.person_id = engineers.person_id) "
+            "ON "
+            "companies.company_id = people.company_id "
+            "WHERE engineers.engineer_name = :engineer_name_1"
             , use_default_dialect=True
         )
 
@@ -319,16 +308,10 @@ class InheritedJoinTest(fixtures.MappedTest, AssertsCompiledSQL):
                 filter(Engineer.name=='dilbert'),
             "SELECT companies.company_id AS companies_company_id, companies.name AS "
             "companies_name "
-            "FROM companies JOIN (SELECT people.person_id AS people_person_id, "
-            "people.company_id AS "
-            "people_company_id, people.name AS people_name, people.type AS "
-            "people_type, engineers.person_id "
-            "AS engineers_person_id, engineers.status AS engineers_status, "
-            "engineers.engineer_name AS engineers_engineer_name, "
-            "engineers.primary_language AS engineers_primary_language FROM people "
+            "FROM companies JOIN (people "
             "JOIN engineers ON people.person_id = "
-            "engineers.person_id) AS anon_1 ON companies.company_id = "
-            "anon_1.people_company_id WHERE anon_1.people_name = :name_1"
+            "engineers.person_id) ON companies.company_id = "
+            "people.company_id WHERE people.name = :name_1"
             , use_default_dialect = True
         )
 
@@ -339,20 +322,14 @@ class InheritedJoinTest(fixtures.MappedTest, AssertsCompiledSQL):
                 filter(Engineer.name=='dilbert').filter(Machine.name=='foo'),
             "SELECT companies.company_id AS companies_company_id, companies.name AS "
             "companies_name "
-            "FROM companies JOIN (SELECT people.person_id AS people_person_id, "
-            "people.company_id AS "
-            "people_company_id, people.name AS people_name, people.type AS people_type,"
-            " engineers.person_id "
-            "AS engineers_person_id, engineers.status AS engineers_status, "
-            "engineers.engineer_name AS engineers_engineer_name, "
-            "engineers.primary_language AS engineers_primary_language FROM people "
+            "FROM companies JOIN (people "
             "JOIN engineers ON people.person_id = "
-            "engineers.person_id) AS anon_1 ON companies.company_id = "
-            "anon_1.people_company_id JOIN "
+            "engineers.person_id) ON companies.company_id = "
+            "people.company_id JOIN "
             "(SELECT machines.machine_id AS machine_id, machines.name AS name, "
             "machines.engineer_id AS engineer_id "
-            "FROM machines) AS anon_2 ON anon_1.engineers_person_id = anon_2.engineer_id "
-            "WHERE anon_1.people_name = :name_1 AND anon_2.name = :name_2"
+            "FROM machines) AS anon_1 ON engineers.person_id = anon_1.engineer_id "
+            "WHERE people.name = :name_1 AND anon_1.name = :name_2"
             , use_default_dialect = True
         )
 
@@ -1364,19 +1341,13 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
 
         assert_raises_message(
             sa_exc.InvalidRequestError,
-            "Could not find a FROM clause to join from.  Tried joining "
-            "to .*?, but got: "
-            "Can't find any foreign key relationships "
-            "between 'users' and 'users'.",
+            "Can't join table/selectable 'users' to itself",
             sess.query(users.c.id).join, User
         )
 
         assert_raises_message(
             sa_exc.InvalidRequestError,
-            "Could not find a FROM clause to join from.  Tried joining "
-            "to .*?, but got: "
-            "Can't find any foreign key relationships "
-            "between 'users' and 'users'.",
+            "Can't join table/selectable 'users' to itself",
             sess.query(users.c.id).select_from(users).join, User
         )
 
@@ -1522,16 +1493,22 @@ class JoinFromSelectableTest(fixtures.MappedTest, AssertsCompiledSQL):
         subq = sess.query(T2.t1_id, func.count(T2.id).label('count')).\
                     group_by(T2.t1_id).subquery()
 
-        # this query is wrong, but verifying behavior stays the same
-        # (or improves, like an error message)
+        assert_raises_message(
+            sa_exc.InvalidRequestError,
+            "Can't join table/selectable 'table1' to itself",
+            sess.query(T1.id, subq.c.count).join, T1, subq.c.t1_id == T1.id
+        )
+
         self.assert_compile(
-            sess.query(T1.id, subq.c.count).join(T1, subq.c.t1_id==T1.id),
-            "SELECT table1.id AS table1_id, anon_1.count AS anon_1_count FROM "
-            "(SELECT table2.t1_id AS t1_id, count(table2.id) AS count FROM "
-            "table2 GROUP BY table2.t1_id) AS anon_1, table1 JOIN table1 "
-            "ON anon_1.t1_id = table1.id"
+            sess.query(T1.id, subq.c.count).select_from(subq).\
+                join(T1, subq.c.t1_id == T1.id),
+            "SELECT table1.id AS table1_id, anon_1.count AS anon_1_count "
+            "FROM (SELECT table2.t1_id AS t1_id, count(table2.id) AS count "
+            "FROM table2 GROUP BY table2.t1_id) AS anon_1 "
+            "JOIN table1 ON anon_1.t1_id = table1.id"
         )
 
+
     def test_mapped_select_to_mapped_explicit_left(self):
         T1, T2 = self.classes.T1, self.classes.T2
 
diff --git a/test/orm/test_mapper.py b/test/orm/test_mapper.py
index 19ff78004..ed09e72c1 100644
--- a/test/orm/test_mapper.py
+++ b/test/orm/test_mapper.py
@@ -582,20 +582,16 @@ class MapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
         self.assert_compile(
             q,
             "SELECT "
-            "anon_1.addresses_id AS anon_1_addresses_id, "
-            "anon_1.users_id AS anon_1_users_id, "
-            "anon_1.users_name AS anon_1_users_name, "
-            "anon_1.addresses_user_id AS anon_1_addresses_user_id, "
-            "anon_1.addresses_email_address AS "
-            "anon_1_addresses_email_address, "
-            "anon_1.users_name || :name_1 AS anon_2 "
-            "FROM addresses JOIN (SELECT users.id AS users_id, "
-            "users.name AS users_name, addresses.id AS addresses_id, "
-            "addresses.user_id AS addresses_user_id, "
-            "addresses.email_address AS addresses_email_address "
-            "FROM users JOIN addresses ON users.id = "
-            "addresses.user_id) AS anon_1 ON "
-            "anon_1.users_id = addresses.user_id"
+            "addresses_1.id AS addresses_1_id, "
+            "users_1.id AS users_1_id, "
+            "users_1.name AS users_1_name, "
+            "addresses_1.user_id AS addresses_1_user_id, "
+            "addresses_1.email_address AS "
+            "addresses_1_email_address, "
+            "users_1.name || :name_1 AS anon_1 "
+            "FROM addresses JOIN (users AS users_1 JOIN addresses AS addresses_1 ON users_1.id = "
+            "addresses_1.user_id) ON "
+            "users_1.id = addresses.user_id"
         )
 
     def test_column_prop_deannotate(self):
diff --git a/test/orm/test_of_type.py b/test/orm/test_of_type.py
index 17ffebc3d..d002fd50f 100644
--- a/test/orm/test_of_type.py
+++ b/test/orm/test_of_type.py
@@ -69,12 +69,16 @@ class _PolymorphicTestBase(object):
                 .filter(Engineer.primary_language == 'java').count(),
             1)
 
+    def test_join_to_subclass_four(self):
+        sess = Session()
         # test [ticket:2093]
         eq_(sess.query(Company.company_id, Engineer)
                 .join(Company.employees.of_type(Engineer))
                 .filter(Engineer.primary_language == 'java').count(),
             1)
 
+    def test_join_to_subclass_five(self):
+        sess = Session()
         eq_(sess.query(Company)
                 .join(Company.employees.of_type(Engineer))
                 .filter(Engineer.primary_language == 'java').count(),
@@ -82,7 +86,7 @@ class _PolymorphicTestBase(object):
 
     def test_with_polymorphic_join_compile_one(self):
         sess = Session()
-
+# MARKMARK
         self.assert_compile(
             sess.query(Company).join(
                     Company.employees.of_type(
@@ -185,17 +189,19 @@ class _PolymorphicTestBase(object):
             )
         self.assert_sql_count(testing.db, go, 3)
 
+
 class PolymorphicPolymorphicTest(_PolymorphicTestBase, _PolymorphicPolymorphic):
     def _polymorphic_join_target(self, cls):
         from sqlalchemy.orm import class_mapper
 
+        from sqlalchemy.sql.expression import FromGrouping
         m, sel = class_mapper(Person)._with_polymorphic_args(cls)
-        sel = sel.alias()
+        sel = FromGrouping(sel.alias(flat=True))
         comp_sel = sel.compile(dialect=default.DefaultDialect())
 
         return \
             comp_sel.process(sel, asfrom=True).replace("\n", "") + \
-            " ON companies.company_id = anon_1.people_company_id"
+            " ON companies.company_id = people_1.company_id"
 
 class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions):
 
@@ -223,13 +229,14 @@ class PolymorphicAliasedJoinsTest(_PolymorphicTestBase, _PolymorphicAliasedJoins
 class PolymorphicJoinsTest(_PolymorphicTestBase, _PolymorphicJoins):
     def _polymorphic_join_target(self, cls):
         from sqlalchemy.orm import class_mapper
+        from sqlalchemy.sql.expression import FromGrouping
 
-        sel = class_mapper(Person)._with_polymorphic_selectable.alias()
+        sel = FromGrouping(class_mapper(Person)._with_polymorphic_selectable.alias(flat=True))
         comp_sel = sel.compile(dialect=default.DefaultDialect())
 
         return \
             comp_sel.process(sel, asfrom=True).replace("\n", "") + \
-            " ON companies.company_id = anon_1.people_company_id"
+            " ON companies.company_id = people_1.company_id"
 
 
 class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeMappedTest):
@@ -448,6 +455,7 @@ class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeM
                             DataContainer.jobs.of_type(Job_P).\
                                 any(Job_P.id < Job.id)
                         )
+
         self.assert_compile(q,
             "SELECT job.id AS job_id, job.type AS job_type, "
             "job.container_id "
@@ -455,11 +463,10 @@ class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeM
             "FROM data_container "
             "JOIN job ON data_container.id = job.container_id "
             "WHERE EXISTS (SELECT 1 "
-            "FROM (SELECT job.id AS job_id, job.type AS job_type, "
-            "job.container_id AS job_container_id, "
-            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
-            "FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) AS anon_1 "
-            "WHERE data_container.id = anon_1.job_container_id AND job.id > anon_1.job_id)"
+            "FROM job AS job_1 LEFT OUTER JOIN subjob AS subjob_1 "
+                "ON job_1.id = subjob_1.id "
+            "WHERE data_container.id = job_1.container_id "
+            "AND job.id > job_1.id)"
         )
 
     def test_any_walias(self):
@@ -501,11 +508,10 @@ class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeM
         self.assert_compile(q,
             "SELECT data_container.id AS data_container_id, "
             "data_container.name AS data_container_name "
-            "FROM data_container JOIN (SELECT job.id AS job_id, "
-            "job.type AS job_type, job.container_id AS job_container_id, "
-            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
-            "FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) "
-            "AS anon_1 ON data_container.id = anon_1.job_container_id")
+            "FROM data_container JOIN "
+            "(job AS job_1 LEFT OUTER JOIN subjob AS subjob_1 "
+                "ON job_1.id = subjob_1.id) "
+            "ON data_container.id = job_1.container_id")
 
     def test_join_wsubclass(self):
         ParentThing, DataContainer, Job, SubJob = \
@@ -524,11 +530,8 @@ class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeM
         self.assert_compile(q,
             "SELECT data_container.id AS data_container_id, "
             "data_container.name AS data_container_name "
-            "FROM data_container JOIN (SELECT job.id AS job_id, "
-            "job.type AS job_type, job.container_id AS job_container_id, "
-            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
-            "FROM job JOIN subjob ON job.id = subjob.id) AS anon_1 "
-            "ON data_container.id = anon_1.job_container_id"
+            "FROM data_container JOIN (job JOIN subjob ON job.id = subjob.id) "
+            "ON data_container.id = job.container_id"
         )
 
     def test_join_wpoly_innerjoin(self):
@@ -545,11 +548,9 @@ class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeM
         self.assert_compile(q,
             "SELECT data_container.id AS data_container_id, "
             "data_container.name AS data_container_name "
-            "FROM data_container JOIN (SELECT job.id AS job_id, "
-            "job.type AS job_type, job.container_id AS job_container_id, "
-            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
-            "FROM job JOIN subjob ON job.id = subjob.id) "
-            "AS anon_1 ON data_container.id = anon_1.job_container_id")
+            "FROM data_container JOIN "
+            "(job AS job_1 JOIN subjob AS subjob_1 ON job_1.id = subjob_1.id) "
+            "ON data_container.id = job_1.container_id")
 
     def test_join_walias(self):
         ParentThing, DataContainer, Job, SubJob = \
@@ -582,9 +583,8 @@ class SubclassRelationshipTest(testing.AssertsCompiledSQL, fixtures.DeclarativeM
         self.assert_compile(q,
             "SELECT data_container.id AS data_container_id, "
             "data_container.name AS data_container_name "
-            "FROM data_container JOIN (SELECT job.id AS job_id, "
-            "job.type AS job_type, job.container_id AS job_container_id, "
-            "subjob.id AS subjob_id, subjob.attr AS subjob_attr "
-            "FROM job LEFT OUTER JOIN subjob ON job.id = subjob.id) "
-            "AS anon_1 ON data_container.id = anon_1.job_container_id")
+            "FROM data_container JOIN "
+            "(job AS job_1 LEFT OUTER JOIN subjob AS subjob_1 "
+            "ON job_1.id = subjob_1.id) "
+            "ON data_container.id = job_1.container_id")
 
diff --git a/test/orm/test_subquery_relations.py b/test/orm/test_subquery_relations.py
index d493e0b7e..a6cc37691 100644
--- a/test/orm/test_subquery_relations.py
+++ b/test/orm/test_subquery_relations.py
@@ -1533,19 +1533,18 @@ class CyclicalInheritingEagerTestTwo(fixtures.DeclarativeMappedTest,
         q = ctx.attributes[('subquery',
                         (inspect(Director), inspect(Director).attrs.movies))]
         self.assert_compile(q,
-            "SELECT anon_1.movie_id AS anon_1_movie_id, "
-            "anon_1.persistent_id AS anon_1_persistent_id, "
-            "anon_1.movie_director_id AS anon_1_movie_director_id, "
-            "anon_1.movie_title AS anon_1_movie_title, "
-            "anon_2.director_id AS anon_2_director_id FROM "
-            "(SELECT director.id AS director_id FROM persistent JOIN director "
-            "ON persistent.id = director.id) AS anon_2 "
-            "JOIN (SELECT persistent.id AS persistent_id, movie.id AS movie_id, "
+            "SELECT movie.id AS movie_id, persistent.id AS persistent_id, "
             "movie.director_id AS movie_director_id, "
-            "movie.title AS movie_title FROM persistent JOIN movie "
-            "ON persistent.id = movie.id) AS anon_1 "
-            "ON anon_2.director_id = anon_1.movie_director_id "
-            "ORDER BY anon_2.director_id")
+            "movie.title AS movie_title, "
+            "anon_1.director_id AS anon_1_director_id "
+            "FROM (SELECT director.id AS director_id "
+                "FROM persistent JOIN director "
+                "ON persistent.id = director.id) AS anon_1 "
+            "JOIN (persistent JOIN movie ON persistent.id = movie.id) "
+            "ON anon_1.director_id = movie.director_id "
+            "ORDER BY anon_1.director_id",
+            dialect="default"
+        )
 
     def test_integrate(self):
         Director = self.classes.Director
diff --git a/test/profiles.txt b/test/profiles.txt
index 62924b895..4d8964639 100644
--- a/test/profiles.txt
+++ b/test/profiles.txt
@@ -1,15 +1,15 @@
 # /mnt/hgfs/classic/dev/sqlalchemy/test/profiles.txt
 # This file is written out on a per-environment basis.
-# For each test in aaa_profiling, the corresponding function and 
+# For each test in aaa_profiling, the corresponding function and
 # environment is located within this file.  If it doesn't exist,
 # the test is skipped.
-# If a callcount does exist, it is compared to what we received. 
+# If a callcount does exist, it is compared to what we received.
 # assertions are raised if the counts do not match.
-# 
-# To add a new callcount test, apply the function_call_count 
-# decorator and re-run the tests using the --write-profiles 
+#
+# To add a new callcount test, apply the function_call_count
+# decorator and re-run the tests using the --write-profiles
 # option - this file will be rewritten including the new count.
-# 
+#
 
 # TEST: test.aaa_profiling.test_compiler.CompileTest.test_insert
 
@@ -109,51 +109,46 @@ test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_
 
 # TEST: test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity
 
-test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.6_sqlite_pysqlite_nocextensions 118319
+test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.6_sqlite_pysqlite_nocextensions 154319
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_mysql_mysqldb_cextensions 124069
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_mysql_mysqldb_nocextensions 126819
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_oracle_cx_oracle_nocextensions 128319
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_postgresql_psycopg2_cextensions 116569
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_postgresql_psycopg2_nocextensions 119319
-test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_sqlite_pysqlite_cextensions 115569
-test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_sqlite_pysqlite_nocextensions 118319
+test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_sqlite_pysqlite_cextensions 151569
+test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 2.7_sqlite_pysqlite_nocextensions 154319
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 3.2_postgresql_psycopg2_nocextensions 121790
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 3.2_sqlite_pysqlite_nocextensions 121822
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 3.3_oracle_cx_oracle_nocextensions 130792
 test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 3.3_postgresql_psycopg2_nocextensions 121822
-test.aaa_profiling.test_orm.LoadManyToOneFromIdentityTest.test_many_to_one_load_no_identity 3.3_sqlite_pysqlite_nocextensions 121822
 
 # TEST: test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks
 
-test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.6_sqlite_pysqlite_nocextensions 19534
+test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.6_sqlite_pysqlite_nocextensions 21744
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_mysql_mysqldb_cextensions 19838
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_mysql_mysqldb_nocextensions 20098
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_oracle_cx_oracle_nocextensions 20152
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_postgresql_psycopg2_cextensions 19237
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_postgresql_psycopg2_nocextensions 19467
-test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_sqlite_pysqlite_cextensions 19274
-test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_sqlite_pysqlite_nocextensions 19534
+test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_sqlite_pysqlite_cextensions 21530
+test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 2.7_sqlite_pysqlite_nocextensions 21790
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 3.2_postgresql_psycopg2_nocextensions 20424
-test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 3.2_sqlite_pysqlite_nocextensions 20447
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 3.3_oracle_cx_oracle_nocextensions 21244
 test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 3.3_postgresql_psycopg2_nocextensions 20344
-test.aaa_profiling.test_orm.MergeBackrefsTest.test_merge_pending_with_all_pks 3.3_sqlite_pysqlite_nocextensions 20433
 
 # TEST: test.aaa_profiling.test_orm.MergeTest.test_merge_load
 
-test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.6_sqlite_pysqlite_nocextensions 1221
+test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.6_sqlite_pysqlite_nocextensions 1521
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_mysql_mysqldb_cextensions 1388
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_mysql_mysqldb_nocextensions 1413
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_oracle_cx_oracle_nocextensions 1349
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_postgresql_psycopg2_cextensions 1296
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_postgresql_psycopg2_nocextensions 1321
-test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_sqlite_pysqlite_cextensions 1196
-test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_sqlite_pysqlite_nocextensions 1221
+test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_sqlite_pysqlite_cextensions 1496
+test.aaa_profiling.test_orm.MergeTest.test_merge_load 2.7_sqlite_pysqlite_nocextensions 1521
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 3.2_postgresql_psycopg2_nocextensions 1332
-test.aaa_profiling.test_orm.MergeTest.test_merge_load 3.2_sqlite_pysqlite_nocextensions 1243
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 3.3_oracle_cx_oracle_nocextensions 1366
 test.aaa_profiling.test_orm.MergeTest.test_merge_load 3.3_postgresql_psycopg2_nocextensions 1357
-test.aaa_profiling.test_orm.MergeTest.test_merge_load 3.3_sqlite_pysqlite_nocextensions 1243
 
 # TEST: test.aaa_profiling.test_orm.MergeTest.test_merge_no_load
 
diff --git a/test/sql/test_join_rewriting.py b/test/sql/test_join_rewriting.py
new file mode 100644
index 000000000..5a9bdd1d3
--- /dev/null
+++ b/test/sql/test_join_rewriting.py
@@ -0,0 +1,292 @@
+from sqlalchemy import Table, Column, Integer, MetaData, ForeignKey, select
+from sqlalchemy.testing import fixtures, AssertsCompiledSQL
+from sqlalchemy import util
+from sqlalchemy.engine import default
+from sqlalchemy import testing
+
+
+m = MetaData()
+
+a = Table('a', m,
+        Column('id', Integer, primary_key=True)
+    )
+
+b = Table('b', m,
+        Column('id', Integer, primary_key=True),
+        Column('a_id', Integer, ForeignKey('a.id'))
+    )
+
+c = Table('c', m,
+        Column('id', Integer, primary_key=True),
+        Column('b_id', Integer, ForeignKey('b.id'))
+    )
+
+d = Table('d', m,
+        Column('id', Integer, primary_key=True),
+        Column('c_id', Integer, ForeignKey('c.id'))
+    )
+
+e = Table('e', m,
+        Column('id', Integer, primary_key=True)
+    )
+
+class _JoinRewriteTestBase(AssertsCompiledSQL):
+    def _test(self, s, assert_):
+        self.assert_compile(
+            s,
+            assert_
+        )
+
+        compiled = s.compile(dialect=self.__dialect__)
+        for key, col in zip([c.key for c in s.c], s.inner_columns):
+            key = key % compiled.anon_map
+            assert col in compiled.result_map[key][1]
+
+
+    def test_a_bc(self):
+        j1 = b.join(c)
+        j2 = a.join(j1)
+
+        # TODO: if we remove 'b' or 'c', shouldn't we get just
+        # the subset of cols from anon_1 ?
+
+        # TODO: do this test also with individual cols, things change
+        # lots based on how you go with this
+
+        s = select([a, b, c], use_labels=True).\
+            select_from(j2).\
+            where(b.c.id == 2).\
+            where(c.c.id == 3).order_by(a.c.id, b.c.id, c.c.id)
+
+        self._test(s, self._a_bc)
+
+    def test_a__b_dc(self):
+        j1 = c.join(d)
+        j2 = b.join(j1)
+        j3 = a.join(j2)
+
+        s = select([a, b, c, d], use_labels=True).\
+            select_from(j3).\
+            where(b.c.id == 2).\
+            where(c.c.id == 3).\
+            where(d.c.id == 4).\
+                order_by(a.c.id, b.c.id, c.c.id, d.c.id)
+
+        self._test(
+            s,
+            self._a__b_dc
+        )
+
+    def test_a_bc_comma_a1_selbc(self):
+        # test here we're emulating is
+        # test.orm.inheritance.test_polymorphic_rel:PolymorphicJoinsTest.test_multi_join
+        j1 = b.join(c)
+        j2 = b.join(c).select(use_labels=True).alias()
+        j3 = a.join(j1)
+        a_a = a.alias()
+        j4 = a_a.join(j2)
+
+        s = select([a, a_a, b, c, j2], use_labels=True).\
+                select_from(j3).select_from(j4).order_by(j2.c.b_id)
+
+        self._test(
+            s,
+            self._a_bc_comma_a1_selbc
+        )
+
+class JoinRewriteTest(_JoinRewriteTestBase, fixtures.TestBase):
+    """test rendering of each join with right-nested rewritten as
+    aliased SELECT statements.."""
+
+    @util.classproperty
+    def __dialect__(cls):
+        dialect = default.DefaultDialect()
+        dialect.supports_right_nested_joins = False
+        return dialect
+
+    _a__b_dc = (
+            "SELECT a.id AS a_id, anon_1.b_id AS b_id, "
+            "anon_1.b_a_id AS b_a_id, anon_1.c_id AS c_id, "
+            "anon_1.c_b_id AS c_b_id, anon_1.d_id AS d_id, "
+            "anon_1.d_c_id AS d_c_id "
+            "FROM a JOIN (SELECT b.id AS b_id, b.a_id AS b_a_id, "
+            "anon_2.c_id AS c_id, anon_2.c_b_id AS c_b_id, "
+            "anon_2.d_id AS d_id, anon_2.d_c_id AS d_c_id "
+            "FROM b JOIN (SELECT c.id AS c_id, c.b_id AS c_b_id, "
+            "d.id AS d_id, d.c_id AS d_c_id "
+            "FROM c JOIN d ON c.id = d.c_id) AS anon_2 "
+            "ON b.id = anon_2.c_b_id) AS anon_1 ON a.id = anon_1.b_a_id "
+            "WHERE anon_1.b_id = :id_1 AND anon_1.c_id = :id_2 AND "
+            "anon_1.d_id = :id_3 "
+            "ORDER BY a.id, anon_1.b_id, anon_1.c_id, anon_1.d_id"
+            )
+
+    _a_bc = (
+            "SELECT a.id AS a_id, anon_1.b_id AS b_id, "
+            "anon_1.b_a_id AS b_a_id, anon_1.c_id AS c_id, "
+            "anon_1.c_b_id AS c_b_id FROM a JOIN "
+            "(SELECT b.id AS b_id, b.a_id AS b_a_id, "
+                "c.id AS c_id, c.b_id AS c_b_id "
+                "FROM b JOIN c ON b.id = c.b_id) AS anon_1 "
+            "ON a.id = anon_1.b_a_id "
+            "WHERE anon_1.b_id = :id_1 AND anon_1.c_id = :id_2 "
+            "ORDER BY a.id, anon_1.b_id, anon_1.c_id"
+            )
+
+    _a_bc_comma_a1_selbc = (
+            "SELECT a.id AS a_id, a_1.id AS a_1_id, anon_1.b_id AS b_id, "
+            "anon_1.b_a_id AS b_a_id, anon_1.c_id AS c_id, "
+            "anon_1.c_b_id AS c_b_id, anon_2.b_id AS anon_2_b_id, "
+            "anon_2.b_a_id AS anon_2_b_a_id, anon_2.c_id AS anon_2_c_id, "
+            "anon_2.c_b_id AS anon_2_c_b_id FROM a "
+            "JOIN (SELECT b.id AS b_id, b.a_id AS b_a_id, c.id AS c_id, "
+            "c.b_id AS c_b_id FROM b JOIN c ON b.id = c.b_id) AS anon_1 "
+            "ON a.id = anon_1.b_a_id, "
+            "a AS a_1 JOIN "
+                "(SELECT b.id AS b_id, b.a_id AS b_a_id, "
+                "c.id AS c_id, c.b_id AS c_b_id "
+                "FROM b JOIN c ON b.id = c.b_id) AS anon_2 "
+            "ON a_1.id = anon_2.b_a_id ORDER BY anon_2.b_id"
+        )
+
+class JoinPlainTest(_JoinRewriteTestBase, fixtures.TestBase):
+    """test rendering of each join with normal nesting."""
+    @util.classproperty
+    def __dialect__(cls):
+        dialect = default.DefaultDialect()
+        return dialect
+
+    _a__b_dc = (
+            "SELECT a.id AS a_id, b.id AS b_id, "
+            "b.a_id AS b_a_id, c.id AS c_id, "
+            "c.b_id AS c_b_id, d.id AS d_id, "
+            "d.c_id AS d_c_id "
+            "FROM a JOIN (b JOIN (c JOIN d ON c.id = d.c_id) "
+            "ON b.id = c.b_id) ON a.id = b.a_id "
+            "WHERE b.id = :id_1 AND c.id = :id_2 AND "
+            "d.id = :id_3 "
+            "ORDER BY a.id, b.id, c.id, d.id"
+            )
+
+
+    _a_bc = (
+            "SELECT a.id AS a_id, b.id AS b_id, "
+            "b.a_id AS b_a_id, c.id AS c_id, "
+            "c.b_id AS c_b_id FROM a JOIN "
+            "(b JOIN c ON b.id = c.b_id) "
+            "ON a.id = b.a_id "
+            "WHERE b.id = :id_1 AND c.id = :id_2 "
+            "ORDER BY a.id, b.id, c.id"
+            )
+
+    _a_bc_comma_a1_selbc = (
+            "SELECT a.id AS a_id, a_1.id AS a_1_id, b.id AS b_id, "
+            "b.a_id AS b_a_id, c.id AS c_id, "
+            "c.b_id AS c_b_id, anon_1.b_id AS anon_1_b_id, "
+            "anon_1.b_a_id AS anon_1_b_a_id, anon_1.c_id AS anon_1_c_id, "
+            "anon_1.c_b_id AS anon_1_c_b_id FROM a "
+            "JOIN (b JOIN c ON b.id = c.b_id) "
+            "ON a.id = b.a_id, "
+            "a AS a_1 JOIN "
+                "(SELECT b.id AS b_id, b.a_id AS b_a_id, "
+                "c.id AS c_id, c.b_id AS c_b_id "
+                "FROM b JOIN c ON b.id = c.b_id) AS anon_1 "
+            "ON a_1.id = anon_1.b_a_id ORDER BY anon_1.b_id"
+        )
+
+class JoinNoUseLabelsTest(_JoinRewriteTestBase, fixtures.TestBase):
+    @util.classproperty
+    def __dialect__(cls):
+        dialect = default.DefaultDialect()
+        dialect.supports_right_nested_joins = False
+        return dialect
+
+    def _test(self, s, assert_):
+        s.use_labels = False
+        self.assert_compile(
+            s,
+            assert_
+        )
+
+    _a__b_dc = (
+            "SELECT a.id, b.id, "
+            "b.a_id, c.id, "
+            "c.b_id, d.id, "
+            "d.c_id "
+            "FROM a JOIN (b JOIN (c JOIN d ON c.id = d.c_id) "
+            "ON b.id = c.b_id) ON a.id = b.a_id "
+            "WHERE b.id = :id_1 AND c.id = :id_2 AND "
+            "d.id = :id_3 "
+            "ORDER BY a.id, b.id, c.id, d.id"
+            )
+
+    _a_bc = (
+            "SELECT a.id, b.id, "
+            "b.a_id, c.id, "
+            "c.b_id FROM a JOIN "
+            "(b JOIN c ON b.id = c.b_id) "
+            "ON a.id = b.a_id "
+            "WHERE b.id = :id_1 AND c.id = :id_2 "
+            "ORDER BY a.id, b.id, c.id"
+            )
+
+    _a_bc_comma_a1_selbc = (
+            "SELECT a.id, a_1.id, b.id, "
+            "b.a_id, c.id, "
+            "c.b_id, anon_1.b_id, "
+            "anon_1.b_a_id, anon_1.c_id, "
+            "anon_1.c_b_id FROM a "
+            "JOIN (b JOIN c ON b.id = c.b_id) "
+            "ON a.id = b.a_id, "
+            "a AS a_1 JOIN "
+                "(SELECT b.id AS b_id, b.a_id AS b_a_id, "
+                "c.id AS c_id, c.b_id AS c_b_id "
+                "FROM b JOIN c ON b.id = c.b_id) AS anon_1 "
+            "ON a_1.id = anon_1.b_a_id ORDER BY anon_1.b_id"
+        )
+
+class JoinExecTest(_JoinRewriteTestBase, fixtures.TestBase):
+    """invoke the SQL on the current backend to ensure compatibility"""
+
+    _a_bc = _a_bc_comma_a1_selbc = _a__b_dc = None
+
+    @classmethod
+    def setup_class(cls):
+        m.create_all(testing.db)
+
+    @classmethod
+    def teardown_class(cls):
+        m.drop_all(testing.db)
+
+    def _test(self, selectable, assert_):
+        testing.db.execute(selectable)
+
+
+class DialectFlagTest(fixtures.TestBase, AssertsCompiledSQL):
+    def test_dialect_flag(self):
+        d1 = default.DefaultDialect(supports_right_nested_joins=True)
+        d2 = default.DefaultDialect(supports_right_nested_joins=False)
+
+        j1 = b.join(c)
+        j2 = a.join(j1)
+
+        s = select([a, b, c], use_labels=True).\
+            select_from(j2)
+
+        self.assert_compile(
+            s,
+            "SELECT a.id AS a_id, b.id AS b_id, b.a_id AS b_a_id, c.id AS c_id, "
+            "c.b_id AS c_b_id FROM a JOIN (b JOIN c ON b.id = c.b_id) "
+            "ON a.id = b.a_id",
+            dialect=d1
+        )
+        self.assert_compile(
+            s,
+            "SELECT a.id AS a_id, anon_1.b_id AS b_id, "
+            "anon_1.b_a_id AS b_a_id, "
+            "anon_1.c_id AS c_id, anon_1.c_b_id AS c_b_id "
+            "FROM a JOIN (SELECT b.id AS b_id, b.a_id AS b_a_id, c.id AS c_id, "
+            "c.b_id AS c_b_id FROM b JOIN c ON b.id = c.b_id) AS anon_1 "
+            "ON a.id = anon_1.b_a_id",
+            dialect=d2
+        )
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 0c5cde9fb..2ac04dce3 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -776,12 +776,13 @@ class AnonLabelTest(fixtures.TestBase):
         c1 = literal_column('x')
         eq_(str(select([c1.label('y')])), "SELECT x AS y")
 
-class JoinConditionTest(fixtures.TestBase, AssertsExecutionResults):
+class JoinConditionTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
 
     def test_join_condition(self):
         m = MetaData()
         t1 = Table('t1', m, Column('id', Integer))
-        t2 = Table('t2', m, Column('id', Integer),
+        t2 = Table('t2', m,
+                    Column('id', Integer),
                     Column('t1id', ForeignKey('t1.id')))
         t3 = Table('t3', m,
                     Column('id', Integer),
@@ -793,6 +794,7 @@ class JoinConditionTest(fixtures.TestBase, AssertsExecutionResults):
                     Column('t1id1', ForeignKey('t1.id')),
                     Column('t1id2', ForeignKey('t1.id')),
             )
+
         t1t2 = t1.join(t2)
         t2t3 = t2.join(t3)
 
@@ -841,21 +843,23 @@ class JoinConditionTest(fixtures.TestBase, AssertsExecutionResults):
                 left.join(right).onclause
             )
 
+        # these are right-nested joins
+        j = t1t2.join(t2t3)
+        assert j.onclause.compare(t2.c.id == t3.c.t2id)
+        self.assert_compile(j,
+                "t1 JOIN t2 ON t1.id = t2.t1id JOIN "
+                "(t2 JOIN t3 ON t2.id = t3.t2id) ON t2.id = t3.t2id")
+
+        st2t3 = t2t3.select(use_labels=True)
+        j = t1t2.join(st2t3)
+        assert j.onclause.compare(t2.c.id == st2t3.c.t3_t2id)
+        self.assert_compile(j,
+                "t1 JOIN t2 ON t1.id = t2.t1id JOIN "
+                "(SELECT t2.id AS t2_id, t2.t1id AS t2_t1id, "
+                "t3.id AS t3_id, t3.t1id AS t3_t1id, t3.t2id AS t3_t2id "
+                "FROM t2 JOIN t3 ON t2.id = t3.t2id) ON t2.id = t3_t2id")
 
 
-        # TODO: this raises due to right side being "grouped", and no
-        # longer has FKs.  Did we want to make FromGrouping friendlier
-        # ?
-
-        assert_raises_message(exc.ArgumentError,
-                              "Perhaps you meant to convert the right "
-                              "side to a subquery using alias\(\)\?",
-                              t1t2.join, t2t3)
-        assert_raises_message(exc.ArgumentError,
-                              "Perhaps you meant to convert the right "
-                              "side to a subquery using alias\(\)\?",
-                              t1t2.join, t2t3.select(use_labels=True))
-
 
     def test_join_cond_no_such_unrelated_table(self):
         m = MetaData()