Merge branch 'master' into pr157

10 files changed, 923 insertions, 217 deletions

diff --git a/lib/sqlalchemy/engine/__init__.py b/lib/sqlalchemy/engine/__init__.py
index 7a14cdfb5..adca6694e 100644
--- a/lib/sqlalchemy/engine/__init__.py
+++ b/lib/sqlalchemy/engine/__init__.py
@@ -1,5 +1,5 @@
 # engine/__init__.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -53,6 +53,7 @@ url.py
 
 from .interfaces import (
     Connectable,
+    CreateEnginePlugin,
     Dialect,
     ExecutionContext,
     ExceptionContext,
@@ -389,14 +390,33 @@ def create_engine(*args, **kwargs):
 def engine_from_config(configuration, prefix='sqlalchemy.', **kwargs):
     """Create a new Engine instance using a configuration dictionary.
 
-    The dictionary is typically produced from a config file where keys
-    are prefixed, such as sqlalchemy.url, sqlalchemy.echo, etc.  The
-    'prefix' argument indicates the prefix to be searched for.
+    The dictionary is typically produced from a config file.
+
+    The keys of interest to ``engine_from_config()`` should be prefixed, e.g.
+    ``sqlalchemy.url``, ``sqlalchemy.echo``, etc.  The 'prefix' argument
+    indicates the prefix to be searched for.  Each matching key (after the
+    prefix is stripped) is treated as though it were the corresponding keyword
+    argument to a :func:`.create_engine` call.
+
+    The only required key is (assuming the default prefix) ``sqlalchemy.url``,
+    which provides the :ref:`database URL <database_urls>`.
 
     A select set of keyword arguments will be "coerced" to their
-    expected type based on string values.  In a future release, this
-    functionality will be expanded and include dialect-specific
-    arguments.
+    expected type based on string values.    The set of arguments
+    is extensible per-dialect using the ``engine_config_types`` accessor.
+
+    :param configuration: A dictionary (typically produced from a config file,
+        but this is not a requirement).  Items whose keys start with the value
+        of 'prefix' will have that prefix stripped, and will then be passed to
+        :ref:`create_engine`.
+
+    :param prefix: Prefix to match and then strip from keys
+        in 'configuration'.
+
+    :param kwargs: Each keyword argument to ``engine_from_config()`` itself
+        overrides the corresponding item taken from the 'configuration'
+        dictionary.  Keyword arguments should *not* be prefixed.
+
     """
 
     options = dict((key[len(prefix):], configuration[key])
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 305fa4620..ef34eef01 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -1,5 +1,5 @@
 # engine/base.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -14,6 +14,7 @@ from __future__ import with_statement
 import sys
 from .. import exc, util, log, interfaces
 from ..sql import util as sql_util
+from ..sql import schema
 from .interfaces import Connectable, ExceptionContext
 from .util import _distill_params
 import contextlib
@@ -44,6 +45,22 @@ class Connection(Connectable):
 
     """
 
+    schema_for_object = schema._schema_getter(None)
+    """Return the ".schema" attribute for an object.
+
+    Used for :class:`.Table`, :class:`.Sequence` and similar objects,
+    and takes into account
+    the :paramref:`.Connection.execution_options.schema_translate_map`
+    parameter.
+
+      .. versionadded:: 1.1
+
+      .. seealso::
+
+          :ref:`schema_translating`
+
+    """
+
     def __init__(self, engine, connection=None, close_with_result=False,
                  _branch_from=None, _execution_options=None,
                  _dispatch=None,
@@ -67,6 +84,7 @@ class Connection(Connectable):
             self.should_close_with_result = False
             self.dispatch = _dispatch
             self._has_events = _branch_from._has_events
+            self.schema_for_object = _branch_from.schema_for_object
         else:
             self.__connection = connection \
                 if connection is not None else engine.raw_connection()
@@ -277,6 +295,19 @@ class Connection(Connectable):
           of many DBAPIs.  The flag is currently understood only by the
           psycopg2 dialect.
 
+        :param schema_translate_map: Available on: Connection, Engine.
+          A dictionary mapping schema names to schema names, that will be
+          applied to the :paramref:`.Table.schema` element of each
+          :class:`.Table` encountered when SQL or DDL expression elements
+          are compiled into strings; the resulting schema name will be
+          converted based on presence in the map of the original name.
+
+          .. versionadded:: 1.1
+
+          .. seealso::
+
+            :ref:`schema_translating`
+
         """
         c = self._clone()
         c._execution_options = c._execution_options.union(opt)
@@ -959,7 +990,10 @@ class Connection(Connectable):
 
         dialect = self.dialect
 
-        compiled = ddl.compile(dialect=dialect)
+        compiled = ddl.compile(
+            dialect=dialect,
+            schema_translate_map=self.schema_for_object
+            if not self.schema_for_object.is_default else None)
         ret = self._execute_context(
             dialect,
             dialect.execution_ctx_cls._init_ddl,
@@ -990,18 +1024,26 @@ class Connection(Connectable):
 
         dialect = self.dialect
         if 'compiled_cache' in self._execution_options:
-            key = dialect, elem, tuple(sorted(keys)), len(distilled_params) > 1
-            if key in self._execution_options['compiled_cache']:
-                compiled_sql = self._execution_options['compiled_cache'][key]
-            else:
+            key = (
+                dialect, elem, tuple(sorted(keys)),
+                self.schema_for_object.hash_key,
+                len(distilled_params) > 1
+            )
+            compiled_sql = self._execution_options['compiled_cache'].get(key)
+            if compiled_sql is None:
                 compiled_sql = elem.compile(
                     dialect=dialect, column_keys=keys,
-                    inline=len(distilled_params) > 1)
+                    inline=len(distilled_params) > 1,
+                    schema_translate_map=self.schema_for_object
+                    if not self.schema_for_object.is_default else None
+                )
                 self._execution_options['compiled_cache'][key] = compiled_sql
         else:
             compiled_sql = elem.compile(
                 dialect=dialect, column_keys=keys,
-                inline=len(distilled_params) > 1)
+                inline=len(distilled_params) > 1,
+                schema_translate_map=self.schema_for_object
+                if not self.schema_for_object.is_default else None)
 
         ret = self._execute_context(
             dialect,
@@ -1156,17 +1198,17 @@ class Connection(Connectable):
         if context.compiled:
             context.post_exec()
 
-        if context.is_crud:
+        if context.is_crud or context.is_text:
             result = context._setup_crud_result_proxy()
         else:
             result = context.get_result_proxy()
             if result._metadata is None:
-                result.close(_autoclose_connection=False)
+                result._soft_close(_autoclose_connection=False)
 
         if context.should_autocommit and self._root.__transaction is None:
             self._root._commit_impl(autocommit=True)
 
-        if result.closed and self.should_close_with_result:
+        if result._soft_closed and self.should_close_with_result:
             self.close()
 
         return result
@@ -1255,12 +1297,15 @@ class Connection(Connectable):
             if context:
                 context.is_disconnect = self._is_disconnect
 
+        invalidate_pool_on_disconnect = True
+
         if self._reentrant_error:
             util.raise_from_cause(
                 exc.DBAPIError.instance(statement,
                                         parameters,
                                         e,
-                                        self.dialect.dbapi.Error),
+                                        self.dialect.dbapi.Error,
+                                        dialect=self.dialect),
                 exc_info
             )
         self._reentrant_error = True
@@ -1276,7 +1321,8 @@ class Connection(Connectable):
                     parameters,
                     e,
                     self.dialect.dbapi.Error,
-                    connection_invalidated=self._is_disconnect)
+                    connection_invalidated=self._is_disconnect,
+                    dialect=self.dialect)
             else:
                 sqlalchemy_exception = None
 
@@ -1317,6 +1363,11 @@ class Connection(Connectable):
                     sqlalchemy_exception.connection_invalidated = \
                         self._is_disconnect = ctx.is_disconnect
 
+                # set up potentially user-defined value for
+                # invalidate pool.
+                invalidate_pool_on_disconnect = \
+                    ctx.invalidate_pool_on_disconnect
+
             if should_wrap and context:
                 context.handle_dbapi_exception(e)
 
@@ -1341,7 +1392,8 @@ class Connection(Connectable):
                 del self._is_disconnect
                 if not self.invalidated:
                     dbapi_conn_wrapper = self.__connection
-                    self.engine.pool._invalidate(dbapi_conn_wrapper, e)
+                    if invalidate_pool_on_disconnect:
+                        self.engine.pool._invalidate(dbapi_conn_wrapper, e)
                     self.invalidate(e)
             if self.should_close_with_result:
                 self.close()
@@ -1522,9 +1574,13 @@ class Transaction(object):
 
     def __init__(self, connection, parent):
         self.connection = connection
-        self._parent = parent or self
+        self._actual_parent = parent
         self.is_active = True
 
+    @property
+    def _parent(self):
+        return self._actual_parent or self
+
     def close(self):
         """Close this :class:`.Transaction`.
 
@@ -1673,6 +1729,22 @@ class Engine(Connectable, log.Identified):
     _has_events = False
     _connection_cls = Connection
 
+    schema_for_object = schema._schema_getter(None)
+    """Return the ".schema" attribute for an object.
+
+    Used for :class:`.Table`, :class:`.Sequence` and similar objects,
+    and takes into account
+    the :paramref:`.Connection.execution_options.schema_translate_map`
+    parameter.
+
+      .. versionadded:: 1.1
+
+      .. seealso::
+
+          :ref:`schema_translating`
+
+    """
+
     def __init__(self, pool, dialect, url,
                  logging_name=None, echo=None, proxy=None,
                  execution_options=None
@@ -1802,29 +1874,28 @@ class Engine(Connectable, log.Identified):
     def dispose(self):
         """Dispose of the connection pool used by this :class:`.Engine`.
 
+        This has the effect of fully closing all **currently checked in**
+        database connections.  Connections that are still checked out
+        will **not** be closed, however they will no longer be associated
+        with this :class:`.Engine`, so when they are closed individually,
+        eventually the :class:`.Pool` which they are associated with will
+        be garbage collected and they will be closed out fully, if
+        not already closed on checkin.
+
         A new connection pool is created immediately after the old one has
         been disposed.   This new pool, like all SQLAlchemy connection pools,
         does not make any actual connections to the database until one is
-        first requested.
+        first requested, so as long as the :class:`.Engine` isn't used again,
+        no new connections will be made.
 
-        This method has two general use cases:
-
-         * When a dropped connection is detected, it is assumed that all
-           connections held by the pool are potentially dropped, and
-           the entire pool is replaced.
-
-         * An application may want to use :meth:`dispose` within a test
-           suite that is creating multiple engines.
+        .. seealso::
 
-        It is critical to note that :meth:`dispose` does **not** guarantee
-        that the application will release all open database connections - only
-        those connections that are checked into the pool are closed.
-        Connections which remain checked out or have been detached from
-        the engine are not affected.
+            :ref:`engine_disposal`
 
         """
         self.pool.dispose()
         self.pool = self.pool.recreate()
+        self.dispatch.engine_disposed(self)
 
     def _execute_default(self, default):
         with self.contextual_connect() as conn:
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 17d2e2531..3ed2d5ee8 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -1,5 +1,5 @@
 # engine/default.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -16,7 +16,7 @@ as the base class for their own corresponding classes.
 import re
 import random
 from . import reflection, interfaces, result
-from ..sql import compiler, expression
+from ..sql import compiler, expression, schema
 from .. import types as sqltypes
 from .. import exc, util, pool, processors
 import codecs
@@ -61,14 +61,13 @@ class DefaultDialect(interfaces.Dialect):
 
     engine_config_types = util.immutabledict([
         ('convert_unicode', util.bool_or_str('force')),
-        ('pool_timeout', int),
+        ('pool_timeout', util.asint),
         ('echo', util.bool_or_str('debug')),
         ('echo_pool', util.bool_or_str('debug')),
-        ('pool_recycle', int),
-        ('pool_size', int),
-        ('max_overflow', int),
-        ('pool_threadlocal', bool),
-        ('use_native_unicode', bool),
+        ('pool_recycle', util.asint),
+        ('pool_size', util.asint),
+        ('max_overflow', util.asint),
+        ('pool_threadlocal', util.asbool),
     ])
 
     # if the NUMERIC type
@@ -157,6 +156,15 @@ class DefaultDialect(interfaces.Dialect):
 
     reflection_options = ()
 
+    dbapi_exception_translation_map = util.immutabledict()
+    """mapping used in the extremely unusual case that a DBAPI's
+    published exceptions don't actually have the __name__ that they
+    are linked towards.
+
+    .. versionadded:: 1.0.5
+
+    """
+
     def __init__(self, convert_unicode=False,
                  encoding='utf-8', paramstyle=None, dbapi=None,
                  implicit_returning=None,
@@ -390,10 +398,22 @@ class DefaultDialect(interfaces.Dialect):
                 if not branch:
                     self._set_connection_isolation(connection, isolation_level)
 
+        if 'schema_translate_map' in opts:
+            getter = schema._schema_getter(opts['schema_translate_map'])
+            engine.schema_for_object = getter
+
+            @event.listens_for(engine, "engine_connect")
+            def set_schema_translate_map(connection, branch):
+                connection.schema_for_object = getter
+
     def set_connection_execution_options(self, connection, opts):
         if 'isolation_level' in opts:
             self._set_connection_isolation(connection, opts['isolation_level'])
 
+        if 'schema_translate_map' in opts:
+            getter = schema._schema_getter(opts['schema_translate_map'])
+            connection.schema_for_object = getter
+
     def _set_connection_isolation(self, connection, level):
         if connection.in_transaction():
             util.warn(
@@ -454,16 +474,34 @@ class DefaultDialect(interfaces.Dialect):
         self.set_isolation_level(dbapi_conn, self.default_isolation_level)
 
 
+class StrCompileDialect(DefaultDialect):
+
+    statement_compiler = compiler.StrSQLCompiler
+    ddl_compiler = compiler.DDLCompiler
+    type_compiler = compiler.StrSQLTypeCompiler
+    preparer = compiler.IdentifierPreparer
+
+    supports_sequences = True
+    sequences_optional = True
+    preexecute_autoincrement_sequences = False
+    implicit_returning = False
+
+    supports_native_boolean = True
+
+    supports_simple_order_by_label = True
+
+
 class DefaultExecutionContext(interfaces.ExecutionContext):
     isinsert = False
     isupdate = False
     isdelete = False
     is_crud = False
+    is_text = False
     isddl = False
     executemany = False
-    result_map = None
     compiled = None
     statement = None
+    result_column_struct = None
     _is_implicit_returning = False
     _is_explicit_returning = False
 
@@ -522,10 +560,9 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         self.execution_options = compiled.statement._execution_options.union(
             connection._execution_options)
 
-        # compiled clauseelement.  process bind params, process table defaults,
-        # track collections used by ResultProxy to target and process results
-
-        self.result_map = compiled.result_map
+        self.result_column_struct = (
+            compiled._result_columns, compiled._ordered_columns,
+            compiled._textual_ordered_columns)
 
         self.unicode_statement = util.text_type(compiled)
         if not dialect.supports_unicode_statements:
@@ -537,6 +574,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         self.isinsert = compiled.isinsert
         self.isupdate = compiled.isupdate
         self.isdelete = compiled.isdelete
+        self.is_text = compiled.isplaintext
 
         if not parameters:
             self.compiled_parameters = [compiled.construct_params()]
@@ -616,6 +654,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         self.root_connection = connection
         self._dbapi_connection = dbapi_connection
         self.dialect = connection.dialect
+        self.is_text = True
 
         # plain text statement
         self.execution_options = connection._execution_options
@@ -817,15 +856,15 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
                 row = result.fetchone()
                 self.returned_defaults = row
                 self._setup_ins_pk_from_implicit_returning(row)
-                result.close(_autoclose_connection=False)
+                result._soft_close(_autoclose_connection=False)
                 result._metadata = None
             elif not self._is_explicit_returning:
-                result.close(_autoclose_connection=False)
+                result._soft_close(_autoclose_connection=False)
                 result._metadata = None
         elif self.isupdate and self._is_implicit_returning:
             row = result.fetchone()
             self.returned_defaults = row
-            result.close(_autoclose_connection=False)
+            result._soft_close(_autoclose_connection=False)
             result._metadata = None
 
         elif result._metadata is None:
@@ -833,7 +872,7 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
             # (which requires open cursor on some drivers
             # such as kintersbasdb, mxodbc)
             result.rowcount
-            result.close(_autoclose_connection=False)
+            result._soft_close(_autoclose_connection=False)
         return result
 
     def _setup_ins_pk_from_lastrowid(self):
@@ -842,18 +881,26 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
         compiled_params = self.compiled_parameters[0]
 
         lastrowid = self.get_lastrowid()
-        autoinc_col = table._autoincrement_column
-        if autoinc_col is not None:
-            # apply type post processors to the lastrowid
-            proc = autoinc_col.type._cached_result_processor(
-                self.dialect, None)
-            if proc is not None:
-                lastrowid = proc(lastrowid)
-        self.inserted_primary_key = [
-            lastrowid if c is autoinc_col else
-            compiled_params.get(key_getter(c), None)
-            for c in table.primary_key
-        ]
+        if lastrowid is not None:
+            autoinc_col = table._autoincrement_column
+            if autoinc_col is not None:
+                # apply type post processors to the lastrowid
+                proc = autoinc_col.type._cached_result_processor(
+                    self.dialect, None)
+                if proc is not None:
+                    lastrowid = proc(lastrowid)
+            self.inserted_primary_key = [
+                lastrowid if c is autoinc_col else
+                compiled_params.get(key_getter(c), None)
+                for c in table.primary_key
+            ]
+        else:
+            # don't have a usable lastrowid, so
+            # do the same as _setup_ins_pk_from_empty
+            self.inserted_primary_key = [
+                compiled_params.get(key_getter(c), None)
+                for c in table.primary_key
+            ]
 
     def _setup_ins_pk_from_empty(self):
         key_getter = self.compiled._key_getters_for_crud_column[2]
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 5f0d74328..26731f9a5 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -1,5 +1,5 @@
 # engine/interfaces.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -7,7 +7,7 @@
 
 """Define core interfaces used by the engine system."""
 
-from .. import util, event
+from .. import util
 
 # backwards compat
 from ..sql.compiler import Compiled, TypeCompiler
@@ -150,6 +150,16 @@ class Dialect(object):
       This will prevent types.Boolean from generating a CHECK
       constraint when that type is used.
 
+    dbapi_exception_translation_map
+       A dictionary of names that will contain as values the names of
+       pep-249 exceptions ("IntegrityError", "OperationalError", etc)
+       keyed to alternate class names, to support the case where a
+       DBAPI has exception classes that aren't named as they are
+       referred to (e.g. IntegrityError = MyException).   In the vast
+       majority of cases this dictionary is empty.
+
+       .. versionadded:: 1.0.5
+
     """
 
     _has_events = False
@@ -242,7 +252,9 @@ class Dialect(object):
 
         sequence
           a dictionary of the form
-              {'name' : str, 'start' :int, 'increment': int}
+              {'name' : str, 'start' :int, 'increment': int, 'minvalue': int,
+               'maxvalue': int, 'nominvalue': bool, 'nomaxvalue': bool,
+               'cycle': bool}
 
         Additional column attributes may be present.
         """
@@ -733,6 +745,146 @@ class Dialect(object):
 
         raise NotImplementedError()
 
+    @classmethod
+    def get_dialect_cls(cls, url):
+        """Given a URL, return the :class:`.Dialect` that will be used.
+
+        This is a hook that allows an external plugin to provide functionality
+        around an existing dialect, by allowing the plugin to be loaded
+        from the url based on an entrypoint, and then the plugin returns
+        the actual dialect to be used.
+
+        By default this just returns the cls.
+
+        .. versionadded:: 1.0.3
+
+        """
+        return cls
+
+    @classmethod
+    def engine_created(cls, engine):
+        """A convenience hook called before returning the final :class:`.Engine`.
+
+        If the dialect returned a different class from the
+        :meth:`.get_dialect_cls`
+        method, then the hook is called on both classes, first on
+        the dialect class returned by the :meth:`.get_dialect_cls` method and
+        then on the class on which the method was called.
+
+        The hook should be used by dialects and/or wrappers to apply special
+        events to the engine or its components.   In particular, it allows
+        a dialect-wrapping class to apply dialect-level events.
+
+        .. versionadded:: 1.0.3
+
+        """
+        pass
+
+
+class CreateEnginePlugin(object):
+    """A set of hooks intended to augment the construction of an
+    :class:`.Engine` object based on entrypoint names in a URL.
+
+    The purpose of :class:`.CreateEnginePlugin` is to allow third-party
+    systems to apply engine, pool and dialect level event listeners without
+    the need for the target application to be modified; instead, the plugin
+    names can be added to the database URL.  Target applications for
+    :class:`.CreateEnginePlugin` include:
+
+    * connection and SQL performance tools, e.g. which use events to track
+      number of checkouts and/or time spent with statements
+
+    * connectivity plugins such as proxies
+
+    Plugins are registered using entry points in a similar way as that
+    of dialects::
+
+        entry_points={
+            'sqlalchemy.plugins': [
+                'myplugin = myapp.plugins:MyPlugin'
+            ]
+
+    A plugin that uses the above names would be invoked from a database
+    URL as in::
+
+        from sqlalchemy import create_engine
+
+        engine = create_engine(
+          "mysql+pymysql://scott:tiger@localhost/test?plugin=myplugin")
+
+    The ``plugin`` argument supports multiple instances, so that a URL
+    may specify multiple plugins; they are loaded in the order stated
+    in the URL::
+
+        engine = create_engine(
+          "mysql+pymysql://scott:tiger@localhost/"
+          "test?plugin=plugin_one&plugin=plugin_twp&plugin=plugin_three")
+
+    A plugin can receive additional arguments from the URL string as
+    well as from the keyword arguments passed to :func:`.create_engine`.
+    The :class:`.URL` object and the keyword dictionary are passed to the
+    constructor so that these arguments can be extracted from the url's
+    :attr:`.URL.query` collection as well as from the dictionary::
+
+        class MyPlugin(CreateEnginePlugin):
+            def __init__(self, url, kwargs):
+                self.my_argument_one = url.query.pop('my_argument_one')
+                self.my_argument_two = url.query.pop('my_argument_two')
+                self.my_argument_three = kwargs.pop('my_argument_three', None)
+
+    Arguments like those illustrated above would be consumed from the
+    following::
+
+        from sqlalchemy import create_engine
+
+        engine = create_engine(
+          "mysql+pymysql://scott:tiger@localhost/"
+          "test?plugin=myplugin&my_argument_one=foo&my_argument_two=bar",
+          my_argument_three='bat')
+
+    The URL and dictionary are used for subsequent setup of the engine
+    as they are, so the plugin can modify their arguments in-place.
+    Arguments that are only understood by the plugin should be popped
+    or otherwise removed so that they aren't interpreted as erroneous
+    arguments afterwards.
+
+    When the engine creation process completes and produces the
+    :class:`.Engine` object, it is again passed to the plugin via the
+    :meth:`.CreateEnginePlugin.engine_created` hook.  In this hook, additional
+    changes can be made to the engine, most typically involving setup of
+    events (e.g. those defined in :ref:`core_event_toplevel`).
+
+    .. versionadded:: 1.1
+
+    """
+    def __init__(self, url, kwargs):
+        """Contruct a new :class:`.CreateEnginePlugin`.
+
+        The plugin object is instantiated individually for each call
+        to :func:`.create_engine`.  A single :class:`.Engine` will be
+        passed to the :meth:`.CreateEnginePlugin.engine_created` method
+        corresponding to this URL.
+
+        :param url: the :class:`.URL` object.  The plugin should inspect
+         what it needs here as well as remove its custom arguments from the
+         :attr:`.URL.query` collection.  The URL can be modified in-place
+         in any other way as well.
+        :param kwargs: The keyword arguments passed to :func`.create_engine`.
+         The plugin can read and modify this dictionary in-place, to affect
+         the ultimate arguments used to create the engine.  It should
+         remove its custom arguments from the dictionary as well.
+
+        """
+        self.url = url
+
+    def engine_created(self, engine):
+        """Receive the :class:`.Engine` object when it is fully constructed.
+
+        The plugin may make additional changes to the engine, such as
+        registering engine or connection pool events.
+
+        """
+
 
 class ExecutionContext(object):
     """A messenger object for a Dialect that corresponds to a single
@@ -1085,3 +1237,21 @@ class ExceptionContext(object):
     changing this flag.
 
     """
+
+    invalidate_pool_on_disconnect = True
+    """Represent whether all connections in the pool should be invalidated
+    when a "disconnect" condition is in effect.
+
+    Setting this flag to False within the scope of the
+    :meth:`.ConnectionEvents.handle_error` event will have the effect such
+    that the full collection of connections in the pool will not be
+    invalidated during a disconnect; only the current connection that is the
+    subject of the error will actually be invalidated.
+
+    The purpose of this flag is for custom disconnect-handling schemes where
+    the invalidation of other connections in the pool is to be performed
+    based on other conditions, or even on a per-connection basis.
+
+    .. versionadded:: 1.0.3
+
+    """
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 6e102aad6..eaa5e2e48 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -1,5 +1,5 @@
 # engine/reflection.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -529,7 +529,8 @@ class Inspector(object):
         """
         dialect = self.bind.dialect
 
-        schema = table.schema
+        schema = self.bind.schema_for_object(table)
+
         table_name = table.name
 
         # get table-level arguments that are specifically
diff --git a/lib/sqlalchemy/engine/result.py b/lib/sqlalchemy/engine/result.py
index 3995942ef..f09b0b40b 100644
--- a/lib/sqlalchemy/engine/result.py
+++ b/lib/sqlalchemy/engine/result.py
@@ -1,5 +1,5 @@
 # engine/result.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -10,7 +10,7 @@ and :class:`.RowProxy."""
 
 
 from .. import exc, util
-from ..sql import expression, sqltypes
+from ..sql import expression, sqltypes, util as sql_util
 import collections
 import operator
 
@@ -35,7 +35,10 @@ except ImportError:
 
 try:
     from sqlalchemy.cresultproxy import BaseRowProxy
+    _baserowproxy_usecext = True
 except ImportError:
+    _baserowproxy_usecext = False
+
     class BaseRowProxy(object):
         __slots__ = ('_parent', '_row', '_processors', '_keymap')
 
@@ -84,8 +87,8 @@ except ImportError:
                     raise
             if index is None:
                 raise exc.InvalidRequestError(
-                    "Ambiguous column name '%s' in result set! "
-                    "try 'use_labels' option on select statement." % key)
+                    "Ambiguous column name '%s' in "
+                    "result set column descriptions" % obj)
             if processor is not None:
                 return processor(self._row[index])
             else:
@@ -150,7 +153,7 @@ class RowProxy(BaseRowProxy):
         return self._op(other, operator.ne)
 
     def __repr__(self):
-        return repr(tuple(self))
+        return repr(sql_util._repr_row(self))
 
     def has_key(self, key):
         """Return True if this RowProxy contains the given key."""
@@ -186,103 +189,324 @@ class ResultMetaData(object):
     """Handle cursor.description, applying additional info from an execution
     context."""
 
-    def __init__(self, parent, metadata):
-        self._processors = processors = []
+    __slots__ = (
+        '_keymap', 'case_sensitive', 'matched_on_name',
+        '_processors', 'keys', '_orig_processors')
 
-        # We do not strictly need to store the processor in the key mapping,
-        # though it is faster in the Python version (probably because of the
-        # saved attribute lookup self._processors)
-        self._keymap = keymap = {}
-        self.keys = []
+    def __init__(self, parent, cursor_description):
         context = parent.context
         dialect = context.dialect
-        typemap = dialect.dbapi_type_map
-        translate_colname = context._translate_colname
         self.case_sensitive = dialect.case_sensitive
+        self.matched_on_name = False
+
+        if context.result_column_struct:
+            result_columns, cols_are_ordered, textual_ordered = \
+                context.result_column_struct
+            num_ctx_cols = len(result_columns)
+        else:
+            result_columns = cols_are_ordered = \
+                num_ctx_cols = textual_ordered = False
+
+        # merge cursor.description with the column info
+        # present in the compiled structure, if any
+        raw = self._merge_cursor_description(
+            context, cursor_description, result_columns,
+            num_ctx_cols, cols_are_ordered, textual_ordered)
+
+        self._keymap = {}
+        if not _baserowproxy_usecext:
+            # keymap indexes by integer index...
+            self._keymap.update([
+                (elem[0], (elem[3], elem[4], elem[0]))
+                for elem in raw
+            ])
+
+        # processors in key order for certain per-row
+        # views like __iter__ and slices
+        self._processors = [elem[3] for elem in raw]
+
+        # keymap by primary string...
+        by_key = dict([
+            (elem[2], (elem[3], elem[4], elem[0]))
+            for elem in raw
+        ])
+
+        # for compiled SQL constructs, copy additional lookup keys into
+        # the key lookup map, such as Column objects, labels,
+        # column keys and other names
+        if num_ctx_cols:
+
+            # if by-primary-string dictionary smaller (or bigger?!) than
+            # number of columns, assume we have dupes, rewrite
+            # dupe records with "None" for index which results in
+            # ambiguous column exception when accessed.
+            if len(by_key) != num_ctx_cols:
+                seen = set()
+                for rec in raw:
+                    key = rec[1]
+                    if key in seen:
+                        # this is an "ambiguous" element, replacing
+                        # the full record in the map
+                        by_key[key] = (None, key, None)
+                    seen.add(key)
+
+                # copy secondary elements from compiled columns
+                # into self._keymap, write in the potentially "ambiguous"
+                # element
+                self._keymap.update([
+                    (obj_elem, by_key[elem[2]])
+                    for elem in raw if elem[4]
+                    for obj_elem in elem[4]
+                ])
+
+                # if we did a pure positional match, then reset the
+                # original "expression element" back to the "unambiguous"
+                # entry.  This is a new behavior in 1.1 which impacts
+                # TextAsFrom but also straight compiled SQL constructs.
+                if not self.matched_on_name:
+                    self._keymap.update([
+                        (elem[4][0], (elem[3], elem[4], elem[0]))
+                        for elem in raw if elem[4]
+                    ])
+            else:
+                # no dupes - copy secondary elements from compiled
+                # columns into self._keymap
+                self._keymap.update([
+                    (obj_elem, (elem[3], elem[4], elem[0]))
+                    for elem in raw if elem[4]
+                    for obj_elem in elem[4]
+                ])
+
+        # update keymap with primary string names taking
+        # precedence
+        self._keymap.update(by_key)
+
+        # update keymap with "translated" names (sqlite-only thing)
+        if not num_ctx_cols and context._translate_colname:
+            self._keymap.update([
+                (elem[5], self._keymap[elem[2]])
+                for elem in raw if elem[5]
+            ])
+
+    def _merge_cursor_description(
+            self, context, cursor_description, result_columns,
+            num_ctx_cols, cols_are_ordered, textual_ordered):
+        """Merge a cursor.description with compiled result column information.
+
+        There are at least four separate strategies used here, selected
+        depending on the type of SQL construct used to start with.
+
+        The most common case is that of the compiled SQL expression construct,
+        which generated the column names present in the raw SQL string and
+        which has the identical number of columns as were reported by
+        cursor.description.  In this case, we assume a 1-1 positional mapping
+        between the entries in cursor.description and the compiled object.
+        This is also the most performant case as we disregard extracting /
+        decoding the column names present in cursor.description since we
+        already have the desired name we generated in the compiled SQL
+        construct.
+
+        The next common case is that of the completely raw string SQL,
+        such as passed to connection.execute().  In this case we have no
+        compiled construct to work with, so we extract and decode the
+        names from cursor.description and index those as the primary
+        result row target keys.
+
+        The remaining fairly common case is that of the textual SQL
+        that includes at least partial column information; this is when
+        we use a :class:`.TextAsFrom` construct.   This contruct may have
+        unordered or ordered column information.  In the ordered case, we
+        merge the cursor.description and the compiled construct's information
+        positionally, and warn if there are additional description names
+        present, however we still decode the names in cursor.description
+        as we don't have a guarantee that the names in the columns match
+        on these.   In the unordered case, we match names in cursor.description
+        to that of the compiled construct based on name matching.
+        In both of these cases, the cursor.description names and the column
+        expression objects and names are indexed as result row target keys.
+
+        The final case is much less common, where we have a compiled
+        non-textual SQL expression construct, but the number of columns
+        in cursor.description doesn't match what's in the compiled
+        construct.  We make the guess here that there might be textual
+        column expressions in the compiled construct that themselves include
+        a comma in them causing them to split.  We do the same name-matching
+        as with textual non-ordered columns.
+
+        The name-matched system of merging is the same as that used by
+        SQLAlchemy for all cases up through te 0.9 series.   Positional
+        matching for compiled SQL expressions was introduced in 1.0 as a
+        major performance feature, and positional matching for textual
+        :class:`.TextAsFrom` objects in 1.1.  As name matching is no longer
+        a common case, it was acceptable to factor it into smaller generator-
+        oriented methods that are easier to understand, but incur slightly
+        more performance overhead.
+
+        """
+
+        case_sensitive = context.dialect.case_sensitive
+
+        if num_ctx_cols and \
+                cols_are_ordered and \
+                not textual_ordered and \
+                num_ctx_cols == len(cursor_description):
+            self.keys = [elem[0] for elem in result_columns]
+            # pure positional 1-1 case; doesn't need to read
+            # the names from cursor.description
+            return [
+                (
+                    idx,
+                    key,
+                    name.lower() if not case_sensitive else name,
+                    context.get_result_processor(
+                        type_, key, cursor_description[idx][1]
+                    ),
+                    obj,
+                    None
+                ) for idx, (key, name, obj, type_)
+                in enumerate(result_columns)
+            ]
+        else:
+            # name-based or text-positional cases, where we need
+            # to read cursor.description names
+            if textual_ordered:
+                # textual positional case
+                raw_iterator = self._merge_textual_cols_by_position(
+                    context, cursor_description, result_columns)
+            elif num_ctx_cols:
+                # compiled SQL with a mismatch of description cols
+                # vs. compiled cols, or textual w/ unordered columns
+                raw_iterator = self._merge_cols_by_name(
+                    context, cursor_description, result_columns)
+            else:
+                # no compiled SQL, just a raw string
+                raw_iterator = self._merge_cols_by_none(
+                    context, cursor_description)
+
+            return [
+                (
+                    idx, colname, colname,
+                    context.get_result_processor(
+                        mapped_type, colname, coltype),
+                    obj, untranslated)
+
+                for idx, colname, mapped_type, coltype, obj, untranslated
+                in raw_iterator
+            ]
+
+    def _colnames_from_description(self, context, cursor_description):
+        """Extract column names and data types from a cursor.description.
+
+        Applies unicode decoding, column translation, "normalization",
+        and case sensitivity rules to the names based on the dialect.
+
+        """
+
+        dialect = context.dialect
+        case_sensitive = dialect.case_sensitive
+        translate_colname = context._translate_colname
+        description_decoder = dialect._description_decoder \
+            if dialect.description_encoding else None
+        normalize_name = dialect.normalize_name \
+            if dialect.requires_name_normalize else None
+        untranslated = None
 
-        # high precedence key values.
-        primary_keymap = {}
+        self.keys = []
 
-        for i, rec in enumerate(metadata):
+        for idx, rec in enumerate(cursor_description):
             colname = rec[0]
             coltype = rec[1]
 
-            if dialect.description_encoding:
-                colname = dialect._description_decoder(colname)
+            if description_decoder:
+                colname = description_decoder(colname)
 
             if translate_colname:
                 colname, untranslated = translate_colname(colname)
 
-            if dialect.requires_name_normalize:
-                colname = dialect.normalize_name(colname)
+            if normalize_name:
+                colname = normalize_name(colname)
+
+            self.keys.append(colname)
+            if not case_sensitive:
+                colname = colname.lower()
+
+            yield idx, colname, untranslated, coltype
 
-            if context.result_map:
-                try:
-                    name, obj, type_ = context.result_map[
-                        colname if self.case_sensitive else colname.lower()]
-                except KeyError:
-                    name, obj, type_ = \
-                        colname, None, typemap.get(coltype, sqltypes.NULLTYPE)
+    def _merge_textual_cols_by_position(
+            self, context, cursor_description, result_columns):
+        dialect = context.dialect
+        typemap = dialect.dbapi_type_map
+        num_ctx_cols = len(result_columns) if result_columns else None
+
+        if num_ctx_cols > len(cursor_description):
+            util.warn(
+                "Number of columns in textual SQL (%d) is "
+                "smaller than number of columns requested (%d)" % (
+                    num_ctx_cols, len(cursor_description)
+                ))
+
+        seen = set()
+        for idx, colname, untranslated, coltype in \
+                self._colnames_from_description(context, cursor_description):
+            if idx < num_ctx_cols:
+                ctx_rec = result_columns[idx]
+                obj = ctx_rec[2]
+                mapped_type = ctx_rec[3]
+                if obj[0] in seen:
+                    raise exc.InvalidRequestError(
+                        "Duplicate column expression requested "
+                        "in textual SQL: %r" % obj[0])
+                seen.add(obj[0])
             else:
-                name, obj, type_ = \
-                    colname, None, typemap.get(coltype, sqltypes.NULLTYPE)
-
-            processor = context.get_result_processor(type_, colname, coltype)
-
-            processors.append(processor)
-            rec = (processor, obj, i)
-
-            # indexes as keys. This is only needed for the Python version of
-            # RowProxy (the C version uses a faster path for integer indexes).
-            primary_keymap[i] = rec
-
-            # populate primary keymap, looking for conflicts.
-            if primary_keymap.setdefault(
-                    name if self.case_sensitive
-                    else name.lower(),
-                    rec) is not rec:
-                # place a record that doesn't have the "index" - this
-                # is interpreted later as an AmbiguousColumnError,
-                # but only when actually accessed.   Columns
-                # colliding by name is not a problem if those names
-                # aren't used; integer access is always
-                # unambiguous.
-                primary_keymap[name
-                               if self.case_sensitive
-                               else name.lower()] = rec = (None, obj, None)
+                mapped_type = typemap.get(coltype, sqltypes.NULLTYPE)
+                obj = None
 
-            self.keys.append(colname)
-            if obj:
-                for o in obj:
-                    keymap[o] = rec
-                    # technically we should be doing this but we
-                    # are saving on callcounts by not doing so.
-                    # if keymap.setdefault(o, rec) is not rec:
-                    #    keymap[o] = (None, obj, None)
-
-            if translate_colname and \
-                    untranslated:
-                keymap[untranslated] = rec
-
-        # overwrite keymap values with those of the
-        # high precedence keymap.
-        keymap.update(primary_keymap)
-
-    @util.pending_deprecation("0.8", "sqlite dialect uses "
-                              "_translate_colname() now")
-    def _set_keymap_synonym(self, name, origname):
-        """Set a synonym for the given name.
-
-        Some dialects (SQLite at the moment) may use this to
-        adjust the column names that are significant within a
-        row.
+            yield idx, colname, mapped_type, coltype, obj, untranslated
 
-        """
-        rec = (processor, obj, i) = self._keymap[origname if
-                                                 self.case_sensitive
-                                                 else origname.lower()]
-        if self._keymap.setdefault(name, rec) is not rec:
-            self._keymap[name] = (processor, obj, None)
+    def _merge_cols_by_name(self, context, cursor_description, result_columns):
+        dialect = context.dialect
+        typemap = dialect.dbapi_type_map
+        case_sensitive = dialect.case_sensitive
+        result_map = self._create_result_map(result_columns, case_sensitive)
+
+        self.matched_on_name = True
+        for idx, colname, untranslated, coltype in \
+                self._colnames_from_description(context, cursor_description):
+            try:
+                ctx_rec = result_map[colname]
+            except KeyError:
+                mapped_type = typemap.get(coltype, sqltypes.NULLTYPE)
+                obj = None
+            else:
+                obj = ctx_rec[1]
+                mapped_type = ctx_rec[2]
+            yield idx, colname, mapped_type, coltype, obj, untranslated
+
+    def _merge_cols_by_none(self, context, cursor_description):
+        dialect = context.dialect
+        typemap = dialect.dbapi_type_map
+        for idx, colname, untranslated, coltype in \
+                self._colnames_from_description(context, cursor_description):
+            mapped_type = typemap.get(coltype, sqltypes.NULLTYPE)
+            yield idx, colname, mapped_type, coltype, None, untranslated
+
+    @classmethod
+    def _create_result_map(cls, result_columns, case_sensitive=True):
+        d = {}
+        for elem in result_columns:
+            key, rec = elem[0], elem[1:]
+            if not case_sensitive:
+                key = key.lower()
+            if key in d:
+                # conflicting keyname, just double up the list
+                # of objects.  this will cause an "ambiguous name"
+                # error if an attempt is made by the result set to
+                # access.
+                e_name, e_obj, e_type = d[key]
+                d[key] = e_name, e_obj + rec[1], e_type
+            else:
+                d[key] = rec
+        return d
 
     def _key_fallback(self, key, raiseerr=True):
         map = self._keymap
@@ -337,19 +561,19 @@ class ResultMetaData(object):
         else:
             return self._key_fallback(key, False) is not None
 
-    def _getter(self, key):
+    def _getter(self, key, raiseerr=True):
         if key in self._keymap:
             processor, obj, index = self._keymap[key]
         else:
-            ret = self._key_fallback(key, False)
+            ret = self._key_fallback(key, raiseerr)
             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)
+                "Ambiguous column name '%s' in "
+                "result set column descriptions" % obj)
 
         return operator.itemgetter(index)
 
@@ -362,6 +586,7 @@ class ResultMetaData(object):
             ),
             'keys': self.keys,
             "case_sensitive": self.case_sensitive,
+            "matched_on_name": self.matched_on_name
         }
 
     def __setstate__(self, state):
@@ -375,7 +600,7 @@ class ResultMetaData(object):
             keymap[key] = (None, None, index)
         self.keys = state['keys']
         self.case_sensitive = state['case_sensitive']
-        self._echo = False
+        self.matched_on_name = state['matched_on_name']
 
 
 class ResultProxy(object):
@@ -403,38 +628,49 @@ class ResultProxy(object):
     out_parameters = None
     _can_close_connection = False
     _metadata = None
+    _soft_closed = False
+    closed = False
 
     def __init__(self, context):
         self.context = context
         self.dialect = context.dialect
-        self.closed = False
         self.cursor = self._saved_cursor = context.cursor
         self.connection = context.root_connection
         self._echo = self.connection._echo and \
             context.engine._should_log_debug()
         self._init_metadata()
 
-    def _getter(self, key):
-        return self._metadata._getter(key)
+    def _getter(self, key, raiseerr=True):
+        try:
+            getter = self._metadata._getter
+        except AttributeError:
+            return self._non_result(None)
+        else:
+            return getter(key, raiseerr)
 
     def _has_key(self, key):
-        return self._metadata._has_key(key)
+        try:
+            has_key = self._metadata._has_key
+        except AttributeError:
+            return self._non_result(None)
+        else:
+            return has_key(key)
 
     def _init_metadata(self):
-        metadata = self._cursor_description()
-        if metadata is not None:
+        cursor_description = self._cursor_description()
+        if cursor_description is not None:
             if self.context.compiled and \
                     'compiled_cache' in self.context.execution_options:
                 if self.context.compiled._cached_metadata:
                     self._metadata = self.context.compiled._cached_metadata
                 else:
                     self._metadata = self.context.compiled._cached_metadata = \
-                        ResultMetaData(self, metadata)
+                        ResultMetaData(self, cursor_description)
             else:
-                self._metadata = ResultMetaData(self, metadata)
+                self._metadata = ResultMetaData(self, cursor_description)
             if self._echo:
                 self.context.engine.logger.debug(
-                    "Col %r", tuple(x[0] for x in metadata))
+                    "Col %r", tuple(x[0] for x in cursor_description))
 
     def keys(self):
         """Return the current set of string keys for rows."""
@@ -544,39 +780,85 @@ class ResultProxy(object):
 
         return self._saved_cursor.description
 
-    def close(self, _autoclose_connection=True):
-        """Close this ResultProxy.
-
-        Closes the underlying DBAPI cursor corresponding to the execution.
-
-        Note that any data cached within this ResultProxy is still available.
-        For some types of results, this may include buffered rows.
+    def _soft_close(self, _autoclose_connection=True):
+        """Soft close this :class:`.ResultProxy`.
 
-        If this ResultProxy was generated from an implicit execution,
-        the underlying Connection will also be closed (returns the
-        underlying DBAPI connection to the connection pool.)
+        This releases all DBAPI cursor resources, but leaves the
+        ResultProxy "open" from a semantic perspective, meaning the
+        fetchXXX() methods will continue to return empty results.
 
         This method is called automatically when:
 
         * all result rows are exhausted using the fetchXXX() methods.
         * cursor.description is None.
 
+        This method is **not public**, but is documented in order to clarify
+        the "autoclose" process used.
+
+        .. versionadded:: 1.0.0
+
+        .. seealso::
+
+            :meth:`.ResultProxy.close`
+
+
+        """
+        if self._soft_closed:
+            return
+        self._soft_closed = True
+        cursor = self.cursor
+        self.connection._safe_close_cursor(cursor)
+        if _autoclose_connection and \
+                self.connection.should_close_with_result:
+            self.connection.close()
+        self.cursor = None
+
+    def close(self):
+        """Close this ResultProxy.
+
+        This closes out the underlying DBAPI cursor corresonding
+        to the statement execution, if one is stil present.  Note that the
+        DBAPI cursor is automatically released when the :class:`.ResultProxy`
+        exhausts all available rows.  :meth:`.ResultProxy.close` is generally
+        an optional method except in the case when discarding a
+        :class:`.ResultProxy` that still has additional rows pending for fetch.
+
+        In the case of a result that is the product of
+        :ref:`connectionless execution <dbengine_implicit>`,
+        the underyling :class:`.Connection` object is also closed, which
+        :term:`releases` DBAPI connection resources.
+
+        After this method is called, it is no longer valid to call upon
+        the fetch methods, which will raise a :class:`.ResourceClosedError`
+        on subsequent use.
+
+        .. versionchanged:: 1.0.0 - the :meth:`.ResultProxy.close` method
+           has been separated out from the process that releases the underlying
+           DBAPI cursor resource.   The "auto close" feature of the
+           :class:`.Connection` now performs a so-called "soft close", which
+           releases the underlying DBAPI cursor, but allows the
+           :class:`.ResultProxy` to still behave as an open-but-exhausted
+           result set; the actual :meth:`.ResultProxy.close` method is never
+           called.    It is still safe to discard a :class:`.ResultProxy`
+           that has been fully exhausted without calling this method.
+
+        .. seealso::
+
+            :ref:`connections_toplevel`
+
+            :meth:`.ResultProxy._soft_close`
+
         """
 
         if not self.closed:
+            self._soft_close()
             self.closed = True
-            self.connection._safe_close_cursor(self.cursor)
-            if _autoclose_connection and \
-                    self.connection.should_close_with_result:
-                self.connection.close()
-            # allow consistent errors
-            self.cursor = None
 
     def __iter__(self):
         while True:
             row = self.fetchone()
             if row is None:
-                raise StopIteration
+                return
             else:
                 yield row
 
@@ -761,7 +1043,7 @@ class ResultProxy(object):
         try:
             return self.cursor.fetchone()
         except AttributeError:
-            self._non_result()
+            return self._non_result(None)
 
     def _fetchmany_impl(self, size=None):
         try:
@@ -770,22 +1052,24 @@ class ResultProxy(object):
             else:
                 return self.cursor.fetchmany(size)
         except AttributeError:
-            self._non_result()
+            return self._non_result([])
 
     def _fetchall_impl(self):
         try:
             return self.cursor.fetchall()
         except AttributeError:
-            self._non_result()
+            return self._non_result([])
 
-    def _non_result(self):
+    def _non_result(self, default):
         if self._metadata is None:
             raise exc.ResourceClosedError(
                 "This result object does not return rows. "
                 "It has been closed automatically.",
             )
-        else:
+        elif self.closed:
             raise exc.ResourceClosedError("This result object is closed.")
+        else:
+            return default
 
     def process_rows(self, rows):
         process_row = self._process_row
@@ -796,7 +1080,7 @@ class ResultProxy(object):
             log = self.context.engine.logger.debug
             l = []
             for row in rows:
-                log("Row %r", row)
+                log("Row %r", sql_util._repr_row(row))
                 l.append(process_row(metadata, row, processors, keymap))
             return l
         else:
@@ -804,11 +1088,25 @@ class ResultProxy(object):
                     for row in rows]
 
     def fetchall(self):
-        """Fetch all rows, just like DB-API ``cursor.fetchall()``."""
+        """Fetch all rows, just like DB-API ``cursor.fetchall()``.
+
+        After all rows have been exhausted, the underlying DBAPI
+        cursor resource is released, and the object may be safely
+        discarded.
+
+        Subsequent calls to :meth:`.ResultProxy.fetchall` will return
+        an empty list.   After the :meth:`.ResultProxy.close` method is
+        called, the method will raise :class:`.ResourceClosedError`.
+
+        .. versionchanged:: 1.0.0 - Added "soft close" behavior which
+           allows the result to be used in an "exhausted" state prior to
+           calling the :meth:`.ResultProxy.close` method.
+
+        """
 
         try:
             l = self.process_rows(self._fetchall_impl())
-            self.close()
+            self._soft_close()
             return l
         except Exception as e:
             self.connection._handle_dbapi_exception(
@@ -819,15 +1117,25 @@ class ResultProxy(object):
         """Fetch many rows, just like DB-API
         ``cursor.fetchmany(size=cursor.arraysize)``.
 
-        If rows are present, the cursor remains open after this is called.
-        Else the cursor is automatically closed and an empty list is returned.
+        After all rows have been exhausted, the underlying DBAPI
+        cursor resource is released, and the object may be safely
+        discarded.
+
+        Calls to :meth:`.ResultProxy.fetchmany` after all rows have been
+        exhuasted will return
+        an empty list.   After the :meth:`.ResultProxy.close` method is
+        called, the method will raise :class:`.ResourceClosedError`.
+
+        .. versionchanged:: 1.0.0 - Added "soft close" behavior which
+           allows the result to be used in an "exhausted" state prior to
+           calling the :meth:`.ResultProxy.close` method.
 
         """
 
         try:
             l = self.process_rows(self._fetchmany_impl(size))
             if len(l) == 0:
-                self.close()
+                self._soft_close()
             return l
         except Exception as e:
             self.connection._handle_dbapi_exception(
@@ -837,8 +1145,18 @@ class ResultProxy(object):
     def fetchone(self):
         """Fetch one row, just like DB-API ``cursor.fetchone()``.
 
-        If a row is present, the cursor remains open after this is called.
-        Else the cursor is automatically closed and None is returned.
+        After all rows have been exhausted, the underlying DBAPI
+        cursor resource is released, and the object may be safely
+        discarded.
+
+        Calls to :meth:`.ResultProxy.fetchone` after all rows have
+        been exhausted will return ``None``.
+        After the :meth:`.ResultProxy.close` method is
+        called, the method will raise :class:`.ResourceClosedError`.
+
+        .. versionchanged:: 1.0.0 - Added "soft close" behavior which
+           allows the result to be used in an "exhausted" state prior to
+           calling the :meth:`.ResultProxy.close` method.
 
         """
         try:
@@ -846,7 +1164,7 @@ class ResultProxy(object):
             if row is not None:
                 return self.process_rows([row])[0]
             else:
-                self.close()
+                self._soft_close()
                 return None
         except Exception as e:
             self.connection._handle_dbapi_exception(
@@ -858,9 +1176,12 @@ class ResultProxy(object):
 
         Returns None if no row is present.
 
+        After calling this method, the object is fully closed,
+        e.g. the :meth:`.ResultProxy.close` method will have been called.
+
         """
         if self._metadata is None:
-            self._non_result()
+            return self._non_result(None)
 
         try:
             row = self._fetchone_impl()
@@ -882,6 +1203,9 @@ class ResultProxy(object):
 
         Returns None if no row is present.
 
+        After calling this method, the object is fully closed,
+        e.g. the :meth:`.ResultProxy.close` method will have been called.
+
         """
         row = self.first()
         if row is not None:
@@ -902,10 +1226,27 @@ class BufferedRowResultProxy(ResultProxy):
 
     The pre-fetching behavior fetches only one row initially, and then
     grows its buffer size by a fixed amount with each successive need
-    for additional rows up to a size of 100.
+    for additional rows up to a size of 1000.
+
+    The size argument is configurable using the ``max_row_buffer``
+    execution option::
+
+        with psycopg2_engine.connect() as conn:
+
+            result = conn.execution_options(
+                stream_results=True, max_row_buffer=50
+                ).execute("select * from table")
+
+    .. versionadded:: 1.0.6 Added the ``max_row_buffer`` option.
+
+    .. seealso::
+
+        :ref:`psycopg2_execution_options`
     """
 
     def _init_metadata(self):
+        self._max_row_buffer = self.context.execution_options.get(
+            'max_row_buffer', None)
         self.__buffer_rows()
         super(BufferedRowResultProxy, self)._init_metadata()
 
@@ -925,13 +1266,21 @@ class BufferedRowResultProxy(ResultProxy):
     }
 
     def __buffer_rows(self):
+        if self.cursor is None:
+            return
         size = getattr(self, '_bufsize', 1)
         self.__rowbuffer = collections.deque(self.cursor.fetchmany(size))
         self._bufsize = self.size_growth.get(size, size)
+        if self._max_row_buffer is not None:
+            self._bufsize = min(self._max_row_buffer, self._bufsize)
+
+    def _soft_close(self, **kw):
+        self.__rowbuffer.clear()
+        super(BufferedRowResultProxy, self)._soft_close(**kw)
 
     def _fetchone_impl(self):
-        if self.closed:
-            return None
+        if self.cursor is None:
+            return self._non_result(None)
         if not self.__rowbuffer:
             self.__buffer_rows()
             if not self.__rowbuffer:
@@ -950,6 +1299,8 @@ class BufferedRowResultProxy(ResultProxy):
         return result
 
     def _fetchall_impl(self):
+        if self.cursor is None:
+            return self._non_result([])
         self.__rowbuffer.extend(self.cursor.fetchall())
         ret = self.__rowbuffer
         self.__rowbuffer = collections.deque()
@@ -972,11 +1323,15 @@ class FullyBufferedResultProxy(ResultProxy):
     def _buffer_rows(self):
         return collections.deque(self.cursor.fetchall())
 
+    def _soft_close(self, **kw):
+        self.__rowbuffer.clear()
+        super(FullyBufferedResultProxy, self)._soft_close(**kw)
+
     def _fetchone_impl(self):
         if self.__rowbuffer:
             return self.__rowbuffer.popleft()
         else:
-            return None
+            return self._non_result(None)
 
     def _fetchmany_impl(self, size=None):
         if size is None:
@@ -990,6 +1345,8 @@ class FullyBufferedResultProxy(ResultProxy):
         return result
 
     def _fetchall_impl(self):
+        if not self.cursor:
+            return self._non_result([])
         ret = self.__rowbuffer
         self.__rowbuffer = collections.deque()
         return ret
diff --git a/lib/sqlalchemy/engine/strategies.py b/lib/sqlalchemy/engine/strategies.py
index fd665ad03..82800a918 100644
--- a/lib/sqlalchemy/engine/strategies.py
+++ b/lib/sqlalchemy/engine/strategies.py
@@ -1,5 +1,5 @@
 # engine/strategies.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -18,8 +18,9 @@ New strategies can be added via new ``EngineStrategy`` classes.
 from operator import attrgetter
 
 from sqlalchemy.engine import base, threadlocal, url
-from sqlalchemy import util, exc, event
+from sqlalchemy import util, event
 from sqlalchemy import pool as poollib
+from sqlalchemy.sql import schema
 
 strategies = {}
 
@@ -48,7 +49,12 @@ class DefaultEngineStrategy(EngineStrategy):
         # create url.URL object
         u = url.make_url(name_or_url)
 
-        dialect_cls = u.get_dialect()
+        plugins = u._instantiate_plugins(kwargs)
+
+        u.query.pop('plugin', None)
+
+        entrypoint = u._get_entrypoint()
+        dialect_cls = entrypoint.get_dialect_cls(u)
 
         if kwargs.pop('_coerce_config', False):
             def pop_kwarg(key, default=None):
@@ -81,11 +87,18 @@ class DefaultEngineStrategy(EngineStrategy):
         # assemble connection arguments
         (cargs, cparams) = dialect.create_connect_args(u)
         cparams.update(pop_kwarg('connect_args', {}))
+        cargs = list(cargs)  # allow mutability
 
         # look for existing pool or create
         pool = pop_kwarg('pool', None)
         if pool is None:
-            def connect():
+            def connect(connection_record=None):
+                if dialect._has_events:
+                    for fn in dialect.dispatch.do_connect:
+                        connection = fn(
+                            dialect, connection_record, cargs, cparams)
+                        if connection is not None:
+                            return connection
                 return dialect.connect(*cargs, **cparams)
 
             creator = pop_kwarg('creator', connect)
@@ -157,6 +170,13 @@ class DefaultEngineStrategy(EngineStrategy):
                 dialect.initialize(c)
             event.listen(pool, 'first_connect', first_connect, once=True)
 
+        dialect_cls.engine_created(engine)
+        if entrypoint is not dialect_cls:
+            entrypoint.engine_created(engine)
+
+        for plugin in plugins:
+            plugin.engine_created(engine)
+
         return engine
 
 
@@ -214,6 +234,8 @@ class MockEngineStrategy(EngineStrategy):
         dialect = property(attrgetter('_dialect'))
         name = property(lambda s: s._dialect.name)
 
+        schema_for_object = schema._schema_getter(None)
+
         def contextual_connect(self, **kwargs):
             return self
 
diff --git a/lib/sqlalchemy/engine/threadlocal.py b/lib/sqlalchemy/engine/threadlocal.py
index e64ab09f4..505d1fadd 100644
--- a/lib/sqlalchemy/engine/threadlocal.py
+++ b/lib/sqlalchemy/engine/threadlocal.py
@@ -1,5 +1,5 @@
 # engine/threadlocal.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
diff --git a/lib/sqlalchemy/engine/url.py b/lib/sqlalchemy/engine/url.py
index 6544cfbf3..cdb3bb7bb 100644
--- a/lib/sqlalchemy/engine/url.py
+++ b/lib/sqlalchemy/engine/url.py
@@ -1,5 +1,5 @@
 # engine/url.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under
@@ -17,7 +17,7 @@ be used directly and is also accepted directly by ``create_engine()``.
 import re
 from .. import exc, util
 from . import Dialect
-from ..dialects import registry
+from ..dialects import registry, plugins
 
 
 class URL(object):
@@ -117,11 +117,21 @@ class URL(object):
         else:
             return self.drivername.split('+')[1]
 
-    def get_dialect(self):
-        """Return the SQLAlchemy database dialect class corresponding
-        to this URL's driver name.
-        """
+    def _instantiate_plugins(self, kwargs):
+        plugin_names = util.to_list(self.query.get('plugin', ()))
+
+        return [
+            plugins.load(plugin_name)(self, kwargs)
+            for plugin_name in plugin_names
+        ]
+
+    def _get_entrypoint(self):
+        """Return the "entry point" dialect class.
+
+        This is normally the dialect itself except in the case when the
+        returned class implements the get_dialect_cls() method.
 
+        """
         if '+' not in self.drivername:
             name = self.drivername
         else:
@@ -137,6 +147,14 @@ class URL(object):
         else:
             return cls
 
+    def get_dialect(self):
+        """Return the SQLAlchemy database dialect class corresponding
+        to this URL's driver name.
+        """
+        entrypoint = self._get_entrypoint()
+        dialect_cls = entrypoint.get_dialect_cls(self)
+        return dialect_cls
+
     def translate_connect_args(self, names=[], **kw):
         """Translate url attributes into a dictionary of connection arguments.
 
diff --git a/lib/sqlalchemy/engine/util.py b/lib/sqlalchemy/engine/util.py
index d9eb1df10..d28d87098 100644
--- a/lib/sqlalchemy/engine/util.py
+++ b/lib/sqlalchemy/engine/util.py
@@ -1,5 +1,5 @@
 # engine/util.py
-# Copyright (C) 2005-2014 the SQLAlchemy authors and contributors
+# Copyright (C) 2005-2016 the SQLAlchemy authors and contributors
 # <see AUTHORS file>
 #
 # This module is part of SQLAlchemy and is released under