Improve rendering of core statements w/ ORM elements

This patch contains a variety of ORM and expression layer
tweaks to support ORM constructs in select() statements,
without the 1.3.x requiremnt in Query that a full
_compile_context() + new select() is needed in order to
get a working statement object.

Includes such tweaks as the ability to implement
aliased class of an aliased class,
as we are looking to fully support ACs against subqueries,
as well as the ability to access anonymously-labeled
ColumnProperty expressions within subqueries by
naming the ".key" of the label after the property
key.   Some tuning to query.join() as well
as ORMJoin internals to allow things to work more
smoothly.

Change-Id: Id810f485c5f7ed971529489b84694e02a3356d6d

8 files changed, 461 insertions, 322 deletions

diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 5589f0e0c..ba30d203b 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -5,7 +5,6 @@
 # This module is part of SQLAlchemy and is released under
 # the MIT License: http://www.opensource.org/licenses/mit-license.php
 
-
 from . import attributes
 from . import interfaces
 from . import loading
@@ -27,6 +26,7 @@ from ..sql import expression
 from ..sql import roles
 from ..sql import util as sql_util
 from ..sql import visitors
+from ..sql.base import _select_iterables
 from ..sql.base import CacheableOptions
 from ..sql.base import CompileState
 from ..sql.base import Options
@@ -90,7 +90,7 @@ class QueryContext(object):
         self.execution_options = execution_options or _EMPTY_DICT
         self.bind_arguments = bind_arguments or _EMPTY_DICT
         self.compile_state = compile_state
-        self.query = query = compile_state.query
+        self.query = query = compile_state.select_statement
         self.session = session
 
         self.propagated_loader_options = {
@@ -119,10 +119,14 @@ class QueryContext(object):
 
 
 class ORMCompileState(CompileState):
+    # note this is a dictionary, but the
+    # default_compile_options._with_polymorphic_adapt_map is a tuple
+    _with_polymorphic_adapt_map = _EMPTY_DICT
+
     class default_compile_options(CacheableOptions):
         _cache_key_traversal = [
             ("_use_legacy_query_style", InternalTraversal.dp_boolean),
-            ("_orm_results", InternalTraversal.dp_boolean),
+            ("_for_statement", InternalTraversal.dp_boolean),
             ("_bake_ok", InternalTraversal.dp_boolean),
             (
                 "_with_polymorphic_adapt_map",
@@ -137,8 +141,18 @@ class ORMCompileState(CompileState):
             ("_for_refresh_state", InternalTraversal.dp_boolean),
         ]
 
+        # set to True by default from Query._statement_20(), to indicate
+        # the rendered query should look like a legacy ORM query.  right
+        # now this basically indicates we should use tablename_columnname
+        # style labels.    Generally indicates the statement originated
+        # from a Query object.
         _use_legacy_query_style = False
-        _orm_results = True
+
+        # set *only* when we are coming from the Query.statement
+        # accessor, or a Query-level equivalent such as
+        # query.subquery().  this supersedes "toplevel".
+        _for_statement = False
+
         _bake_ok = True
         _with_polymorphic_adapt_map = ()
         _current_path = _path_registry
@@ -149,42 +163,24 @@ class ORMCompileState(CompileState):
         _set_base_alias = False
         _for_refresh_state = False
 
-        @classmethod
-        def merge(cls, other):
-            return cls + other._state_dict()
-
     current_path = _path_registry
 
     def __init__(self, *arg, **kw):
         raise NotImplementedError()
 
-    def dispose(self):
-        self.attributes.clear()
-
     @classmethod
     def create_for_statement(cls, statement_container, compiler, **kw):
-        raise NotImplementedError()
+        """Create a context for a statement given a :class:`.Compiler`.
 
-    @classmethod
-    def _create_for_legacy_query(cls, query, toplevel, for_statement=False):
-        stmt = query._statement_20(orm_results=not for_statement)
-
-        # this chooses between ORMFromStatementCompileState and
-        # ORMSelectCompileState.  We could also base this on
-        # query._statement is not None as we have the ORM Query here
-        # however this is the more general path.
-        compile_state_cls = CompileState._get_plugin_class_for_plugin(
-            stmt, "orm"
-        )
+        This method is always invoked in the context of SQLCompiler.process().
 
-        return compile_state_cls._create_for_statement_or_query(
-            stmt, toplevel, for_statement=for_statement
-        )
+        For a Select object, this would be invoked from
+        SQLCompiler.visit_select(). For the special FromStatement object used
+        by Query to indicate "Query.from_statement()", this is called by
+        FromStatement._compiler_dispatch() that would be called by
+        SQLCompiler.process().
 
-    @classmethod
-    def _create_for_statement_or_query(
-        cls, statement_container, for_statement=False,
-    ):
+        """
         raise NotImplementedError()
 
     @classmethod
@@ -266,21 +262,20 @@ class ORMCompileState(CompileState):
             and ext_info.mapper.persist_selectable
             not in self._polymorphic_adapters
         ):
-            self._mapper_loads_polymorphically_with(
-                ext_info.mapper,
-                sql_util.ColumnAdapter(
-                    selectable, ext_info.mapper._equivalent_columns
-                ),
-            )
+            for mp in ext_info.mapper.iterate_to_root():
+                self._mapper_loads_polymorphically_with(
+                    mp,
+                    sql_util.ColumnAdapter(selectable, mp._equivalent_columns),
+                )
 
     def _mapper_loads_polymorphically_with(self, mapper, adapter):
         for m2 in mapper._with_polymorphic_mappers or [mapper]:
             self._polymorphic_adapters[m2] = adapter
-            for m in m2.iterate_to_root():
+            for m in m2.iterate_to_root():  # TODO: redundant ?
                 self._polymorphic_adapters[m.local_table] = adapter
 
 
-@sql.base.CompileState.plugin_for("orm", "grouping")
+@sql.base.CompileState.plugin_for("orm", "orm_from_statement")
 class ORMFromStatementCompileState(ORMCompileState):
     _aliased_generations = util.immutabledict()
     _from_obj_alias = None
@@ -294,31 +289,23 @@ class ORMFromStatementCompileState(ORMCompileState):
 
     @classmethod
     def create_for_statement(cls, statement_container, compiler, **kw):
-        compiler._rewrites_selected_columns = True
-        toplevel = not compiler.stack
-        return cls._create_for_statement_or_query(
-            statement_container, toplevel
-        )
 
-    @classmethod
-    def _create_for_statement_or_query(
-        cls, statement_container, toplevel, for_statement=False,
-    ):
-        # from .query import FromStatement
-
-        # assert isinstance(statement_container, FromStatement)
+        if compiler is not None:
+            compiler._rewrites_selected_columns = True
+            toplevel = not compiler.stack
+        else:
+            toplevel = True
 
         self = cls.__new__(cls)
         self._primary_entity = None
 
-        self.use_orm_style = (
+        self.use_legacy_query_style = (
             statement_container.compile_options._use_legacy_query_style
         )
-        self.statement_container = self.query = statement_container
-        self.requested_statement = statement_container.element
+        self.statement_container = self.select_statement = statement_container
+        self.requested_statement = statement = statement_container.element
 
         self._entities = []
-        self._with_polymorphic_adapt_map = {}
         self._polymorphic_adapters = {}
         self._no_yield_pers = set()
 
@@ -349,12 +336,6 @@ class ORMFromStatementCompileState(ORMCompileState):
         self.create_eager_joins = []
         self._fallback_from_clauses = []
 
-        self._setup_for_statement()
-
-        return self
-
-    def _setup_for_statement(self):
-        statement = self.requested_statement
         if (
             isinstance(statement, expression.SelectBase)
             and not statement._is_textual
@@ -392,6 +373,8 @@ class ORMFromStatementCompileState(ORMCompileState):
             #            for entity in self._entities:
             #                entity.setup_compile_state(self)
 
+        return self
+
     def _adapt_col_list(self, cols, current_adapter):
         return cols
 
@@ -401,7 +384,8 @@ class ORMFromStatementCompileState(ORMCompileState):
 
 @sql.base.CompileState.plugin_for("orm", "select")
 class ORMSelectCompileState(ORMCompileState, SelectState):
-    _joinpath = _joinpoint = util.immutabledict()
+    _joinpath = _joinpoint = _EMPTY_DICT
+
     _from_obj_alias = None
     _has_mapper_entities = False
 
@@ -417,77 +401,71 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
 
     @classmethod
     def create_for_statement(cls, statement, compiler, **kw):
+        """compiler hook, we arrive here from compiler.visit_select() only."""
+
         if not statement._is_future:
             return SelectState(statement, compiler, **kw)
 
-        toplevel = not compiler.stack
+        if compiler is not None:
+            toplevel = not compiler.stack
+            compiler._rewrites_selected_columns = True
+        else:
+            toplevel = True
 
-        compiler._rewrites_selected_columns = True
+        select_statement = statement
 
-        orm_state = cls._create_for_statement_or_query(
-            statement, for_statement=True, toplevel=toplevel
-        )
-        SelectState.__init__(orm_state, orm_state.statement, compiler, **kw)
-        return orm_state
-
-    @classmethod
-    def _create_for_statement_or_query(
-        cls, query, toplevel, for_statement=False, _entities_only=False
-    ):
-        assert isinstance(query, future.Select)
-
-        query.compile_options = cls.default_compile_options.merge(
-            query.compile_options
+        # if we are a select() that was never a legacy Query, we won't
+        # have ORM level compile options.
+        statement.compile_options = cls.default_compile_options.safe_merge(
+            statement.compile_options
         )
 
         self = cls.__new__(cls)
 
-        self._primary_entity = None
-
-        self.query = query
-        self.use_orm_style = query.compile_options._use_legacy_query_style
+        self.select_statement = select_statement
 
-        self.select_statement = select_statement = query
+        # indicates this select() came from Query.statement
+        self.for_statement = (
+            for_statement
+        ) = select_statement.compile_options._for_statement
 
-        if not hasattr(select_statement.compile_options, "_orm_results"):
-            select_statement.compile_options = cls.default_compile_options
-            select_statement.compile_options += {"_orm_results": for_statement}
-        else:
-            for_statement = not select_statement.compile_options._orm_results
+        if not for_statement and not toplevel:
+            # for subqueries, turn off eagerloads.
+            # if "for_statement" mode is set, Query.subquery()
+            # would have set this flag to False already if that's what's
+            # desired
+            select_statement.compile_options += {
+                "_enable_eagerloads": False,
+            }
 
-        self.query = query
+        # generally if we are from Query or directly from a select()
+        self.use_legacy_query_style = (
+            select_statement.compile_options._use_legacy_query_style
+        )
 
         self._entities = []
-
+        self._primary_entity = None
         self._aliased_generations = {}
         self._polymorphic_adapters = {}
         self._no_yield_pers = set()
 
         # legacy: only for query.with_polymorphic()
-        self._with_polymorphic_adapt_map = wpam = dict(
-            select_statement.compile_options._with_polymorphic_adapt_map
-        )
-        if wpam:
+        if select_statement.compile_options._with_polymorphic_adapt_map:
+            self._with_polymorphic_adapt_map = dict(
+                select_statement.compile_options._with_polymorphic_adapt_map
+            )
             self._setup_with_polymorphics()
 
         _QueryEntity.to_compile_state(self, select_statement._raw_columns)
 
-        if _entities_only:
-            return self
-
-        self.compile_options = query.compile_options
-
-        # TODO: the name of this flag "for_statement" has to change,
-        # as it is difficult to distinguish from the "query._statement" use
-        # case which is something totally different
-        self.for_statement = for_statement
+        self.compile_options = select_statement.compile_options
 
         # determine label style.   we can make different decisions here.
         # at the moment, trying to see if we can always use DISAMBIGUATE_ONLY
         # rather than LABEL_STYLE_NONE, and if we can use disambiguate style
         # for new style ORM selects too.
         if self.select_statement._label_style is LABEL_STYLE_NONE:
-            if self.use_orm_style and not for_statement:
+            if self.use_legacy_query_style and not self.for_statement:
                 self.label_style = LABEL_STYLE_TABLENAME_PLUS_COL
             else:
                 self.label_style = LABEL_STYLE_DISAMBIGUATE_ONLY
@@ -522,129 +500,16 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
             info.selectable for info in select_statement._from_obj
         ]
 
+        # this is a fairly arbitrary break into a second method,
+        # so it might be nicer to break up create_for_statement()
+        # and _setup_for_generate into three or four logical sections
         self._setup_for_generate()
 
-        return self
-
-    @classmethod
-    def _create_entities_collection(cls, query):
-        """Creates a partial ORMSelectCompileState that includes
-        the full collection of _MapperEntity and other _QueryEntity objects.
-
-        Supports a few remaining use cases that are pre-compilation
-        but still need to gather some of the column  / adaption information.
-
-        """
-        self = cls.__new__(cls)
-
-        self._entities = []
-        self._primary_entity = None
-        self._aliased_generations = {}
-        self._polymorphic_adapters = {}
-
-        # legacy: only for query.with_polymorphic()
-        self._with_polymorphic_adapt_map = wpam = dict(
-            query.compile_options._with_polymorphic_adapt_map
-        )
-        if wpam:
-            self._setup_with_polymorphics()
+        if compiler is not None:
+            SelectState.__init__(self, self.statement, compiler, **kw)
 
-        _QueryEntity.to_compile_state(self, query._raw_columns)
         return self
 
-    @classmethod
-    def determine_last_joined_entity(cls, statement):
-        setup_joins = statement._setup_joins
-
-        if not setup_joins:
-            return None
-
-        (target, onclause, from_, flags) = setup_joins[-1]
-
-        if isinstance(target, interfaces.PropComparator):
-            return target.entity
-        else:
-            return target
-
-    def _setup_with_polymorphics(self):
-        # legacy: only for query.with_polymorphic()
-        for ext_info, wp in self._with_polymorphic_adapt_map.items():
-            self._mapper_loads_polymorphically_with(ext_info, wp._adapter)
-
-    def _set_select_from_alias(self):
-
-        query = self.select_statement  # query
-
-        assert self.compile_options._set_base_alias
-        assert len(query._from_obj) == 1
-
-        adapter = self._get_select_from_alias_from_obj(query._from_obj[0])
-        if adapter:
-            self.compile_options += {"_enable_single_crit": False}
-            self._from_obj_alias = adapter
-
-    def _get_select_from_alias_from_obj(self, from_obj):
-        info = from_obj
-
-        if "parententity" in info._annotations:
-            info = info._annotations["parententity"]
-
-        if hasattr(info, "mapper"):
-            if not info.is_aliased_class:
-                raise sa_exc.ArgumentError(
-                    "A selectable (FromClause) instance is "
-                    "expected when the base alias is being set."
-                )
-            else:
-                return info._adapter
-
-        elif isinstance(info.selectable, sql.selectable.AliasedReturnsRows):
-            equivs = self._all_equivs()
-            return sql_util.ColumnAdapter(info, equivs)
-        else:
-            return None
-
-    def _mapper_zero(self):
-        """return the Mapper associated with the first QueryEntity."""
-        return self._entities[0].mapper
-
-    def _entity_zero(self):
-        """Return the 'entity' (mapper or AliasedClass) associated
-        with the first QueryEntity, or alternatively the 'select from'
-        entity if specified."""
-
-        for ent in self.from_clauses:
-            if "parententity" in ent._annotations:
-                return ent._annotations["parententity"]
-        for qent in self._entities:
-            if qent.entity_zero:
-                return qent.entity_zero
-
-        return None
-
-    def _only_full_mapper_zero(self, methname):
-        if self._entities != [self._primary_entity]:
-            raise sa_exc.InvalidRequestError(
-                "%s() can only be used against "
-                "a single mapped class." % methname
-            )
-        return self._primary_entity.entity_zero
-
-    def _only_entity_zero(self, rationale=None):
-        if len(self._entities) > 1:
-            raise sa_exc.InvalidRequestError(
-                rationale
-                or "This operation requires a Query "
-                "against a single mapper."
-            )
-        return self._entity_zero()
-
-    def _all_equivs(self):
-        equivs = {}
-        for ent in self._mapper_entities:
-            equivs.update(ent.mapper._equivalent_columns)
-        return equivs
-
     def _setup_for_generate(self):
         query = self.select_statement
 
@@ -772,6 +637,140 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
                     {"deepentity": ezero}
                 )
 
+    @classmethod
+    def _create_entities_collection(cls, query):
+        """Creates a partial ORMSelectCompileState that includes
+        the full collection of _MapperEntity and other _QueryEntity objects.
+
+        Supports a few remaining use cases that are pre-compilation
+        but still need to gather some of the column  / adaption information.
+
+        """
+        self = cls.__new__(cls)
+
+        self._entities = []
+        self._primary_entity = None
+        self._aliased_generations = {}
+        self._polymorphic_adapters = {}
+
+        # legacy: only for query.with_polymorphic()
+        if query.compile_options._with_polymorphic_adapt_map:
+            self._with_polymorphic_adapt_map = dict(
+                query.compile_options._with_polymorphic_adapt_map
+            )
+            self._setup_with_polymorphics()
+
+        _QueryEntity.to_compile_state(self, query._raw_columns)
+        return self
+
+    @classmethod
+    def determine_last_joined_entity(cls, statement):
+        setup_joins = statement._setup_joins
+
+        if not setup_joins:
+            return None
+
+        (target, onclause, from_, flags) = setup_joins[-1]
+
+        if isinstance(target, interfaces.PropComparator):
+            return target.entity
+        else:
+            return target
+
+    @classmethod
+    def exported_columns_iterator(cls, statement):
+        for element in statement._raw_columns:
+            if (
+                element.is_selectable
+                and "entity_namespace" in element._annotations
+            ):
+                for elem in _select_iterables(
+                    element._annotations["entity_namespace"].columns
+                ):
+                    yield elem
+            else:
+                for elem in _select_iterables([element]):
+                    yield elem
+
+    def _setup_with_polymorphics(self):
+        # legacy: only for query.with_polymorphic()
+        for ext_info, wp in self._with_polymorphic_adapt_map.items():
+            self._mapper_loads_polymorphically_with(ext_info, wp._adapter)
+
+    def _set_select_from_alias(self):
+
+        query = self.select_statement  # query
+
+        assert self.compile_options._set_base_alias
+        assert len(query._from_obj) == 1
+
+        adapter = self._get_select_from_alias_from_obj(query._from_obj[0])
+        if adapter:
+            self.compile_options += {"_enable_single_crit": False}
+            self._from_obj_alias = adapter
+
+    def _get_select_from_alias_from_obj(self, from_obj):
+        info = from_obj
+
+        if "parententity" in info._annotations:
+            info = info._annotations["parententity"]
+
+        if hasattr(info, "mapper"):
+            if not info.is_aliased_class:
+                raise sa_exc.ArgumentError(
+                    "A selectable (FromClause) instance is "
+                    "expected when the base alias is being set."
+                )
+            else:
+                return info._adapter
+
+        elif isinstance(info.selectable, sql.selectable.AliasedReturnsRows):
+            equivs = self._all_equivs()
+            return sql_util.ColumnAdapter(info, equivs)
+        else:
+            return None
+
+    def _mapper_zero(self):
+        """return the Mapper associated with the first QueryEntity."""
+        return self._entities[0].mapper
+
+    def _entity_zero(self):
+        """Return the 'entity' (mapper or AliasedClass) associated
+        with the first QueryEntity, or alternatively the 'select from'
+        entity if specified."""
+
+        for ent in self.from_clauses:
+            if "parententity" in ent._annotations:
+                return ent._annotations["parententity"]
+        for qent in self._entities:
+            if qent.entity_zero:
+                return qent.entity_zero
+
+        return None
+
+    def _only_full_mapper_zero(self, methname):
+        if self._entities != [self._primary_entity]:
+            raise sa_exc.InvalidRequestError(
+                "%s() can only be used against "
+                "a single mapped class." % methname
+            )
+        return self._primary_entity.entity_zero
+
+    def _only_entity_zero(self, rationale=None):
+        if len(self._entities) > 1:
+            raise sa_exc.InvalidRequestError(
+                rationale
+                or "This operation requires a Query "
+                "against a single mapper."
+            )
+        return self._entity_zero()
+
+    def _all_equivs(self):
+        equivs = {}
+        for ent in self._mapper_entities:
+            equivs.update(ent.mapper._equivalent_columns)
+        return equivs
+
     def _compound_eager_statement(self):
         # for eager joins present and LIMIT/OFFSET/DISTINCT,
         # wrap the query inside a select,
@@ -920,6 +919,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         statement = Select.__new__(Select)
         statement._raw_columns = raw_columns
         statement._from_obj = from_obj
+
         statement._label_style = label_style
 
         if where_criteria:
@@ -1653,31 +1653,10 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
                     "target."
                 )
 
-        aliased_entity = (
-            right_mapper
-            and not right_is_aliased
-            and (
-                # TODO: there is a reliance here on aliasing occurring
-                # when we join to a polymorphic mapper that doesn't actually
-                # need aliasing.  When this condition is present, we should
-                # be able to say mapper_loads_polymorphically_with()
-                # and render the straight polymorphic selectable.  this
-                # does not appear to be possible at the moment as the
-                # adapter no longer takes place on the rest of the query
-                # and it's not clear where that's failing to happen.
-                (
-                    right_mapper.with_polymorphic
-                    and isinstance(
-                        right_mapper._with_polymorphic_selectable,
-                        expression.AliasedReturnsRows,
-                    )
-                )
-                or overlap
-                # test for overlap:
-                # orm/inheritance/relationships.py
-                # SelfReferentialM2MTest
-            )
-        )
+        # test for overlap:
+        # orm/inheritance/relationships.py
+        # SelfReferentialM2MTest
+        aliased_entity = right_mapper and not right_is_aliased and overlap
 
         if not need_adapter and (create_aliases or aliased_entity):
             # there are a few places in the ORM that automatic aliasing
@@ -1707,7 +1686,30 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
                 self._aliased_generations[aliased_generation] = (
                     adapter,
                 ) + self._aliased_generations.get(aliased_generation, ())
-
+        elif (
+            not r_info.is_clause_element
+            and not right_is_aliased
+            and right_mapper.with_polymorphic
+            and isinstance(
+                right_mapper._with_polymorphic_selectable,
+                expression.AliasedReturnsRows,
+            )
+        ):
+            # for the case where the target mapper has a with_polymorphic
+            # set up, ensure an adapter is set up for criteria that works
+            # against this mapper.  Previously, this logic used to
+            # use the "create_aliases or aliased_entity" case to generate
+            # an aliased() object, but this creates an alias that isn't
+            # strictly necessary.
+            # see test/orm/test_core_compilation.py
+            # ::RelNaturalAliasedJoinsTest::test_straight
+            # and similar
+            self._mapper_loads_polymorphically_with(
+                right_mapper,
+                sql_util.ColumnAdapter(
+                    right_mapper.selectable, right_mapper._equivalent_columns,
+                ),
+            )
         # if the onclause is a ClauseElement, adapt it with any
         # adapters that are in place right now
         if isinstance(onclause, expression.ClauseElement):
@@ -1755,8 +1757,8 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
             "offset_clause": self.select_statement._offset_clause,
             "distinct": self.distinct,
             "distinct_on": self.distinct_on,
-            "prefixes": self.query._prefixes,
-            "suffixes": self.query._suffixes,
+            "prefixes": self.select_statement._prefixes,
+            "suffixes": self.select_statement._suffixes,
             "group_by": self.group_by or None,
         }
 
@@ -2036,7 +2038,14 @@ class _MapperEntity(_QueryEntity):
             self._with_polymorphic_mappers = ext_info.with_polymorphic_mappers
             self._polymorphic_discriminator = ext_info.polymorphic_on
 
-            if mapper.with_polymorphic or mapper._requires_row_aliasing:
+            if (
+                mapper.with_polymorphic
+                # controversy - only if inheriting mapper is also
+                # polymorphic?
+                # or (mapper.inherits and mapper.inherits.with_polymorphic)
+                or mapper.inherits
+                or mapper._requires_row_aliasing
+            ):
                 compile_state._create_with_polymorphic_adapter(
                     ext_info, self.selectable
                 )
@@ -2361,7 +2370,7 @@ class _ORMColumnEntity(_ColumnEntity):
 
         _entity._post_inspect
         self.entity_zero = self.entity_zero_or_selectable = ezero = _entity
-        self.mapper = _entity.mapper
+        self.mapper = mapper = _entity.mapper
 
         if parent_bundle:
             parent_bundle._entities.append(self)
@@ -2373,7 +2382,11 @@ class _ORMColumnEntity(_ColumnEntity):
 
         self._extra_entities = (self.expr, self.column)
 
-        if self.mapper.with_polymorphic:
+        if (
+            mapper.with_polymorphic
+            or mapper.inherits
+            or mapper._requires_row_aliasing
+        ):
             compile_state._create_with_polymorphic_adapter(
                 ezero, ezero.selectable
             )
@@ -2414,6 +2427,7 @@ class _ORMColumnEntity(_ColumnEntity):
             column = current_adapter(self.column, False)
         else:
             column = self.column
+
         ezero = self.entity_zero
 
         single_table_crit = self.mapper._single_table_criterion
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 88d01eb0f..424ed5dfe 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -345,7 +345,7 @@ def load_on_pk_identity(
     if load_options is None:
         load_options = QueryContext.default_load_options
 
-    compile_options = ORMCompileState.default_compile_options.merge(
+    compile_options = ORMCompileState.default_compile_options.safe_merge(
         q.compile_options
     )
 
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 7bfe70c36..4166e6d2a 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -1743,6 +1743,11 @@ class Mapper(
                     or prop.columns[0] is self.polymorphic_on
                 )
 
+            if isinstance(col, expression.Label):
+                # new in 1.4, get column property against expressions
+                # to be addressable in subqueries
+                col.key = col._key_label = key
+
             self.columns.add(col, key)
             for col in prop.columns + prop._orig_columns:
                 for col in col.proxy_set:
@@ -2282,6 +2287,29 @@ class Mapper(
             )
         )
 
+    def _columns_plus_keys(self, polymorphic_mappers=()):
+        if polymorphic_mappers:
+            poly_properties = self._iterate_polymorphic_properties(
+                polymorphic_mappers
+            )
+        else:
+            poly_properties = self._polymorphic_properties
+
+        return [
+            (prop.key, prop.columns[0])
+            for prop in poly_properties
+            if isinstance(prop, properties.ColumnProperty)
+        ]
+
+    @HasMemoized.memoized_attribute
+    def _polymorphic_adapter(self):
+        if self.with_polymorphic:
+            return sql_util.ColumnAdapter(
+                self.selectable, equivalents=self._equivalent_columns
+            )
+        else:
+            return None
+
     def _iterate_polymorphic_properties(self, mappers=None):
         """Return an iterator of MapperProperty objects which will render into
         a SELECT."""
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index 4cf501e3f..02f0752a5 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -264,6 +264,7 @@ class ColumnProperty(StrategizedProperty):
 
     def do_init(self):
         super(ColumnProperty, self).do_init()
+
         if len(self.columns) > 1 and set(self.parent.primary_key).issuperset(
             self.columns
         ):
@@ -339,28 +340,51 @@ class ColumnProperty(StrategizedProperty):
 
         __slots__ = "__clause_element__", "info", "expressions"
 
+        def _orm_annotate_column(self, column):
+            """annotate and possibly adapt a column to be returned
+            as the mapped-attribute exposed version of the column.
+
+            The column in this context needs to act as much like the
+            column in an ORM mapped context as possible, so includes
+            annotations to give hints to various ORM functions as to
+            the source entity of this column.   It also adapts it
+            to the mapper's with_polymorphic selectable if one is
+            present.
+
+            """
+
+            pe = self._parententity
+            annotations = {
+                "entity_namespace": pe,
+                "parententity": pe,
+                "parentmapper": pe,
+                "orm_key": self.prop.key,
+            }
+
+            col = column
+
+            # for a mapper with polymorphic_on and an adapter, return
+            # the column against the polymorphic selectable.
+            # see also orm.util._orm_downgrade_polymorphic_columns
+            # for the reverse operation.
+            if self._parentmapper._polymorphic_adapter:
+                mapper_local_col = col
+                col = self._parentmapper._polymorphic_adapter.traverse(col)
+
+                # this is a clue to the ORM Query etc. that this column
+                # was adapted to the mapper's polymorphic_adapter.  the
+                # ORM uses this hint to know which column its adapting.
+                annotations["adapt_column"] = mapper_local_col
+
+            return col._annotate(annotations)._set_propagate_attrs(
+                {"compile_state_plugin": "orm", "plugin_subject": pe}
+            )
+
         def _memoized_method___clause_element__(self):
             if self.adapter:
                 return self.adapter(self.prop.columns[0], self.prop.key)
             else:
-                pe = self._parententity
-                # no adapter, so we aren't aliased
-                # assert self._parententity is self._parentmapper
-                return (
-                    self.prop.columns[0]
-                    ._annotate(
-                        {
-                            "entity_namespace": pe,
-                            "parententity": pe,
-                            "parentmapper": pe,
-                            "orm_key": self.prop.key,
-                            "compile_state_plugin": "orm",
-                        }
-                    )
-                    ._set_propagate_attrs(
-                        {"compile_state_plugin": "orm", "plugin_subject": pe}
-                    )
-                )
+                return self._orm_annotate_column(self.prop.columns[0])
 
         def _memoized_attr_info(self):
             """The .info dictionary for this attribute."""
@@ -384,23 +408,8 @@ class ColumnProperty(StrategizedProperty):
                     for col in self.prop.columns
                 ]
             else:
-                # no adapter, so we aren't aliased
-                # assert self._parententity is self._parentmapper
                 return [
-                    col._annotate(
-                        {
-                            "parententity": self._parententity,
-                            "parentmapper": self._parententity,
-                            "orm_key": self.prop.key,
-                            "compile_state_plugin": "orm",
-                        }
-                    )._set_propagate_attrs(
-                        {
-                            "compile_state_plugin": "orm",
-                            "plugin_subject": self._parententity,
-                        }
-                    )
-                    for col in self.prop.columns
+                    self._orm_annotate_column(col) for col in self.prop.columns
                 ]
 
         def _fallback_getattr(self, key):
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 97a81e30f..5137f9b1d 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -360,7 +360,7 @@ class Query(
         ):
             # if we don't have legacy top level aliasing features in use
             # then convert to a future select() directly
-            stmt = self._statement_20()
+            stmt = self._statement_20(for_statement=True)
         else:
             stmt = self._compile_state(for_statement=True).statement
 
@@ -371,7 +371,24 @@ class Query(
 
         return stmt
 
-    def _statement_20(self, orm_results=False):
+    def _final_statement(self, legacy_query_style=True):
+        """Return the 'final' SELECT statement for this :class:`.Query`.
+
+        This is the Core-only select() that will be rendered by a complete
+        compilation of this query, and is what .statement used to return
+        in 1.3.
+
+        This method creates a complete compile state so is fairly expensive.
+
+        """
+
+        q = self._clone()
+
+        return q._compile_state(
+            use_legacy_query_style=legacy_query_style
+        ).statement
+
+    def _statement_20(self, for_statement=False, use_legacy_query_style=True):
         # TODO: this event needs to be deprecated, as it currently applies
         # only to ORM query and occurs at this spot that is now more
         # or less an artificial spot
@@ -384,7 +401,10 @@ class Query(
                         self.compile_options += {"_bake_ok": False}
 
         compile_options = self.compile_options
-        compile_options += {"_use_legacy_query_style": True}
+        compile_options += {
+            "_for_statement": for_statement,
+            "_use_legacy_query_style": use_legacy_query_style,
+        }
 
         if self._statement is not None:
             stmt = FromStatement(self._raw_columns, self._statement)
@@ -404,13 +424,16 @@ class Query(
                 compile_options=compile_options,
             )
 
-        if not orm_results:
-            stmt.compile_options += {"_orm_results": False}
-
         stmt._propagate_attrs = self._propagate_attrs
         return stmt
 
-    def subquery(self, name=None, with_labels=False, reduce_columns=False):
+    def subquery(
+        self,
+        name=None,
+        with_labels=False,
+        reduce_columns=False,
+        _legacy_core_statement=False,
+    ):
         """return the full SELECT statement represented by
         this :class:`_query.Query`, embedded within an
         :class:`_expression.Alias`.
@@ -436,7 +459,11 @@ class Query(
         q = self.enable_eagerloads(False)
         if with_labels:
             q = q.with_labels()
-        q = q.statement
+
+        if _legacy_core_statement:
+            q = q._compile_state(for_statement=True).statement
+        else:
+            q = q.statement
 
         if reduce_columns:
             q = q.reduce_columns()
@@ -943,7 +970,7 @@ class Query(
         # tablename_colname style is used which at the moment is asserted
         # in a lot of unit tests :)
 
-        statement = self._statement_20(orm_results=True).apply_labels()
+        statement = self._statement_20().apply_labels()
         return db_load_fn(
             self.session,
             statement,
@@ -1328,13 +1355,13 @@ class Query(
             self.with_labels()
             .enable_eagerloads(False)
             .correlate(None)
-            .subquery()
+            .subquery(_legacy_core_statement=True)
             ._anonymous_fromclause()
         )
 
         parententity = self._raw_columns[0]._annotations.get("parententity")
         if parententity:
-            ac = aliased(parententity, alias=fromclause)
+            ac = aliased(parententity.mapper, alias=fromclause)
             q = self._from_selectable(ac)
         else:
             q = self._from_selectable(fromclause)
@@ -2782,7 +2809,7 @@ class Query(
     def _iter(self):
         # new style execution.
         params = self.load_options._params
-        statement = self._statement_20(orm_results=True)
+        statement = self._statement_20()
         result = self.session.execute(
             statement,
             params,
@@ -2808,7 +2835,7 @@ class Query(
         )
 
     def __str__(self):
-        statement = self._statement_20(orm_results=True)
+        statement = self._statement_20()
 
         try:
             bind = (
@@ -2879,9 +2906,8 @@ class Query(
                 "for linking ORM results to arbitrary select constructs.",
                 version="1.4",
             )
-            compile_state = ORMCompileState._create_for_legacy_query(
-                self, toplevel=True
-            )
+            compile_state = self._compile_state(for_statement=False)
+
             context = QueryContext(
                 compile_state, self.session, self.load_options
             )
@@ -3294,10 +3320,35 @@ class Query(
         return update_op.rowcount
 
     def _compile_state(self, for_statement=False, **kw):
-        return ORMCompileState._create_for_legacy_query(
-            self, toplevel=True, for_statement=for_statement, **kw
+        """Create an out-of-compiler ORMCompileState object.
+
+        The ORMCompileState object is normally created directly as a result
+        of the SQLCompiler.process() method being handed a Select()
+        or FromStatement() object that uses the "orm" plugin.   This method
+        provides a means of creating this ORMCompileState object directly
+        without using the compiler.
+
+        This method is used only for deprecated cases, which include
+        the .from_self() method for a Query that has multiple levels
+        of .from_self() in use, as well as the instances() method.  It is
+        also used within the test suite to generate ORMCompileState objects
+        for test purposes.
+
+        """
+
+        stmt = self._statement_20(for_statement=for_statement, **kw)
+        assert for_statement == stmt.compile_options._for_statement
+
+        # this chooses between ORMFromStatementCompileState and
+        # ORMSelectCompileState.  We could also base this on
+        # query._statement is not None as we have the ORM Query here
+        # however this is the more general path.
+        compile_state_cls = ORMCompileState._get_plugin_class_for_plugin(
+            stmt, "orm"
         )
 
+        return compile_state_cls.create_for_statement(stmt, None)
+
     def _compile_context(self, for_statement=False):
         compile_state = self._compile_state(for_statement=for_statement)
         context = QueryContext(compile_state, self.session, self.load_options)
@@ -3311,6 +3362,8 @@ class FromStatement(SelectStatementGrouping, Executable):
 
     """
 
+    __visit_name__ = "orm_from_statement"
+
     compile_options = ORMFromStatementCompileState.default_compile_options
 
     _compile_state_factory = ORMFromStatementCompileState.create_for_statement
@@ -3329,6 +3382,14 @@ class FromStatement(SelectStatementGrouping, Executable):
         super(FromStatement, self).__init__(element)
 
     def _compiler_dispatch(self, compiler, **kw):
+
+        """provide a fixed _compiler_dispatch method.
+
+        This is roughly similar to using the sqlalchemy.ext.compiler
+        ``@compiles`` extension.
+
+        """
+
         compile_state = self._compile_state_factory(self, compiler, **kw)
 
         toplevel = not compiler.stack
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index e82cd174f..683f2b978 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -1170,9 +1170,9 @@ class RelationshipProperty(StrategizedProperty):
         def __clause_element__(self):
             adapt_from = self._source_selectable()
             if self._of_type:
-                of_type_mapper = inspect(self._of_type).mapper
+                of_type_entity = inspect(self._of_type)
             else:
-                of_type_mapper = None
+                of_type_entity = None
 
             (
                 pj,
@@ -1184,7 +1184,7 @@ class RelationshipProperty(StrategizedProperty):
             ) = self.property._create_joins(
                 source_selectable=adapt_from,
                 source_polymorphic=True,
-                of_type_mapper=of_type_mapper,
+                of_type_entity=of_type_entity,
                 alias_secondary=True,
             )
             if sj is not None:
@@ -1311,7 +1311,6 @@ class RelationshipProperty(StrategizedProperty):
                 secondary,
                 target_adapter,
             ) = self.property._create_joins(
-                dest_polymorphic=True,
                 dest_selectable=to_selectable,
                 source_selectable=source_selectable,
             )
@@ -2424,9 +2423,8 @@ class RelationshipProperty(StrategizedProperty):
         self,
         source_polymorphic=False,
         source_selectable=None,
-        dest_polymorphic=False,
         dest_selectable=None,
-        of_type_mapper=None,
+        of_type_entity=None,
         alias_secondary=False,
     ):
 
@@ -2439,9 +2437,17 @@ class RelationshipProperty(StrategizedProperty):
             if source_polymorphic and self.parent.with_polymorphic:
                 source_selectable = self.parent._with_polymorphic_selectable
 
+        if of_type_entity:
+            dest_mapper = of_type_entity.mapper
+            if dest_selectable is None:
+                dest_selectable = of_type_entity.selectable
+                aliased = True
+        else:
+            dest_mapper = self.mapper
+
         if dest_selectable is None:
             dest_selectable = self.entity.selectable
-            if dest_polymorphic and self.mapper.with_polymorphic:
+            if self.mapper.with_polymorphic:
                 aliased = True
 
             if self._is_self_referential and source_selectable is None:
@@ -2453,8 +2459,6 @@ class RelationshipProperty(StrategizedProperty):
         ):
             aliased = True
 
-        dest_mapper = of_type_mapper or self.mapper
-
         single_crit = dest_mapper._single_table_criterion
         aliased = aliased or (
             source_selectable is not None
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 626018997..2b8c384c9 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -1143,7 +1143,7 @@ class SubqueryLoader(PostLoader):
         ) = self._get_leftmost(subq_path)
 
         orig_query = compile_state.attributes.get(
-            ("orig_query", SubqueryLoader), compile_state.query
+            ("orig_query", SubqueryLoader), compile_state.select_statement
         )
 
         # generate a new Query from the original, then
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index ce37d962e..85f4f85d1 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -466,7 +466,7 @@ class AliasedClass(object):
 
     def __init__(
         self,
-        cls,
+        mapped_class_or_ac,
         alias=None,
         name=None,
         flat=False,
@@ -478,7 +478,9 @@ class AliasedClass(object):
         use_mapper_path=False,
         represents_outer_join=False,
     ):
-        mapper = _class_to_mapper(cls)
+        insp = inspection.inspect(mapped_class_or_ac)
+        mapper = insp.mapper
+
         if alias is None:
             alias = mapper._with_polymorphic_selectable._anonymous_fromclause(
                 name=name, flat=flat
@@ -486,7 +488,7 @@ class AliasedClass(object):
 
         self._aliased_insp = AliasedInsp(
             self,
-            mapper,
+            insp,
             alias,
             name,
             with_polymorphic_mappers
@@ -617,7 +619,7 @@ class AliasedInsp(
     def __init__(
         self,
         entity,
-        mapper,
+        inspected,
         selectable,
         name,
         with_polymorphic_mappers,
@@ -627,6 +629,10 @@ class AliasedInsp(
         adapt_on_names,
         represents_outer_join,
     ):
+
+        mapped_class_or_ac = inspected.entity
+        mapper = inspected.mapper
+
         self._weak_entity = weakref.ref(entity)
         self.mapper = mapper
         self.selectable = (
@@ -665,9 +671,12 @@ class AliasedInsp(
             adapt_on_names=adapt_on_names,
             anonymize_labels=True,
         )
+        if inspected.is_aliased_class:
+            self._adapter = inspected._adapter.wrap(self._adapter)
 
         self._adapt_on_names = adapt_on_names
-        self._target = mapper.class_
+        self._target = mapped_class_or_ac
+        # self._target = mapper.class_  # mapped_class_or_ac
 
     @property
     def entity(self):
@@ -795,6 +804,21 @@ class AliasedInsp(
     def _memoized_values(self):
         return {}
 
+    @util.memoized_property
+    def columns(self):
+        if self._is_with_polymorphic:
+            cols_plus_keys = self.mapper._columns_plus_keys(
+                [ent.mapper for ent in self._with_polymorphic_entities]
+            )
+        else:
+            cols_plus_keys = self.mapper._columns_plus_keys()
+
+        cols_plus_keys = [
+            (key, self._adapt_element(col)) for key, col in cols_plus_keys
+        ]
+
+        return ColumnCollection(cols_plus_keys)
+
     def _memo(self, key, callable_, *args, **kw):
         if key in self._memoized_values:
             return self._memoized_values[key]
@@ -1290,8 +1314,7 @@ class _ORMJoin(expression.Join):
                 source_selectable=adapt_from,
                 dest_selectable=adapt_to,
                 source_polymorphic=True,
-                dest_polymorphic=True,
-                of_type_mapper=right_info.mapper,
+                of_type_entity=right_info,
                 alias_secondary=True,
             )