- Added a new extension suite :mod:`sqlalchemy.ext.baked`. This

simple but unusual system allows for a dramatic savings in Python
overhead for the construction and processing of orm :class:`.Query`
objects, from query construction up through rendering of a string
SQL statement.
fixes #3054

8 files changed, 711 insertions, 128 deletions

diff --git a/lib/sqlalchemy/ext/__init__.py b/lib/sqlalchemy/ext/__init__.py
index ff95c07d6..60a17c65e 100644
--- a/lib/sqlalchemy/ext/__init__.py
+++ b/lib/sqlalchemy/ext/__init__.py
@@ -4,3 +4,8 @@
 #
 # This module is part of SQLAlchemy and is released under
 # the MIT License: http://www.opensource.org/licenses/mit-license.php
+
+from .. import util as _sa_util
+
+_sa_util.dependencies.resolve_all("sqlalchemy.ext")
+
diff --git a/lib/sqlalchemy/ext/baked.py b/lib/sqlalchemy/ext/baked.py
new file mode 100644
index 000000000..65d6a8603
--- /dev/null
+++ b/lib/sqlalchemy/ext/baked.py
@@ -0,0 +1,499 @@
+# sqlalchemy/ext/baked.py
+# Copyright (C) 2005-2015 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: http://www.opensource.org/licenses/mit-license.php
+"""Baked query extension.
+
+Provides a creational pattern for the :class:`.query.Query` object which
+allows the fully constructed object, Core select statement, and string
+compiled result to be fully cached.
+
+
+"""
+
+from ..orm.query import Query
+from ..orm import strategies, attributes, properties, \
+    strategy_options, util as orm_util, interfaces
+from .. import log as sqla_log
+from ..sql import util as sql_util
+from ..orm import exc as orm_exc
+from .. import exc as sa_exc
+from .. import util
+
+import copy
+import logging
+
+log = logging.getLogger(__name__)
+
+
+class BakedQuery(object):
+    """A builder object for :class:`.query.Query` objects."""
+
+    __slots__ = 'steps', '_bakery', '_cache_key', '_spoiled'
+
+    def __init__(self, bakery, initial_fn, args=()):
+        if args:
+            self._cache_key = tuple(args)
+        else:
+            self._cache_key = ()
+        self._update_cache_key(initial_fn)
+        self.steps = [initial_fn]
+        self._spoiled = False
+        self._bakery = bakery
+
+    @classmethod
+    def bakery(cls, size=200):
+        """Construct a new bakery."""
+
+        _bakery = util.LRUCache(size)
+
+        def call(initial_fn):
+            return cls(_bakery, initial_fn)
+
+        return call
+
+    def _clone(self):
+        b1 = BakedQuery.__new__(BakedQuery)
+        b1._cache_key = self._cache_key
+        b1.steps = list(self.steps)
+        b1._bakery = self._bakery
+        b1._spoiled = self._spoiled
+        return b1
+
+    def _update_cache_key(self, fn, args=()):
+        self._cache_key += (fn.__code__,) + args
+
+    def __iadd__(self, other):
+        if isinstance(other, tuple):
+            self.add_criteria(*other)
+        else:
+            self.add_criteria(other)
+        return self
+
+    def __add__(self, other):
+        if isinstance(other, tuple):
+            return self.with_criteria(*other)
+        else:
+            return self.with_criteria(other)
+
+    def add_criteria(self, fn, *args):
+        """Add a criteria function to this :class:`.BakedQuery`.
+
+        This is equivalent to using the ``+=`` operator to
+        modify a :class:`.BakedQuery` in-place.
+
+        """
+        self._update_cache_key(fn, args)
+        self.steps.append(fn)
+        return self
+
+    def with_criteria(self, fn, *args):
+        """Add a criteria function to a :class:`.BakedQuery` cloned from this one.
+
+        This is equivalent to using the ``+`` operator to
+        produce a new :class:`.BakedQuery` with modifications.
+
+        """
+        return self._clone().add_criteria(fn, *args)
+
+    def for_session(self, session):
+        """Return a :class:`.Result` object for this :class:`.BakedQuery`.
+
+        This is equivalent to calling the :class:`.BakedQuery` as a
+        Python callable, e.g. ``result = my_baked_query(session)``.
+
+        """
+        return Result(self, session)
+
+    def __call__(self, session):
+        return self.for_session(session)
+
+    def spoil(self, full=False):
+        """Cancel any query caching that will occur on this BakedQuery object.
+
+        The BakedQuery can continue to be used normally, however additional
+        creational functions will not be cached; they will be called
+        on every invocation.
+
+        This is to support the case where a particular step in constructing
+        a baked query disqualifies the query from being cacheable, such
+        as a variant that relies upon some uncacheable value.
+
+        :param full: if False, only functions added to this
+         :class:`.BakedQuery` object subsequent to the spoil step will be
+         non-cached; the state of the :class:`.BakedQuery` up until
+         this point will be pulled from the cache.   If True, then the
+         entire :class:`.Query` object is built from scratch each
+         time, with all creational functions being called on each
+         invocation.
+
+        """
+        if not full:
+            _spoil_point = self._clone()
+            _spoil_point._cache_key += ('_query_only', )
+            self.steps = [_spoil_point._retrieve_baked_query]
+        self._spoiled = True
+        return self
+
+    def _retrieve_baked_query(self, session):
+        query = self._bakery.get(self._cache_key, None)
+        if query is None:
+            query = self._as_query(session)
+            self._bakery[self._cache_key] = query.with_session(None)
+        return query.with_session(session)
+
+    def _bake(self, session):
+        query = self._as_query(session)
+
+        context = query._compile_context()
+        self._bake_subquery_loaders(session, context)
+        context.session = None
+        context.query = query = context.query.with_session(None)
+        query._execution_options = query._execution_options.union(
+            {"compiled_cache": self._bakery}
+        )
+        # we'll be holding onto the query for some of its state,
+        # so delete some compilation-use-only attributes that can take up
+        # space
+        for attr in (
+                '_correlate', '_from_obj', '_mapper_adapter_map',
+                '_joinpath', '_joinpoint'):
+            query.__dict__.pop(attr, None)
+        self._bakery[self._cache_key] = context
+        return context
+
+    def _as_query(self, session):
+        query = self.steps[0](session)
+
+        for step in self.steps[1:]:
+            query = step(query)
+        return query
+
+    def _bake_subquery_loaders(self, session, context):
+        """convert subquery eager loaders in the cache into baked queries.
+
+        For subquery eager loading to work, all we need here is that the
+        Query point to the correct session when it is run.  However, since
+        we are "baking" anyway, we may as well also turn the query into
+        a "baked" query so that we save on performance too.
+
+        """
+        context.attributes['baked_queries'] = baked_queries = []
+        for k, v in list(context.attributes.items()):
+            if isinstance(v, Query):
+                if 'subquery' in k:
+                    bk = BakedQuery(self._bakery, lambda *args: v)
+                    bk._cache_key = self._cache_key + k
+                    bk._bake(session)
+                    baked_queries.append((k, bk._cache_key, v))
+                del context.attributes[k]
+
+    def _unbake_subquery_loaders(self, session, context, params):
+        """Retrieve subquery eager loaders stored by _bake_subquery_loaders
+        and turn them back into Result objects that will iterate just
+        like a Query object.
+
+        """
+        for k, cache_key, query in context.attributes["baked_queries"]:
+            bk = BakedQuery(self._bakery, lambda sess: query.with_session(sess))
+            bk._cache_key = cache_key
+            context.attributes[k] = bk.for_session(session).params(**params)
+
+
+class Result(object):
+    """Invokes a :class:`.BakedQuery` against a :class:`.Session`.
+
+    The :class:`.Result` object is where the actual :class:`.query.Query`
+    object gets created, or retrieved from the cache,
+    against a target :class:`.Session`, and is then invoked for results.
+
+    """
+    __slots__ = 'bq', 'session', '_params'
+
+    def __init__(self, bq, session):
+        self.bq = bq
+        self.session = session
+        self._params = {}
+
+    def params(self, *args, **kw):
+        """Specify parameters to be replaced into the string SQL statement."""
+
+        if len(args) == 1:
+            kw.update(args[0])
+        elif len(args) > 0:
+            raise sa_exc.ArgumentError(
+                "params() takes zero or one positional argument, "
+                "which is a dictionary.")
+        self._params.update(kw)
+        return self
+
+    def _as_query(self):
+        return self.bq._as_query(self.session).params(self._params)
+
+    def __str__(self):
+        return str(self._as_query())
+
+    def __iter__(self):
+        bq = self.bq
+        if bq._spoiled:
+            return iter(self._as_query())
+
+        baked_context = bq._bakery.get(bq._cache_key, None)
+        if baked_context is None:
+            baked_context = bq._bake(self.session)
+
+        context = copy.copy(baked_context)
+        context.session = self.session
+        context.attributes = context.attributes.copy()
+
+        bq._unbake_subquery_loaders(self.session, context, self._params)
+
+        context.statement.use_labels = True
+        if context.autoflush and not context.populate_existing:
+            self.session._autoflush()
+        return context.query.params(self._params).\
+            with_session(self.session)._execute_and_instances(context)
+
+    def first(self):
+        """Return the first row.
+
+        Equivalent to :meth:`.Query.first`.
+
+        """
+        bq = self.bq.with_criteria(lambda q: q.slice(0, 1))
+        ret = list(bq.for_session(self.session).params(self._params))
+        if len(ret) > 0:
+            return ret[0]
+        else:
+            return None
+
+    def one(self):
+        """Return exactly one result or raise an exception.
+
+        Equivalent to :meth:`.Query.one`.
+
+        """
+        ret = list(self)
+
+        l = len(ret)
+        if l == 1:
+            return ret[0]
+        elif l == 0:
+            raise orm_exc.NoResultFound("No row was found for one()")
+        else:
+            raise orm_exc.MultipleResultsFound(
+                "Multiple rows were found for one()")
+
+    def all(self):
+        """Return all rows.
+
+        Equivalent to :meth:`.Query.all`.
+
+        """
+        return list(self)
+
+    def get(self, ident):
+        """Retrieve an object based on identity.
+
+        Equivalent to :meth:`.Query.get`.
+
+        """
+
+        query = self.bq.steps[0](self.session)
+        return query._get_impl(ident, self._load_on_ident)
+
+    def _load_on_ident(self, query, key):
+        """Load the given identity key from the database."""
+
+        ident = key[1]
+
+        mapper = query._mapper_zero()
+
+        _get_clause, _get_params = mapper._get_clause
+
+        def setup(query):
+            _lcl_get_clause = _get_clause
+            q = query._clone()
+            q._get_condition()
+            q._order_by = None
+
+            # None present in ident - turn those comparisons
+            # into "IS NULL"
+            if None in ident:
+                nones = set([
+                    _get_params[col].key for col, value in
+                    zip(mapper.primary_key, ident) if value is None
+                ])
+                _lcl_get_clause = sql_util.adapt_criterion_to_null(
+                    _lcl_get_clause, nones)
+
+            _lcl_get_clause = q._adapt_clause(_lcl_get_clause, True, False)
+            q._criterion = _lcl_get_clause
+            return q
+
+        # cache the query against a key that includes
+        # which positions in the primary key are NULL
+        # (remember, we can map to an OUTER JOIN)
+        bq = self.bq
+
+        bq = bq.with_criteria(setup, tuple(elem is None for elem in ident))
+
+        params = dict([
+            (_get_params[primary_key].key, id_val)
+            for id_val, primary_key in zip(ident, mapper.primary_key)
+        ])
+
+        result = list(bq.for_session(self.session).params(**params))
+        l = len(result)
+        if l > 1:
+            raise orm_exc.MultipleResultsFound()
+        elif l:
+            return result[0]
+        else:
+            return None
+
+
+def bake_lazy_loaders():
+    """Enable the use of baked queries for all lazyloaders systemwide.
+
+    This operation should be safe for all lazy loaders, and will reduce
+    Python overhead for these operations.
+
+    """
+    strategies.LazyLoader._strategy_keys[:] = []
+    BakedLazyLoader._strategy_keys[:] = []
+
+    properties.RelationshipProperty.strategy_for(
+        lazy="select")(BakedLazyLoader)
+    properties.RelationshipProperty.strategy_for(
+        lazy=True)(BakedLazyLoader)
+    properties.RelationshipProperty.strategy_for(
+        lazy="baked_select")(BakedLazyLoader)
+
+
+def unbake_lazy_loaders():
+    """Disable the use of baked queries for all lazyloaders systemwide.
+
+    This operation reverts the changes produced by :func:`.bake_lazy_loaders`.
+
+    """
+    strategies.LazyLoader._strategy_keys[:] = []
+    BakedLazyLoader._strategy_keys[:] = []
+
+    properties.RelationshipProperty.strategy_for(
+        lazy="select")(strategies.LazyLoader)
+    properties.RelationshipProperty.strategy_for(
+        lazy=True)(strategies.LazyLoader)
+    properties.RelationshipProperty.strategy_for(
+        lazy="baked_select")(BakedLazyLoader)
+    assert strategies.LazyLoader._strategy_keys
+
+
+@sqla_log.class_logger
+@properties.RelationshipProperty.strategy_for(lazy="baked_select")
+class BakedLazyLoader(strategies.LazyLoader):
+
+    def _emit_lazyload(self, session, state, ident_key, passive):
+        q = BakedQuery(
+            self.mapper._compiled_cache,
+            lambda session: session.query(self.mapper))
+        q.add_criteria(
+            lambda q: q._adapt_all_clauses()._with_invoke_all_eagers(False),
+            self.parent_property)
+
+        if not self.parent_property.bake_queries:
+            q.spoil(full=True)
+
+        if self.parent_property.secondary is not None:
+            q.add_criteria(
+                lambda q:
+                q.select_from(self.mapper, self.parent_property.secondary))
+
+        pending = not state.key
+
+        # don't autoflush on pending
+        if pending or passive & attributes.NO_AUTOFLUSH:
+            q.add_criteria(lambda q: q.autoflush(False))
+
+        if state.load_path:
+            q.spoil()
+            q.add_criteria(
+                lambda q:
+                q._with_current_path(state.load_path[self.parent_property]))
+
+        if state.load_options:
+            q.spoil()
+            q.add_criteria(
+                lambda q: q._conditional_options(*state.load_options))
+
+        if self.use_get:
+            return q(session)._load_on_ident(
+                session.query(self.mapper), ident_key)
+
+        if self.parent_property.order_by:
+            q.add_criteria(
+                lambda q:
+                q.order_by(*util.to_list(self.parent_property.order_by)))
+
+        for rev in self.parent_property._reverse_property:
+            # reverse props that are MANYTOONE are loading *this*
+            # object from get(), so don't need to eager out to those.
+            if rev.direction is interfaces.MANYTOONE and \
+                rev._use_get and \
+                    not isinstance(rev.strategy, strategies.LazyLoader):
+                q.add_criteria(
+                    lambda q:
+                    q.options(
+                        strategy_options.Load(
+                            rev.parent).baked_lazyload(rev.key)))
+
+        lazy_clause, params = self._generate_lazy_clause(state, passive)
+
+        if pending:
+            if orm_util._none_set.intersection(params.values()):
+                return None
+
+        q.add_criteria(lambda q: q.filter(lazy_clause))
+        result = q(session).params(**params).all()
+        if self.uselist:
+            return result
+        else:
+            l = len(result)
+            if l:
+                if l > 1:
+                    util.warn(
+                        "Multiple rows returned with "
+                        "uselist=False for lazily-loaded attribute '%s' "
+                        % self.parent_property)
+
+                return result[0]
+            else:
+                return None
+
+
+@strategy_options.loader_option()
+def baked_lazyload(loadopt, attr):
+    """Indicate that the given attribute should be loaded using "lazy"
+    loading with a "baked" query used in the load.
+
+    """
+    return loadopt.set_relationship_strategy(attr, {"lazy": "baked_select"})
+
+
+@baked_lazyload._add_unbound_fn
+def baked_lazyload(*keys):
+    return strategy_options._UnboundLoad._from_keys(
+        strategy_options._UnboundLoad.baked_lazyload, keys, False, {})
+
+
+@baked_lazyload._add_unbound_all_fn
+def baked_lazyload_all(*keys):
+    return strategy_options._UnboundLoad._from_keys(
+        strategy_options._UnboundLoad.baked_lazyload, keys, True, {})
+
+baked_lazyload = baked_lazyload._unbound_fn
+baked_lazyload_all = baked_lazyload_all._unbound_all_fn
+
+bakery = BakedQuery.bakery
diff --git a/lib/sqlalchemy/orm/base.py b/lib/sqlalchemy/orm/base.py
index 01981f26f..c259878f0 100644
--- a/lib/sqlalchemy/orm/base.py
+++ b/lib/sqlalchemy/orm/base.py
@@ -329,8 +329,7 @@ def _is_mapped_class(entity):
     return insp is not None and \
         not insp.is_clause_element and \
         (
-            insp.is_mapper
-            or insp.is_aliased_class
+            insp.is_mapper or insp.is_aliased_class
         )
 
 
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index 39bc53adb..6cc613baa 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -27,6 +27,7 @@ from .base import (ONETOMANY, MANYTOONE, MANYTOMANY,
 from .base import (InspectionAttr, InspectionAttr,
     InspectionAttrInfo, _MappedAttribute)
 import collections
+from .. import inspect
 
 # imported later
 MapperExtension = SessionExtension = AttributeExtension = None
@@ -333,11 +334,11 @@ class PropComparator(operators.ColumnOperators):
 
     """
 
-    __slots__ = 'prop', 'property', '_parentmapper', '_adapt_to_entity'
+    __slots__ = 'prop', 'property', '_parententity', '_adapt_to_entity'
 
     def __init__(self, prop, parentmapper, adapt_to_entity=None):
         self.prop = self.property = prop
-        self._parentmapper = parentmapper
+        self._parententity = parentmapper
         self._adapt_to_entity = adapt_to_entity
 
     def __clause_element__(self):
@@ -350,7 +351,13 @@ class PropComparator(operators.ColumnOperators):
         """Return a copy of this PropComparator which will use the given
         :class:`.AliasedInsp` to produce corresponding expressions.
         """
-        return self.__class__(self.prop, self._parentmapper, adapt_to_entity)
+        return self.__class__(self.prop, self._parententity, adapt_to_entity)
+
+    @property
+    def _parentmapper(self):
+        """legacy; this is renamed to _parententity to be
+        compatible with QueryableAttribute."""
+        return inspect(self._parententity).mapper
 
     @property
     def adapter(self):
@@ -523,7 +530,9 @@ class StrategizedProperty(MapperProperty):
     @classmethod
     def strategy_for(cls, **kw):
         def decorate(dec_cls):
-            if not hasattr(dec_cls, '_strategy_keys'):
+            # ensure each subclass of the strategy has its
+            # own _strategy_keys collection
+            if '_strategy_keys' not in dec_cls.__dict__:
                 dec_cls._strategy_keys = []
             key = tuple(sorted(kw.items()))
             cls._all_strategies[cls][key] = dec_cls
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index 238ac83a9..5694f7255 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -246,8 +246,8 @@ class ColumnProperty(StrategizedProperty):
                 return self.adapter(self.prop.columns[0])
             else:
                 return self.prop.columns[0]._annotate({
-                    "parententity": self._parentmapper,
-                    "parentmapper": self._parentmapper})
+                    "parententity": self._parententity,
+                    "parentmapper": self._parententity})
 
         def _memoized_attr_info(self):
             ce = self.__clause_element__()
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index c6fdf479e..05349cf0b 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -26,7 +26,7 @@ from . import (
     exc as orm_exc, loading
 )
 from .base import _entity_descriptor, _is_aliased_class, \
-    _is_mapped_class, _orm_columns, _generative
+    _is_mapped_class, _orm_columns, _generative, InspectionAttr
 from .path_registry import PathRegistry
 from .util import (
     AliasedClass, ORMAdapter, join as orm_join, with_parent, aliased
@@ -831,7 +831,9 @@ class Query(object):
         :return: The object instance, or ``None``.
 
         """
+        return self._get_impl(ident, loading.load_on_ident)
 
+    def _get_impl(self, ident, fallback_fn):
         # convert composite types to individual args
         if hasattr(ident, '__composite_values__'):
             ident = ident.__composite_values__()
@@ -862,7 +864,7 @@ class Query(object):
                     return None
                 return instance
 
-        return loading.load_on_ident(self, key)
+        return fallback_fn(self, key)
 
     @_generative()
     def correlate(self, *args):
@@ -3332,7 +3334,7 @@ class _MapperEntity(_QueryEntity):
 
 
 @inspection._self_inspects
-class Bundle(object):
+class Bundle(InspectionAttr):
     """A grouping of SQL expressions that are returned by a :class:`.Query`
     under one namespace.
 
@@ -3528,14 +3530,20 @@ class _ColumnEntity(_QueryEntity):
     def __init__(self, query, column, namespace=None):
         self.expr = column
         self.namespace = namespace
+        search_entities = True
 
         if isinstance(column, util.string_types):
             column = sql.literal_column(column)
             self._label_name = column.name
+            search_entities = False
+            _entity = None
         elif isinstance(column, (
             attributes.QueryableAttribute,
             interfaces.PropComparator
         )):
+            _entity = column._parententity
+            if _entity is not None:
+                search_entities = False
             self._label_name = column.key
             column = column._query_clause_element()
             if isinstance(column, Bundle):
@@ -3558,6 +3566,7 @@ class _ColumnEntity(_QueryEntity):
             )
         else:
             self._label_name = getattr(column, 'key', None)
+            search_entities = True
 
         self.type = type_ = column.type
         if type_.hashable:
@@ -3588,30 +3597,38 @@ class _ColumnEntity(_QueryEntity):
         # leaking out their entities into the main select construct
         self.actual_froms = actual_froms = set(column._from_objects)
 
-        all_elements = [
-            elem for elem in visitors.iterate(column, {})
-            if 'parententity' in elem._annotations
-        ]
-
-        self.entities = util.unique_list([
-            elem._annotations['parententity']
-            for elem in all_elements
-            if 'parententity' in elem._annotations
-        ])
-
-        self._from_entities = set([
-            elem._annotations['parententity']
-            for elem in all_elements
-            if 'parententity' in elem._annotations
-            and actual_froms.intersection(elem._from_objects)
-        ])
-
-        if self.entities:
-            self.entity_zero = self.entities[0]
-        elif self.namespace is not None:
-            self.entity_zero = self.namespace
+        if not search_entities:
+            self.entity_zero = _entity
+            if _entity:
+                self.entities = [_entity]
+            else:
+                self.entities = []
+            self._from_entities = set(self.entities)
         else:
-            self.entity_zero = None
+            all_elements = [
+                elem for elem in visitors.iterate(column, {})
+                if 'parententity' in elem._annotations
+            ]
+
+            self.entities = util.unique_list([
+                elem._annotations['parententity']
+                for elem in all_elements
+                if 'parententity' in elem._annotations
+            ])
+
+            self._from_entities = set([
+                elem._annotations['parententity']
+                for elem in all_elements
+                if 'parententity' in elem._annotations
+                and actual_froms.intersection(elem._from_objects)
+            ])
+
+            if self.entities:
+                self.entity_zero = self.entities[0]
+            elif self.namespace is not None:
+                self.entity_zero = self.namespace
+            else:
+                self.entity_zero = None
 
     supports_single_entity = False
 
@@ -3673,10 +3690,15 @@ class _ColumnEntity(_QueryEntity):
 
 
 class QueryContext(object):
-    multi_row_eager_loaders = False
-    adapter = None
-    froms = ()
-    for_update = None
+    __slots__ = (
+        'multi_row_eager_loaders', 'adapter', 'froms', 'for_update',
+        'query', 'session', 'autoflush', 'populate_existing',
+        'invoke_all_eagers', 'version_check', 'refresh_state',
+        'primary_columns', 'secondary_columns', 'eager_order_by',
+        'eager_joins', 'create_eager_joins', 'propagate_options',
+        'attributes', 'statement', 'from_clause', 'whereclause',
+        'order_by', 'labels', '_for_update_arg', 'runid', 'partials'
+    )
 
     def __init__(self, query):
 
@@ -3693,8 +3715,13 @@ class QueryContext(object):
             self.whereclause = query._criterion
             self.order_by = query._order_by
 
+        self.multi_row_eager_loaders = False
+        self.adapter = None
+        self.froms = ()
+        self.for_update = None
         self.query = query
         self.session = query.session
+        self.autoflush = query._autoflush
         self.populate_existing = query._populate_existing
         self.invoke_all_eagers = query._invoke_all_eagers
         self.version_check = query._version_check
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index afd524f7b..e36a644da 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -23,7 +23,7 @@ from . import attributes
 from ..sql.util import (
     ClauseAdapter,
     join_condition, _shallow_annotate, visit_binary_product,
-    _deep_deannotate, selectables_overlap
+    _deep_deannotate, selectables_overlap, adapt_criterion_to_null
 )
 from ..sql import operators, expression, visitors
 from .interfaces import (MANYTOMANY, MANYTOONE, ONETOMANY,
@@ -113,6 +113,7 @@ class RelationshipProperty(StrategizedProperty):
                  active_history=False,
                  cascade_backrefs=True,
                  load_on_pending=False,
+                 bake_queries=True,
                  strategy_class=None, _local_remote_pairs=None,
                  query_class=None,
                  info=None):
@@ -274,6 +275,15 @@ class RelationshipProperty(StrategizedProperty):
             :paramref:`~.relationship.backref` - alternative form
             of backref specification.
 
+        :param bake_queries:
+          Use the :class:`.BakedQuery` cache to cache queries used in lazy
+          loads.  True by default, as this typically improves performance
+          significantly.  Set to False to reduce ORM memory use, or
+          if unresolved stability issues are observed with the baked query
+          cache system.
+
+          .. versionadded:: 1.0.0
+
         :param cascade:
           a comma-separated list of cascade rules which determines how
           Session operations should be "cascaded" from parent to child.
@@ -802,6 +812,7 @@ class RelationshipProperty(StrategizedProperty):
         self.join_depth = join_depth
         self.local_remote_pairs = _local_remote_pairs
         self.extension = extension
+        self.bake_queries = bake_queries
         self.load_on_pending = load_on_pending
         self.comparator_factory = comparator_factory or \
             RelationshipProperty.Comparator
@@ -873,13 +884,13 @@ class RelationshipProperty(StrategizedProperty):
 
             """
             self.prop = prop
-            self._parentmapper = parentmapper
+            self._parententity = parentmapper
             self._adapt_to_entity = adapt_to_entity
             if of_type:
                 self._of_type = of_type
 
         def adapt_to_entity(self, adapt_to_entity):
-            return self.__class__(self.property, self._parentmapper,
+            return self.__class__(self.property, self._parententity,
                                   adapt_to_entity=adapt_to_entity,
                                   of_type=self._of_type)
 
@@ -931,7 +942,7 @@ class RelationshipProperty(StrategizedProperty):
             """
             return RelationshipProperty.Comparator(
                 self.property,
-                self._parentmapper,
+                self._parententity,
                 adapt_to_entity=self._adapt_to_entity,
                 of_type=cls)
 
@@ -1315,16 +1326,69 @@ class RelationshipProperty(StrategizedProperty):
         return self._optimized_compare(
             instance, value_is_parent=True, alias_secondary=alias_secondary)
 
-    def _optimized_compare(self, value, value_is_parent=False,
+    def _optimized_compare(self, state, value_is_parent=False,
                            adapt_source=None,
                            alias_secondary=True):
-        if value is not None:
-            value = attributes.instance_state(value)
-        return self._lazy_strategy.lazy_clause(
-            value,
-            reverse_direction=not value_is_parent,
-            alias_secondary=alias_secondary,
-            adapt_source=adapt_source)
+        if state is not None:
+            state = attributes.instance_state(state)
+
+        reverse_direction = not value_is_parent
+
+        if state is None:
+            return self._lazy_none_clause(
+                reverse_direction,
+                adapt_source=adapt_source)
+
+        if not reverse_direction:
+            criterion, bind_to_col = \
+                self._lazy_strategy._lazywhere, \
+                self._lazy_strategy._bind_to_col
+        else:
+            criterion, bind_to_col = \
+                self._lazy_strategy._rev_lazywhere, \
+                self._lazy_strategy._rev_bind_to_col
+
+        if reverse_direction:
+            mapper = self.mapper
+        else:
+            mapper = self.parent
+
+        dict_ = attributes.instance_dict(state.obj())
+
+        def visit_bindparam(bindparam):
+            if bindparam._identifying_key in bind_to_col:
+                bindparam.callable = \
+                    lambda: mapper._get_state_attr_by_column(
+                        state, dict_,
+                        bind_to_col[bindparam._identifying_key])
+
+        if self.secondary is not None and alias_secondary:
+            criterion = ClauseAdapter(
+                self.secondary.alias()).\
+                traverse(criterion)
+
+        criterion = visitors.cloned_traverse(
+            criterion, {}, {'bindparam': visit_bindparam})
+
+        if adapt_source:
+            criterion = adapt_source(criterion)
+        return criterion
+
+    def _lazy_none_clause(self, reverse_direction=False, adapt_source=None):
+        if not reverse_direction:
+            criterion, bind_to_col = \
+                self._lazy_strategy._lazywhere, \
+                self._lazy_strategy._bind_to_col
+        else:
+            criterion, bind_to_col = \
+                self._lazy_strategy._rev_lazywhere, \
+                self._lazy_strategy._rev_bind_to_col
+
+        criterion = adapt_criterion_to_null(criterion, bind_to_col)
+
+        if adapt_source:
+            criterion = adapt_source(criterion)
+        return criterion
 
     def __str__(self):
         return str(self.parent.class_.__name__) + "." + self.key
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 611635333..0b2672d66 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -353,7 +353,7 @@ class NoLoader(AbstractRelationshipLoader):
 @log.class_logger
 @properties.RelationshipProperty.strategy_for(lazy=True)
 @properties.RelationshipProperty.strategy_for(lazy="select")
-class LazyLoader(AbstractRelationshipLoader):
+class LazyLoader(AbstractRelationshipLoader, util.MemoizedSlots):
     """Provide loading behavior for a :class:`.RelationshipProperty`
     with "lazy=True", that is loads when first accessed.
 
@@ -421,78 +421,54 @@ class LazyLoader(AbstractRelationshipLoader):
             active_history=active_history
         )
 
-    def lazy_clause(
-            self, state, reverse_direction=False,
-            alias_secondary=False,
-            adapt_source=None,
-            passive=None):
-        if state is None:
-            return self._lazy_none_clause(
-                reverse_direction,
-                adapt_source=adapt_source)
-
-        if not reverse_direction:
-            criterion, bind_to_col = \
-                self._lazywhere, \
-                self._bind_to_col
-        else:
-            criterion, bind_to_col = \
-                self._rev_lazywhere, \
-                self._rev_bind_to_col
+    def _memoized_attr__simple_lazy_clause(self):
+        criterion, bind_to_col = (
+            self._lazywhere,
+            self._bind_to_col
+        )
 
-        if reverse_direction:
-            mapper = self.parent_property.mapper
-        else:
-            mapper = self.parent_property.parent
+        params = []
 
-        o = state.obj()  # strong ref
-        dict_ = attributes.instance_dict(o)
+        def visit_bindparam(bindparam):
+            bindparam.unique = False
+            if bindparam._identifying_key in bind_to_col:
+                params.append((
+                    bindparam.key, bind_to_col[bindparam._identifying_key],
+                    None))
+            else:
+                params.append((bindparam.key, None, bindparam.value))
+
+        criterion = visitors.cloned_traverse(
+            criterion, {}, {'bindparam': visit_bindparam}
+        )
 
-        # use the "committed state" only if we're in a flush
-        # for this state.
+        return criterion, params
 
-        if passive and passive & attributes.LOAD_AGAINST_COMMITTED:
-            def visit_bindparam(bindparam):
-                if bindparam._identifying_key in bind_to_col:
-                    bindparam.callable = \
-                        lambda: mapper._get_committed_state_attr_by_column(
-                            state, dict_,
-                            bind_to_col[bindparam._identifying_key])
-        else:
-            def visit_bindparam(bindparam):
-                if bindparam._identifying_key in bind_to_col:
-                    bindparam.callable = \
-                        lambda: mapper._get_state_attr_by_column(
-                            state, dict_,
-                            bind_to_col[bindparam._identifying_key])
-
-        if self.parent_property.secondary is not None and alias_secondary:
-            criterion = sql_util.ClauseAdapter(
-                self.parent_property.secondary.alias()).\
-                traverse(criterion)
+    def _generate_lazy_clause(self, state, passive):
+        criterion, param_keys = self._simple_lazy_clause
 
-        criterion = visitors.cloned_traverse(
-            criterion, {}, {'bindparam': visit_bindparam})
+        if state is None:
+            return sql_util.adapt_criterion_to_null(
+                criterion, [key for key, ident, value in param_keys])
 
-        if adapt_source:
-            criterion = adapt_source(criterion)
-        return criterion
+        mapper = self.parent_property.parent
 
-    def _lazy_none_clause(self, reverse_direction=False, adapt_source=None):
-        if not reverse_direction:
-            criterion, bind_to_col = \
-                self._lazywhere, \
-                self._bind_to_col
-        else:
-            criterion, bind_to_col = \
-                self._rev_lazywhere, \
-                self._rev_bind_to_col
+        o = state.obj()  # strong ref
+        dict_ = attributes.instance_dict(o)
+
+        params = {}
+        for key, ident, value in param_keys:
+            if ident is not None:
+                if passive and passive & attributes.LOAD_AGAINST_COMMITTED:
+                    value = mapper._get_committed_state_attr_by_column(
+                        state, dict_, ident)
+                else:
+                    value = mapper._get_state_attr_by_column(
+                        state, dict_, ident)
 
-        criterion = sql_util.adapt_criterion_to_null(criterion, bind_to_col)
+            params[key] = value
 
-        if adapt_source:
-            criterion = adapt_source(criterion)
-        return criterion
+        return criterion, params
 
     def _load_for_state(self, state, passive):
         if not state.key and (
@@ -569,10 +545,9 @@ class LazyLoader(AbstractRelationshipLoader):
 
     @util.dependencies("sqlalchemy.orm.strategy_options")
     def _emit_lazyload(
-            self, strategy_options, session, state,
-            ident_key, passive):
-        q = session.query(self.mapper)._adapt_all_clauses()
+            self, strategy_options, session, state, ident_key, passive):
 
+        q = session.query(self.mapper)._adapt_all_clauses()
         if self.parent_property.secondary is not None:
             q = q.select_from(self.mapper, self.parent_property.secondary)
 
@@ -603,17 +578,15 @@ class LazyLoader(AbstractRelationshipLoader):
                 rev._use_get and \
                     not isinstance(rev.strategy, LazyLoader):
                 q = q.options(
-                    strategy_options.Load(rev.parent).
-                    lazyload(rev.key))
+                    strategy_options.Load(rev.parent).lazyload(rev.key))
 
-        lazy_clause = self.lazy_clause(state, passive=passive)
+        lazy_clause, params = self._generate_lazy_clause(
+            state, passive=passive)
 
-        if pending:
-            bind_values = sql_util.bind_values(lazy_clause)
-            if orm_util._none_set.intersection(bind_values):
-                return None
+        if pending and orm_util._none_set.intersection(params.values()):
+            return None
 
-        q = q.filter(lazy_clause)
+        q = q.filter(lazy_clause).params(params)
 
         result = q.all()
         if self.uselist:
@@ -646,7 +619,7 @@ class LazyLoader(AbstractRelationshipLoader):
             # class-level lazyloader installed.
             set_lazy_callable = InstanceState._instance_level_callable_processor(
                 mapper.class_manager,
-                LoadLazyAttribute(key), key)
+                LoadLazyAttribute(key, self._strategy_keys[0]), key)
 
             populators["new"].append((self.key, set_lazy_callable))
         elif context.populate_existing or mapper.always_refresh:
@@ -667,14 +640,15 @@ class LazyLoader(AbstractRelationshipLoader):
 class LoadLazyAttribute(object):
     """serializable loader object used by LazyLoader"""
 
-    def __init__(self, key):
+    def __init__(self, key, strategy_key=(('lazy', 'select'),)):
         self.key = key
+        self.strategy_key = strategy_key
 
     def __call__(self, state, passive=attributes.PASSIVE_OFF):
         key = self.key
         instance_mapper = state.manager.mapper
         prop = instance_mapper._props[key]
-        strategy = prop._strategies[LazyLoader]
+        strategy = prop._strategies[self.strategy_key]
 
         return strategy._load_for_state(state, passive)
 
@@ -1029,6 +1003,12 @@ class SubqueryLoader(AbstractRelationshipLoader):
         if subq is None:
             return
 
+        assert subq.session is context.session, (
+            "Subquery session doesn't refer to that of "
+            "our context.  Are there broken context caching "
+            "schemes being used?"
+        )
+
         local_cols = self.parent_property.local_columns
 
         # cache the loaded collections in the context