- squash-merge the final row_proc integration branch. this is

a much more modest outcome than what we started with.   The
work of create_row_processor() for ColumnProperty objects
is essentially done at query setup time combined with some
lookups in _instance_processor().
- to allow this change for deferred columns, deferred columns
no longer search for themselves in the result.   If they've been
set up as deferred without any explicit directive to undefer them,
then this is what was asked for.  if we don't do this,
then we're stuck with this performance penalty for all deferred
columns which in the vast majority of typical use cases (e.g. loading
large, legacy tables or tables with many/large very seldom
used values) won't be present in the result and won't be accessed at all.

11 files changed, 211 insertions, 100 deletions

diff --git a/doc/build/changelog/changelog_10.rst b/doc/build/changelog/changelog_10.rst
index 48b94c07a..e1c22c019 100644
--- a/doc/build/changelog/changelog_10.rst
+++ b/doc/build/changelog/changelog_10.rst
@@ -24,6 +24,17 @@
     on compatibility concerns, see :doc:`/changelog/migration_10`.
 
     .. change::
+        :tags: change, orm
+
+        Mapped attributes marked as deferred without explicit undeferral
+        will now remain "deferred" even if their column is otherwise
+        present in the result set in some way.   This is a performance
+        enhancement in that an ORM load no longer spends time searching
+        for each deferred column when the result set is obtained.  However,
+        for an application that has been relying upon this, an explicit
+        :func:`.undefer` or similar option should now be used.
+
+    .. change::
         :tags: feature, orm
         :tickets: 3307
 
diff --git a/doc/build/changelog/migration_10.rst b/doc/build/changelog/migration_10.rst
index 651679dfd..7783c90c0 100644
--- a/doc/build/changelog/migration_10.rst
+++ b/doc/build/changelog/migration_10.rst
@@ -8,7 +8,7 @@ What's New in SQLAlchemy 1.0?
     undergoing maintenance releases as of May, 2014,
     and SQLAlchemy version 1.0, as of yet unreleased.
 
-    Document last updated: January 30, 2015
+    Document last updated: March 1, 2015
 
 Introduction
 ============
@@ -1088,6 +1088,18 @@ as all the subclasses normally refer to the same table::
 :ticket:`3233`
 
 
+Deferred Columns No Longer Implicitly Undefer
+---------------------------------------------
+
+Mapped attributes marked as deferred without explicit undeferral
+will now remain "deferred" even if their column is otherwise
+present in the result set in some way.   This is a performance
+enhancement in that an ORM load no longer spends time searching
+for each deferred column when the result set is obtained.  However,
+for an application that has been relying upon this, an explicit
+:func:`.undefer` or similar option should now be used, in order
+to prevent a SELECT from being emitted when the attribute is accessed.
+
 
 .. _migration_deprecated_orm_events:
 
diff --git a/lib/sqlalchemy/orm/base.py b/lib/sqlalchemy/orm/base.py
index 7bfafdc2b..875443e60 100644
--- a/lib/sqlalchemy/orm/base.py
+++ b/lib/sqlalchemy/orm/base.py
@@ -183,6 +183,10 @@ NOT_EXTENSION = util.symbol(
 
 _none_set = frozenset([None, NEVER_SET, PASSIVE_NO_RESULT])
 
+_SET_DEFERRED_EXPIRED = util.symbol("SET_DEFERRED_EXPIRED")
+
+_DEFER_FOR_STATE = util.symbol("DEFER_FOR_STATE")
+
 
 def _generative(*assertions):
     """Mark a method as generative, e.g. method-chained."""
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index b3b8d612d..cd9fa150e 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -488,7 +488,8 @@ class StrategizedProperty(MapperProperty):
     def _get_strategy_by_cls(self, cls):
         return self._get_strategy(cls._strategy_keys[0])
 
-    def setup(self, context, entity, path, adapter, **kwargs):
+    def setup(
+            self, context, entity, path, adapter, **kwargs):
         loader = self._get_context_loader(context, path)
         if loader and loader.strategy:
             strat = self._get_strategy(loader.strategy)
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index c59257039..64c7e171c 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -18,6 +18,7 @@ from .. import util
 from . import attributes, exc as orm_exc
 from ..sql import util as sql_util
 from .util import _none_set, state_str
+from .base import _SET_DEFERRED_EXPIRED, _DEFER_FOR_STATE
 from .. import exc as sa_exc
 import collections
 
@@ -218,10 +219,56 @@ def load_on_ident(query, key,
         return None
 
 
-def instance_processor(mapper, context, result, path, adapter,
-                       only_load_props=None, refresh_state=None,
-                       polymorphic_discriminator=None,
-                       _polymorphic_from=None):
+def _setup_entity_query(
+    context, mapper, query_entity,
+        path, adapter, column_collection,
+        with_polymorphic=None, only_load_props=None,
+        polymorphic_discriminator=None, **kw):
+
+    if with_polymorphic:
+        poly_properties = mapper._iterate_polymorphic_properties(
+            with_polymorphic)
+    else:
+        poly_properties = mapper._polymorphic_properties
+
+    quick_populators = {}
+
+    path.set(
+        context.attributes,
+        "memoized_setups",
+        quick_populators)
+
+    for value in poly_properties:
+        if only_load_props and \
+                value.key not in only_load_props:
+            continue
+        value.setup(
+            context,
+            query_entity,
+            path,
+            adapter,
+            only_load_props=only_load_props,
+            column_collection=column_collection,
+            memoized_populators=quick_populators,
+            **kw
+        )
+
+    if polymorphic_discriminator is not None and \
+        polymorphic_discriminator \
+            is not mapper.polymorphic_on:
+
+        if adapter:
+            pd = adapter.columns[polymorphic_discriminator]
+        else:
+            pd = polymorphic_discriminator
+        column_collection.append(pd)
+
+
+def _instance_processor(
+        mapper, context, result, path, adapter,
+        only_load_props=None, refresh_state=None,
+        polymorphic_discriminator=None,
+        _polymorphic_from=None):
     """Produce a mapper level row processor callable
        which processes rows into mapped instances."""
 
@@ -240,13 +287,41 @@ def instance_processor(mapper, context, result, path, adapter,
 
     populators = collections.defaultdict(list)
 
-    props = mapper._props.values()
+    props = mapper._prop_set
     if only_load_props is not None:
-        props = (p for p in props if p.key in only_load_props)
+        props = props.intersection(
+            mapper._props[k] for k in only_load_props)
+
+    quick_populators = path.get(
+        context.attributes, "memoized_setups", _none_set)
 
     for prop in props:
-        prop.create_row_processor(
-            context, path, mapper, result, adapter, populators)
+        if prop in quick_populators:
+            # this is an inlined path just for column-based attributes.
+            col = quick_populators[prop]
+            if col is _DEFER_FOR_STATE:
+                populators["new"].append(
+                    (prop.key, prop._deferred_column_loader))
+            elif col is _SET_DEFERRED_EXPIRED:
+                # note that in this path, we are no longer
+                # searching in the result to see if the column might
+                # be present in some unexpected way.
+                populators["expire"].append((prop.key, False))
+            else:
+                if adapter:
+                    col = adapter.columns[col]
+                getter = result._getter(col)
+                if getter:
+                    populators["quick"].append((prop.key, getter))
+                else:
+                    # fall back to the ColumnProperty itself, which
+                    # will iterate through all of its columns
+                    # to see if one fits
+                    prop.create_row_processor(
+                        context, path, mapper, result, adapter, populators)
+        else:
+            prop.create_row_processor(
+                context, path, mapper, result, adapter, populators)
 
     propagate_options = context.propagate_options
     if propagate_options:
@@ -388,7 +463,7 @@ def instance_processor(mapper, context, result, path, adapter,
 
         return instance
 
-    if not _polymorphic_from and not refresh_state:
+    if mapper.polymorphic_map and not _polymorphic_from and not refresh_state:
         # if we are doing polymorphic, dispatch to a different _instance()
         # method specific to the subclass mapper
         _instance = _decorate_polymorphic_switch(
@@ -503,7 +578,7 @@ def _decorate_polymorphic_switch(
             if sub_mapper is mapper:
                 return None
 
-            return instance_processor(
+            return _instance_processor(
                 sub_mapper, context, result,
                 path, adapter, _polymorphic_from=mapper)
 
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index eb5abbd4f..df67ff147 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -1501,6 +1501,10 @@ class Mapper(InspectionAttr):
 
         return identities
 
+    @_memoized_configured_property
+    def _prop_set(self):
+        return frozenset(self._props.values())
+
     def _adapt_inherited_property(self, key, prop, init):
         if not self.concrete:
             self._configure_property(key, prop, init=False, setparent=False)
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index d51b6920d..31e9c7f3f 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -39,7 +39,7 @@ class ColumnProperty(StrategizedProperty):
         'instrument', 'comparator_factory', 'descriptor', 'extension',
         'active_history', 'expire_on_flush', 'info', 'doc',
         'strategy_class', '_creation_order', '_is_polymorphic_discriminator',
-        '_mapped_by_synonym')
+        '_mapped_by_synonym', '_deferred_loader')
 
     def __init__(self, *columns, **kwargs):
         """Provide a column-level property for use with a Mapper.
@@ -157,6 +157,12 @@ class ColumnProperty(StrategizedProperty):
             ("instrument", self.instrument)
         )
 
+    @util.dependencies("sqlalchemy.orm.state", "sqlalchemy.orm.strategies")
+    def _memoized_attr__deferred_column_loader(self, state, strategies):
+        return state.InstanceState._instance_level_callable_processor(
+            self.parent.class_manager,
+            strategies.LoadDeferredColumns(self.key), self.key)
+
     @property
     def expression(self):
         """Return the primary column or expression for this ColumnProperty.
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 205a5539f..eac2da083 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -3272,25 +3272,21 @@ class _MapperEntity(_QueryEntity):
                 self.mapper._equivalent_columns)
 
         if query._primary_entity is self:
-            _instance = loading.instance_processor(
-                self.mapper,
-                context,
-                result,
-                self.path,
-                adapter,
-                only_load_props=query._only_load_props,
-                refresh_state=context.refresh_state,
-                polymorphic_discriminator=self._polymorphic_discriminator
-            )
+            only_load_props = query._only_load_props
+            refresh_state = context.refresh_state
         else:
-            _instance = loading.instance_processor(
-                self.mapper,
-                context,
-                result,
-                self.path,
-                adapter,
-                polymorphic_discriminator=self._polymorphic_discriminator
-            )
+            only_load_props = refresh_state = None
+
+        _instance = loading._instance_processor(
+            self.mapper,
+            context,
+            result,
+            self.path,
+            adapter,
+            only_load_props=only_load_props,
+            refresh_state=refresh_state,
+            polymorphic_discriminator=self._polymorphic_discriminator
+        )
 
         return _instance, self._label_name
 
@@ -3311,34 +3307,12 @@ class _MapperEntity(_QueryEntity):
                     )
                 )
 
-        if self._with_polymorphic:
-            poly_properties = self.mapper._iterate_polymorphic_properties(
-                self._with_polymorphic)
-        else:
-            poly_properties = self.mapper._polymorphic_properties
-
-        for value in poly_properties:
-            if query._only_load_props and \
-                    value.key not in query._only_load_props:
-                continue
-            value.setup(
-                context,
-                self,
-                self.path,
-                adapter,
-                only_load_props=query._only_load_props,
-                column_collection=context.primary_columns
-            )
-
-        if self._polymorphic_discriminator is not None and \
-            self._polymorphic_discriminator \
-                is not self.mapper.polymorphic_on:
-
-            if adapter:
-                pd = adapter.columns[self._polymorphic_discriminator]
-            else:
-                pd = self._polymorphic_discriminator
-            context.primary_columns.append(pd)
+        loading._setup_entity_query(
+            context, self.mapper, self,
+            self.path, adapter, context.primary_columns,
+            with_polymorphic=self._with_polymorphic,
+            only_load_props=query._only_load_props,
+            polymorphic_discriminator=self._polymorphic_discriminator)
 
     def __str__(self):
         return str(self.mapper)
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 0444c63ae..a0b9bd31e 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -22,6 +22,7 @@ from . import properties
 from .interfaces import (
     LoaderStrategy, StrategizedProperty
 )
+from .base import _SET_DEFERRED_EXPIRED, _DEFER_FOR_STATE
 from .session import _state_session
 import itertools
 
@@ -139,12 +140,18 @@ class ColumnLoader(LoaderStrategy):
 
     def setup_query(
             self, context, entity, path, loadopt,
-            adapter, column_collection, **kwargs):
+            adapter, column_collection, memoized_populators, **kwargs):
+
         for c in self.columns:
             if adapter:
                 c = adapter.columns[c]
             column_collection.append(c)
 
+        fetch = self.columns[0]
+        if adapter:
+            fetch = adapter.columns[fetch]
+        memoized_populators[self.parent_property] = fetch
+
     def init_class_attribute(self, mapper):
         self.is_class_level = True
         coltype = self.columns[0].type
@@ -193,23 +200,14 @@ class DeferredColumnLoader(LoaderStrategy):
     def create_row_processor(
             self, context, path, loadopt,
             mapper, result, adapter, populators):
-        col = self.columns[0]
-        if adapter:
-            col = adapter.columns[col]
-
-        # TODO: put a result-level contains here
-        getter = result._getter(col)
-        if getter:
-            self.parent_property._get_strategy_by_cls(ColumnLoader).\
-                create_row_processor(
-                    context, path, loadopt, mapper, result,
-                    adapter, populators)
 
-        elif not self.is_class_level:
+        # this path currently does not check the result
+        # for the column; this is because in most cases we are
+        # working just with the setup_query() directive which does
+        # not support this, and the behavior here should be consistent.
+        if not self.is_class_level:
             set_deferred_for_local_state = \
-                InstanceState._instance_level_callable_processor(
-                    mapper.class_manager,
-                    LoadDeferredColumns(self.key), self.key)
+                self.parent_property._deferred_column_loader
             populators["new"].append((self.key, set_deferred_for_local_state))
         else:
             populators["expire"].append((self.key, False))
@@ -225,8 +223,9 @@ class DeferredColumnLoader(LoaderStrategy):
         )
 
     def setup_query(
-            self, context, entity, path, loadopt, adapter,
-            only_load_props=None, **kwargs):
+            self, context, entity, path, loadopt,
+            adapter, column_collection, memoized_populators,
+            only_load_props=None, **kw):
 
         if (
             (
@@ -248,7 +247,12 @@ class DeferredColumnLoader(LoaderStrategy):
         ):
             self.parent_property._get_strategy_by_cls(ColumnLoader).\
                 setup_query(context, entity,
-                            path, loadopt, adapter, **kwargs)
+                            path, loadopt, adapter,
+                            column_collection, memoized_populators, **kw)
+        elif self.is_class_level:
+            memoized_populators[self.parent_property] = _SET_DEFERRED_EXPIRED
+        else:
+            memoized_populators[self.parent_property] = _DEFER_FOR_STATE
 
     def _load_for_state(self, state, passive):
         if not state.key:
@@ -1153,16 +1157,12 @@ class JoinedLoader(AbstractRelationshipLoader):
 
         path = path[self.mapper]
 
-        for value in self.mapper._iterate_polymorphic_properties(
-                mappers=with_polymorphic):
-            value.setup(
-                context,
-                entity,
-                path,
-                clauses,
-                parentmapper=self.mapper,
-                column_collection=add_to_collection,
-                chained_from_outerjoin=chained_from_outerjoin)
+        loading._setup_entity_query(
+            context, self.mapper, entity,
+            path, clauses, add_to_collection,
+            with_polymorphic=with_polymorphic,
+            parentmapper=self.mapper,
+            chained_from_outerjoin=chained_from_outerjoin)
 
         if with_poly_info is not None and \
                 None in set(context.secondary_columns):
@@ -1454,7 +1454,7 @@ class JoinedLoader(AbstractRelationshipLoader):
         if eager_adapter is not False:
             key = self.key
 
-            _instance = loading.instance_processor(
+            _instance = loading._instance_processor(
                 self.mapper,
                 context,
                 result,
diff --git a/test/orm/test_deferred.py b/test/orm/test_deferred.py
index 1b777b527..29087fdb8 100644
--- a/test/orm/test_deferred.py
+++ b/test/orm/test_deferred.py
@@ -341,7 +341,8 @@ class DeferredOptionsTest(AssertsCompiledSQL, _fixtures.FixtureTest):
             )
 
     def test_locates_col(self):
-        """Manually adding a column to the result undefers the column."""
+        """changed in 1.0 - we don't search for deferred cols in the result
+        now.  """
 
         orders, Order = self.tables.orders, self.classes.Order
 
@@ -350,18 +351,40 @@ class DeferredOptionsTest(AssertsCompiledSQL, _fixtures.FixtureTest):
             'description': deferred(orders.c.description)})
 
         sess = create_session()
-        o1 = sess.query(Order).order_by(Order.id).first()
+        o1 = (sess.query(Order).
+              order_by(Order.id).
+              add_column(orders.c.description).first())[0]
         def go():
             eq_(o1.description, 'order 1')
+        # prior to 1.0 we'd search in the result for this column
+        # self.sql_count_(0, go)
         self.sql_count_(1, go)
 
+    def test_locates_col_rowproc_only(self):
+        """changed in 1.0 - we don't search for deferred cols in the result
+        now.
+
+        Because the loading for ORM Query and Query from a core select
+        is now split off, we test loading from a plain select()
+        separately.
+
+        """
+
+        orders, Order = self.tables.orders, self.classes.Order
+
+
+        mapper(Order, orders, properties={
+            'description': deferred(orders.c.description)})
+
         sess = create_session()
+        stmt = sa.select([Order]).order_by(Order.id)
         o1 = (sess.query(Order).
-              order_by(Order.id).
-              add_column(orders.c.description).first())[0]
+              from_statement(stmt).all())[0]
         def go():
             eq_(o1.description, 'order 1')
-        self.sql_count_(0, go)
+        # prior to 1.0 we'd search in the result for this column
+        # self.sql_count_(0, go)
+        self.sql_count_(1, go)
 
     def test_deep_options(self):
         users, items, order_items, Order, Item, User, orders = (self.tables.users,
diff --git a/test/orm/test_eager_relations.py b/test/orm/test_eager_relations.py
index 4c6d9bbe1..ea8db8fda 100644
--- a/test/orm/test_eager_relations.py
+++ b/test/orm/test_eager_relations.py
@@ -294,20 +294,21 @@ class EagerTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
         sess.expunge_all()
         a = sess.query(Address).filter(Address.id == 1).all()[0]
 
+        # 1.0 change!  we don't automatically undefer user_id here.
+        # if the user wants a column undeferred, add the option.
         def go():
             eq_(a.user_id, 7)
-        # assert that the eager loader added 'user_id' to the row and deferred
-        # loading of that col was disabled
-        self.assert_sql_count(testing.db, go, 0)
+        # self.assert_sql_count(testing.db, go, 0)
+        self.assert_sql_count(testing.db, go, 1)
 
         sess.expunge_all()
         a = sess.query(Address).filter(Address.id == 1).first()
 
         def go():
             eq_(a.user_id, 7)
-        # assert that the eager loader added 'user_id' to the row and deferred
-        # loading of that col was disabled
-        self.assert_sql_count(testing.db, go, 0)
+        # same, 1.0 doesn't check these
+        # self.assert_sql_count(testing.db, go, 0)
+        self.assert_sql_count(testing.db, go, 1)
 
         # do the mapping in reverse
         # (we would have just used an "addresses" backref but the test