Merge branch 'master' into ticket_3100

40 files changed, 1087 insertions, 958 deletions

diff --git a/lib/sqlalchemy/__init__.py b/lib/sqlalchemy/__init__.py
index 2ab717996..853566172 100644
--- a/lib/sqlalchemy/__init__.py
+++ b/lib/sqlalchemy/__init__.py
@@ -15,6 +15,7 @@ from .sql import (
     case,
     cast,
     collate,
+    column,
     delete,
     desc,
     distinct,
@@ -39,6 +40,7 @@ from .sql import (
     over,
     select,
     subquery,
+    table,
     text,
     true,
     tuple_,
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 374960765..012d178e7 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -190,15 +190,13 @@ SQLAlchemy standardizes the DBAPI ``cursor.rowcount`` attribute to be the
 usual definition of "number of rows matched by an UPDATE or DELETE" statement.
 This is in contradiction to the default setting on most MySQL DBAPI drivers,
 which is "number of rows actually modified/deleted".  For this reason, the
-SQLAlchemy MySQL dialects always set the ``constants.CLIENT.FOUND_ROWS`` flag,
-or whatever is equivalent for the DBAPI in use, on connect, unless the flag
-value is overridden using DBAPI-specific options
-(such as ``client_flag`` for the MySQL-Python driver, ``found_rows`` for the
-OurSQL driver).
+SQLAlchemy MySQL dialects always add the ``constants.CLIENT.FOUND_ROWS``
+flag, or whatever is equivalent for the target dialect, upon connection.
+This setting is currently hardcoded.
 
-See also:
+.. seealso::
 
-:attr:`.ResultProxy.rowcount`
+    :attr:`.ResultProxy.rowcount`
 
 
 CAST Support
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index c2b1d66f4..f1418f903 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -417,6 +417,42 @@ of :class:`.PGInspector`, which offers additional methods::
 .. autoclass:: PGInspector
     :members:
 
+.. _postgresql_table_options:
+
+PostgreSQL Table Options
+-------------------------
+
+Several options for CREATE TABLE are supported directly by the PostgreSQL
+dialect in conjunction with the :class:`.Table` construct:
+
+* ``TABLESPACE``::
+
+    Table("some_table", metadata, ..., postgresql_tablespace='some_tablespace')
+
+* ``ON COMMIT``::
+
+    Table("some_table", metadata, ..., postgresql_on_commit='PRESERVE ROWS')
+
+* ``WITH OIDS``::
+
+    Table("some_table", metadata, ..., postgresql_with_oids=True)
+
+* ``WITHOUT OIDS``::
+
+    Table("some_table", metadata, ..., postgresql_with_oids=False)
+
+* ``INHERITS``::
+
+    Table("some_table", metadata, ..., postgresql_inherits="some_supertable")
+
+    Table("some_table", metadata, ..., postgresql_inherits=("t1", "t2", ...))
+
+.. versionadded:: 1.0.0
+
+.. seealso::
+
+    `Postgresql CREATE TABLE options
+    <http://www.postgresql.org/docs/9.3/static/sql-createtable.html>`_
 
 """
 from collections import defaultdict
@@ -1448,6 +1484,36 @@ class PGDDLCompiler(compiler.DDLCompiler):
         text += self.define_constraint_deferrability(constraint)
         return text
 
+    def post_create_table(self, table):
+        table_opts = []
+        pg_opts = table.dialect_options['postgresql']
+
+        inherits = pg_opts.get('inherits')
+        if inherits is not None:
+            if not isinstance(inherits, (list, tuple)):
+                inherits = (inherits, )
+            table_opts.append(
+                '\n INHERITS ( ' +
+                ', '.join(self.preparer.quote(name) for name in inherits) +
+                ' )')
+
+        if pg_opts['with_oids'] is True:
+            table_opts.append('\n WITH OIDS')
+        elif pg_opts['with_oids'] is False:
+            table_opts.append('\n WITHOUT OIDS')
+
+        if pg_opts['on_commit']:
+            on_commit_options = pg_opts['on_commit'].replace("_", " ").upper()
+            table_opts.append('\n ON COMMIT %s' % on_commit_options)
+
+        if pg_opts['tablespace']:
+            tablespace_name = pg_opts['tablespace']
+            table_opts.append(
+                '\n TABLESPACE %s' % self.preparer.quote(tablespace_name)
+            )
+
+        return ''.join(table_opts)
+
 
 class PGTypeCompiler(compiler.GenericTypeCompiler):
 
@@ -1707,7 +1773,11 @@ class PGDialect(default.DefaultDialect):
             "ops": {}
         }),
         (schema.Table, {
-            "ignore_search_path": False
+            "ignore_search_path": False,
+            "tablespace": None,
+            "with_oids": None,
+            "on_commit": None,
+            "inherits": None
         })
     ]
 
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 012d1d35d..cf1f2d3dd 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -578,18 +578,27 @@ class Inspector(object):
             name = index_d['name']
             columns = index_d['column_names']
             unique = index_d['unique']
-            flavor = index_d.get('type', 'unknown type')
+            flavor = index_d.get('type', 'index')
             if include_columns and \
                     not set(columns).issubset(include_columns):
                 util.warn(
-                    "Omitting %s KEY for (%s), key covers omitted columns." %
+                    "Omitting %s key for (%s), key covers omitted columns." %
                     (flavor, ', '.join(columns)))
                 continue
             # look for columns by orig name in cols_by_orig_name,
             # but support columns that are in-Python only as fallback
-            sa_schema.Index(name, *[
-                cols_by_orig_name[c] if c in cols_by_orig_name
-                else table.c[c]
-                for c in columns
-            ],
-                **dict(unique=unique))
+            idx_cols = []
+            for c in columns:
+                try:
+                    idx_col = cols_by_orig_name[c] \
+                        if c in cols_by_orig_name else table.c[c]
+                except KeyError:
+                    util.warn(
+                        "%s key '%s' was not located in "
+                        "columns for table '%s'" % (
+                            flavor, c, table_name
+                        ))
+                else:
+                    idx_cols.append(idx_col)
+
+            sa_schema.Index(name, *idx_cols, **dict(unique=unique))
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 06a81aa6c..3995942ef 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -110,7 +110,7 @@ class RowProxy(BaseRowProxy):
     __slots__ = ()
 
     def __contains__(self, key):
-        return self._parent._has_key(self._row, key)
+        return self._parent._has_key(key)
 
     def __getstate__(self):
         return {
@@ -155,7 +155,7 @@ class RowProxy(BaseRowProxy):
     def has_key(self, key):
         """Return True if this RowProxy contains the given key."""
 
-        return self._parent._has_key(self._row, key)
+        return self._parent._has_key(key)
 
     def items(self):
         """Return a list of tuples, each tuple containing a key/value pair."""
@@ -331,12 +331,28 @@ class ResultMetaData(object):
             map[key] = result
         return result
 
-    def _has_key(self, row, key):
+    def _has_key(self, key):
         if key in self._keymap:
             return True
         else:
             return self._key_fallback(key, False) is not None
 
+    def _getter(self, key):
+        if key in self._keymap:
+            processor, obj, index = self._keymap[key]
+        else:
+            ret = self._key_fallback(key, False)
+            if ret is None:
+                return None
+            processor, obj, index = ret
+
+        if index is None:
+            raise exc.InvalidRequestError(
+                "Ambiguous column name '%s' in result set! "
+                "try 'use_labels' option on select statement." % key)
+
+        return operator.itemgetter(index)
+
     def __getstate__(self):
         return {
             '_pickled_keymap': dict(
@@ -398,6 +414,12 @@ class ResultProxy(object):
             context.engine._should_log_debug()
         self._init_metadata()
 
+    def _getter(self, key):
+        return self._metadata._getter(key)
+
+    def _has_key(self, key):
+        return self._metadata._has_key(key)
+
     def _init_metadata(self):
         metadata = self._cursor_description()
         if metadata is not None:
diff --git a/lib/sqlalchemy/ext/mutable.py b/lib/sqlalchemy/ext/mutable.py
index 7469bcbda..e49e9ea8b 100644
--- a/lib/sqlalchemy/ext/mutable.py
+++ b/lib/sqlalchemy/ext/mutable.py
@@ -621,16 +621,20 @@ class MutableDict(Mutable, dict):
         dict.__delitem__(self, key)
         self.changed()
 
+    def update(self, *a, **kw):
+        dict.update(self, *a, **kw)
+        self.changed()
+
     def clear(self):
         dict.clear(self)
         self.changed()
 
     @classmethod
     def coerce(cls, key, value):
-        """Convert plain dictionary to MutableDict."""
-        if not isinstance(value, MutableDict):
+        """Convert plain dictionary to instance of this class."""
+        if not isinstance(value, cls):
             if isinstance(value, dict):
-                return MutableDict(value)
+                return cls(value)
             return Mutable.coerce(key, value)
         else:
             return value
diff --git a/lib/sqlalchemy/orm/deprecated_interfaces.py b/lib/sqlalchemy/orm/deprecated_interfaces.py
index fa693c968..275582323 100644
--- a/lib/sqlalchemy/orm/deprecated_interfaces.py
+++ b/lib/sqlalchemy/orm/deprecated_interfaces.py
@@ -67,10 +67,6 @@ class MapperExtension(object):
             (
                 'init_instance',
                 'init_failed',
-                'translate_row',
-                'create_instance',
-                'append_result',
-                'populate_instance',
                 'reconstruct_instance',
                 'before_insert',
                 'after_insert',
@@ -156,108 +152,6 @@ class MapperExtension(object):
         """
         return EXT_CONTINUE
 
-    def translate_row(self, mapper, context, row):
-        """Perform pre-processing on the given result row and return a
-        new row instance.
-
-        This is called when the mapper first receives a row, before
-        the object identity or the instance itself has been derived
-        from that row.   The given row may or may not be a
-        ``RowProxy`` object - it will always be a dictionary-like
-        object which contains mapped columns as keys.  The
-        returned object should also be a dictionary-like object
-        which recognizes mapped columns as keys.
-
-        If the ultimate return value is EXT_CONTINUE, the row
-        is not translated.
-
-        """
-        return EXT_CONTINUE
-
-    def create_instance(self, mapper, selectcontext, row, class_):
-        """Receive a row when a new object instance is about to be
-        created from that row.
-
-        The method can choose to create the instance itself, or it can return
-        EXT_CONTINUE to indicate normal object creation should take place.
-
-        mapper
-          The mapper doing the operation
-
-        selectcontext
-          The QueryContext generated from the Query.
-
-        row
-          The result row from the database
-
-        class\_
-          The class we are mapping.
-
-        return value
-          A new object instance, or EXT_CONTINUE
-
-        """
-        return EXT_CONTINUE
-
-    def append_result(self, mapper, selectcontext, row, instance,
-                      result, **flags):
-        """Receive an object instance before that instance is appended
-        to a result list.
-
-        If this method returns EXT_CONTINUE, result appending will proceed
-        normally.  if this method returns any other value or None,
-        result appending will not proceed for this instance, giving
-        this extension an opportunity to do the appending itself, if
-        desired.
-
-        mapper
-          The mapper doing the operation.
-
-        selectcontext
-          The QueryContext generated from the Query.
-
-        row
-          The result row from the database.
-
-        instance
-          The object instance to be appended to the result.
-
-        result
-          List to which results are being appended.
-
-        \**flags
-          extra information about the row, same as criterion in
-          ``create_row_processor()`` method of
-          :class:`~sqlalchemy.orm.interfaces.MapperProperty`
-        """
-
-        return EXT_CONTINUE
-
-    def populate_instance(self, mapper, selectcontext, row,
-                          instance, **flags):
-        """Receive an instance before that instance has
-        its attributes populated.
-
-        This usually corresponds to a newly loaded instance but may
-        also correspond to an already-loaded instance which has
-        unloaded attributes to be populated.  The method may be called
-        many times for a single instance, as multiple result rows are
-        used to populate eagerly loaded collections.
-
-        If this method returns EXT_CONTINUE, instance population will
-        proceed normally.  If any other value or None is returned,
-        instance population will not proceed, giving this extension an
-        opportunity to populate the instance itself, if desired.
-
-        .. deprecated:: 0.5
-            Most usages of this hook are obsolete.  For a
-            generic "object has been newly created from a row" hook, use
-            ``reconstruct_instance()``, or the ``@orm.reconstructor``
-            decorator.
-
-        """
-        return EXT_CONTINUE
-
     def reconstruct_instance(self, mapper, instance):
         """Receive an object instance after it has been created via
         ``__new__``, and after initial attribute population has
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index f0f9a6468..19ff71f73 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -372,9 +372,9 @@ class CompositeProperty(DescriptorProperty):
                 property.key, *expr)
 
         def create_row_processor(self, query, procs, labels):
-            def proc(row, result):
+            def proc(row):
                 return self.property.composite_class(
-                    *[proc(row, result) for proc in procs])
+                    *[proc(row) for proc in procs])
             return proc
 
     class Comparator(PropComparator):
diff --git a/lib/sqlalchemy/orm/events.py b/lib/sqlalchemy/orm/events.py
index aa99673ba..daf705040 100644
--- a/lib/sqlalchemy/orm/events.py
+++ b/lib/sqlalchemy/orm/events.py
@@ -293,18 +293,6 @@ class InstanceEvents(event.Events):
 
         """
 
-    def resurrect(self, target):
-        """Receive an object instance as it is 'resurrected' from
-        garbage collection, which occurs when a "dirty" state falls
-        out of scope.
-
-        :param target: the mapped instance.  If
-         the event is configured with ``raw=True``, this will
-         instead be the :class:`.InstanceState` state-management
-         object associated with the instance.
-
-        """
-
     def pickle(self, target, state_dict):
         """Receive an object instance when its associated state is
         being pickled.
@@ -664,145 +652,6 @@ class MapperEvents(event.Events):
 
         """
 
-    def translate_row(self, mapper, context, row):
-        """Perform pre-processing on the given result row and return a
-        new row instance.
-
-        .. deprecated:: 0.9 the :meth:`.translate_row` event should
-           be considered as legacy.  The row as delivered in a mapper
-           load operation typically requires that highly technical
-           details be accommodated in order to identity the correct
-           column keys are present in the row, rendering this particular
-           event hook as difficult to use and unreliable.
-
-        This listener is typically registered with ``retval=True``.
-        It is called when the mapper first receives a row, before
-        the object identity or the instance itself has been derived
-        from that row.   The given row may or may not be a
-        :class:`.RowProxy` object - it will always be a dictionary-like
-        object which contains mapped columns as keys.  The
-        returned object should also be a dictionary-like object
-        which recognizes mapped columns as keys.
-
-        :param mapper: the :class:`.Mapper` which is the target
-         of this event.
-        :param context: the :class:`.QueryContext`, which includes
-         a handle to the current :class:`.Query` in progress as well
-         as additional state information.
-        :param row: the result row being handled.  This may be
-         an actual :class:`.RowProxy` or may be a dictionary containing
-         :class:`.Column` objects as keys.
-        :return: When configured with ``retval=True``, the function
-         should return a dictionary-like row object, or ``EXT_CONTINUE``,
-         indicating the original row should be used.
-
-
-        """
-
-    def create_instance(self, mapper, context, row, class_):
-        """Receive a row when a new object instance is about to be
-        created from that row.
-
-        .. deprecated:: 0.9 the :meth:`.create_instance` event should
-           be considered as legacy.  Manipulation of the object construction
-           mechanics during a load should not be necessary.
-
-        The method can choose to create the instance itself, or it can return
-        EXT_CONTINUE to indicate normal object creation should take place.
-        This listener is typically registered with ``retval=True``.
-
-        :param mapper: the :class:`.Mapper` which is the target
-         of this event.
-        :param context: the :class:`.QueryContext`, which includes
-         a handle to the current :class:`.Query` in progress as well
-         as additional state information.
-        :param row: the result row being handled.  This may be
-         an actual :class:`.RowProxy` or may be a dictionary containing
-         :class:`.Column` objects as keys.
-        :param class\_: the mapped class.
-        :return: When configured with ``retval=True``, the return value
-         should be a newly created instance of the mapped class,
-         or ``EXT_CONTINUE`` indicating that default object construction
-         should take place.
-
-        """
-
-    def append_result(self, mapper, context, row, target,
-                      result, **flags):
-        """Receive an object instance before that instance is appended
-        to a result list.
-
-        .. deprecated:: 0.9 the :meth:`.append_result` event should
-           be considered as legacy.  It is a difficult to use method
-           whose original purpose is better suited by custom collection
-           classes.
-
-        This is a rarely used hook which can be used to alter
-        the construction of a result list returned by :class:`.Query`.
-
-        :param mapper: the :class:`.Mapper` which is the target
-         of this event.
-        :param context: the :class:`.QueryContext`, which includes
-         a handle to the current :class:`.Query` in progress as well
-         as additional state information.
-        :param row: the result row being handled.  This may be
-         an actual :class:`.RowProxy` or may be a dictionary containing
-         :class:`.Column` objects as keys.
-        :param target: the mapped instance being populated.  If
-         the event is configured with ``raw=True``, this will
-         instead be the :class:`.InstanceState` state-management
-         object associated with the instance.
-        :param result: a list-like object where results are being
-         appended.
-        :param \**flags: Additional state information about the
-         current handling of the row.
-        :return: If this method is registered with ``retval=True``,
-         a return value of ``EXT_STOP`` will prevent the instance
-         from being appended to the given result list, whereas a
-         return value of ``EXT_CONTINUE`` will result in the default
-         behavior of appending the value to the result list.
-
-        """
-
-    def populate_instance(self, mapper, context, row,
-                          target, **flags):
-        """Receive an instance before that instance has
-        its attributes populated.
-
-        .. deprecated:: 0.9 the :meth:`.populate_instance` event should
-           be considered as legacy.  The mechanics of instance population
-           should not need modification; special "on load" rules can as always
-           be accommodated by the :class:`.InstanceEvents.load` event.
-
-        This usually corresponds to a newly loaded instance but may
-        also correspond to an already-loaded instance which has
-        unloaded attributes to be populated.  The method may be called
-        many times for a single instance, as multiple result rows are
-        used to populate eagerly loaded collections.
-
-        Most usages of this hook are obsolete.  For a
-        generic "object has been newly created from a row" hook, use
-        :meth:`.InstanceEvents.load`.
-
-        :param mapper: the :class:`.Mapper` which is the target
-         of this event.
-        :param context: the :class:`.QueryContext`, which includes
-         a handle to the current :class:`.Query` in progress as well
-         as additional state information.
-        :param row: the result row being handled.  This may be
-         an actual :class:`.RowProxy` or may be a dictionary containing
-         :class:`.Column` objects as keys.
-        :param target: the mapped instance.  If
-         the event is configured with ``raw=True``, this will
-         instead be the :class:`.InstanceState` state-management
-         object associated with the instance.
-        :return: When configured with ``retval=True``, a return
-         value of ``EXT_STOP`` will bypass instance population by
-         the mapper. A value of ``EXT_CONTINUE`` indicates that
-         default instance population should take place.
-
-        """
-
     def before_insert(self, mapper, connection, target):
         """Receive an object instance before an INSERT statement
         is emitted corresponding to that instance.
diff --git a/lib/sqlalchemy/orm/identity.py b/lib/sqlalchemy/orm/identity.py
index 0fa541194..24dd47859 100644
--- a/lib/sqlalchemy/orm/identity.py
+++ b/lib/sqlalchemy/orm/identity.py
@@ -187,6 +187,12 @@ class WeakInstanceDict(IdentityMap):
             return list(self._dict.values())
 
     def discard(self, state):
+        st = self._dict.pop(state.key, None)
+        if st:
+            assert st is state
+            self._manage_removed_state(state)
+
+    def safe_discard(self, state):
         if state.key in self._dict:
             st = self._dict[state.key]
             if st is state:
@@ -259,6 +265,13 @@ class StrongInstanceDict(IdentityMap):
         state._instance_dict = self._wr
 
     def discard(self, state):
+        obj = self._dict.pop(state.key, None)
+        if obj is not None:
+            self._manage_removed_state(state)
+            st = attributes.instance_state(obj)
+            assert st is state
+
+    def safe_discard(self, state):
         if state.key in self._dict:
             obj = self._dict[state.key]
             st = attributes.instance_state(obj)
diff --git a/lib/sqlalchemy/orm/instrumentation.py b/lib/sqlalchemy/orm/instrumentation.py
index eb5b65baa..ad7d2d53d 100644
--- a/lib/sqlalchemy/orm/instrumentation.py
+++ b/lib/sqlalchemy/orm/instrumentation.py
@@ -41,6 +41,8 @@ class ClassManager(dict):
     MANAGER_ATTR = base.DEFAULT_MANAGER_ATTR
     STATE_ATTR = base.DEFAULT_STATE_ATTR
 
+    _state_setter = staticmethod(util.attrsetter(STATE_ATTR))
+
     deferred_scalar_loader = None
 
     original_init = object.__init__
@@ -288,15 +290,15 @@ class ClassManager(dict):
 
     def new_instance(self, state=None):
         instance = self.class_.__new__(self.class_)
-        setattr(instance, self.STATE_ATTR,
-                self._state_constructor(instance, self)
-                if not state else state)
+        if state is None:
+            state = self._state_constructor(instance, self)
+        self._state_setter(instance, state)
         return instance
 
     def setup_instance(self, instance, state=None):
-        setattr(instance, self.STATE_ATTR,
-                self._state_constructor(instance, self)
-                if not state else state)
+        if state is None:
+            state = self._state_constructor(instance, self)
+        self._state_setter(instance, state)
 
     def teardown_instance(self, instance):
         delattr(instance, self.STATE_ATTR)
@@ -323,7 +325,7 @@ class ClassManager(dict):
                 _new_state_if_none(instance)
         else:
             state = self._state_constructor(instance, self)
-            setattr(instance, self.STATE_ATTR, state)
+            self._state_setter(instance, state)
             return state
 
     def has_state(self, instance):
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index 49ec99ce4..47ee4c076 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -82,11 +82,11 @@ class MapperProperty(_MappedAttribute, InspectionAttr):
         pass
 
     def create_row_processor(self, context, path,
-                             mapper, row, adapter):
+                             mapper, result, adapter, populators):
         """Return a 3-tuple consisting of three row processing functions.
 
         """
-        return None, None, None
+        pass
 
     def cascade_iterator(self, type_, state, visited_instances=None,
                          halt_on=None):
@@ -443,14 +443,17 @@ class StrategizedProperty(MapperProperty):
             strat = self.strategy
         strat.setup_query(context, entity, path, loader, adapter, **kwargs)
 
-    def create_row_processor(self, context, path, mapper, row, adapter):
+    def create_row_processor(
+            self, context, path, mapper,
+            result, adapter, populators):
         loader = self._get_context_loader(context, path)
         if loader and loader.strategy:
             strat = self._get_strategy(loader.strategy)
         else:
             strat = self.strategy
-        return strat.create_row_processor(context, path, loader,
-                                          mapper, row, adapter)
+        strat.create_row_processor(
+            context, path, loader,
+            mapper, result, adapter, populators)
 
     def do_init(self):
         self._strategies = {}
@@ -543,14 +546,14 @@ class LoaderStrategy(object):
         pass
 
     def create_row_processor(self, context, path, loadopt, mapper,
-                             row, adapter):
+                             result, adapter, populators):
         """Return row processing functions which fulfill the contract
         specified by MapperProperty.create_row_processor.
 
         StrategizedProperty delegates its create_row_processor method
         directly to this method. """
 
-        return None, None, None
+        pass
 
     def __str__(self):
         return str(self.parent_property)
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 232eb89de..380afcdc7 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -12,26 +12,24 @@ the functions here are called primarily by Query, Mapper,
 as well as some of the attribute loading strategies.
 
 """
-
+from __future__ import absolute_import
 
 from .. import util
-from . import attributes, exc as orm_exc, state as statelib
-from .interfaces import EXT_CONTINUE
+from . import attributes, exc as orm_exc
 from ..sql import util as sql_util
 from .util import _none_set, state_str
 from .. import exc as sa_exc
+import collections
 
 _new_runid = util.counter()
 
 
 def instances(query, cursor, context):
     """Return an ORM result as an iterator."""
-    session = query.session
 
     context.runid = _new_runid()
 
-    filter_fns = [ent.filter_fn
-                  for ent in query._entities]
+    filter_fns = [ent.filter_fn for ent in query._entities]
     filtered = id in filter_fns
 
     single_entity = len(query._entities) == 1 and \
@@ -44,18 +42,17 @@ def instances(query, cursor, context):
             def filter_fn(row):
                 return tuple(fn(x) for x, fn in zip(row, filter_fns))
 
-    custom_rows = single_entity and \
-        query._entities[0].custom_rows
-
     (process, labels) = \
         list(zip(*[
             query_entity.row_processor(query,
-                                       context, custom_rows)
+                                       context, cursor)
             for query_entity in query._entities
         ]))
 
+    if not single_entity:
+        keyed_tuple = util.lightweight_named_tuple('result', labels)
+
     while True:
-        context.progress = {}
         context.partials = {}
 
         if query._yield_per:
@@ -65,33 +62,16 @@ def instances(query, cursor, context):
         else:
             fetch = cursor.fetchall()
 
-        if custom_rows:
-            rows = []
-            for row in fetch:
-                process[0](row, rows)
-        elif single_entity:
-            rows = [process[0](row, None) for row in fetch]
+        if single_entity:
+            proc = process[0]
+            rows = [proc(row) for row in fetch]
         else:
-            rows = [util.KeyedTuple([proc(row, None) for proc in process],
-                                    labels) for row in fetch]
+            rows = [keyed_tuple([proc(row) for proc in process])
+                    for row in fetch]
 
         if filtered:
             rows = util.unique_list(rows, filter_fn)
 
-        if context.refresh_state and query._only_load_props \
-                and context.refresh_state in context.progress:
-            context.refresh_state._commit(
-                context.refresh_state.dict, query._only_load_props)
-            context.progress.pop(context.refresh_state)
-
-        statelib.InstanceState._commit_all_states(
-            context.progress.items(),
-            session.identity_map
-        )
-
-        for state, (dict_, attrs) in context.partials.items():
-            state._commit(dict_, attrs)
-
         for row in rows:
             yield row
 
@@ -126,6 +106,7 @@ def merge_result(querylib, query, iterator, load=True):
                                if isinstance(e, querylib._MapperEntity)]
             result = []
             keys = [ent._label_name for ent in query._entities]
+            keyed_tuple = util.lightweight_named_tuple('result', keys)
             for row in iterator:
                 newrow = list(row)
                 for i in mapped_entities:
@@ -134,7 +115,7 @@ def merge_result(querylib, query, iterator, load=True):
                             attributes.instance_state(newrow[i]),
                             attributes.instance_dict(newrow[i]),
                             load=load, _recursive={})
-                result.append(util.KeyedTuple(newrow, keys))
+                result.append(keyed_tuple(newrow))
 
         return iter(result)
     finally:
@@ -233,11 +214,10 @@ def load_on_ident(query, key,
         return None
 
 
-def instance_processor(mapper, context, path, adapter,
-                       polymorphic_from=None,
-                       only_load_props=None,
-                       refresh_state=None,
-                       polymorphic_discriminator=None):
+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."""
 
@@ -249,292 +229,264 @@ def instance_processor(mapper, context, path, adapter,
 
     pk_cols = mapper.primary_key
 
-    if polymorphic_from or refresh_state:
-        polymorphic_on = None
-    else:
-        if polymorphic_discriminator is not None:
-            polymorphic_on = polymorphic_discriminator
-        else:
-            polymorphic_on = mapper.polymorphic_on
-        polymorphic_instances = util.PopulateDict(
-            _configure_subclass_mapper(
-                mapper,
-                context, path, adapter)
-        )
-
-    version_id_col = mapper.version_id_col
-
     if adapter:
         pk_cols = [adapter.columns[c] for c in pk_cols]
-        if polymorphic_on is not None:
-            polymorphic_on = adapter.columns[polymorphic_on]
-        if version_id_col is not None:
-            version_id_col = adapter.columns[version_id_col]
 
     identity_class = mapper._identity_class
 
-    new_populators = []
-    existing_populators = []
-    eager_populators = []
+    populators = collections.defaultdict(list)
 
-    load_path = context.query._current_path + path \
-        if context.query._current_path.path \
-        else path
+    props = mapper._props.values()
+    if only_load_props is not None:
+        props = (p for p in props if p.key in only_load_props)
 
-    def populate_state(state, dict_, row, isnew, only_load_props):
-        if isnew:
-            if context.propagate_options:
-                state.load_options = context.propagate_options
-            if state.load_options:
-                state.load_path = load_path
+    for prop in props:
+        prop.create_row_processor(
+            context, path, mapper, result, adapter, populators)
 
-        if not new_populators:
-            _populators(mapper, context, path, row, adapter,
-                        new_populators,
-                        existing_populators,
-                        eager_populators
-                        )
-
-        if isnew:
-            populators = new_populators
-        else:
-            populators = existing_populators
-
-        if only_load_props is None:
-            for key, populator in populators:
-                populator(state, dict_, row)
-        elif only_load_props:
-            for key, populator in populators:
-                if key in only_load_props:
-                    populator(state, dict_, row)
+    propagate_options = context.propagate_options
+    if propagate_options:
+        load_path = context.query._current_path + path \
+            if context.query._current_path.path else path
 
     session_identity_map = context.session.identity_map
 
-    listeners = mapper.dispatch
-
-    # legacy events - I'd very much like to yank these totally
-    translate_row = listeners.translate_row or None
-    create_instance = listeners.create_instance or None
-    populate_instance = listeners.populate_instance or None
-    append_result = listeners.append_result or None
-    ####
-
     populate_existing = context.populate_existing or mapper.always_refresh
-    invoke_all_eagers = context.invoke_all_eagers
-    load_evt = mapper.class_manager.dispatch.load or None
-    refresh_evt = mapper.class_manager.dispatch.refresh or None
+    load_evt = bool(mapper.class_manager.dispatch.load)
+    refresh_evt = bool(mapper.class_manager.dispatch.refresh)
     instance_state = attributes.instance_state
     instance_dict = attributes.instance_dict
+    session_id = context.session.hash_key
+    version_check = context.version_check
+    runid = context.runid
+
+    if refresh_state:
+        refresh_identity_key = refresh_state.key
+        if refresh_identity_key is None:
+            # super-rare condition; a refresh is being called
+            # on a non-instance-key instance; this is meant to only
+            # occur within a flush()
+            refresh_identity_key = \
+                mapper._identity_key_from_state(refresh_state)
+    else:
+        refresh_identity_key = None
 
     if mapper.allow_partial_pks:
         is_not_primary_key = _none_set.issuperset
     else:
         is_not_primary_key = _none_set.intersection
 
-    def _instance(row, result):
-        if not new_populators and invoke_all_eagers:
-            _populators(mapper, context, path, row, adapter,
-                        new_populators,
-                        existing_populators,
-                        eager_populators)
-
-        if translate_row:
-            for fn in translate_row:
-                ret = fn(mapper, context, row)
-                if ret is not EXT_CONTINUE:
-                    row = ret
-                    break
-
-        if polymorphic_on is not None:
-            discriminator = row[polymorphic_on]
-            if discriminator is not None:
-                _instance = polymorphic_instances[discriminator]
-                if _instance:
-                    return _instance(row, result)
-
-        # determine identity key
-        if refresh_state:
-            identitykey = refresh_state.key
-            if identitykey is None:
-                # super-rare condition; a refresh is being called
-                # on a non-instance-key instance; this is meant to only
-                # occur within a flush()
-                identitykey = mapper._identity_key_from_state(refresh_state)
+    def _instance(row):
+
+        # determine the state that we'll be populating
+        if refresh_identity_key:
+            # fixed state that we're refreshing
+            state = refresh_state
+            instance = state.obj()
+            dict_ = instance_dict(instance)
+            isnew = state.runid != runid
+            currentload = True
+            loaded_instance = False
         else:
+            # look at the row, see if that identity is in the
+            # session, or we have to create a new one
             identitykey = (
                 identity_class,
                 tuple([row[column] for column in pk_cols])
             )
 
-        instance = session_identity_map.get(identitykey)
+            instance = session_identity_map.get(identitykey)
 
-        if instance is not None:
-            state = instance_state(instance)
-            dict_ = instance_dict(instance)
+            if instance is not None:
+                # existing instance
+                state = instance_state(instance)
+                dict_ = instance_dict(instance)
 
-            isnew = state.runid != context.runid
-            currentload = not isnew
-            loaded_instance = False
+                isnew = state.runid != runid
+                currentload = not isnew
+                loaded_instance = False
 
-            if not currentload and \
-                    version_id_col is not None and \
-                    context.version_check and \
-                    mapper._get_state_attr_by_column(
-                        state,
-                        dict_,
-                        mapper.version_id_col) != \
-                    row[version_id_col]:
-
-                raise orm_exc.StaleDataError(
-                    "Instance '%s' has version id '%s' which "
-                    "does not match database-loaded version id '%s'."
-                    % (state_str(state),
-                       mapper._get_state_attr_by_column(
-                        state, dict_,
-                        mapper.version_id_col),
-                       row[version_id_col]))
-        elif refresh_state:
-            # out of band refresh_state detected (i.e. its not in the
-            # session.identity_map) honor it anyway.  this can happen
-            # if a _get() occurs within save_obj(), such as
-            # when eager_defaults is True.
-            state = refresh_state
-            instance = state.obj()
-            dict_ = instance_dict(instance)
-            isnew = state.runid != context.runid
-            currentload = True
-            loaded_instance = False
-        else:
-            # check for non-NULL values in the primary key columns,
-            # else no entity is returned for the row
-            if is_not_primary_key(identitykey[1]):
-                return None
+                if version_check and not currentload:
+                    _validate_version_id(mapper, state, dict_, row, adapter)
 
-            isnew = True
-            currentload = True
-            loaded_instance = True
-
-            if create_instance:
-                for fn in create_instance:
-                    instance = fn(mapper, context,
-                                  row, mapper.class_)
-                    if instance is not EXT_CONTINUE:
-                        manager = attributes.manager_of_class(
-                            instance.__class__)
-                        # TODO: if manager is None, raise a friendly error
-                        # about returning instances of unmapped types
-                        manager.setup_instance(instance)
-                        break
-                else:
-                    instance = mapper.class_manager.new_instance()
             else:
+                # create a new instance
+
+                # check for non-NULL values in the primary key columns,
+                # else no entity is returned for the row
+                if is_not_primary_key(identitykey[1]):
+                    return None
+
+                isnew = True
+                currentload = True
+                loaded_instance = True
+
                 instance = mapper.class_manager.new_instance()
 
-            dict_ = instance_dict(instance)
-            state = instance_state(instance)
-            state.key = identitykey
+                dict_ = instance_dict(instance)
+                state = instance_state(instance)
+                state.key = identitykey
 
-            # attach instance to session.
-            state.session_id = context.session.hash_key
-            session_identity_map._add_unpresent(state, identitykey)
+                # attach instance to session.
+                state.session_id = session_id
+                session_identity_map._add_unpresent(state, identitykey)
 
+        # populate.  this looks at whether this state is new
+        # for this load or was existing, and whether or not this
+        # row is the first row with this identity.
         if currentload or populate_existing:
-            # state is being fully loaded, so populate.
-            # add to the "context.progress" collection.
+            # full population routines.  Objects here are either
+            # just created, or we are doing a populate_existing
+
+            if isnew and propagate_options:
+                state.load_options = propagate_options
+                state.load_path = load_path
+
+            _populate_full(
+                context, row, state, dict_, isnew,
+                loaded_instance, populate_existing, populators)
+
             if isnew:
-                state.runid = context.runid
-                context.progress[state] = dict_
-
-            if populate_instance:
-                for fn in populate_instance:
-                    ret = fn(mapper, context, row, state,
-                             only_load_props=only_load_props,
-                             instancekey=identitykey, isnew=isnew)
-                    if ret is not EXT_CONTINUE:
-                        break
-                else:
-                    populate_state(state, dict_, row, isnew, only_load_props)
-            else:
-                populate_state(state, dict_, row, isnew, only_load_props)
-
-            if loaded_instance and load_evt:
-                state.manager.dispatch.load(state, context)
-            elif isnew and refresh_evt:
-                state.manager.dispatch.refresh(
-                    state, context, only_load_props)
-
-        elif state in context.partials or state.unloaded or eager_populators:
-            # state is having a partial set of its attributes
-            # refreshed.  Populate those attributes,
-            # and add to the "context.partials" collection.
-            if state in context.partials:
-                isnew = False
-                (d_, attrs) = context.partials[state]
-            else:
-                isnew = True
-                attrs = state.unloaded
-                context.partials[state] = (dict_, attrs)
-
-            if populate_instance:
-                for fn in populate_instance:
-                    ret = fn(mapper, context, row, state,
-                             only_load_props=attrs,
-                             instancekey=identitykey, isnew=isnew)
-                    if ret is not EXT_CONTINUE:
-                        break
-                else:
-                    populate_state(state, dict_, row, isnew, attrs)
-            else:
-                populate_state(state, dict_, row, isnew, attrs)
-
-            for key, pop in eager_populators:
-                if key not in state.unloaded:
-                    pop(state, dict_, row)
-
-            if isnew and refresh_evt:
-                state.manager.dispatch.refresh(state, context, attrs)
-
-        if result is not None:
-            if append_result:
-                for fn in append_result:
-                    if fn(mapper, context, row, state,
-                          result, instancekey=identitykey,
-                          isnew=isnew) is not EXT_CONTINUE:
-                        break
-                else:
-                    result.append(instance)
-            else:
-                result.append(instance)
+                if loaded_instance and load_evt:
+                    state.manager.dispatch.load(state, context)
+                elif refresh_evt:
+                    state.manager.dispatch.refresh(
+                        state, context, only_load_props)
+
+                if populate_existing or state.modified:
+                    if refresh_state and only_load_props:
+                        state._commit(dict_, only_load_props)
+                    else:
+                        state._commit_all(dict_, session_identity_map)
+
+        else:
+            # partial population routines, for objects that were already
+            # in the Session, but a row matches them; apply eager loaders
+            # on existing objects, etc.
+            unloaded = state.unloaded
+            isnew = state not in context.partials
+
+            if not isnew or unloaded or populators["eager"]:
+                # state is having a partial set of its attributes
+                # refreshed.  Populate those attributes,
+                # and add to the "context.partials" collection.
+
+                to_load = _populate_partial(
+                    context, row, state, dict_, isnew,
+                    unloaded, populators)
+
+                if isnew:
+                    if refresh_evt:
+                        state.manager.dispatch.refresh(
+                            state, context, to_load)
+
+                    state._commit(dict_, to_load)
 
         return instance
+
+    if 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(
+            _instance, context, mapper, result, path,
+            polymorphic_discriminator, adapter)
+
     return _instance
 
 
-def _populators(mapper, context, path, row, adapter,
-                new_populators, existing_populators, eager_populators):
-    """Produce a collection of attribute level row processor
-    callables."""
+def _populate_full(
+        context, row, state, dict_, isnew,
+        loaded_instance, populate_existing, populators):
+    if isnew:
+        # first time we are seeing a row with this identity.
+        state.runid = context.runid
+
+        for key, getter in populators["quick"]:
+            dict_[key] = getter(row)
+        if populate_existing:
+            for key, set_callable in populators["expire"]:
+                dict_.pop(key, None)
+                if set_callable:
+                    state.callables[key] = state
+        else:
+            for key, set_callable in populators["expire"]:
+                if set_callable:
+                    state.callables[key] = state
+        for key, populator in populators["new"]:
+            populator(state, dict_, row)
+        for key, populator in populators["delayed"]:
+            populator(state, dict_, row)
+    else:
+        # have already seen rows with this identity.
+        for key, populator in populators["existing"]:
+            populator(state, dict_, row)
+
+
+def _populate_partial(
+        context, row, state, dict_, isnew,
+        unloaded, populators):
+    if not isnew:
+        to_load = context.partials[state]
+        for key, populator in populators["existing"]:
+            if key in to_load:
+                populator(state, dict_, row)
+    else:
+        to_load = unloaded
+        context.partials[state] = to_load
+
+        for key, getter in populators["quick"]:
+            if key in to_load:
+                dict_[key] = getter(row)
+        for key, set_callable in populators["expire"]:
+            if key in to_load:
+                dict_.pop(key, None)
+                if set_callable:
+                    state.callables[key] = state
+        for key, populator in populators["new"]:
+            if key in to_load:
+                populator(state, dict_, row)
+        for key, populator in populators["delayed"]:
+            if key in to_load:
+                populator(state, dict_, row)
+    for key, populator in populators["eager"]:
+        if key not in unloaded:
+            populator(state, dict_, row)
+
+    return to_load
 
-    delayed_populators = []
-    pops = (new_populators, existing_populators, delayed_populators,
-            eager_populators)
 
-    for prop in mapper._props.values():
+def _validate_version_id(mapper, state, dict_, row, adapter):
 
-        for i, pop in enumerate(prop.create_row_processor(
-                context,
-                path,
-                mapper, row, adapter)):
-            if pop is not None:
-                pops[i].append((prop.key, pop))
+    version_id_col = mapper.version_id_col
 
-    if delayed_populators:
-        new_populators.extend(delayed_populators)
+    if version_id_col is None:
+        return
 
+    if adapter:
+        version_id_col = adapter.columns[version_id_col]
+
+    if mapper._get_state_attr_by_column(
+            state, dict_, mapper.version_id_col) != row[version_id_col]:
+        raise orm_exc.StaleDataError(
+            "Instance '%s' has version id '%s' which "
+            "does not match database-loaded version id '%s'."
+            % (state_str(state), mapper._get_state_attr_by_column(
+               state, dict_, mapper.version_id_col),
+               row[version_id_col]))
+
+
+def _decorate_polymorphic_switch(
+        instance_fn, context, mapper, result, path,
+        polymorphic_discriminator, adapter):
+    if polymorphic_discriminator is not None:
+        polymorphic_on = polymorphic_discriminator
+    else:
+        polymorphic_on = mapper.polymorphic_on
+    if polymorphic_on is None:
+        return instance_fn
 
-def _configure_subclass_mapper(mapper, context, path, adapter):
-    """Produce a mapper level row processor callable factory for mappers
-    inheriting this one."""
+    if adapter:
+        polymorphic_on = adapter.columns[polymorphic_on]
 
     def configure_subclass_mapper(discriminator):
         try:
@@ -543,16 +495,26 @@ def _configure_subclass_mapper(mapper, context, path, adapter):
             raise AssertionError(
                 "No such polymorphic_identity %r is defined" %
                 discriminator)
-        if sub_mapper is mapper:
-            return None
+        else:
+            if sub_mapper is mapper:
+                return None
+
+            return instance_processor(
+                sub_mapper, context, result,
+                path, adapter, _polymorphic_from=mapper)
+
+    polymorphic_instances = util.PopulateDict(
+        configure_subclass_mapper
+    )
 
-        return instance_processor(
-            sub_mapper,
-            context,
-            path,
-            adapter,
-            polymorphic_from=mapper)
-    return configure_subclass_mapper
+    def polymorphic_instance(row):
+        discriminator = row[polymorphic_on]
+        if discriminator is not None:
+            _instance = polymorphic_instances[discriminator]
+            if _instance:
+                return _instance(row)
+        return instance_fn(row)
+    return polymorphic_instance
 
 
 def load_scalar_attributes(mapper, state, attribute_names):
@@ -600,10 +562,11 @@ def load_scalar_attributes(mapper, state, attribute_names):
         if (_none_set.issubset(identity_key) and
                 not mapper.allow_partial_pks) or \
                 _none_set.issuperset(identity_key):
-            util.warn("Instance %s to be refreshed doesn't "
-                      "contain a full primary key - can't be refreshed "
-                      "(and shouldn't be expired, either)."
-                      % state_str(state))
+            util.warn_limited(
+                "Instance %s to be refreshed doesn't "
+                "contain a full primary key - can't be refreshed "
+                "(and shouldn't be expired, either).",
+                state_str(state))
             return
 
         result = load_on_ident(
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 31c17e69e..984f05256 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -1127,7 +1127,6 @@ class Mapper(InspectionAttr):
 
         event.listen(manager, 'first_init', _event_on_first_init, raw=True)
         event.listen(manager, 'init', _event_on_init, raw=True)
-        event.listen(manager, 'resurrect', _event_on_resurrect, raw=True)
 
         for key, method in util.iterate_attributes(self.class_):
             if isinstance(method, types.FunctionType):
@@ -1453,13 +1452,11 @@ class Mapper(InspectionAttr):
                 if polymorphic_key in dict_ and \
                         dict_[polymorphic_key] not in \
                         mapper._acceptable_polymorphic_identities:
-                    util.warn(
+                    util.warn_limited(
                         "Flushing object %s with "
                         "incompatible polymorphic identity %r; the "
-                        "object may not refresh and/or load correctly" % (
-                            state_str(state),
-                            dict_[polymorphic_key]
-                        )
+                        "object may not refresh and/or load correctly",
+                        (state_str(state), dict_[polymorphic_key])
                     )
 
             self._set_polymorphic_identity = _set_polymorphic_identity
@@ -2287,6 +2284,16 @@ class Mapper(InspectionAttr):
     def primary_base_mapper(self):
         return self.class_manager.mapper.base_mapper
 
+    def _result_has_identity_key(self, result, adapter=None):
+        pk_cols = self.primary_key
+        if adapter:
+            pk_cols = [adapter.columns[c] for c in pk_cols]
+        for col in pk_cols:
+            if not result._has_key(col):
+                return False
+        else:
+            return True
+
     def identity_key_from_row(self, row, adapter=None):
         """Return an identity-map key for use in storing/retrieving an
         item from the identity map.
@@ -2770,16 +2777,6 @@ def _event_on_init(state, args, kwargs):
             instrumenting_mapper._set_polymorphic_identity(state)
 
 
-def _event_on_resurrect(state):
-    # re-populate the primary key elements
-    # of the dict based on the mapping.
-    instrumenting_mapper = state.manager.info.get(_INSTRUMENTOR)
-    if instrumenting_mapper:
-        for col, val in zip(instrumenting_mapper.primary_key, state.key[1]):
-            instrumenting_mapper._set_state_attr_by_column(
-                state, state.dict, col, val)
-
-
 class _ColumnMapping(dict):
     """Error reporting helper for mapper._columntoproperty."""
 
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index aa10da9f4..198eeb46f 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -659,8 +659,7 @@ def _emit_update_statements(base_mapper, uowtransaction,
         elif needs_version_id:
             util.warn("Dialect %s does not support updated rowcount "
                       "- versioning cannot be verified." %
-                      c.dialect.dialect_description,
-                      stacklevel=12)
+                      c.dialect.dialect_description)
 
 
 def _emit_insert_statements(base_mapper, uowtransaction,
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 12e11b26c..ba557ef79 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -218,7 +218,7 @@ class Query(object):
     def _adapt_col_list(self, cols):
         return [
             self._adapt_clause(
-                expression._literal_as_text(o),
+                expression._literal_as_label_reference(o),
                 True, True)
             for o in cols
         ]
@@ -595,11 +595,19 @@ class Query(object):
 
         This is used primarily when nesting the Query's
         statement into a subquery or other
-        selectable.
+        selectable, or when using :meth:`.Query.yield_per`.
 
         """
         self._enable_eagerloads = value
 
+    def _no_yield_per(self, message):
+        raise sa_exc.InvalidRequestError(
+            "The yield_per Query option is currently not "
+            "compatible with %s eager loading.  Please "
+            "specify lazyload('*') or query.enable_eagerloads(False) in "
+            "order to "
+            "proceed with query.yield_per()." % message)
+
     @_generative()
     def with_labels(self):
         """Apply column labels to the return value of Query.statement.
@@ -705,24 +713,58 @@ class Query(object):
     def yield_per(self, count):
         """Yield only ``count`` rows at a time.
 
-        WARNING: use this method with caution; if the same instance is present
-        in more than one batch of rows, end-user changes to attributes will be
-        overwritten.
+        The purpose of this method is when fetching very large result sets
+        (> 10K rows), to batch results in sub-collections and yield them
+        out partially, so that the Python interpreter doesn't need to declare
+        very large areas of memory which is both time consuming and leads
+        to excessive memory use.   The performance from fetching hundreds of
+        thousands of rows can often double when a suitable yield-per setting
+        (e.g. approximately 1000) is used, even with DBAPIs that buffer
+        rows (which are most).
+
+        The :meth:`.Query.yield_per` method **is not compatible with most
+        eager loading schemes, including subqueryload and joinedload with
+        collections**.  For this reason, it may be helpful to disable
+        eager loads, either unconditionally with
+        :meth:`.Query.enable_eagerloads`::
+
+            q = sess.query(Object).yield_per(100).enable_eagerloads(False)
+
+        Or more selectively using :func:`.lazyload`; such as with
+        an asterisk to specify the default loader scheme::
+
+            q = sess.query(Object).yield_per(100).\\
+                options(lazyload('*'), joinedload(Object.some_related))
+
+        .. warning::
+
+            Use this method with caution; if the same instance is
+            present in more than one batch of rows, end-user changes
+            to attributes will be overwritten.
+
+            In particular, it's usually impossible to use this setting
+            with eagerly loaded collections (i.e. any lazy='joined' or
+            'subquery') since those collections will be cleared for a
+            new load when encountered in a subsequent result batch.
+            In the case of 'subquery' loading, the full result for all
+            rows is fetched which generally defeats the purpose of
+            :meth:`~sqlalchemy.orm.query.Query.yield_per`.
+
+            Also note that while
+            :meth:`~sqlalchemy.orm.query.Query.yield_per` will set the
+            ``stream_results`` execution option to True, currently
+            this is only understood by
+            :mod:`~sqlalchemy.dialects.postgresql.psycopg2` dialect
+            which will stream results using server side cursors
+            instead of pre-buffer all rows for this query. Other
+            DBAPIs **pre-buffer all rows** before making them
+            available.  The memory use of raw database rows is much less
+            than that of an ORM-mapped object, but should still be taken into
+            consideration when benchmarking.
 
-        In particular, it's usually impossible to use this setting with
-        eagerly loaded collections (i.e. any lazy='joined' or 'subquery')
-        since those collections will be cleared for a new load when
-        encountered in a subsequent result batch.   In the case of 'subquery'
-        loading, the full result for all rows is fetched which generally
-        defeats the purpose of :meth:`~sqlalchemy.orm.query.Query.yield_per`.
+        .. seealso::
 
-        Also note that while :meth:`~sqlalchemy.orm.query.Query.yield_per`
-        will set the ``stream_results`` execution option to True, currently
-        this is only understood by
-        :mod:`~sqlalchemy.dialects.postgresql.psycopg2` dialect which will
-        stream results using server side cursors instead of pre-buffer all
-        rows for this query. Other DBAPIs pre-buffer all rows before making
-        them available.
+            :meth:`.Query.enable_eagerloads`
 
         """
         self._yield_per = count
@@ -945,9 +987,9 @@ class Query(object):
 
         """
         fromclause = self.with_labels().enable_eagerloads(False).\
-            _set_enable_single_crit(False).\
             statement.correlate(None)
         q = self._from_selectable(fromclause)
+        q._enable_single_crit = False
         if entities:
             q._set_entities(entities)
         return q
@@ -1240,7 +1282,7 @@ class Query(object):
 
         """
         for criterion in list(criterion):
-            criterion = expression._literal_as_text(criterion)
+            criterion = expression._expression_literal_as_text(criterion)
 
             criterion = self._adapt_clause(criterion, True, True)
 
@@ -1339,8 +1381,7 @@ class Query(object):
 
         """
 
-        if isinstance(criterion, util.string_types):
-            criterion = sql.text(criterion)
+        criterion = expression._expression_literal_as_text(criterion)
 
         if criterion is not None and \
                 not isinstance(criterion, sql.ClauseElement):
@@ -2306,13 +2347,18 @@ class Query(object):
         This method bypasses all internal statement compilation, and the
         statement is executed without modification.
 
-        The statement argument is either a string, a ``select()`` construct,
-        or a ``text()`` construct, and should return the set of columns
-        appropriate to the entity class represented by this ``Query``.
+        The statement is typically either a :func:`~.expression.text`
+        or :func:`~.expression.select` construct, and should return the set
+        of columns
+        appropriate to the entity class represented by this :class:`.Query`.
+
+        .. seealso::
+
+            :ref:`orm_tutorial_literal_sql` - usage examples in the
+            ORM tutorial
 
         """
-        if isinstance(statement, util.string_types):
-            statement = sql.text(statement)
+        statement = expression._expression_literal_as_text(statement)
 
         if not isinstance(statement,
                           (expression.TextClause,
@@ -2558,7 +2604,7 @@ class Query(object):
         # .with_only_columns() after we have a core select() so that
         # we get just "SELECT 1" without any entities.
         return sql.exists(self.add_columns('1').with_labels().
-                          statement.with_only_columns(['1']))
+                          statement.with_only_columns([1]))
 
     def count(self):
         """Return a count of rows this Query would return.
@@ -3003,7 +3049,6 @@ class _MapperEntity(_QueryEntity):
         else:
             self._label_name = self.mapper.class_.__name__
         self.path = self.entity_zero._path_registry
-        self.custom_rows = bool(self.mapper.dispatch.append_result)
 
     def set_with_polymorphic(self, query, cls_or_mappers,
                              selectable, polymorphic_on):
@@ -3082,7 +3127,7 @@ class _MapperEntity(_QueryEntity):
 
         return ret
 
-    def row_processor(self, query, context, custom_rows):
+    def row_processor(self, query, context, result):
         adapter = self._get_entity_clauses(query, context)
 
         if context.adapter and adapter:
@@ -3102,6 +3147,7 @@ class _MapperEntity(_QueryEntity):
             _instance = loading.instance_processor(
                 self.mapper,
                 context,
+                result,
                 self.path,
                 adapter,
                 only_load_props=query._only_load_props,
@@ -3112,6 +3158,7 @@ class _MapperEntity(_QueryEntity):
             _instance = loading.instance_processor(
                 self.mapper,
                 context,
+                result,
                 self.path,
                 adapter,
                 polymorphic_discriminator=self._polymorphic_discriminator
@@ -3275,9 +3322,10 @@ class Bundle(object):
             :ref:`bundles` - includes an example of subclassing.
 
         """
-        def proc(row, result):
-            return util.KeyedTuple(
-                [proc(row, None) for proc in procs], labels)
+        keyed_tuple = util.lightweight_named_tuple('result', labels)
+
+        def proc(row):
+            return keyed_tuple([proc(row) for proc in procs])
         return proc
 
 
@@ -3302,7 +3350,6 @@ class _BundleEntity(_QueryEntity):
 
         self.supports_single_entity = self.bundle.single_entity
 
-    custom_rows = False
 
     @property
     def entity_zero(self):
@@ -3344,9 +3391,9 @@ class _BundleEntity(_QueryEntity):
         for ent in self._entities:
             ent.setup_context(query, context)
 
-    def row_processor(self, query, context, custom_rows):
+    def row_processor(self, query, context, result):
         procs, labels = zip(
-            *[ent.row_processor(query, context, custom_rows)
+            *[ent.row_processor(query, context, result)
               for ent in self._entities]
         )
 
@@ -3436,7 +3483,6 @@ class _ColumnEntity(_QueryEntity):
             self.entity_zero = None
 
     supports_single_entity = False
-    custom_rows = False
 
     @property
     def entity_zero_or_selectable(self):
@@ -3473,17 +3519,15 @@ class _ColumnEntity(_QueryEntity):
     def _resolve_expr_against_query_aliases(self, query, expr, context):
         return query._adapt_clause(expr, False, True)
 
-    def row_processor(self, query, context, custom_rows):
+    def row_processor(self, query, context, result):
         column = self._resolve_expr_against_query_aliases(
             query, self.column, context)
 
         if context.adapter:
             column = context.adapter.columns[column]
 
-        def proc(row, result):
-            return row[column]
-
-        return proc, self._label_name
+        getter = result._getter(column)
+        return getter, self._label_name
 
     def setup_context(self, query, context):
         column = self._resolve_expr_against_query_aliases(
diff --git a/lib/sqlalchemy/orm/relationships.py b/lib/sqlalchemy/orm/relationships.py
index c2debda03..2bcb3f4a1 100644
--- a/lib/sqlalchemy/orm/relationships.py
+++ b/lib/sqlalchemy/orm/relationships.py
@@ -459,22 +459,18 @@ class RelationshipProperty(StrategizedProperty):
           nullable, or when the reference is one-to-one or a collection that
           is guaranteed to have one or at least one entry.
 
-          If the joined-eager load is chained onto an existing LEFT OUTER
-          JOIN, ``innerjoin=True`` will be bypassed and the join will continue
-          to chain as LEFT OUTER JOIN so that the results don't change.  As an
-          alternative, specify the value ``"nested"``.  This will instead nest
-          the join on the right side, e.g. using the form "a LEFT OUTER JOIN
-          (b JOIN c)".
-
-          .. versionadded:: 0.9.4 Added ``innerjoin="nested"`` option to
-             support nesting of eager "inner" joins.
+          The option supports the same "nested" and "unnested" options as
+          that of :paramref:`.joinedload.innerjoin`.  See that flag
+          for details on nested / unnested behaviors.
 
           .. seealso::
 
+            :paramref:`.joinedload.innerjoin` - the option as specified by
+            loader option, including detail on nesting behavior.
+
             :ref:`what_kind_of_loading` - Discussion of some details of
             various loader options.
 
-            :paramref:`.joinedload.innerjoin` - loader option version
 
         :param join_depth:
           when non-``None``, an integer value indicating how many levels
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 968868e84..e075b9c71 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -271,7 +271,7 @@ class SessionTransaction(object):
                 del s.key
 
         for s, (oldkey, newkey) in self._key_switches.items():
-            self.session.identity_map.discard(s)
+            self.session.identity_map.safe_discard(s)
             s.key = oldkey
             self.session.identity_map.replace(s)
 
@@ -1397,7 +1397,7 @@ class Session(_SessionClassMethods):
             self._new.pop(state)
             state._detach()
         elif self.identity_map.contains_state(state):
-            self.identity_map.discard(state)
+            self.identity_map.safe_discard(state)
             self._deleted.pop(state, None)
             state._detach()
         elif self.transaction:
@@ -1430,10 +1430,10 @@ class Session(_SessionClassMethods):
                 if state.key is None:
                     state.key = instance_key
                 elif state.key != instance_key:
-                    # primary key switch. use discard() in case another
+                    # primary key switch. use safe_discard() in case another
                     # state has already replaced this one in the identity
                     # map (see test/orm/test_naturalpks.py ReversePKsTest)
-                    self.identity_map.discard(state)
+                    self.identity_map.safe_discard(state)
                     if state in self.transaction._key_switches:
                         orig_key = self.transaction._key_switches[state][0]
                     else:
@@ -1467,7 +1467,7 @@ class Session(_SessionClassMethods):
             if self._enable_transaction_accounting and self.transaction:
                 self.transaction._deleted[state] = True
 
-            self.identity_map.discard(state)
+            self.identity_map.safe_discard(state)
             self._deleted.pop(state, None)
             state.deleted = True
 
diff --git a/lib/sqlalchemy/orm/state.py b/lib/sqlalchemy/orm/state.py
index fe8ccd222..3c12fda1a 100644
--- a/lib/sqlalchemy/orm/state.py
+++ b/lib/sqlalchemy/orm/state.py
@@ -58,7 +58,6 @@ class InstanceState(interfaces.InspectionAttr):
     expired = False
     deleted = False
     _load_pending = False
-
     is_instance = True
 
     def __init__(self, obj, manager):
@@ -221,7 +220,7 @@ class InstanceState(interfaces.InspectionAttr):
 
     def _cleanup(self, ref):
         instance_dict = self._instance_dict()
-        if instance_dict:
+        if instance_dict is not None:
             instance_dict.discard(self)
 
         self.callables.clear()
@@ -335,20 +334,6 @@ class InstanceState(interfaces.InspectionAttr):
             self.manager[key].impl._invalidate_collection(old)
         self.callables.pop(key, None)
 
-    def _expire_attribute_pre_commit(self, dict_, key):
-        """a fast expire that can be called by column loaders during a load.
-
-        The additional bookkeeping is finished up in commit_all().
-
-        Should only be called for scalar attributes.
-
-        This method is actually called a lot with joined-table
-        loading, when the second table isn't present in the result.
-
-        """
-        dict_.pop(key, None)
-        self.callables[key] = self
-
     @classmethod
     def _row_processor(cls, manager, fn, key):
         impl = manager[key].impl
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 1e8020dd9..2159d9135 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -119,8 +119,8 @@ class UninstrumentedColumnLoader(LoaderStrategy):
 
     def create_row_processor(
             self, context, path, loadopt,
-            mapper, row, adapter):
-        return None, None, None
+            mapper, result, adapter, populators):
+        pass
 
 
 @log.class_logger
@@ -157,21 +157,18 @@ class ColumnLoader(LoaderStrategy):
 
     def create_row_processor(
             self, context, path,
-            loadopt, mapper, row, adapter):
-        key = self.key
+            loadopt, mapper, result, adapter, populators):
         # look through list of columns represented here
         # to see which, if any, is present in the row.
         for col in self.columns:
             if adapter:
                 col = adapter.columns[col]
-            if col is not None and col in row:
-                def fetch_col(state, dict_, row):
-                    dict_[key] = row[col]
-                return fetch_col, None, None
+            getter = result._getter(col)
+            if getter:
+                populators["quick"].append((self.key, getter))
+                break
         else:
-            def expire_for_non_present_col(state, dict_, row):
-                state._expire_attribute_pre_commit(dict_, key)
-            return expire_for_non_present_col, None, None
+            populators["expire"].append((self.key, True))
 
 
 @log.class_logger
@@ -189,28 +186,26 @@ class DeferredColumnLoader(LoaderStrategy):
 
     def create_row_processor(
             self, context, path, loadopt,
-            mapper, row, adapter):
+            mapper, result, adapter, populators):
         col = self.columns[0]
         if adapter:
             col = adapter.columns[col]
 
-        key = self.key
-        if col in row:
-            return self.parent_property._get_strategy_by_cls(ColumnLoader).\
+        # 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, row, adapter)
+                    context, path, loadopt, mapper, result,
+                    adapter, populators)
 
         elif not self.is_class_level:
             set_deferred_for_local_state = InstanceState._row_processor(
                 mapper.class_manager,
-                LoadDeferredColumns(key), key)
-            return set_deferred_for_local_state, None, None
+                LoadDeferredColumns(self.key), self.key)
+            populators["new"].append((self.key, set_deferred_for_local_state))
         else:
-            def reset_col_for_deferred(state, dict_, row):
-                # reset state on the key so that deferred callables
-                # fire off on next access.
-                state._reset(dict_, key)
-            return reset_col_for_deferred, None, None
+            populators["expire"].append((self.key, False))
 
     def init_class_attribute(self, mapper):
         self.is_class_level = True
@@ -333,10 +328,10 @@ class NoLoader(AbstractRelationshipLoader):
 
     def create_row_processor(
             self, context, path, loadopt, mapper,
-            row, adapter):
+            result, adapter, populators):
         def invoke_no_load(state, dict_, row):
             state._initialize(self.key)
-        return invoke_no_load, None, None
+        populators["new"].append((self.key, invoke_no_load))
 
 
 @log.class_logger
@@ -618,7 +613,7 @@ class LazyLoader(AbstractRelationshipLoader):
 
     def create_row_processor(
             self, context, path, loadopt,
-            mapper, row, adapter):
+            mapper, result, adapter, populators):
         key = self.key
         if not self.is_class_level:
             # we are not the primary manager for this attribute
@@ -633,8 +628,8 @@ class LazyLoader(AbstractRelationshipLoader):
                 mapper.class_manager,
                 LoadLazyAttribute(key), key)
 
-            return set_lazy_callable, None, None
-        else:
+            populators["new"].append((self.key, set_lazy_callable))
+        elif context.populate_existing or mapper.always_refresh:
             def reset_for_lazy_callable(state, dict_, row):
                 # we are the primary manager for this attribute on
                 # this class - reset its
@@ -646,7 +641,7 @@ class LazyLoader(AbstractRelationshipLoader):
                 # any existing state.
                 state._reset(dict_, key)
 
-            return reset_for_lazy_callable, None, None
+            populators["new"].append((self.key, reset_for_lazy_callable))
 
 
 class LoadLazyAttribute(object):
@@ -679,11 +674,11 @@ class ImmediateLoader(AbstractRelationshipLoader):
 
     def create_row_processor(
             self, context, path, loadopt,
-            mapper, row, adapter):
+            mapper, result, adapter, populators):
         def load_immediate(state, dict_, row):
             state.get_impl(self.key).get(state, dict_)
 
-        return None, None, load_immediate
+        populators["delayed"].append((self.key, load_immediate))
 
 
 @log.class_logger
@@ -706,6 +701,8 @@ class SubqueryLoader(AbstractRelationshipLoader):
 
         if not context.query._enable_eagerloads:
             return
+        elif context.query._yield_per:
+            context.query._no_yield_per("subquery")
 
         path = path[self.parent_property]
 
@@ -994,7 +991,7 @@ class SubqueryLoader(AbstractRelationshipLoader):
 
     def create_row_processor(
             self, context, path, loadopt,
-            mapper, row, adapter):
+            mapper, result, adapter, populators):
         if not self.parent.class_manager[self.key].impl.supports_population:
             raise sa_exc.InvalidRequestError(
                 "'%s' does not support object "
@@ -1006,7 +1003,7 @@ class SubqueryLoader(AbstractRelationshipLoader):
         subq = path.get(context.attributes, 'subquery')
 
         if subq is None:
-            return None, None, None
+            return
 
         local_cols = self.parent_property.local_columns
 
@@ -1022,11 +1019,14 @@ class SubqueryLoader(AbstractRelationshipLoader):
             local_cols = [adapter.columns[c] for c in local_cols]
 
         if self.uselist:
-            return self._create_collection_loader(collections, local_cols)
+            self._create_collection_loader(
+                context, collections, local_cols, populators)
         else:
-            return self._create_scalar_loader(collections, local_cols)
+            self._create_scalar_loader(
+                context, collections, local_cols, populators)
 
-    def _create_collection_loader(self, collections, local_cols):
+    def _create_collection_loader(
+            self, context, collections, local_cols, populators):
         def load_collection_from_subq(state, dict_, row):
             collection = collections.get(
                 tuple([row[col] for col in local_cols]),
@@ -1035,9 +1035,12 @@ class SubqueryLoader(AbstractRelationshipLoader):
             state.get_impl(self.key).\
                 set_committed_value(state, dict_, collection)
 
-        return load_collection_from_subq, None, None, collections.loader
+        populators["new"].append((self.key, load_collection_from_subq))
+        if context.invoke_all_eagers:
+            populators["eager"].append((self.key, collections.loader))
 
-    def _create_scalar_loader(self, collections, local_cols):
+    def _create_scalar_loader(
+            self, context, collections, local_cols, populators):
         def load_scalar_from_subq(state, dict_, row):
             collection = collections.get(
                 tuple([row[col] for col in local_cols]),
@@ -1053,7 +1056,9 @@ class SubqueryLoader(AbstractRelationshipLoader):
             state.get_impl(self.key).\
                 set_committed_value(state, dict_, scalar)
 
-        return load_scalar_from_subq, None, None, collections.loader
+        populators["new"].append((self.key, load_scalar_from_subq))
+        if context.invoke_all_eagers:
+            populators["eager"].append((self.key, collections.loader))
 
 
 @log.class_logger
@@ -1081,6 +1086,8 @@ class JoinedLoader(AbstractRelationshipLoader):
 
         if not context.query._enable_eagerloads:
             return
+        elif context.query._yield_per and self.uselist:
+            context.query._no_yield_per("joined collection")
 
         path = path[self.parent_property]
 
@@ -1324,7 +1331,8 @@ class JoinedLoader(AbstractRelationshipLoader):
         join_to_outer = innerjoin and isinstance(towrap, sql.Join) and \
             towrap.isouter
 
-        if chained_from_outerjoin and join_to_outer and innerjoin == 'nested':
+        if chained_from_outerjoin and \
+                join_to_outer and innerjoin != 'unnested':
             inner = orm_util.join(
                 towrap.right,
                 clauses.aliased_class,
@@ -1377,7 +1385,7 @@ class JoinedLoader(AbstractRelationshipLoader):
                     )
                 )
 
-    def _create_eager_adapter(self, context, row, adapter, path, loadopt):
+    def _create_eager_adapter(self, context, result, adapter, path, loadopt):
         user_defined_adapter = self._init_user_defined_eager_proc(
             loadopt, context) if loadopt else False
 
@@ -1395,17 +1403,16 @@ class JoinedLoader(AbstractRelationshipLoader):
             if decorator is None:
                 return False
 
-        try:
-            self.mapper.identity_key_from_row(row, decorator)
+        if self.mapper._result_has_identity_key(result, decorator):
             return decorator
-        except KeyError:
+        else:
             # no identity key - don't return a row
             # processor, will cause a degrade to lazy
             return False
 
     def create_row_processor(
             self, context, path, loadopt, mapper,
-            row, adapter):
+            result, adapter, populators):
         if not self.parent.class_manager[self.key].impl.supports_population:
             raise sa_exc.InvalidRequestError(
                 "'%s' does not support object "
@@ -1417,7 +1424,7 @@ class JoinedLoader(AbstractRelationshipLoader):
 
         eager_adapter = self._create_eager_adapter(
             context,
-            row,
+            result,
             adapter, our_path, loadopt)
 
         if eager_adapter is not False:
@@ -1426,27 +1433,31 @@ class JoinedLoader(AbstractRelationshipLoader):
             _instance = loading.instance_processor(
                 self.mapper,
                 context,
+                result,
                 our_path[self.mapper],
                 eager_adapter)
 
             if not self.uselist:
-                return self._create_scalar_loader(context, key, _instance)
+                self._create_scalar_loader(context, key, _instance, populators)
             else:
-                return self._create_collection_loader(context, key, _instance)
+                self._create_collection_loader(
+                    context, key, _instance, populators)
         else:
-            return self.parent_property._get_strategy_by_cls(LazyLoader).\
+            self.parent_property._get_strategy_by_cls(LazyLoader).\
                 create_row_processor(
                     context, path, loadopt,
-                    mapper, row, adapter)
+                    mapper, result, adapter, populators)
 
-    def _create_collection_loader(self, context, key, _instance):
+    def _create_collection_loader(self, context, key, _instance, populators):
         def load_collection_from_joined_new_row(state, dict_, row):
             collection = attributes.init_state_collection(
                 state, dict_, key)
             result_list = util.UniqueAppender(collection,
                                               'append_without_event')
             context.attributes[(state, key)] = result_list
-            _instance(row, result_list)
+            inst = _instance(row)
+            if inst is not None:
+                result_list.append(inst)
 
         def load_collection_from_joined_existing_row(state, dict_, row):
             if (state, key) in context.attributes:
@@ -1462,25 +1473,30 @@ class JoinedLoader(AbstractRelationshipLoader):
                     collection,
                     'append_without_event')
                 context.attributes[(state, key)] = result_list
-            _instance(row, result_list)
+            inst = _instance(row)
+            if inst is not None:
+                result_list.append(inst)
 
         def load_collection_from_joined_exec(state, dict_, row):
-            _instance(row, None)
+            _instance(row)
 
-        return load_collection_from_joined_new_row, \
-            load_collection_from_joined_existing_row, \
-            None, load_collection_from_joined_exec
+        populators["new"].append((self.key, load_collection_from_joined_new_row))
+        populators["existing"].append(
+            (self.key, load_collection_from_joined_existing_row))
+        if context.invoke_all_eagers:
+            populators["eager"].append(
+                (self.key, load_collection_from_joined_exec))
 
-    def _create_scalar_loader(self, context, key, _instance):
+    def _create_scalar_loader(self, context, key, _instance, populators):
         def load_scalar_from_joined_new_row(state, dict_, row):
             # set a scalar object instance directly on the parent
             # object, bypassing InstrumentedAttribute event handlers.
-            dict_[key] = _instance(row, None)
+            dict_[key] = _instance(row)
 
         def load_scalar_from_joined_existing_row(state, dict_, row):
             # call _instance on the row, even though the object has
             # been created, so that we further descend into properties
-            existing = _instance(row, None)
+            existing = _instance(row)
             if existing is not None \
                 and key in dict_ \
                     and existing is not dict_[key]:
@@ -1490,11 +1506,13 @@ class JoinedLoader(AbstractRelationshipLoader):
                     % self)
 
         def load_scalar_from_joined_exec(state, dict_, row):
-            _instance(row, None)
+            _instance(row)
 
-        return load_scalar_from_joined_new_row, \
-            load_scalar_from_joined_existing_row, \
-            None, load_scalar_from_joined_exec
+        populators["new"].append((self.key, load_scalar_from_joined_new_row))
+        populators["existing"].append(
+            (self.key, load_scalar_from_joined_existing_row))
+        if context.invoke_all_eagers:
+            populators["eager"].append((self.key, load_scalar_from_joined_exec))
 
 
 def single_parent_validator(desc, prop):
diff --git a/lib/sqlalchemy/orm/strategy_options.py b/lib/sqlalchemy/orm/strategy_options.py
index 392f7cec2..4f986193e 100644
--- a/lib/sqlalchemy/orm/strategy_options.py
+++ b/lib/sqlalchemy/orm/strategy_options.py
@@ -1,4 +1,3 @@
-# orm/strategy_options.py
 # Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
@@ -631,15 +630,47 @@ def joinedload(loadopt, attr, innerjoin=None):
 
         query(Order).options(joinedload(Order.user, innerjoin=True))
 
-     If the joined-eager load is chained onto an existing LEFT OUTER JOIN,
-     ``innerjoin=True`` will be bypassed and the join will continue to
-     chain as LEFT OUTER JOIN so that the results don't change.  As an
-     alternative, specify the value ``"nested"``.  This will instead nest the
-     join on the right side, e.g. using the form "a LEFT OUTER JOIN
-     (b JOIN c)".
+     In order to chain multiple eager joins together where some may be
+     OUTER and others INNER, right-nested joins are used to link them::
 
-     .. versionadded:: 0.9.4 Added ``innerjoin="nested"`` option to support
-        nesting of eager "inner" joins.
+        query(A).options(
+            joinedload(A.bs, innerjoin=False).
+                joinedload(B.cs, innerjoin=True)
+        )
+
+     The above query, linking A.bs via "outer" join and B.cs via "inner" join
+     would render the joins as "a LEFT OUTER JOIN (b JOIN c)".   When using
+     SQLite, this form of JOIN is translated to use full subqueries as this
+     syntax is otherwise not directly supported.
+
+     The ``innerjoin`` flag can also be stated with the term ``"unnested"``.
+     This will prevent joins from being right-nested, and will instead
+     link an "innerjoin" eagerload to an "outerjoin" eagerload by bypassing
+     the "inner" join.   Using this form as follows::
+
+        query(A).options(
+            joinedload(A.bs, innerjoin=False).
+                joinedload(B.cs, innerjoin="unnested")
+        )
+
+     Joins will be rendered as "a LEFT OUTER JOIN b LEFT OUTER JOIN c", so that
+     all of "a" is matched rather than being incorrectly limited by a "b" that
+     does not contain a "c".
+
+     .. note:: The "unnested" flag does **not** affect the JOIN rendered
+        from a many-to-many association table, e.g. a table configured
+        as :paramref:`.relationship.secondary`, to the target table; for
+        correctness of results, these joins are always INNER and are
+        therefore right-nested if linked to an OUTER join.
+
+     .. versionadded:: 0.9.4 Added support for "nesting" of eager "inner"
+        joins.  See :ref:`feature_2976`.
+
+     .. versionchanged:: 1.0.0 ``innerjoin=True`` now implies
+        ``innerjoin="nested"``, whereas in 0.9 it implied
+        ``innerjoin="unnested"``.  In order to achieve the pre-1.0 "unnested"
+        inner join behavior, use the value ``innerjoin="unnested"``.
+        See :ref:`migration_3008`.
 
     .. note::
 
diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 89cddfc31..bc9affe4a 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -305,7 +305,7 @@ class Pool(log.Identified):
         """Return a new :class:`.Pool`, of the same class as this one
         and configured with identical creation arguments.
 
-        This method is used in conjunection with :meth:`dispose`
+        This method is used in conjunction with :meth:`dispose`
         to close out an entire :class:`.Pool` and create a new one in
         its place.
 
diff --git a/lib/sqlalchemy/sql/annotation.py b/lib/sqlalchemy/sql/annotation.py
index 02f5c3c1c..3df4257d4 100644
--- a/lib/sqlalchemy/sql/annotation.py
+++ b/lib/sqlalchemy/sql/annotation.py
@@ -46,6 +46,7 @@ class Annotated(object):
         self.__dict__ = element.__dict__.copy()
         self.__element = element
         self._annotations = values
+        self._hash = hash(element)
 
     def _annotate(self, values):
         _values = self._annotations.copy()
@@ -87,7 +88,7 @@ class Annotated(object):
             return self.__class__(clone, self._annotations)
 
     def __hash__(self):
-        return hash(self.__element)
+        return self._hash
 
     def __eq__(self, other):
         if isinstance(self.__element, operators.ColumnOperators):
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index e45510aa4..23e5456a7 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -494,6 +494,28 @@ class SQLCompiler(Compiled):
     def visit_grouping(self, grouping, asfrom=False, **kwargs):
         return "(" + grouping.element._compiler_dispatch(self, **kwargs) + ")"
 
+    def visit_label_reference(self, element, **kwargs):
+        if not self.stack:
+            # compiling the element outside of the context of a SELECT
+            return self.process(
+                element._text_clause
+            )
+
+        selectable = self.stack[-1]['selectable']
+        try:
+            col = selectable._inner_column_dict[element.text]
+        except KeyError:
+            # treat it like text()
+            util.warn_limited(
+                "Can't resolve label reference %r; converting to text()",
+                util.ellipses_string(element.text))
+            return self.process(
+                element._text_clause
+            )
+        else:
+            kwargs['render_label_as_label'] = col
+            return self.process(col, **kwargs)
+
     def visit_label(self, label,
                     add_to_result_map=None,
                     within_label_clause=False,
@@ -761,7 +783,8 @@ class SQLCompiler(Compiled):
             {
                 'correlate_froms': entry['correlate_froms'],
                 'iswrapper': toplevel,
-                'asfrom_froms': entry['asfrom_froms']
+                'asfrom_froms': entry['asfrom_froms'],
+                'selectable': cs
             })
 
         keyword = self.compound_keywords.get(cs.keyword)
@@ -1480,7 +1503,8 @@ class SQLCompiler(Compiled):
         new_entry = {
             'asfrom_froms': new_correlate_froms,
             'iswrapper': iswrapper,
-            'correlate_froms': all_correlate_froms
+            'correlate_froms': all_correlate_froms,
+            'selectable': select,
         }
         self.stack.append(new_entry)
 
@@ -1791,7 +1815,8 @@ class SQLCompiler(Compiled):
         self.stack.append(
             {'correlate_froms': set([update_stmt.table]),
              "iswrapper": False,
-             "asfrom_froms": set([update_stmt.table])})
+             "asfrom_froms": set([update_stmt.table]),
+             "selectable": update_stmt})
 
         self.isupdate = True
 
@@ -1981,11 +2006,13 @@ class SQLCompiler(Compiled):
 
         need_pks = self.isinsert and \
             not self.inline and \
-            not stmt._returning
+            not stmt._returning and \
+            not stmt._has_multi_parameters
 
         implicit_returning = need_pks and \
             self.dialect.implicit_returning and \
             stmt.table.implicit_returning
+
         if self.isinsert:
             implicit_return_defaults = (implicit_returning and
                                         stmt._return_defaults)
@@ -2245,7 +2272,8 @@ class SQLCompiler(Compiled):
     def visit_delete(self, delete_stmt, **kw):
         self.stack.append({'correlate_froms': set([delete_stmt.table]),
                            "iswrapper": False,
-                           "asfrom_froms": set([delete_stmt.table])})
+                           "asfrom_froms": set([delete_stmt.table]),
+                           "selectable": delete_stmt})
         self.isdelete = True
 
         text = "DELETE "
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index f7e033d85..1934d0776 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -269,6 +269,13 @@ class ValuesBase(UpdateBase):
          .. versionadded:: 0.8
              Support for multiple-VALUES INSERT statements.
 
+        .. versionchanged:: 1.0.0 an INSERT that uses a multiple-VALUES
+           clause, even a list of length one,
+           implies that the :paramref:`.Insert.inline` flag is set to
+           True, indicating that the statement will not attempt to fetch
+           the "last inserted primary key" or other defaults.  The statement
+           deals with an arbitrary number of rows, so the
+           :attr:`.ResultProxy.inserted_primary_key` accessor does not apply.
 
         .. seealso::
 
@@ -434,8 +441,13 @@ class Insert(ValuesBase):
          dynamically render the VALUES clause at execution time based on
          the parameters passed to :meth:`.Connection.execute`.
 
-        :param inline: if True, SQL defaults will be compiled 'inline' into
-         the statement and not pre-executed.
+        :param inline: if True, no attempt will be made to retrieve the
+         SQL-generated default values to be provided within the statement;
+         in particular,
+         this allows SQL expressions to be rendered 'inline' within the
+         statement without the need to pre-execute them beforehand; for
+         backends that support "returning", this turns off the "implicit
+         returning" feature for the statement.
 
         If both `values` and compile-time bind parameters are present, the
         compile-time bind parameters override the information specified
@@ -495,17 +507,12 @@ class Insert(ValuesBase):
          would normally raise an exception if these column lists don't
          correspond.
 
-        .. note::
-
-           Depending on backend, it may be necessary for the :class:`.Insert`
-           statement to be constructed using the ``inline=True`` flag; this
-           flag will prevent the implicit usage of ``RETURNING`` when the
-           ``INSERT`` statement is rendered, which isn't supported on a
-           backend such as Oracle in conjunction with an ``INSERT..SELECT``
-           combination::
-
-             sel = select([table1.c.a, table1.c.b]).where(table1.c.c > 5)
-             ins = table2.insert(inline=True).from_select(['a', 'b'], sel)
+        .. versionchanged:: 1.0.0 an INSERT that uses FROM SELECT
+           implies that the :paramref:`.insert.inline` flag is set to
+           True, indicating that the statement will not attempt to fetch
+           the "last inserted primary key" or other defaults.  The statement
+           deals with an arbitrary number of rows, so the
+           :attr:`.ResultProxy.inserted_primary_key` accessor does not apply.
 
         .. note::
 
@@ -525,6 +532,7 @@ class Insert(ValuesBase):
             self._process_colparams(dict((n, Null()) for n in names))
 
         self.select_names = names
+        self.inline = True
         self.select = _interpret_as_select(select)
 
     def _copy_internals(self, clone=_clone, **kw):
@@ -728,10 +736,10 @@ class Delete(UpdateBase):
         :meth:`~.TableClause.delete` method on
         :class:`~.schema.Table`.
 
-        :param table: The table to be updated.
+        :param table: The table to delete rows from.
 
         :param whereclause: A :class:`.ClauseElement` describing the ``WHERE``
-          condition of the ``UPDATE`` statement. Note that the
+          condition of the ``DELETE`` statement. Note that the
           :meth:`~Delete.where()` generative method may be used instead.
 
         .. seealso::
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 6cbf583cc..984cfe0ee 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -19,7 +19,8 @@ from .visitors import Visitable, cloned_traverse, traverse
 from .annotation import Annotated
 import itertools
 from .base import Executable, PARSE_AUTOCOMMIT, Immutable, NO_ARG
-from .base import _generative, Generative
+from .base import _generative
+import numbers
 
 import re
 import operator
@@ -624,7 +625,7 @@ class ColumnElement(operators.ColumnOperators, ClauseElement):
     __visit_name__ = 'column'
     primary_key = False
     foreign_keys = []
-    _label = None
+    _label = _columns_clause_label = None
     _key_label = key = None
     _alt_names = ()
 
@@ -1180,6 +1181,10 @@ class TextClause(Executable, ClauseElement):
 
     _hide_froms = []
 
+    # help in those cases where text() is
+    # interpreted in a column expression situation
+    key = _label = _columns_clause_label = None
+
     def __init__(
             self,
             text,
@@ -1694,13 +1699,16 @@ class ClauseList(ClauseElement):
         self.operator = kwargs.pop('operator', operators.comma_op)
         self.group = kwargs.pop('group', True)
         self.group_contents = kwargs.pop('group_contents', True)
+        text_converter = kwargs.pop(
+            '_literal_as_text',
+            _expression_literal_as_text)
         if self.group_contents:
             self.clauses = [
-                _literal_as_text(clause).self_group(against=self.operator)
+                text_converter(clause).self_group(against=self.operator)
                 for clause in clauses]
         else:
             self.clauses = [
-                _literal_as_text(clause)
+                text_converter(clause)
                 for clause in clauses]
 
     def __iter__(self):
@@ -1767,7 +1775,7 @@ class BooleanClauseList(ClauseList, ColumnElement):
 
         clauses = util.coerce_generator_arg(clauses)
         for clause in clauses:
-            clause = _literal_as_text(clause)
+            clause = _expression_literal_as_text(clause)
 
             if isinstance(clause, continue_on):
                 continue
@@ -2133,14 +2141,15 @@ class Case(ColumnElement):
 
 
 def literal_column(text, type_=None):
-    """Return a textual column expression, as would be in the columns
-    clause of a ``SELECT`` statement.
-
-    The object returned supports further expressions in the same way as any
-    other column object, including comparison, math and string operations.
-    The type\_ parameter is important to determine proper expression behavior
-    (such as, '+' means string concatenation or numerical addition based on
-    the type).
+    """Produce a :class:`.ColumnClause` object that has the
+    :paramref:`.column.is_literal` flag set to True.
+
+    :func:`.literal_column` is similar to :func:`.column`, except that
+    it is more often used as a "standalone" column expression that renders
+    exactly as stated; while :func:`.column` stores a string name that
+    will be assumed to be part of a table and may be quoted as such,
+    :func:`.literal_column` can be that, or any other arbitrary column-oriented
+    expression.
 
     :param text: the text of the expression; can be any SQL expression.
       Quoting rules will not be applied. To specify a column-name expression
@@ -2152,6 +2161,14 @@ def literal_column(text, type_=None):
       provide result-set translation and additional expression semantics for
       this column. If left as None the type will be NullType.
 
+    .. seealso::
+
+        :func:`.column`
+
+        :func:`.text`
+
+        :ref:`sqlexpression_literal_column`
+
     """
     return ColumnClause(text, type_=type_, is_literal=True)
 
@@ -2271,6 +2288,17 @@ class Extract(ColumnElement):
         return self.expr._from_objects
 
 
+class _label_reference(ColumnElement):
+    __visit_name__ = 'label_reference'
+
+    def __init__(self, text):
+        self.text = self.key = text
+
+    @util.memoized_property
+    def _text_clause(self):
+        return TextClause._create_text(self.text)
+
+
 class UnaryExpression(ColumnElement):
     """Define a 'unary' expression.
 
@@ -2334,7 +2362,8 @@ class UnaryExpression(ColumnElement):
 
         """
         return UnaryExpression(
-            _literal_as_text(column), modifier=operators.nullsfirst_op)
+            _literal_as_label_reference(column),
+            modifier=operators.nullsfirst_op)
 
     @classmethod
     def _create_nullslast(cls, column):
@@ -2374,7 +2403,8 @@ class UnaryExpression(ColumnElement):
 
         """
         return UnaryExpression(
-            _literal_as_text(column), modifier=operators.nullslast_op)
+            _literal_as_label_reference(column),
+            modifier=operators.nullslast_op)
 
     @classmethod
     def _create_desc(cls, column):
@@ -2412,7 +2442,7 @@ class UnaryExpression(ColumnElement):
 
         """
         return UnaryExpression(
-            _literal_as_text(column), modifier=operators.desc_op)
+            _literal_as_label_reference(column), modifier=operators.desc_op)
 
     @classmethod
     def _create_asc(cls, column):
@@ -2449,7 +2479,7 @@ class UnaryExpression(ColumnElement):
 
         """
         return UnaryExpression(
-            _literal_as_text(column), modifier=operators.asc_op)
+            _literal_as_label_reference(column), modifier=operators.asc_op)
 
     @classmethod
     def _create_distinct(cls, expr):
@@ -2733,9 +2763,13 @@ class Over(ColumnElement):
         """
         self.func = func
         if order_by is not None:
-            self.order_by = ClauseList(*util.to_list(order_by))
+            self.order_by = ClauseList(
+                *util.to_list(order_by),
+                _literal_as_text=_literal_as_label_reference)
         if partition_by is not None:
-            self.partition_by = ClauseList(*util.to_list(partition_by))
+            self.partition_by = ClauseList(
+                *util.to_list(partition_by),
+                _literal_as_text=_literal_as_label_reference)
 
     @util.memoized_property
     def type(self):
@@ -2795,7 +2829,8 @@ class Label(ColumnElement):
             self.name = _anonymous_label(
                 '%%(%d %s)s' % (id(self), getattr(element, 'name', 'anon'))
             )
-        self.key = self._label = self._key_label = self.name
+        self.key = self._label = self._key_label = \
+            self._columns_clause_label = self.name
         self._element = element
         self._type = type_
         self._proxies = [element]
@@ -2860,7 +2895,7 @@ class ColumnClause(Immutable, ColumnElement):
     :class:`.Column` class, is typically invoked using the
     :func:`.column` function, as in::
 
-        from sqlalchemy.sql import column
+        from sqlalchemy import column
 
         id, name = column("id"), column("name")
         stmt = select([id, name]).select_from("user")
@@ -2900,7 +2935,7 @@ class ColumnClause(Immutable, ColumnElement):
         :class:`.Column` class.  The :func:`.column` function can
         be invoked with just a name alone, as in::
 
-            from sqlalchemy.sql import column
+            from sqlalchemy import column
 
             id, name = column("id"), column("name")
             stmt = select([id, name]).select_from("user")
@@ -2932,7 +2967,7 @@ class ColumnClause(Immutable, ColumnElement):
         (which is the lightweight analogue to :class:`.Table`) to produce
         a working table construct with minimal boilerplate::
 
-            from sqlalchemy.sql import table, column
+            from sqlalchemy import table, column, select
 
             user = table("user",
                     column("id"),
@@ -2948,6 +2983,10 @@ class ColumnClause(Immutable, ColumnElement):
         :class:`.schema.MetaData`, DDL, or events, unlike its
         :class:`.Table` counterpart.
 
+        .. versionchanged:: 1.0.0 :func:`.expression.column` can now
+           be imported from the plain ``sqlalchemy`` namespace like any
+           other SQL element.
+
         :param text: the text of the element.
 
         :param type: :class:`.types.TypeEngine` object which can associate
@@ -2965,9 +3004,11 @@ class ColumnClause(Immutable, ColumnElement):
 
             :func:`.literal_column`
 
+            :func:`.table`
+
             :func:`.text`
 
-            :ref:`metadata_toplevel`
+            :ref:`sqlexpression_literal_column`
 
         """
 
@@ -3024,6 +3065,13 @@ class ColumnClause(Immutable, ColumnElement):
     def _label(self):
         return self._gen_label(self.name)
 
+    @_memoized_property
+    def _columns_clause_label(self):
+        if self.table is None:
+            return None
+        else:
+            return self._label
+
     def _gen_label(self, name):
         t = self.table
 
@@ -3427,12 +3475,29 @@ def _clause_element_as_expr(element):
         return element
 
 
-def _literal_as_text(element):
+def _literal_as_label_reference(element):
+    if isinstance(element, util.string_types):
+        return _label_reference(element)
+    else:
+        return _literal_as_text(element)
+
+
+def _expression_literal_as_text(element):
+    return _literal_as_text(element, warn=True)
+
+
+def _literal_as_text(element, warn=False):
     if isinstance(element, Visitable):
         return element
     elif hasattr(element, '__clause_element__'):
         return element.__clause_element__()
     elif isinstance(element, util.string_types):
+        if warn:
+            util.warn_limited(
+                "Textual SQL expression %(expr)r should be "
+                "explicitly declared as text(%(expr)r)",
+                {"expr": util.ellipses_string(element)})
+
         return TextClause(util.text_type(element))
     elif isinstance(element, (util.NoneType, bool)):
         return _const_expr(element)
@@ -3487,6 +3552,8 @@ def _literal_as_binds(element, name=None, type_=None):
     else:
         return element
 
+_guess_straight_column = re.compile(r'^\w\S*$', re.I)
+
 
 def _interpret_as_column_or_from(element):
     if isinstance(element, Visitable):
@@ -3501,7 +3568,31 @@ def _interpret_as_column_or_from(element):
     elif hasattr(insp, "selectable"):
         return insp.selectable
 
-    return ColumnClause(str(element), is_literal=True)
+    # be forgiving as this is an extremely common
+    # and known expression
+    if element == "*":
+        guess_is_literal = True
+    elif isinstance(element, (numbers.Number)):
+        return ColumnClause(str(element), is_literal=True)
+    else:
+        element = str(element)
+        # give into temptation, as this fact we are guessing about
+        # is not one we've previously ever needed our users tell us;
+        # but let them know we are not happy about it
+        guess_is_literal = not _guess_straight_column.match(element)
+        util.warn_limited(
+            "Textual column expression %(column)r should be "
+            "explicitly declared with text(%(column)r), "
+            "or use %(literal_column)s(%(column)r) "
+            "for more specificity",
+            {
+                "column": util.ellipses_string(element),
+                "literal_column": "literal_column"
+                if guess_is_literal else "column"
+            })
+    return ColumnClause(
+        element,
+        is_literal=guess_is_literal)
 
 
 def _const_expr(element):
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index fd57f9be8..d96f048b9 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -106,7 +106,8 @@ from .elements import _literal_as_text, _clause_element_as_expr,\
     _is_column, _labeled, _only_column_elements, _string_or_unprintable, \
     _truncated_label, _clone, _cloned_difference, _cloned_intersection,\
     _column_as_key, _literal_as_binds, _select_iterables, \
-    _corresponding_column_or_error
+    _corresponding_column_or_error, _literal_as_label_reference, \
+    _expression_literal_as_text
 from .selectable import _interpret_as_from
 
 
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index c8e815d24..d9fd37f92 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -1124,8 +1124,12 @@ class Column(SchemaItem, ColumnClause):
             else:
                 if getattr(self.type, '_warn_on_bytestring', False):
                     if isinstance(self.default, util.binary_type):
-                        util.warn("Unicode column received non-unicode "
-                                  "default value.")
+                        util.warn(
+                            "Unicode column '%s' has non-unicode "
+                            "default value %r specified." % (
+                                self.key,
+                                self.default
+                            ))
                 args.append(ColumnDefault(self.default))
 
         if self.server_default is not None:
@@ -2429,7 +2433,7 @@ class CheckConstraint(Constraint):
 
         super(CheckConstraint, self).\
             __init__(name, deferrable, initially, _create_rule, info=info)
-        self.sqltext = _literal_as_text(sqltext)
+        self.sqltext = _literal_as_text(sqltext, warn=False)
         if table is not None:
             self._set_parent_with_dispatch(table)
         elif _autoattach:
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 4808a3935..a49493995 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -15,8 +15,8 @@ from .elements import ClauseElement, TextClause, ClauseList, \
 from .elements import _clone, \
     _literal_as_text, _interpret_as_column_or_from, _expand_cloned,\
     _select_iterables, _anonymous_label, _clause_element_as_expr,\
-    _cloned_intersection, _cloned_difference, True_, _only_column_elements,\
-    TRUE
+    _cloned_intersection, _cloned_difference, True_, \
+    _literal_as_label_reference
 from .base import Immutable, Executable, _generative, \
     ColumnCollection, ColumnSet, _from_objects, Generative
 from . import type_api
@@ -36,6 +36,12 @@ def _interpret_as_from(element):
     insp = inspection.inspect(element, raiseerr=False)
     if insp is None:
         if isinstance(element, util.string_types):
+            util.warn_limited(
+                "Textual SQL FROM expression %(expr)r should be "
+                "explicitly declared as text(%(expr)r), "
+                "or use table(%(expr)r) for more specificity",
+                {"expr": util.ellipses_string(element)})
+
             return TextClause(util.text_type(element))
     elif hasattr(insp, "selectable"):
         return insp.selectable
@@ -1177,7 +1183,7 @@ class TableClause(Immutable, FromClause):
     collection of columns, which are typically produced
     by the :func:`.expression.column` function::
 
-        from sqlalchemy.sql import table, column
+        from sqlalchemy import table, column
 
         user = table("user",
                 column("id"),
@@ -1218,11 +1224,9 @@ class TableClause(Immutable, FromClause):
         :class:`~.schema.Table` object.
         It may be used to construct lightweight table constructs.
 
-        Note that the :func:`.expression.table` function is not part of
-        the ``sqlalchemy`` namespace.  It must be imported from the
-        ``sql`` package::
-
-            from sqlalchemy.sql import table, column
+        .. versionchanged:: 1.0.0 :func:`.expression.table` can now
+           be imported from the plain ``sqlalchemy`` namespace like any
+           other SQL element.
 
         :param name: Name of the table.
 
@@ -1626,9 +1630,13 @@ class GenerativeSelect(SelectBase):
         self._bind = bind
 
         if order_by is not None:
-            self._order_by_clause = ClauseList(*util.to_list(order_by))
+            self._order_by_clause = ClauseList(
+                *util.to_list(order_by),
+                _literal_as_text=_literal_as_label_reference)
         if group_by is not None:
-            self._group_by_clause = ClauseList(*util.to_list(group_by))
+            self._group_by_clause = ClauseList(
+                *util.to_list(group_by),
+                _literal_as_text=_literal_as_label_reference)
 
     @property
     def for_update(self):
@@ -1784,7 +1792,8 @@ class GenerativeSelect(SelectBase):
         else:
             if getattr(self, '_order_by_clause', None) is not None:
                 clauses = list(self._order_by_clause) + list(clauses)
-            self._order_by_clause = ClauseList(*clauses)
+            self._order_by_clause = ClauseList(
+                *clauses, _literal_as_text=_literal_as_label_reference)
 
     def append_group_by(self, *clauses):
         """Append the given GROUP BY criterion applied to this selectable.
@@ -1801,7 +1810,12 @@ class GenerativeSelect(SelectBase):
         else:
             if getattr(self, '_group_by_clause', None) is not None:
                 clauses = list(self._group_by_clause) + list(clauses)
-            self._group_by_clause = ClauseList(*clauses)
+            self._group_by_clause = ClauseList(
+                *clauses, _literal_as_text=_literal_as_label_reference)
+
+    @property
+    def _inner_column_dict(self):
+        raise NotImplementedError()
 
     def _copy_internals(self, clone=_clone, **kw):
         if self._limit_clause is not None:
@@ -1869,6 +1883,12 @@ class CompoundSelect(GenerativeSelect):
 
         GenerativeSelect.__init__(self, **kwargs)
 
+    @property
+    def _inner_column_dict(self):
+        return dict(
+            (c.key, c) for c in self.c
+        )
+
     @classmethod
     def _create_union(cls, *selects, **kwargs):
         """Return a ``UNION`` of multiple selectables.
@@ -2092,7 +2112,7 @@ class HasPrefixes(object):
 
     def _setup_prefixes(self, prefixes, dialect=None):
         self._prefixes = self._prefixes + tuple(
-            [(_literal_as_text(p), dialect) for p in prefixes])
+            [(_literal_as_text(p, warn=False), dialect) for p in prefixes])
 
 
 class Select(HasPrefixes, GenerativeSelect):
@@ -2477,6 +2497,15 @@ class Select(HasPrefixes, GenerativeSelect):
         """
         return _select_iterables(self._raw_columns)
 
+    @_memoized_property
+    def _inner_column_dict(self):
+        d = dict(
+            (c._label or c.key, c)
+            for c in _select_iterables(self._raw_columns))
+        d.update((c.key, c) for c in _select_iterables(self.froms))
+
+        return d
+
     def is_derived_from(self, fromclause):
         if self in fromclause._cloned_set:
             return True
@@ -2706,7 +2735,7 @@ class Select(HasPrefixes, GenerativeSelect):
 
         """
         if expr:
-            expr = [_literal_as_text(e) for e in expr]
+            expr = [_literal_as_label_reference(e) for e in expr]
             if isinstance(self._distinct, list):
                 self._distinct = self._distinct + expr
             else:
@@ -2945,9 +2974,10 @@ class Select(HasPrefixes, GenerativeSelect):
             names = set()
 
             def name_for_col(c):
-                if c._label is None:
+                if c._columns_clause_label is None:
                     return (None, c)
-                name = c._label
+
+                name = c._columns_clause_label
                 if name in names:
                     name = c.anon_label
                 else:
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index a7f25bbfa..2729bc83e 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -180,8 +180,10 @@ class String(Concatenable, TypeEngine):
                 if self._warn_on_bytestring:
                     def process(value):
                         if isinstance(value, util.binary_type):
-                            util.warn("Unicode type received non-unicode"
-                                      "bind param value.")
+                            util.warn_limited(
+                                "Unicode type received non-unicode "
+                                "bind param value %r.",
+                                (util.ellipses_string(value),))
                         return value
                     return process
                 else:
@@ -194,8 +196,10 @@ class String(Concatenable, TypeEngine):
                     if isinstance(value, util.text_type):
                         return encoder(value, self.unicode_error)[0]
                     elif warn_on_bytestring and value is not None:
-                        util.warn("Unicode type received non-unicode bind "
-                                  "param value")
+                        util.warn_limited(
+                            "Unicode type received non-unicode bind "
+                            "param value %r.",
+                            (util.ellipses_string(value),))
                     return value
             return process
         else:
diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index 8f8f56412..1f37b4b45 100644
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -6,7 +6,7 @@
 # the MIT License: http://www.opensource.org/licenses/mit-license.php
 
 
-from .warnings import testing_warn, assert_warnings, resetwarnings
+from .warnings import assert_warnings
 
 from . import config
 
@@ -21,7 +21,7 @@ def against(*queries):
 from .assertions import emits_warning, emits_warning_on, uses_deprecated, \
     eq_, ne_, is_, is_not_, startswith_, assert_raises, \
     assert_raises_message, AssertsCompiledSQL, ComparesTables, \
-    AssertsExecutionResults, expect_deprecated
+    AssertsExecutionResults, expect_deprecated, expect_warnings
 
 from .util import run_as_contextmanager, rowset, fail, provide_metadata, adict
 
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index 79411af7e..bf7c27a89 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -9,79 +9,77 @@ from __future__ import absolute_import
 
 from . import util as testutil
 from sqlalchemy import pool, orm, util
-from sqlalchemy.engine import default, create_engine, url
-from sqlalchemy import exc as sa_exc
+from sqlalchemy.engine import default, url
 from sqlalchemy.util import decorator
-from sqlalchemy import types as sqltypes, schema
+from sqlalchemy import types as sqltypes, schema, exc as sa_exc
 import warnings
 import re
-from .warnings import resetwarnings
 from .exclusions import db_spec, _is_excluded
 from . import assertsql
 from . import config
-import itertools
 from .util import fail
 import contextlib
+from . import mock
 
 
-def emits_warning(*messages):
-    """Mark a test as emitting a warning.
+def expect_warnings(*messages):
+    """Context manager which expects one or more warnings.
+
+    With no arguments, squelches all SAWarnings emitted via
+    sqlalchemy.util.warn and sqlalchemy.util.warn_limited.   Otherwise
+    pass string expressions that will match selected warnings via regex;
+    all non-matching warnings are sent through.
+
+    Note that the test suite sets SAWarning warnings to raise exceptions.
+
+    """
+    return _expect_warnings(sa_exc.SAWarning, messages)
+
+
+@contextlib.contextmanager
+def expect_warnings_on(db, *messages):
+    """Context manager which expects one or more warnings on specific
+    dialects.
 
-    With no arguments, squelches all SAWarning failures.  Or pass one or more
-    strings; these will be matched to the root of the warning description by
-    warnings.filterwarnings().
     """
-    # TODO: it would be nice to assert that a named warning was
-    # emitted. should work with some monkeypatching of warnings,
-    # and may work on non-CPython if they keep to the spirit of
-    # warnings.showwarning's docstring.
-    # - update: jython looks ok, it uses cpython's module
+    spec = db_spec(db)
+
+    if isinstance(db, util.string_types) and not spec(config._current):
+        yield
+    elif not _is_excluded(*db):
+        yield
+    else:
+        with expect_warnings(*messages):
+            yield
+
+
+def emits_warning(*messages):
+    """Decorator form of expect_warnings()."""
 
     @decorator
     def decorate(fn, *args, **kw):
-        # todo: should probably be strict about this, too
-        filters = [dict(action='ignore',
-                        category=sa_exc.SAPendingDeprecationWarning)]
-        if not messages:
-            filters.append(dict(action='ignore',
-                                category=sa_exc.SAWarning))
-        else:
-            filters.extend(dict(action='ignore',
-                                message=message,
-                                category=sa_exc.SAWarning)
-                           for message in messages)
-        for f in filters:
-            warnings.filterwarnings(**f)
-        try:
+        with expect_warnings(*messages):
             return fn(*args, **kw)
-        finally:
-            resetwarnings()
+
     return decorate
 
 
-def emits_warning_on(db, *warnings):
+def expect_deprecated(*messages):
+    return _expect_warnings(sa_exc.SADeprecationWarning, messages)
+
+
+def emits_warning_on(db, *messages):
     """Mark a test as emitting a warning on a specific dialect.
 
     With no arguments, squelches all SAWarning failures.  Or pass one or more
     strings; these will be matched to the root of the warning description by
     warnings.filterwarnings().
     """
-    spec = db_spec(db)
-
     @decorator
     def decorate(fn, *args, **kw):
-        if isinstance(db, util.string_types):
-            if not spec(config._current):
-                return fn(*args, **kw)
-            else:
-                wrapped = emits_warning(*warnings)(fn)
-                return wrapped(*args, **kw)
-        else:
-            if not _is_excluded(*db):
-                return fn(*args, **kw)
-            else:
-                wrapped = emits_warning(*warnings)(fn)
-                return wrapped(*args, **kw)
+        with expect_warnings_on(db, *messages):
+            return fn(*args, **kw)
+
     return decorate
 
 
@@ -105,29 +103,27 @@ def uses_deprecated(*messages):
 
 
 @contextlib.contextmanager
-def expect_deprecated(*messages):
-    # todo: should probably be strict about this, too
-    filters = [dict(action='ignore',
-                    category=sa_exc.SAPendingDeprecationWarning)]
-    if not messages:
-        filters.append(dict(action='ignore',
-                            category=sa_exc.SADeprecationWarning))
-    else:
-        filters.extend(
-            [dict(action='ignore',
-                  message=message,
-                  category=sa_exc.SADeprecationWarning)
-             for message in
-             [(m.startswith('//') and
-               ('Call to deprecated function ' + m[2:]) or m)
-              for m in messages]])
-
-    for f in filters:
-        warnings.filterwarnings(**f)
-    try:
+def _expect_warnings(exc_cls, messages):
+
+    filters = [re.compile(msg, re.I) for msg in messages]
+
+    real_warn = warnings.warn
+
+    def our_warn(msg, exception, *arg, **kw):
+        if not issubclass(exception, exc_cls):
+            return real_warn(msg, exception, *arg, **kw)
+
+        if not filters:
+            return
+
+        for filter_ in filters:
+            if filter_.match(msg):
+                break
+        else:
+            real_warn(msg, exception, *arg, **kw)
+
+    with mock.patch("warnings.warn", our_warn):
         yield
-    finally:
-        resetwarnings()
 
 
 def global_cleanup_assertions():
diff --git a/lib/sqlalchemy/testing/plugin/plugin_base.py b/lib/sqlalchemy/testing/plugin/plugin_base.py
index c02f0556b..7ba31d3e3 100644
--- a/lib/sqlalchemy/testing/plugin/plugin_base.py
+++ b/lib/sqlalchemy/testing/plugin/plugin_base.py
@@ -181,7 +181,7 @@ def post_begin():
     from sqlalchemy.testing import fixtures, engines, exclusions, \
         assertions, warnings, profiling, config
     from sqlalchemy import util
-
+    warnings.setup_filters()
 
 def _log(opt_str, value, parser):
     global logging
@@ -491,13 +491,11 @@ def before_test(test, test_module_name, test_class, test_name):
 
     id_ = "%s.%s.%s" % (test_module_name, name, test_name)
 
-    warnings.resetwarnings()
     profiling._current_test = id_
 
 
 def after_test(test):
     engines.testing_reaper._after_test_ctx()
-    warnings.resetwarnings()
 
 
 def _possible_configs_for_cls(cls, reasons=None):
diff --git a/lib/sqlalchemy/testing/replay_fixture.py b/lib/sqlalchemy/testing/replay_fixture.py
index b8a0f6df1..b50f52e3d 100644
--- a/lib/sqlalchemy/testing/replay_fixture.py
+++ b/lib/sqlalchemy/testing/replay_fixture.py
@@ -29,9 +29,11 @@ class ReplayFixtureTest(fixtures.TestBase):
         self.session = Session(engine)
 
         self.setup_engine()
-        self._run_steps(ctx=self._dummy_ctx)
-        self.teardown_engine()
-        engine.dispose()
+        try:
+            self._run_steps(ctx=self._dummy_ctx)
+        finally:
+            self.teardown_engine()
+            engine.dispose()
 
         player = lambda: dbapi_session.player()
         engine = create_engine(
@@ -43,8 +45,11 @@ class ReplayFixtureTest(fixtures.TestBase):
         self.session = Session(engine)
 
         self.setup_engine()
-        self._run_steps(ctx=profiling.count_functions)
-        self.teardown_engine()
+        try:
+            self._run_steps(ctx=profiling.count_functions)
+        finally:
+            self.session.close()
+            engine.dispose()
 
     def setup_engine(self):
         pass
diff --git a/lib/sqlalchemy/testing/util.py b/lib/sqlalchemy/testing/util.py
index fc8390a79..7b3f721a6 100644
--- a/lib/sqlalchemy/testing/util.py
+++ b/lib/sqlalchemy/testing/util.py
@@ -203,5 +203,7 @@ class adict(dict):
         except KeyError:
             return dict.__getattribute__(self, key)
 
-    def get_all(self, *keys):
+    def __call__(self, *keys):
         return tuple([self[key] for key in keys])
+
+    get_all = __call__
diff --git a/lib/sqlalchemy/testing/warnings.py b/lib/sqlalchemy/testing/warnings.py
index b3314de6e..47f1e1404 100644
--- a/lib/sqlalchemy/testing/warnings.py
+++ b/lib/sqlalchemy/testing/warnings.py
@@ -9,25 +9,11 @@ from __future__ import absolute_import
 
 import warnings
 from .. import exc as sa_exc
-from .. import util
 import re
 
 
-def testing_warn(msg, stacklevel=3):
-    """Replaces sqlalchemy.util.warn during tests."""
-
-    filename = "sqlalchemy.testing.warnings"
-    lineno = 1
-    if isinstance(msg, util.string_types):
-        warnings.warn_explicit(msg, sa_exc.SAWarning, filename, lineno)
-    else:
-        warnings.warn_explicit(msg, filename, lineno)
-
-
-def resetwarnings():
-    """Reset warning behavior to testing defaults."""
-
-    util.warn = util.langhelpers.warn = testing_warn
+def setup_filters():
+    """Set global warning behavior for the test suite."""
 
     warnings.filterwarnings('ignore',
                             category=sa_exc.SAPendingDeprecationWarning)
@@ -35,24 +21,20 @@ def resetwarnings():
     warnings.filterwarnings('error', category=sa_exc.SAWarning)
 
 
-def assert_warnings(fn, warnings, regex=False):
+def assert_warnings(fn, warning_msgs, regex=False):
     """Assert that each of the given warnings are emitted by fn."""
 
-    from .assertions import eq_, emits_warning
+    from .assertions import eq_
 
-    canary = []
-    orig_warn = util.warn
+    with warnings.catch_warnings(record=True) as log:
+        # ensure that nothing is going into __warningregistry__
+        warnings.filterwarnings("always")
 
-    def capture_warnings(*args, **kw):
-        orig_warn(*args, **kw)
-        popwarn = warnings.pop(0)
-        canary.append(popwarn)
+        result = fn()
+    for warning in log:
+        popwarn = warning_msgs.pop(0)
         if regex:
-            assert re.match(popwarn, args[0])
+            assert re.match(popwarn, str(warning.message))
         else:
-            eq_(args[0], popwarn)
-    util.warn = util.langhelpers.warn = capture_warnings
-
-    result = emits_warning()(fn)()
-    assert canary, "No warning was emitted"
+            eq_(popwarn, str(warning.message))
     return result
diff --git a/lib/sqlalchemy/util/__init__.py b/lib/sqlalchemy/util/__init__.py
index 15b2ac38e..c963b18c3 100644
--- a/lib/sqlalchemy/util/__init__.py
+++ b/lib/sqlalchemy/util/__init__.py
@@ -21,7 +21,7 @@ from ._collections import KeyedTuple, ImmutableContainer, immutabledict, \
     UniqueAppender, PopulateDict, EMPTY_SET, to_list, to_set, \
     to_column_set, update_copy, flatten_iterator, \
     LRUCache, ScopedRegistry, ThreadLocalRegistry, WeakSequence, \
-    coerce_generator_arg
+    coerce_generator_arg, lightweight_named_tuple
 
 from .langhelpers import iterate_attributes, class_hierarchy, \
     portable_instancemethod, unbound_method_to_callable, \
@@ -34,7 +34,8 @@ from .langhelpers import iterate_attributes, class_hierarchy, \
     classproperty, set_creation_order, warn_exception, warn, NoneType,\
     constructor_copy, methods_equivalent, chop_traceback, asint,\
     generic_repr, counter, PluginLoader, hybridmethod, safe_reraise,\
-    get_callable_argspec, only_once
+    get_callable_argspec, only_once, attrsetter, ellipses_string, \
+    warn_limited
 
 from .deprecations import warn_deprecated, warn_pending_deprecation, \
     deprecated, pending_deprecation, inject_docstring_text
diff --git a/lib/sqlalchemy/util/_collections.py b/lib/sqlalchemy/util/_collections.py
index 0904d454e..a1fbc0fa0 100644
--- a/lib/sqlalchemy/util/_collections.py
+++ b/lib/sqlalchemy/util/_collections.py
@@ -17,7 +17,20 @@ import types
 EMPTY_SET = frozenset()
 
 
-class KeyedTuple(tuple):
+class AbstractKeyedTuple(tuple):
+    def keys(self):
+        """Return a list of string key names for this :class:`.KeyedTuple`.
+
+        .. seealso::
+
+            :attr:`.KeyedTuple._fields`
+
+        """
+
+        return list(self._fields)
+
+
+class KeyedTuple(AbstractKeyedTuple):
     """``tuple`` subclass that adds labeled names.
 
     E.g.::
@@ -56,23 +69,13 @@ class KeyedTuple(tuple):
 
     def __new__(cls, vals, labels=None):
         t = tuple.__new__(cls, vals)
-        t._labels = []
         if labels:
             t.__dict__.update(zip(labels, vals))
-            t._labels = labels
+        else:
+            labels = []
+        t.__dict__['_labels'] = labels
         return t
 
-    def keys(self):
-        """Return a list of string key names for this :class:`.KeyedTuple`.
-
-        .. seealso::
-
-            :attr:`.KeyedTuple._fields`
-
-        """
-
-        return [l for l in self._labels if l is not None]
-
     @property
     def _fields(self):
         """Return a tuple of string key names for this :class:`.KeyedTuple`.
@@ -86,7 +89,10 @@ class KeyedTuple(tuple):
             :meth:`.KeyedTuple.keys`
 
         """
-        return tuple(self.keys())
+        return tuple([l for l in self._labels if l is not None])
+
+    def __setattr__(self, key, value):
+        raise AttributeError("Can't set attribute: %s" % key)
 
     def _asdict(self):
         """Return the contents of this :class:`.KeyedTuple` as a dictionary.
@@ -100,6 +106,40 @@ class KeyedTuple(tuple):
         return dict((key, self.__dict__[key]) for key in self.keys())
 
 
+class _LW(AbstractKeyedTuple):
+    __slots__ = ()
+
+    def __new__(cls, vals):
+        return tuple.__new__(cls, vals)
+
+    def __reduce__(self):
+        # for pickling, degrade down to the regular
+        # KeyedTuple, thus avoiding anonymous class pickling
+        # difficulties
+        return KeyedTuple, (list(self), self._real_fields)
+
+    def _asdict(self):
+        """Return the contents of this :class:`.KeyedTuple` as a dictionary."""
+
+        d = dict(zip(self._real_fields, self))
+        d.pop(None, None)
+        return d
+
+
+def lightweight_named_tuple(name, fields):
+
+    tp_cls = type(name, (_LW,), {})
+    for idx, field in enumerate(fields):
+        if field is None:
+            continue
+        setattr(tp_cls, field, property(operator.itemgetter(idx)))
+
+    tp_cls._real_fields = fields
+    tp_cls._fields = tuple([f for f in fields if f is not None])
+
+    return tp_cls
+
+
 class ImmutableContainer(object):
     def _immutable(self, *arg, **kw):
         raise TypeError("%s object is immutable" % self.__class__.__name__)
diff --git a/lib/sqlalchemy/util/deprecations.py b/lib/sqlalchemy/util/deprecations.py
index d48efbaaa..124f304fc 100644
--- a/lib/sqlalchemy/util/deprecations.py
+++ b/lib/sqlalchemy/util/deprecations.py
@@ -102,7 +102,7 @@ def _decorate_with_warning(func, wtype, message, docstring_header=None):
 
     @decorator
     def warned(fn, *args, **kwargs):
-        warnings.warn(wtype(message), stacklevel=3)
+        warnings.warn(message, wtype, stacklevel=3)
         return fn(*args, **kwargs)
 
     doc = func.__doc__ is not None and func.__doc__ or ''
diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py
index 828e8f1f3..76f85f605 100644
--- a/lib/sqlalchemy/util/langhelpers.py
+++ b/lib/sqlalchemy/util/langhelpers.py
@@ -1189,24 +1189,55 @@ def warn_exception(func, *args, **kwargs):
         warn("%s('%s') ignored" % sys.exc_info()[0:2])
 
 
-def warn(msg, stacklevel=3):
+def ellipses_string(value, len_=25):
+    if len(value) > len_:
+        return "%s..." % value[0:len_]
+    else:
+        return value
+
+
+class _hash_limit_string(compat.text_type):
+    """A string subclass that can only be hashed on a maximum amount
+    of unique values.
+
+    This is used for warnings so that we can send out parameterized warnings
+    without the __warningregistry__ of the module,  or the non-overridable
+    "once" registry within warnings.py, overloading memory,
+
+
+    """
+    def __new__(cls, value, num, args):
+        interpolated = (value % args) + \
+            (" (this warning may be suppressed after %d occurrences)" % num)
+        self = super(_hash_limit_string, cls).__new__(cls, interpolated)
+        self._hash = hash("%s_%d" % (value, hash(interpolated) % num))
+        return self
+
+    def __hash__(self):
+        return self._hash
+
+    def __eq__(self, other):
+        return hash(self) == hash(other)
+
+
+def warn(msg):
     """Issue a warning.
 
     If msg is a string, :class:`.exc.SAWarning` is used as
     the category.
 
-    .. note::
+    """
+    warnings.warn(msg, exc.SAWarning, stacklevel=2)
+
 
-       This function is swapped out when the test suite
-       runs, with a compatible version that uses
-       warnings.warn_explicit, so that the warnings registry can
-       be controlled.
+def warn_limited(msg, args):
+    """Issue a warning with a paramterized string, limiting the number
+    of registrations.
 
     """
-    if isinstance(msg, compat.string_types):
-        warnings.warn(msg, exc.SAWarning, stacklevel=stacklevel)
-    else:
-        warnings.warn(msg, stacklevel=stacklevel)
+    if args:
+        msg = _hash_limit_string(msg, 10, args)
+    warnings.warn(msg, exc.SAWarning, stacklevel=2)
 
 
 def only_once(fn):
@@ -1249,3 +1280,11 @@ def chop_traceback(tb, exclude_prefix=_UNITTEST_RE, exclude_suffix=_SQLA_RE):
     return tb[start:end + 1]
 
 NoneType = type(None)
+
+def attrsetter(attrname):
+    code = \
+        "def set(obj, value):"\
+        "    obj.%s = value" % attrname
+    env = locals().copy()
+    exec(code, env)
+    return env['set']