Merge branch 'master' into ticket_3100

37 files changed, 887 insertions, 209 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index dad02ee0f..5d84975c0 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -226,6 +226,53 @@ The DATE and TIME types are not available for MSSQL 2005 and
 previous - if a server version below 2008 is detected, DDL
 for these types will be issued as DATETIME.
 
+.. _mssql_large_type_deprecation:
+
+Large Text/Binary Type Deprecation
+----------------------------------
+
+Per `SQL Server 2012/2014 Documentation <http://technet.microsoft.com/en-us/library/ms187993.aspx>`_,
+the ``NTEXT``, ``TEXT`` and ``IMAGE`` datatypes are to be removed from SQL Server
+in a future release.   SQLAlchemy normally relates these types to the
+:class:`.UnicodeText`, :class:`.Text` and :class:`.LargeBinary` datatypes.
+
+In order to accommodate this change, a new flag ``deprecate_large_types``
+is added to the dialect, which will be automatically set based on detection
+of the server version in use, if not otherwise set by the user.  The
+behavior of this flag is as follows:
+
+* When this flag is ``True``, the :class:`.UnicodeText`, :class:`.Text` and
+  :class:`.LargeBinary` datatypes, when used to render DDL, will render the
+  types ``NVARCHAR(max)``, ``VARCHAR(max)``, and ``VARBINARY(max)``,
+  respectively.  This is a new behavior as of the addition of this flag.
+
+* When this flag is ``False``, the :class:`.UnicodeText`, :class:`.Text` and
+  :class:`.LargeBinary` datatypes, when used to render DDL, will render the
+  types ``NTEXT``, ``TEXT``, and ``IMAGE``,
+  respectively.  This is the long-standing behavior of these types.
+
+* The flag begins with the value ``None``, before a database connection is
+  established.   If the dialect is used to render DDL without the flag being
+  set, it is interpreted the same as ``False``.
+
+* On first connection, the dialect detects if SQL Server version 2012 or greater
+  is in use; if the flag is still at ``None``, it sets it to ``True`` or
+  ``False`` based on whether 2012 or greater is detected.
+
+* The flag can be set to either ``True`` or ``False`` when the dialect
+  is created, typically via :func:`.create_engine`::
+
+        eng = create_engine("mssql+pymssql://user:pass@host/db",
+                        deprecate_large_types=True)
+
+* Complete control over whether the "old" or "new" types are rendered is
+  available in all SQLAlchemy versions by using the UPPERCASE type objects
+  instead: :class:`.NVARCHAR`, :class:`.VARCHAR`, :class:`.types.VARBINARY`,
+  :class:`.TEXT`, :class:`.mssql.NTEXT`, :class:`.mssql.IMAGE` will always remain
+  fixed and always output exactly that type.
+
+.. versionadded:: 1.0.0
+
 .. _mssql_indexes:
 
 Clustered Index Support
@@ -367,19 +414,20 @@ import operator
 import re
 
 from ... import sql, schema as sa_schema, exc, util
-from ...sql import compiler, expression, \
-    util as sql_util, cast
+from ...sql import compiler, expression, util as sql_util
 from ... import engine
 from ...engine import reflection, default
 from ... import types as sqltypes
 from ...types import INTEGER, BIGINT, SMALLINT, DECIMAL, NUMERIC, \
     FLOAT, TIMESTAMP, DATETIME, DATE, BINARY,\
-    VARBINARY, TEXT, VARCHAR, NVARCHAR, CHAR, NCHAR
+    TEXT, VARCHAR, NVARCHAR, CHAR, NCHAR
 
 
 from ...util import update_wrapper
 from . import information_schema as ischema
 
+# http://sqlserverbuilds.blogspot.com/
+MS_2012_VERSION = (11,)
 MS_2008_VERSION = (10,)
 MS_2005_VERSION = (9,)
 MS_2000_VERSION = (8,)
@@ -545,6 +593,26 @@ class NTEXT(sqltypes.UnicodeText):
     __visit_name__ = 'NTEXT'
 
 
+class VARBINARY(sqltypes.VARBINARY, sqltypes.LargeBinary):
+    """The MSSQL VARBINARY type.
+
+    This type extends both :class:`.types.VARBINARY` and
+    :class:`.types.LargeBinary`.   In "deprecate_large_types" mode,
+    the :class:`.types.LargeBinary` type will produce ``VARBINARY(max)``
+    on SQL Server.
+
+    .. versionadded:: 1.0.0
+
+    .. seealso::
+
+        :ref:`mssql_large_type_deprecation`
+
+
+
+    """
+    __visit_name__ = 'VARBINARY'
+
+
 class IMAGE(sqltypes.LargeBinary):
     __visit_name__ = 'IMAGE'
 
@@ -683,8 +751,17 @@ class MSTypeCompiler(compiler.GenericTypeCompiler):
     def visit_unicode(self, type_):
         return self.visit_NVARCHAR(type_)
 
+    def visit_text(self, type_):
+        if self.dialect.deprecate_large_types:
+            return self.visit_VARCHAR(type_)
+        else:
+            return self.visit_TEXT(type_)
+
     def visit_unicode_text(self, type_):
-        return self.visit_NTEXT(type_)
+        if self.dialect.deprecate_large_types:
+            return self.visit_NVARCHAR(type_)
+        else:
+            return self.visit_NTEXT(type_)
 
     def visit_NTEXT(self, type_):
         return self._extend("NTEXT", type_)
@@ -717,7 +794,10 @@ class MSTypeCompiler(compiler.GenericTypeCompiler):
             return self.visit_TIME(type_)
 
     def visit_large_binary(self, type_):
-        return self.visit_IMAGE(type_)
+        if self.dialect.deprecate_large_types:
+            return self.visit_VARBINARY(type_)
+        else:
+            return self.visit_IMAGE(type_)
 
     def visit_IMAGE(self, type_):
         return "IMAGE"
@@ -1370,13 +1450,15 @@ class MSDialect(default.DefaultDialect):
                  query_timeout=None,
                  use_scope_identity=True,
                  max_identifier_length=None,
-                 schema_name="dbo", **opts):
+                 schema_name="dbo",
+                 deprecate_large_types=None, **opts):
         self.query_timeout = int(query_timeout or 0)
         self.schema_name = schema_name
 
         self.use_scope_identity = use_scope_identity
         self.max_identifier_length = int(max_identifier_length or 0) or \
             self.max_identifier_length
+        self.deprecate_large_types = deprecate_large_types
         super(MSDialect, self).__init__(**opts)
 
     def do_savepoint(self, connection, name):
@@ -1390,6 +1472,9 @@ class MSDialect(default.DefaultDialect):
 
     def initialize(self, connection):
         super(MSDialect, self).initialize(connection)
+        self._setup_version_attributes()
+
+    def _setup_version_attributes(self):
         if self.server_version_info[0] not in list(range(8, 17)):
             # FreeTDS with version 4.2 seems to report here
             # a number like "95.10.255".  Don't know what
@@ -1405,6 +1490,9 @@ class MSDialect(default.DefaultDialect):
             self.implicit_returning = True
         if self.server_version_info >= MS_2008_VERSION:
             self.supports_multivalues_insert = True
+        if self.deprecate_large_types is None:
+            self.deprecate_large_types = \
+                self.server_version_info >= MS_2012_VERSION
 
     def _get_default_schema_name(self, connection):
         if self.server_version_info < MS_2005_VERSION:
@@ -1592,12 +1680,11 @@ class MSDialect(default.DefaultDialect):
             if coltype in (MSString, MSChar, MSNVarchar, MSNChar, MSText,
                            MSNText, MSBinary, MSVarBinary,
                            sqltypes.LargeBinary):
+                if charlen == -1:
+                    charlen = 'max'
                 kwargs['length'] = charlen
                 if collation:
                     kwargs['collation'] = collation
-                if coltype == MSText or \
-                        (coltype in (MSString, MSNVarchar) and charlen == -1):
-                    kwargs.pop('length')
 
             if coltype is None:
                 util.warn(
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 2fb054d0c..c868f58b2 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -602,6 +602,14 @@ class _StringType(sqltypes.String):
                                  to_inspect=[_StringType, sqltypes.String])
 
 
+class _MatchType(sqltypes.Float, sqltypes.MatchType):
+    def __init__(self, **kw):
+        # TODO: float arguments?
+        sqltypes.Float.__init__(self)
+        sqltypes.MatchType.__init__(self)
+
+
+
 class NUMERIC(_NumericType, sqltypes.NUMERIC):
     """MySQL NUMERIC type."""
 
@@ -1544,6 +1552,7 @@ colspecs = {
     sqltypes.Float: FLOAT,
     sqltypes.Time: TIME,
     sqltypes.Enum: ENUM,
+    sqltypes.MatchType: _MatchType
 }
 
 # Everything 3.23 through 5.1 excepting OpenGIS types.
@@ -2593,7 +2602,7 @@ class MySQLDialect(default.DefaultDialect):
                 pass
             else:
                 self.logger.info(
-                    "Converting unknown KEY type %s to a plain KEY" % flavor)
+                    "Converting unknown KEY type %s to a plain KEY", flavor)
                 pass
             index_d = {}
             index_d['name'] = spec['name']
diff --git a/lib/sqlalchemy/dialects/mysql/mysqldb.py b/lib/sqlalchemy/dialects/mysql/mysqldb.py
index 73210d67a..893c6a9e2 100644
--- a/lib/sqlalchemy/dialects/mysql/mysqldb.py
+++ b/lib/sqlalchemy/dialects/mysql/mysqldb.py
@@ -77,7 +77,7 @@ class MySQLIdentifierPreparer_mysqldb(MySQLIdentifierPreparer):
 
 class MySQLDialect_mysqldb(MySQLDialect):
     driver = 'mysqldb'
-    supports_unicode_statements = False
+    supports_unicode_statements = True
     supports_sane_rowcount = True
     supports_sane_multi_rowcount = True
 
diff --git a/lib/sqlalchemy/dialects/mysql/pymysql.py b/lib/sqlalchemy/dialects/mysql/pymysql.py
index 31226cea0..8df2ba03f 100644
--- a/lib/sqlalchemy/dialects/mysql/pymysql.py
+++ b/lib/sqlalchemy/dialects/mysql/pymysql.py
@@ -31,8 +31,7 @@ class MySQLDialect_pymysql(MySQLDialect_mysqldb):
     driver = 'pymysql'
 
     description_encoding = None
-    if py3k:
-        supports_unicode_statements = True
+    supports_unicode_statements = True
 
     @classmethod
     def dbapi(cls):
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 6df38e57e..9f375da94 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -213,6 +213,8 @@ is reflected and the type is reported as ``DATE``, the time-supporting
    examining the type of column for use in special Python translations or
    for migrating schemas to other database backends.
 
+.. _oracle_table_options:
+
 Oracle Table Options
 -------------------------
 
@@ -228,15 +230,63 @@ in conjunction with the :class:`.Table` construct:
 
 .. versionadded:: 1.0.0
 
+* ``COMPRESS``::
+
+    Table('mytable', metadata, Column('data', String(32)),
+        oracle_compress=True)
+
+    Table('mytable', metadata, Column('data', String(32)),
+        oracle_compress=6)
+
+   The ``oracle_compress`` parameter accepts either an integer compression
+   level, or ``True`` to use the default compression level.
+
+.. versionadded:: 1.0.0
+
+.. _oracle_index_options:
+
+Oracle Specific Index Options
+-----------------------------
+
+Bitmap Indexes
+~~~~~~~~~~~~~~
+
+You can specify the ``oracle_bitmap`` parameter to create a bitmap index
+instead of a B-tree index::
+
+    Index('my_index', my_table.c.data, oracle_bitmap=True)
+
+Bitmap indexes cannot be unique and cannot be compressed. SQLAlchemy will not
+check for such limitations, only the database will.
+
+.. versionadded:: 1.0.0
+
+Index compression
+~~~~~~~~~~~~~~~~~
+
+Oracle has a more efficient storage mode for indexes containing lots of
+repeated values. Use the ``oracle_compress`` parameter to turn on key c
+ompression::
+
+    Index('my_index', my_table.c.data, oracle_compress=True)
+
+    Index('my_index', my_table.c.data1, my_table.c.data2, unique=True,
+           oracle_compress=1)
+
+The ``oracle_compress`` parameter accepts either an integer specifying the
+number of prefix columns to compress, or ``True`` to use the default (all
+columns for non-unique indexes, all but the last column for unique indexes).
+
+.. versionadded:: 1.0.0
+
 """
 
 import re
 
 from sqlalchemy import util, sql
-from sqlalchemy.engine import default, base, reflection
+from sqlalchemy.engine import default, reflection
 from sqlalchemy.sql import compiler, visitors, expression
-from sqlalchemy.sql import (operators as sql_operators,
-                            functions as sql_functions)
+from sqlalchemy.sql import operators as sql_operators
 from sqlalchemy import types as sqltypes, schema as sa_schema
 from sqlalchemy.types import VARCHAR, NVARCHAR, CHAR, \
     BLOB, CLOB, TIMESTAMP, FLOAT
@@ -549,6 +599,9 @@ class OracleCompiler(compiler.SQLCompiler):
     def visit_false(self, expr, **kw):
         return '0'
 
+    def get_cte_preamble(self, recursive):
+        return "WITH"
+
     def get_select_hint_text(self, byfroms):
         return " ".join(
             "/*+ %s */" % text for table, text in byfroms.items()
@@ -619,22 +672,10 @@ class OracleCompiler(compiler.SQLCompiler):
         return (self.dialect.identifier_preparer.format_sequence(seq) +
                 ".nextval")
 
-    def visit_alias(self, alias, asfrom=False, ashint=False, **kwargs):
-        """Oracle doesn't like ``FROM table AS alias``.  Is the AS standard
-        SQL??
-        """
-
-        if asfrom or ashint:
-            alias_name = isinstance(alias.name, expression._truncated_label) and \
-                self._truncated_identifier("alias", alias.name) or alias.name
+    def get_render_as_alias_suffix(self, alias_name_text):
+        """Oracle doesn't like ``FROM table AS alias``"""
 
-        if ashint:
-            return alias_name
-        elif asfrom:
-            return self.process(alias.original, asfrom=asfrom, **kwargs) + \
-                " " + self.preparer.format_alias(alias, alias_name)
-        else:
-            return self.process(alias.original, **kwargs)
+        return " " + alias_name_text
 
     def returning_clause(self, stmt, returning_cols):
         columns = []
@@ -795,9 +836,32 @@ class OracleDDLCompiler(compiler.DDLCompiler):
 
         return text
 
-    def visit_create_index(self, create, **kw):
-        return super(OracleDDLCompiler, self).\
-            visit_create_index(create, include_schema=True)
+    def visit_create_index(self, create):
+        index = create.element
+        self._verify_index_table(index)
+        preparer = self.preparer
+        text = "CREATE "
+        if index.unique:
+            text += "UNIQUE "
+        if index.dialect_options['oracle']['bitmap']:
+            text += "BITMAP "
+        text += "INDEX %s ON %s (%s)" % (
+            self._prepared_index_name(index, include_schema=True),
+            preparer.format_table(index.table, use_schema=True),
+            ', '.join(
+                self.sql_compiler.process(
+                    expr,
+                    include_table=False, literal_binds=True)
+                for expr in index.expressions)
+        )
+        if index.dialect_options['oracle']['compress'] is not False:
+            if index.dialect_options['oracle']['compress'] is True:
+                text += " COMPRESS"
+            else:
+                text += " COMPRESS %d" % (
+                    index.dialect_options['oracle']['compress']
+                )
+        return text
 
     def post_create_table(self, table):
         table_opts = []
@@ -807,6 +871,14 @@ class OracleDDLCompiler(compiler.DDLCompiler):
             on_commit_options = opts['on_commit'].replace("_", " ").upper()
             table_opts.append('\n ON COMMIT %s' % on_commit_options)
 
+        if opts['compress']:
+            if opts['compress'] is True:
+                table_opts.append("\n COMPRESS")
+            else:
+                table_opts.append("\n COMPRESS FOR %s" % (
+                    opts['compress']
+                ))
+
         return ''.join(table_opts)
 
 
@@ -870,7 +942,12 @@ class OracleDialect(default.DefaultDialect):
     construct_arguments = [
         (sa_schema.Table, {
             "resolve_synonyms": False,
-            "on_commit": None
+            "on_commit": None,
+            "compress": False
+        }),
+        (sa_schema.Index, {
+            "bitmap": False,
+            "compress": False
         })
     ]
 
@@ -902,6 +979,16 @@ class OracleDialect(default.DefaultDialect):
             self.server_version_info < (9, )
 
     @property
+    def _supports_table_compression(self):
+        return self.server_version_info and \
+            self.server_version_info >= (9, 2, )
+
+    @property
+    def _supports_table_compress_for(self):
+        return self.server_version_info and \
+            self.server_version_info >= (11, )
+
+    @property
     def _supports_char_length(self):
         return not self._is_oracle_8
 
@@ -1084,6 +1171,50 @@ class OracleDialect(default.DefaultDialect):
         return [self.normalize_name(row[0]) for row in cursor]
 
     @reflection.cache
+    def get_table_options(self, connection, table_name, schema=None, **kw):
+        options = {}
+
+        resolve_synonyms = kw.get('oracle_resolve_synonyms', False)
+        dblink = kw.get('dblink', '')
+        info_cache = kw.get('info_cache')
+
+        (table_name, schema, dblink, synonym) = \
+            self._prepare_reflection_args(connection, table_name, schema,
+                                          resolve_synonyms, dblink,
+                                          info_cache=info_cache)
+
+        params = {"table_name": table_name}
+
+        columns = ["table_name"]
+        if self._supports_table_compression:
+            columns.append("compression")
+        if self._supports_table_compress_for:
+            columns.append("compress_for")
+
+        text = "SELECT %(columns)s "\
+            "FROM ALL_TABLES%(dblink)s "\
+            "WHERE table_name = :table_name"
+
+        if schema is not None:
+            params['owner'] = schema
+            text += " AND owner = :owner "
+        text = text % {'dblink': dblink, 'columns': ", ".join(columns)}
+
+        result = connection.execute(sql.text(text), **params)
+
+        enabled = dict(DISABLED=False, ENABLED=True)
+
+        row = result.first()
+        if row:
+            if "compression" in row and enabled.get(row.compression, False):
+                if "compress_for" in row:
+                    options['oracle_compress'] = row.compress_for
+                else:
+                    options['oracle_compress'] = True
+
+        return options
+
+    @reflection.cache
     def get_columns(self, connection, table_name, schema=None, **kw):
         """
 
@@ -1168,7 +1299,8 @@ class OracleDialect(default.DefaultDialect):
 
         params = {'table_name': table_name}
         text = \
-            "SELECT a.index_name, a.column_name, b.uniqueness "\
+            "SELECT a.index_name, a.column_name, "\
+            "\nb.index_type, b.uniqueness, b.compression, b.prefix_length "\
             "\nFROM ALL_IND_COLUMNS%(dblink)s a, "\
             "\nALL_INDEXES%(dblink)s b "\
             "\nWHERE "\
@@ -1194,6 +1326,7 @@ class OracleDialect(default.DefaultDialect):
             dblink=dblink, info_cache=kw.get('info_cache'))
         pkeys = pk_constraint['constrained_columns']
         uniqueness = dict(NONUNIQUE=False, UNIQUE=True)
+        enabled = dict(DISABLED=False, ENABLED=True)
 
         oracle_sys_col = re.compile(r'SYS_NC\d+\$', re.IGNORECASE)
 
@@ -1213,10 +1346,15 @@ class OracleDialect(default.DefaultDialect):
             if rset.index_name != last_index_name:
                 remove_if_primary_key(index)
                 index = dict(name=self.normalize_name(rset.index_name),
-                             column_names=[])
+                             column_names=[], dialect_options={})
                 indexes.append(index)
             index['unique'] = uniqueness.get(rset.uniqueness, False)
 
+            if rset.index_type in ('BITMAP', 'FUNCTION-BASED BITMAP'):
+                index['dialect_options']['oracle_bitmap'] = True
+            if enabled.get(rset.compression, False):
+                index['dialect_options']['oracle_compress'] = rset.prefix_length
+
             # filter out Oracle SYS_NC names.  could also do an outer join
             # to the all_tab_columns table and check for real col names there.
             if not oracle_sys_col.match(rset.column_name):
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index baa640eaa..034ee9076 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1942,7 +1942,8 @@ class PGDialect(default.DefaultDialect):
             cursor = connection.execute(
                 sql.text(
                     "select relname from pg_class c join pg_namespace n on "
-                    "n.oid=c.relnamespace where n.nspname=current_schema() "
+                    "n.oid=c.relnamespace where "
+                    "pg_catalog.pg_table_is_visible(c.oid) "
                     "and relname=:name",
                     bindparams=[
                         sql.bindparam('name', util.text_type(table_name),
diff --git a/lib/sqlalchemy/dialects/postgresql/psycopg2.py b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
index 1a2a1ffe4..f67b2e3b0 100644
--- a/lib/sqlalchemy/dialects/postgresql/psycopg2.py
+++ b/lib/sqlalchemy/dialects/postgresql/psycopg2.py
@@ -159,6 +159,55 @@ defaults to ``utf-8``.
 SQLAlchemy's own unicode encode/decode functionality is steadily becoming
 obsolete as most DBAPIs now support unicode fully.
 
+Bound Parameter Styles
+----------------------
+
+The default parameter style for the psycopg2 dialect is "pyformat", where
+SQL is rendered using ``%(paramname)s`` style.   This format has the limitation
+that it does not accommodate the unusual case of parameter names that
+actually contain percent or parenthesis symbols; as SQLAlchemy in many cases
+generates bound parameter names based on the name of a column, the presence
+of these characters in a column name can lead to problems.
+
+There are two solutions to the issue of a :class:`.schema.Column` that contains
+one of these characters in its name.  One is to specify the
+:paramref:`.schema.Column.key` for columns that have such names::
+
+    measurement = Table('measurement', metadata,
+        Column('Size (meters)', Integer, key='size_meters')
+    )
+
+Above, an INSERT statement such as ``measurement.insert()`` will use
+``size_meters`` as the parameter name, and a SQL expression such as
+``measurement.c.size_meters > 10`` will derive the bound parameter name
+from the ``size_meters`` key as well.
+
+.. versionchanged:: 1.0.0 - SQL expressions will use :attr:`.Column.key`
+   as the source of naming when anonymous bound parameters are created
+   in SQL expressions; previously, this behavior only applied to
+   :meth:`.Table.insert` and :meth:`.Table.update` parameter names.
+
+The other solution is to use a positional format; psycopg2 allows use of the
+"format" paramstyle, which can be passed to
+:paramref:`.create_engine.paramstyle`::
+
+    engine = create_engine(
+        'postgresql://scott:tiger@localhost:5432/test', paramstyle='format')
+
+With the above engine, instead of a statement like::
+
+    INSERT INTO measurement ("Size (meters)") VALUES (%(Size (meters))s)
+    {'Size (meters)': 1}
+
+we instead see::
+
+    INSERT INTO measurement ("Size (meters)") VALUES (%s)
+    (1, )
+
+Where above, the dictionary style is converted into a tuple with positional
+style.
+
+
 Transactions
 ------------
 
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 335b35c94..ccd7f2539 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -9,6 +9,7 @@
 .. dialect:: sqlite
     :name: SQLite
 
+.. _sqlite_datetime:
 
 Date and Time Types
 -------------------
@@ -23,6 +24,20 @@ These types represent dates and times as ISO formatted strings, which also
 nicely support ordering. There's no reliance on typical "libc" internals for
 these functions so historical dates are fully supported.
 
+Ensuring Text affinity
+^^^^^^^^^^^^^^^^^^^^^^
+
+The DDL rendered for these types is the standard ``DATE``, ``TIME``
+and ``DATETIME`` indicators.    However, custom storage formats can also be
+applied to these types.   When the
+storage format is detected as containing no alpha characters, the DDL for
+these types is rendered as ``DATE_CHAR``, ``TIME_CHAR``, and ``DATETIME_CHAR``,
+so that the column continues to have textual affinity.
+
+.. seealso::
+
+    `Type Affinity <http://www.sqlite.org/datatype3.html#affinity>`_ - in the SQLite documentation
+
 .. _sqlite_autoincrement:
 
 SQLite Auto Incrementing Behavior
@@ -255,7 +270,7 @@ from ... import util
 from ...engine import default, reflection
 from ...sql import compiler
 
-from ...types import (BLOB, BOOLEAN, CHAR, DATE, DECIMAL, FLOAT,
+from ...types import (BLOB, BOOLEAN, CHAR, DECIMAL, FLOAT,
                       INTEGER, REAL, NUMERIC, SMALLINT, TEXT,
                       TIMESTAMP, VARCHAR)
 
@@ -271,6 +286,25 @@ class _DateTimeMixin(object):
         if storage_format is not None:
             self._storage_format = storage_format
 
+    @property
+    def format_is_text_affinity(self):
+        """return True if the storage format will automatically imply
+        a TEXT affinity.
+
+        If the storage format contains no non-numeric characters,
+        it will imply a NUMERIC storage format on SQLite; in this case,
+        the type will generate its DDL as DATE_CHAR, DATETIME_CHAR,
+        TIME_CHAR.
+
+        .. versionadded:: 1.0.0
+
+        """
+        spec = self._storage_format % {
+            "year": 0, "month": 0, "day": 0, "hour": 0,
+            "minute": 0, "second": 0, "microsecond": 0
+        }
+        return bool(re.search(r'[^0-9]', spec))
+
     def adapt(self, cls, **kw):
         if issubclass(cls, _DateTimeMixin):
             if self._storage_format:
@@ -526,7 +560,9 @@ ischema_names = {
     'BOOLEAN': sqltypes.BOOLEAN,
     'CHAR': sqltypes.CHAR,
     'DATE': sqltypes.DATE,
+    'DATE_CHAR': sqltypes.DATE,
     'DATETIME': sqltypes.DATETIME,
+    'DATETIME_CHAR': sqltypes.DATETIME,
     'DOUBLE': sqltypes.FLOAT,
     'DECIMAL': sqltypes.DECIMAL,
     'FLOAT': sqltypes.FLOAT,
@@ -537,6 +573,7 @@ ischema_names = {
     'SMALLINT': sqltypes.SMALLINT,
     'TEXT': sqltypes.TEXT,
     'TIME': sqltypes.TIME,
+    'TIME_CHAR': sqltypes.TIME,
     'TIMESTAMP': sqltypes.TIMESTAMP,
     'VARCHAR': sqltypes.VARCHAR,
     'NVARCHAR': sqltypes.NVARCHAR,
@@ -646,8 +683,8 @@ class SQLiteDDLCompiler(compiler.DDLCompiler):
 
     def visit_foreign_key_constraint(self, constraint):
 
-        local_table = list(constraint._elements.values())[0].parent.table
-        remote_table = list(constraint._elements.values())[0].column.table
+        local_table = constraint.elements[0].parent.table
+        remote_table = constraint.elements[0].column.table
 
         if local_table.schema != remote_table.schema:
             return None
@@ -670,6 +707,27 @@ class SQLiteTypeCompiler(compiler.GenericTypeCompiler):
     def visit_large_binary(self, type_):
         return self.visit_BLOB(type_)
 
+    def visit_DATETIME(self, type_):
+        if not isinstance(type_, _DateTimeMixin) or \
+                type_.format_is_text_affinity:
+            return super(SQLiteTypeCompiler, self).visit_DATETIME(type_)
+        else:
+            return "DATETIME_CHAR"
+
+    def visit_DATE(self, type_):
+        if not isinstance(type_, _DateTimeMixin) or \
+                type_.format_is_text_affinity:
+            return super(SQLiteTypeCompiler, self).visit_DATE(type_)
+        else:
+            return "DATE_CHAR"
+
+    def visit_TIME(self, type_):
+        if not isinstance(type_, _DateTimeMixin) or \
+                type_.format_is_text_affinity:
+            return super(SQLiteTypeCompiler, self).visit_TIME(type_)
+        else:
+            return "TIME_CHAR"
+
 
 class SQLiteIdentifierPreparer(compiler.IdentifierPreparer):
     reserved_words = set([
diff --git a/lib/sqlalchemy/engine/__init__.py b/lib/sqlalchemy/engine/__init__.py
index 68145f5cd..cf75871bf 100644
--- a/lib/sqlalchemy/engine/__init__.py
+++ b/lib/sqlalchemy/engine/__init__.py
@@ -292,6 +292,17 @@ def create_engine(*args, **kwargs):
         be used instead. Can be used for testing of DBAPIs as well as to
         inject "mock" DBAPI implementations into the :class:`.Engine`.
 
+    :param paramstyle=None: The `paramstyle <http://legacy.python.org/dev/peps/pep-0249/#paramstyle>`_
+        to use when rendering bound parameters.  This style defaults to the
+        one recommended by the DBAPI itself, which is retrieved from the
+        ``.paramstyle`` attribute of the DBAPI.  However, most DBAPIs accept
+        more than one paramstyle, and in particular it may be desirable
+        to change a "named" paramstyle into a "positional" one, or vice versa.
+        When this attribute is passed, it should be one of the values
+        ``"qmark"``, ``"numeric"``, ``"named"``, ``"format"`` or
+        ``"pyformat"``, and should correspond to a parameter style known
+        to be supported by the DBAPI in use.
+
     :param pool=None: an already-constructed instance of
         :class:`~sqlalchemy.pool.Pool`, such as a
         :class:`~sqlalchemy.pool.QueuePool` instance. If non-None, this
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index dd82be1d1..23348469d 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -265,18 +265,18 @@ class Connection(Connectable):
         try:
             return self.__connection
         except AttributeError:
-            return self._revalidate_connection()
+            return self._revalidate_connection(_wrap=True)
 
-    def _revalidate_connection(self):
+    def _revalidate_connection(self, _wrap):
         if self.__branch_from:
-            return self.__branch_from._revalidate_connection()
-
+            return self.__branch_from._revalidate_connection(_wrap=_wrap)
         if self.__can_reconnect and self.__invalid:
             if self.__transaction is not None:
                 raise exc.InvalidRequestError(
                     "Can't reconnect until invalid "
                     "transaction is rolled back")
-            self.__connection = self.engine.raw_connection()
+            self.__connection = self.engine.raw_connection(
+                _connection=self, _wrap=_wrap)
             self.__invalid = False
             return self.__connection
         raise exc.ResourceClosedError("This Connection is closed")
@@ -817,7 +817,7 @@ class Connection(Connectable):
             try:
                 conn = self.__connection
             except AttributeError:
-                conn = self._revalidate_connection()
+                conn = self._revalidate_connection(_wrap=False)
 
             dialect = self.dialect
             ctx = dialect.execution_ctx_cls._init_default(
@@ -955,13 +955,14 @@ class Connection(Connectable):
             try:
                 conn = self.__connection
             except AttributeError:
-                conn = self._revalidate_connection()
+                conn = self._revalidate_connection(_wrap=False)
 
             context = constructor(dialect, self, conn, *args)
         except Exception as e:
-            self._handle_dbapi_exception(e,
-                                         util.text_type(statement), parameters,
-                                         None, None)
+            self._handle_dbapi_exception(
+                e,
+                util.text_type(statement), parameters,
+                None, None)
 
         if context.compiled:
             context.pre_exec()
@@ -1149,7 +1150,10 @@ class Connection(Connectable):
             self._is_disconnect =  \
                 isinstance(e, self.dialect.dbapi.Error) and \
                 not self.closed and \
-                self.dialect.is_disconnect(e, self.__connection, cursor)
+                self.dialect.is_disconnect(
+                    e,
+                    self.__connection if not self.invalidated else None,
+                    cursor)
             if context:
                 context.is_disconnect = self._is_disconnect
 
@@ -1194,7 +1198,8 @@ class Connection(Connectable):
 
                 # new handle_error event
                 ctx = ExceptionContextImpl(
-                    e, sqlalchemy_exception, self, cursor, statement,
+                    e, sqlalchemy_exception, self.engine,
+                    self, cursor, statement,
                     parameters, context, self._is_disconnect)
 
                 for fn in self.dispatch.handle_error:
@@ -1242,6 +1247,59 @@ class Connection(Connectable):
             if self.should_close_with_result:
                 self.close()
 
+    @classmethod
+    def _handle_dbapi_exception_noconnection(
+            cls, e, dialect, engine, connection):
+
+        exc_info = sys.exc_info()
+
+        is_disconnect = dialect.is_disconnect(e, None, None)
+
+        should_wrap = isinstance(e, dialect.dbapi.Error)
+
+        if should_wrap:
+            sqlalchemy_exception = exc.DBAPIError.instance(
+                None,
+                None,
+                e,
+                dialect.dbapi.Error,
+                connection_invalidated=is_disconnect)
+        else:
+            sqlalchemy_exception = None
+
+        newraise = None
+
+        if engine._has_events:
+            ctx = ExceptionContextImpl(
+                e, sqlalchemy_exception, engine, connection, None, None,
+                None, None, is_disconnect)
+            for fn in engine.dispatch.handle_error:
+                try:
+                    # handler returns an exception;
+                    # call next handler in a chain
+                    per_fn = fn(ctx)
+                    if per_fn is not None:
+                        ctx.chained_exception = newraise = per_fn
+                except Exception as _raised:
+                    # handler raises an exception - stop processing
+                    newraise = _raised
+                    break
+
+            if sqlalchemy_exception and \
+                    is_disconnect != ctx.is_disconnect:
+                sqlalchemy_exception.connection_invalidated = \
+                    is_disconnect = ctx.is_disconnect
+
+        if newraise:
+            util.raise_from_cause(newraise, exc_info)
+        elif should_wrap:
+            util.raise_from_cause(
+                sqlalchemy_exception,
+                exc_info
+            )
+        else:
+            util.reraise(*exc_info)
+
     def default_schema_name(self):
         return self.engine.dialect.get_default_schema_name(self)
 
@@ -1320,8 +1378,9 @@ class ExceptionContextImpl(ExceptionContext):
     """Implement the :class:`.ExceptionContext` interface."""
 
     def __init__(self, exception, sqlalchemy_exception,
-                 connection, cursor, statement, parameters,
+                 engine, connection, cursor, statement, parameters,
                  context, is_disconnect):
+        self.engine = engine
         self.connection = connection
         self.sqlalchemy_exception = sqlalchemy_exception
         self.original_exception = exception
@@ -1898,7 +1957,17 @@ class Engine(Connectable, log.Identified):
         """
         return self.run_callable(self.dialect.has_table, table_name, schema)
 
-    def raw_connection(self):
+    def _wrap_pool_connect(self, fn, connection, wrap=True):
+        if not wrap:
+            return fn()
+        dialect = self.dialect
+        try:
+            return fn()
+        except dialect.dbapi.Error as e:
+            Connection._handle_dbapi_exception_noconnection(
+                e, dialect, self, connection)
+
+    def raw_connection(self, _connection=None, _wrap=True):
         """Return a "raw" DBAPI connection from the connection pool.
 
         The returned object is a proxied version of the DBAPI
@@ -1914,8 +1983,8 @@ class Engine(Connectable, log.Identified):
         :meth:`.Engine.connect` method.
 
         """
-
-        return self.pool.unique_connection()
+        return self._wrap_pool_connect(
+            self.pool.unique_connection, _connection, _wrap)
 
 
 class OptionEngine(Engine):
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index 0ad2efae0..5f66e54b5 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -917,7 +917,23 @@ class ExceptionContext(object):
     connection = None
     """The :class:`.Connection` in use during the exception.
 
-    This member is always present.
+    This member is present, except in the case of a failure when
+    first connecting.
+
+    .. seealso::
+
+        :attr:`.ExceptionContext.engine`
+
+
+    """
+
+    engine = None
+    """The :class:`.Engine` in use during the exception.
+
+    This member should always be present, even in the case of a failure
+    when first connecting.
+
+    .. versionadded:: 1.0.0
 
     """
 
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 2a1def86a..25f084c15 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -394,6 +394,12 @@ class Inspector(object):
         unique
           boolean
 
+        dialect_options
+          dict of dialect-specific index options.  May not be present
+          for all dialects.
+
+          .. versionadded:: 1.0.0
+
         :param table_name: string name of the table.  For special quoting,
          use :class:`.quoted_name`.
 
@@ -642,6 +648,8 @@ class Inspector(object):
             columns = index_d['column_names']
             unique = index_d['unique']
             flavor = index_d.get('type', 'index')
+            dialect_options = index_d.get('dialect_options', {})
+
             duplicates = index_d.get('duplicates_constraint')
             if include_columns and \
                     not set(columns).issubset(include_columns):
@@ -667,7 +675,10 @@ class Inspector(object):
                 else:
                     idx_cols.append(idx_col)
 
-            sa_schema.Index(name, *idx_cols, **dict(unique=unique))
+            sa_schema.Index(
+                name, *idx_cols,
+                **dict(list(dialect_options.items()) + [('unique', unique)])
+            )
 
     def _reflect_unique_constraints(
         self, table_name, schema, table, cols_by_orig_name,
diff --git a/lib/sqlalchemy/engine/strategies.py b/lib/sqlalchemy/engine/strategies.py
index 398ef8df6..fd665ad03 100644
--- a/lib/sqlalchemy/engine/strategies.py
+++ b/lib/sqlalchemy/engine/strategies.py
@@ -86,16 +86,7 @@ class DefaultEngineStrategy(EngineStrategy):
         pool = pop_kwarg('pool', None)
         if pool is None:
             def connect():
-                try:
-                    return dialect.connect(*cargs, **cparams)
-                except dialect.dbapi.Error as e:
-                    invalidated = dialect.is_disconnect(e, None, None)
-                    util.raise_from_cause(
-                        exc.DBAPIError.instance(
-                            None, None, e, dialect.dbapi.Error,
-                            connection_invalidated=invalidated
-                        )
-                    )
+                return dialect.connect(*cargs, **cparams)
 
             creator = pop_kwarg('creator', connect)
 
diff --git a/lib/sqlalchemy/engine/threadlocal.py b/lib/sqlalchemy/engine/threadlocal.py
index 637523a0e..824b68fdf 100644
--- a/lib/sqlalchemy/engine/threadlocal.py
+++ b/lib/sqlalchemy/engine/threadlocal.py
@@ -59,7 +59,10 @@ class TLEngine(base.Engine):
             # guards against pool-level reapers, if desired.
             # or not connection.connection.is_valid:
             connection = self._tl_connection_cls(
-                self, self.pool.connect(), **kw)
+                self,
+                self._wrap_pool_connect(
+                    self.pool.connect, connection, wrap=True),
+                **kw)
             self._connections.conn = weakref.ref(connection)
 
         return connection._increment_connect()
diff --git a/lib/sqlalchemy/events.py b/lib/sqlalchemy/events.py
index b4f057b0a..8600c20f5 100644
--- a/lib/sqlalchemy/events.py
+++ b/lib/sqlalchemy/events.py
@@ -420,6 +420,12 @@ class ConnectionEvents(event.Events):
                                             context, executemany):
                 log.info("Received statement: %s" % statement)
 
+    When the methods are called with a `statement` parameter, such as in
+    :meth:`.after_cursor_execute`, :meth:`.before_cursor_execute` and
+    :meth:`.dbapi_error`, the statement is the exact SQL string that was
+    prepared for transmission to the DBAPI ``cursor`` in the connection's
+    :class:`.Dialect`.
+
     The :meth:`.before_execute` and :meth:`.before_cursor_execute`
     events can also be established with the ``retval=True`` flag, which
     allows modification of the statement and parameters to be sent
@@ -549,9 +555,8 @@ class ConnectionEvents(event.Events):
     def before_cursor_execute(self, conn, cursor, statement,
                               parameters, context, executemany):
         """Intercept low-level cursor execute() events before execution,
-        receiving the string
-        SQL statement and DBAPI-specific parameter list to be invoked
-        against a cursor.
+        receiving the string SQL statement and DBAPI-specific parameter list to
+        be invoked against a cursor.
 
         This event is a good choice for logging as well as late modifications
         to the SQL string.  It's less ideal for parameter modifications except
@@ -571,7 +576,7 @@ class ConnectionEvents(event.Events):
 
         :param conn: :class:`.Connection` object
         :param cursor: DBAPI cursor object
-        :param statement: string SQL statement
+        :param statement: string SQL statement, as to be passed to the DBAPI
         :param parameters: Dictionary, tuple, or list of parameters being
          passed to the ``execute()`` or ``executemany()`` method of the
          DBAPI ``cursor``.  In some cases may be ``None``.
@@ -596,7 +601,7 @@ class ConnectionEvents(event.Events):
         :param cursor: DBAPI cursor object.  Will have results pending
          if the statement was a SELECT, but these should not be consumed
          as they will be needed by the :class:`.ResultProxy`.
-        :param statement: string SQL statement
+        :param statement: string SQL statement, as passed to the DBAPI
         :param parameters: Dictionary, tuple, or list of parameters being
          passed to the ``execute()`` or ``executemany()`` method of the
          DBAPI ``cursor``.  In some cases may be ``None``.
@@ -640,7 +645,7 @@ class ConnectionEvents(event.Events):
 
         :param conn: :class:`.Connection` object
         :param cursor: DBAPI cursor object
-        :param statement: string SQL statement
+        :param statement: string SQL statement, as passed to the DBAPI
         :param parameters: Dictionary, tuple, or list of parameters being
          passed to the ``execute()`` or ``executemany()`` method of the
          DBAPI ``cursor``.  In some cases may be ``None``.
@@ -734,6 +739,12 @@ class ConnectionEvents(event.Events):
         .. versionadded:: 0.9.7 Added the
             :meth:`.ConnectionEvents.handle_error` hook.
 
+        .. versionchanged:: 1.0.0 The :meth:`.handle_error` event is now
+           invoked when an :class:`.Engine` fails during the initial
+           call to :meth:`.Engine.connect`, as well as when a
+           :class:`.Connection` object encounters an error during a
+           reconnect operation.
+
         .. versionchanged:: 1.0.0 The :meth:`.handle_error` event is
            not fired off when a dialect makes use of the
            ``skip_user_error_events`` execution option.   This is used
diff --git a/lib/sqlalchemy/orm/mapper.py b/lib/sqlalchemy/orm/mapper.py
index 082dae054..c61d93230 100644
--- a/lib/sqlalchemy/orm/mapper.py
+++ b/lib/sqlalchemy/orm/mapper.py
@@ -1581,6 +1581,8 @@ class Mapper(InspectionAttr):
                           self,
                           prop,
                       ))
+            oldprop = self._props[key]
+            self._path_registry.pop(oldprop, None)
 
         self._props[key] = prop
 
diff --git a/lib/sqlalchemy/orm/path_registry.py b/lib/sqlalchemy/orm/path_registry.py
index f10a125a8..d4dbf29a0 100644
--- a/lib/sqlalchemy/orm/path_registry.py
+++ b/lib/sqlalchemy/orm/path_registry.py
@@ -13,6 +13,9 @@ from .. import util
 from .. import exc
 from itertools import chain
 from .base import class_mapper
+import logging
+
+log = logging.getLogger(__name__)
 
 
 def _unreduce_path(path):
@@ -54,9 +57,11 @@ class PathRegistry(object):
             self.path == other.path
 
     def set(self, attributes, key, value):
+        log.debug("set '%s' on path '%s' to '%s'", key, self, value)
         attributes[(key, self.path)] = value
 
     def setdefault(self, attributes, key, value):
+        log.debug("setdefault '%s' on path '%s' to '%s'", key, self, value)
         attributes.setdefault((key, self.path), value)
 
     def get(self, attributes, key, value=None):
@@ -184,6 +189,11 @@ class PropRegistry(PathRegistry):
         self.parent = parent
         self.path = parent.path + (prop,)
 
+    def __str__(self):
+        return " -> ".join(
+            str(elem) for elem in self.path
+        )
+
     @util.memoized_property
     def has_entity(self):
         return hasattr(self.prop, "mapper")
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index f07060825..9b7747e15 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -75,6 +75,7 @@ class Query(object):
     _having = None
     _distinct = False
     _prefixes = None
+    _suffixes = None
     _offset = None
     _limit = None
     _for_update_arg = None
@@ -1003,7 +1004,7 @@ class Query(object):
                 '_limit', '_offset',
                 '_joinpath', '_joinpoint',
                 '_distinct', '_having',
-                '_prefixes',
+                '_prefixes', '_suffixes'
         ):
             self.__dict__.pop(attr, None)
         self._set_select_from([fromclause], True)
@@ -1740,6 +1741,14 @@ class Query(object):
          anonymously aliased.  Subsequent calls to :meth:`~.Query.filter`
          and similar will adapt the incoming criterion to the target
          alias, until :meth:`~.Query.reset_joinpoint` is called.
+        :param isouter=False: If True, the join used will be a left outer join,
+         just as if the :meth:`.Query.outerjoin` method were called.  This
+         flag is here to maintain consistency with the same flag as accepted
+         by :meth:`.FromClause.join` and other Core constructs.
+
+
+         .. versionadded:: 1.0.0
+
         :param from_joinpoint=False: When using ``aliased=True``, a setting
          of True here will cause the join to be from the most recent
          joined target, rather than starting back from the original
@@ -1757,13 +1766,15 @@ class Query(object):
             SQLAlchemy versions was the primary ORM-level joining interface.
 
         """
-        aliased, from_joinpoint = kwargs.pop('aliased', False),\
-            kwargs.pop('from_joinpoint', False)
+        aliased, from_joinpoint, isouter = kwargs.pop('aliased', False),\
+            kwargs.pop('from_joinpoint', False),\
+            kwargs.pop('isouter', False)
         if kwargs:
             raise TypeError("unknown arguments: %s" %
                             ','.join(kwargs.keys))
+        isouter = isouter
         return self._join(props,
-                          outerjoin=False, create_aliases=aliased,
+                          outerjoin=isouter, create_aliases=aliased,
                           from_joinpoint=from_joinpoint)
 
     def outerjoin(self, *props, **kwargs):
@@ -2349,12 +2360,38 @@ class Query(object):
 
         .. versionadded:: 0.7.7
 
+        .. seealso::
+
+            :meth:`.HasPrefixes.prefix_with`
+
         """
         if self._prefixes:
             self._prefixes += prefixes
         else:
             self._prefixes = prefixes
 
+    @_generative()
+    def suffix_with(self, *suffixes):
+        """Apply the suffix to the query and return the newly resulting
+        ``Query``.
+
+        :param \*suffixes: optional suffixes, typically strings,
+         not using any commas.
+
+        .. versionadded:: 1.0.0
+
+        .. seealso::
+
+            :meth:`.Query.prefix_with`
+
+            :meth:`.HasSuffixes.suffix_with`
+
+        """
+        if self._suffixes:
+            self._suffixes += suffixes
+        else:
+            self._suffixes = suffixes
+
     def all(self):
         """Return the results represented by this ``Query`` as a list.
 
@@ -2591,6 +2628,7 @@ class Query(object):
             'offset': self._offset,
             'distinct': self._distinct,
             'prefixes': self._prefixes,
+            'suffixes': self._suffixes,
             'group_by': self._group_by or None,
             'having': self._having
         }
@@ -3396,7 +3434,6 @@ class _BundleEntity(_QueryEntity):
 
         self.supports_single_entity = self.bundle.single_entity
 
-
     @property
     def entity_zero(self):
         for ent in self._entities:
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index cdb501c14..d95f17f64 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -373,7 +373,7 @@ class LazyLoader(AbstractRelationshipLoader):
                         self._equated_columns[c] = self._equated_columns[col]
 
             self.logger.info("%s will use query.get() to "
-                             "optimize instance loads" % self)
+                             "optimize instance loads", self)
 
     def init_class_attribute(self, mapper):
         self.is_class_level = True
diff --git a/lib/sqlalchemy/orm/strategy_options.py b/lib/sqlalchemy/orm/strategy_options.py
index 4f986193e..276da2ae0 100644
--- a/lib/sqlalchemy/orm/strategy_options.py
+++ b/lib/sqlalchemy/orm/strategy_options.py
@@ -161,11 +161,14 @@ class Load(Generative, MapperOption):
                 ext_info = inspect(ac)
 
                 path_element = ext_info.mapper
+                existing = path.entity_path[prop].get(
+                    self.context, "path_with_polymorphic")
                 if not ext_info.is_aliased_class:
                     ac = orm_util.with_polymorphic(
                         ext_info.mapper.base_mapper,
                         ext_info.mapper, aliased=True,
-                        _use_mapper_path=True)
+                        _use_mapper_path=True,
+                        _existing_alias=existing)
                 path.entity_path[prop].set(
                     self.context, "path_with_polymorphic", inspect(ac))
                 path = path[prop][path_element]
@@ -176,6 +179,9 @@ class Load(Generative, MapperOption):
             path = path.entity_path
         return path
 
+    def __str__(self):
+        return "Load(strategy=%r)" % self.strategy
+
     def _coerce_strat(self, strategy):
         if strategy is not None:
             strategy = tuple(sorted(strategy.items()))
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index ad610a4ac..4be8d19ff 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -543,8 +543,13 @@ class AliasedInsp(InspectionAttr):
                 mapper, self)
 
     def __repr__(self):
-        return '<AliasedInsp at 0x%x; %s>' % (
-            id(self), self.class_.__name__)
+        if self.with_polymorphic_mappers:
+            with_poly = "(%s)" % ", ".join(
+                mp.class_.__name__ for mp in self.with_polymorphic_mappers)
+        else:
+            with_poly = ""
+        return '<AliasedInsp at 0x%x; %s%s>' % (
+            id(self), self.class_.__name__, with_poly)
 
 
 inspection._inspects(AliasedClass)(lambda target: target._aliased_insp)
@@ -648,7 +653,8 @@ def aliased(element, alias=None, name=None, flat=False, adapt_on_names=False):
 def with_polymorphic(base, classes, selectable=False,
                      flat=False,
                      polymorphic_on=None, aliased=False,
-                     innerjoin=False, _use_mapper_path=False):
+                     innerjoin=False, _use_mapper_path=False,
+                     _existing_alias=None):
     """Produce an :class:`.AliasedClass` construct which specifies
     columns for descendant mappers of the given base.
 
@@ -713,6 +719,16 @@ def with_polymorphic(base, classes, selectable=False,
        only be specified if querying for one specific subtype only
     """
     primary_mapper = _class_to_mapper(base)
+    if _existing_alias:
+        assert _existing_alias.mapper is primary_mapper
+        classes = util.to_set(classes)
+        new_classes = set([
+            mp.class_ for mp in
+            _existing_alias.with_polymorphic_mappers])
+        if classes == new_classes:
+            return _existing_alias
+        else:
+            classes = classes.union(new_classes)
     mappers, selectable = primary_mapper.\
         _with_polymorphic_args(classes, selectable,
                                innerjoin=innerjoin)
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 5fa78ad0f..9304bba9f 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -82,6 +82,7 @@ OPERATORS = {
     operators.eq: ' = ',
     operators.concat_op: ' || ',
     operators.match_op: ' MATCH ',
+    operators.notmatch_op: ' NOT MATCH ',
     operators.in_op: ' IN ',
     operators.notin_op: ' NOT IN ',
     operators.comma_op: ', ',
@@ -862,14 +863,18 @@ class SQLCompiler(Compiled):
         else:
             return "%s = 0" % self.process(element.element, **kw)
 
-    def visit_binary(self, binary, **kw):
+    def visit_notmatch_op_binary(self, binary, operator, **kw):
+        return "NOT %s" % self.visit_binary(
+            binary, override_operator=operators.match_op)
+
+    def visit_binary(self, binary, override_operator=None, **kw):
         # don't allow "? = ?" to render
         if self.ansi_bind_rules and \
                 isinstance(binary.left, elements.BindParameter) and \
                 isinstance(binary.right, elements.BindParameter):
             kw['literal_binds'] = True
 
-        operator_ = binary.operator
+        operator_ = override_operator or binary.operator
         disp = getattr(self, "visit_%s_binary" % operator_.__name__, None)
         if disp:
             return disp(binary, operator_, **kw)
@@ -1188,12 +1193,16 @@ class SQLCompiler(Compiled):
                     self, asfrom=True, **kwargs
                 )
 
+            if cte._suffixes:
+                text += " " + self._generate_prefixes(
+                    cte, cte._suffixes, **kwargs)
+
             self.ctes[cte] = text
 
         if asfrom:
             if cte_alias_name:
                 text = self.preparer.format_alias(cte, cte_alias_name)
-                text += " AS " + cte_name
+                text += self.get_render_as_alias_suffix(cte_name)
             else:
                 return self.preparer.format_alias(cte, cte_name)
             return text
@@ -1212,8 +1221,8 @@ class SQLCompiler(Compiled):
         elif asfrom:
             ret = alias.original._compiler_dispatch(self,
                                                     asfrom=True, **kwargs) + \
-                " AS " + \
-                self.preparer.format_alias(alias, alias_name)
+                self.get_render_as_alias_suffix(
+                    self.preparer.format_alias(alias, alias_name))
 
             if fromhints and alias in fromhints:
                 ret = self.format_from_hint_text(ret, alias,
@@ -1223,6 +1232,9 @@ class SQLCompiler(Compiled):
         else:
             return alias.original._compiler_dispatch(self, **kwargs)
 
+    def get_render_as_alias_suffix(self, alias_name_text):
+        return " AS " + alias_name_text
+
     def _add_to_result_map(self, keyname, name, objects, type_):
         if not self.dialect.case_sensitive:
             keyname = keyname.lower()
@@ -1549,6 +1561,10 @@ class SQLCompiler(Compiled):
                 compound_index == 0 and toplevel:
             text = self._render_cte_clause() + text
 
+        if select._suffixes:
+            text += " " + self._generate_prefixes(
+                select, select._suffixes, **kwargs)
+
         self.stack.pop(-1)
 
         if asfrom and parens:
@@ -1729,6 +1745,12 @@ class SQLCompiler(Compiled):
         )
 
     def visit_insert(self, insert_stmt, **kw):
+        self.stack.append(
+            {'correlate_froms': set(),
+             "iswrapper": False,
+             "asfrom_froms": set(),
+             "selectable": insert_stmt})
+
         self.isinsert = True
         crud_params = crud._get_crud_params(self, insert_stmt, **kw)
 
@@ -1812,6 +1834,8 @@ class SQLCompiler(Compiled):
         if self.returning and not self.returning_precedes_values:
             text += " " + returning_clause
 
+        self.stack.pop(-1)
+
         return text
 
     def update_limit_clause(self, update_stmt):
@@ -2278,14 +2302,14 @@ class DDLCompiler(Compiled):
             formatted_name = self.preparer.format_constraint(constraint)
             if formatted_name is not None:
                 text += "CONSTRAINT %s " % formatted_name
-        remote_table = list(constraint._elements.values())[0].column.table
+        remote_table = list(constraint.elements)[0].column.table
         text += "FOREIGN KEY(%s) REFERENCES %s (%s)" % (
             ', '.join(preparer.quote(f.parent.name)
-                      for f in constraint._elements.values()),
+                      for f in constraint.elements),
             self.define_constraint_remote_table(
                 constraint, remote_table, preparer),
             ', '.join(preparer.quote(f.column.name)
-                      for f in constraint._elements.values())
+                      for f in constraint.elements)
         )
         text += self.define_constraint_match(constraint)
         text += self.define_constraint_cascades(constraint)
diff --git a/lib/sqlalchemy/sql/default_comparator.py b/lib/sqlalchemy/sql/default_comparator.py
index 4f53e2979..d26fdc455 100644
--- a/lib/sqlalchemy/sql/default_comparator.py
+++ b/lib/sqlalchemy/sql/default_comparator.py
@@ -68,8 +68,12 @@ class _DefaultColumnComparator(operators.ColumnOperators):
 
     def _boolean_compare(self, expr, op, obj, negate=None, reverse=False,
                          _python_is_types=(util.NoneType, bool),
+                         result_type = None,
                          **kwargs):
 
+        if result_type is None:
+            result_type = type_api.BOOLEANTYPE
+
         if isinstance(obj, _python_is_types + (Null, True_, False_)):
 
             # allow x ==/!= True/False to be treated as a literal.
@@ -80,7 +84,7 @@ class _DefaultColumnComparator(operators.ColumnOperators):
                 return BinaryExpression(expr,
                                         _literal_as_text(obj),
                                         op,
-                                        type_=type_api.BOOLEANTYPE,
+                                        type_=result_type,
                                         negate=negate, modifiers=kwargs)
             else:
                 # all other None/True/False uses IS, IS NOT
@@ -103,13 +107,13 @@ class _DefaultColumnComparator(operators.ColumnOperators):
             return BinaryExpression(obj,
                                     expr,
                                     op,
-                                    type_=type_api.BOOLEANTYPE,
+                                    type_=result_type,
                                     negate=negate, modifiers=kwargs)
         else:
             return BinaryExpression(expr,
                                     obj,
                                     op,
-                                    type_=type_api.BOOLEANTYPE,
+                                    type_=result_type,
                                     negate=negate, modifiers=kwargs)
 
     def _binary_operate(self, expr, op, obj, reverse=False, result_type=None,
@@ -125,7 +129,8 @@ class _DefaultColumnComparator(operators.ColumnOperators):
             op, result_type = left.comparator._adapt_expression(
                 op, right.comparator)
 
-        return BinaryExpression(left, right, op, type_=result_type)
+        return BinaryExpression(
+            left, right, op, type_=result_type, modifiers=kw)
 
     def _conjunction_operate(self, expr, op, other, **kw):
         if op is operators.and_:
@@ -216,11 +221,16 @@ class _DefaultColumnComparator(operators.ColumnOperators):
 
     def _match_impl(self, expr, op, other, **kw):
         """See :meth:`.ColumnOperators.match`."""
+
         return self._boolean_compare(
             expr, operators.match_op,
             self._check_literal(
                 expr, operators.match_op, other),
-            **kw)
+            result_type=type_api.MATCHTYPE,
+            negate=operators.notmatch_op
+            if op is operators.match_op else operators.match_op,
+            **kw
+        )
 
     def _distinct_impl(self, expr, op, **kw):
         """See :meth:`.ColumnOperators.distinct`."""
@@ -282,6 +292,7 @@ class _DefaultColumnComparator(operators.ColumnOperators):
         "isnot": (_boolean_compare, operators.isnot),
         "collate": (_collate_impl,),
         "match_op": (_match_impl,),
+        "notmatch_op": (_match_impl,),
         "distinct_op": (_distinct_impl,),
         "between_op": (_between_impl, ),
         "notbetween_op": (_between_impl, ),
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index fa9b66024..30965c801 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -1092,7 +1092,7 @@ class BindParameter(ColumnElement):
         """
         if isinstance(key, ColumnClause):
             type_ = key.type
-            key = key.name
+            key = key.key
         if required is NO_ARG:
             required = (value is NO_ARG and callable_ is None)
         if value is NO_ARG:
@@ -2763,7 +2763,7 @@ class BinaryExpression(ColumnElement):
                 self.right,
                 self.negate,
                 negate=self.operator,
-                type_=type_api.BOOLEANTYPE,
+                type_=self.type,
                 modifiers=self.modifiers)
         else:
             return super(BinaryExpression, self)._negate()
@@ -3335,7 +3335,7 @@ class ColumnClause(Immutable, ColumnElement):
             return name
 
     def _bind_param(self, operator, obj):
-        return BindParameter(self.name, obj,
+        return BindParameter(self.key, obj,
                              _compared_to_operator=operator,
                              _compared_to_type=self.type,
                              unique=True)
diff --git a/lib/sqlalchemy/sql/operators.py b/lib/sqlalchemy/sql/operators.py
index 945356328..b08e44ab8 100644
--- a/lib/sqlalchemy/sql/operators.py
+++ b/lib/sqlalchemy/sql/operators.py
@@ -767,6 +767,10 @@ def match_op(a, b, **kw):
     return a.match(b, **kw)
 
 
+def notmatch_op(a, b, **kw):
+    return a.notmatch(b, **kw)
+
+
 def comma_op(a, b):
     raise NotImplementedError()
 
@@ -834,6 +838,7 @@ _PRECEDENCE = {
 
     concat_op: 6,
     match_op: 6,
+    notmatch_op: 6,
 
     ilike_op: 6,
     notilike_op: 6,
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 96cabbf4f..b90f7fc53 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -824,7 +824,7 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
                 table.append_constraint(
                     c.copy(schema=fk_constraint_schema, target_table=table))
 
-            else:
+            elif not c._type_bound:
                 table.append_constraint(
                     c.copy(schema=schema, target_table=table))
         for index in self.indexes:
@@ -1295,7 +1295,7 @@ class Column(SchemaItem, ColumnClause):
 
         # Constraint objects plus non-constraint-bound ForeignKey objects
         args = \
-            [c.copy(**kw) for c in self.constraints] + \
+            [c.copy(**kw) for c in self.constraints if not c._type_bound] + \
             [c.copy(**kw) for c in self.foreign_keys if not c.constraint]
 
         type_ = self.type
@@ -1804,7 +1804,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
                 match=self.match,
                 **self._unvalidated_dialect_kw
             )
-            self.constraint._elements[self.parent] = self
+            self.constraint._append_element(column, self)
             self.constraint._set_parent_with_dispatch(table)
         table.foreign_keys.add(self)
 
@@ -2254,7 +2254,7 @@ class Constraint(DialectKWArgs, SchemaItem):
     __visit_name__ = 'constraint'
 
     def __init__(self, name=None, deferrable=None, initially=None,
-                 _create_rule=None, info=None,
+                 _create_rule=None, info=None, _type_bound=False,
                  **dialect_kw):
         """Create a SQL constraint.
 
@@ -2304,6 +2304,7 @@ class Constraint(DialectKWArgs, SchemaItem):
         if info:
             self.info = info
         self._create_rule = _create_rule
+        self._type_bound = _type_bound
         util.set_creation_order(self)
         self._validate_dialect_kwargs(dialect_kw)
 
@@ -2420,7 +2421,7 @@ class CheckConstraint(Constraint):
 
     def __init__(self, sqltext, name=None, deferrable=None,
                  initially=None, table=None, info=None, _create_rule=None,
-                 _autoattach=True):
+                 _autoattach=True, _type_bound=False):
         """Construct a CHECK constraint.
 
         :param sqltext:
@@ -2450,7 +2451,9 @@ class CheckConstraint(Constraint):
         """
 
         super(CheckConstraint, self).\
-            __init__(name, deferrable, initially, _create_rule, info=info)
+            __init__(
+                name, deferrable, initially, _create_rule, info=info,
+                _type_bound=_type_bound)
         self.sqltext = _literal_as_text(sqltext, warn=False)
         if table is not None:
             self._set_parent_with_dispatch(table)
@@ -2485,11 +2488,12 @@ class CheckConstraint(Constraint):
                             deferrable=self.deferrable,
                             _create_rule=self._create_rule,
                             table=target_table,
-                            _autoattach=False)
+                            _autoattach=False,
+                            _type_bound=self._type_bound)
         return self._schema_item_copy(c)
 
 
-class ForeignKeyConstraint(Constraint):
+class ForeignKeyConstraint(ColumnCollectionConstraint):
     """A table-level FOREIGN KEY constraint.
 
     Defines a single column or composite FOREIGN KEY ... REFERENCES
@@ -2564,9 +2568,10 @@ class ForeignKeyConstraint(Constraint):
             .. versionadded:: 0.9.2
 
         """
-        super(ForeignKeyConstraint, self).\
-            __init__(name, deferrable, initially, info=info, **dialect_kw)
 
+        Constraint.__init__(
+            self, name=name, deferrable=deferrable, initially=initially,
+            info=info, **dialect_kw)
         self.onupdate = onupdate
         self.ondelete = ondelete
         self.link_to_name = link_to_name
@@ -2575,14 +2580,12 @@ class ForeignKeyConstraint(Constraint):
         self.use_alter = use_alter
         self.match = match
 
-        self._elements = util.OrderedDict()
-
         # standalone ForeignKeyConstraint - create
         # associated ForeignKey objects which will be applied to hosted
         # Column objects (in col.foreign_keys), either now or when attached
         # to the Table for string-specified names
-        for col, refcol in zip(columns, refcolumns):
-            self._elements[col] = ForeignKey(
+        self.elements = [
+            ForeignKey(
                 refcol,
                 _constraint=self,
                 name=self.name,
@@ -2594,25 +2597,36 @@ class ForeignKeyConstraint(Constraint):
                 deferrable=self.deferrable,
                 initially=self.initially,
                 **self.dialect_kwargs
-            )
+            ) for refcol in refcolumns
+        ]
 
+        ColumnCollectionMixin.__init__(self, *columns)
         if table is not None:
+            if hasattr(self, "parent"):
+                assert table is self.parent
             self._set_parent_with_dispatch(table)
-        elif columns and \
-            isinstance(columns[0], Column) and \
-                columns[0].table is not None:
-            self._set_parent_with_dispatch(columns[0].table)
+
+    def _append_element(self, column, fk):
+        self.columns.add(column)
+        self.elements.append(fk)
+
+    @property
+    def _elements(self):
+        # legacy - provide a dictionary view of (column_key, fk)
+        return util.OrderedDict(
+            zip(self.column_keys, self.elements)
+        )
 
     @property
     def _referred_schema(self):
-        for elem in self._elements.values():
+        for elem in self.elements:
             return elem._referred_schema
         else:
             return None
 
     def _validate_dest_table(self, table):
         table_keys = set([elem._table_key()
-                          for elem in self._elements.values()])
+                          for elem in self.elements])
         if None not in table_keys and len(table_keys) > 1:
             elem0, elem1 = sorted(table_keys)[0:2]
             raise exc.ArgumentError(
@@ -2625,38 +2639,48 @@ class ForeignKeyConstraint(Constraint):
                 ))
 
     @property
-    def _col_description(self):
-        return ", ".join(self._elements)
+    def column_keys(self):
+        """Return a list of string keys representing the local
+        columns in this :class:`.ForeignKeyConstraint`.
 
-    @property
-    def columns(self):
-        return list(self._elements)
+        This list is either the original string arguments sent
+        to the constructor of the :class:`.ForeignKeyConstraint`,
+        or if the constraint has been initialized with :class:`.Column`
+        objects, is the string .key of each element.
+
+        .. versionadded:: 1.0.0
+
+        """
+        if hasattr(self, "parent"):
+            return self.columns.keys()
+        else:
+            return [
+                col.key if isinstance(col, ColumnElement)
+                else str(col) for col in self._pending_colargs
+            ]
 
     @property
-    def elements(self):
-        return list(self._elements.values())
+    def _col_description(self):
+        return ", ".join(self.column_keys)
 
     def _set_parent(self, table):
-        super(ForeignKeyConstraint, self)._set_parent(table)
-
-        self._validate_dest_table(table)
+        Constraint._set_parent(self, table)
 
-        for col, fk in self._elements.items():
-            # string-specified column names now get
-            # resolved to Column objects
-            if isinstance(col, util.string_types):
-                try:
-                    col = table.c[col]
-                except KeyError:
-                    raise exc.ArgumentError(
-                        "Can't create ForeignKeyConstraint "
-                        "on table '%s': no column "
-                        "named '%s' is present." % (table.description, col))
+        try:
+            ColumnCollectionConstraint._set_parent(self, table)
+        except KeyError as ke:
+            raise exc.ArgumentError(
+                "Can't create ForeignKeyConstraint "
+                "on table '%s': no column "
+                "named '%s' is present." % (table.description, ke.args[0]))
 
+        for col, fk in zip(self.columns, self.elements):
             if not hasattr(fk, 'parent') or \
                     fk.parent is not col:
                 fk._set_parent_with_dispatch(col)
 
+        self._validate_dest_table(table)
+
         if self.use_alter:
             def supports_alter(ddl, event, schema_item, bind, **kw):
                 return table in set(kw['tables']) and \
@@ -2669,14 +2693,14 @@ class ForeignKeyConstraint(Constraint):
 
     def copy(self, schema=None, target_table=None, **kw):
         fkc = ForeignKeyConstraint(
-            [x.parent.key for x in self._elements.values()],
+            [x.parent.key for x in self.elements],
             [x._get_colspec(
                 schema=schema,
                 table_name=target_table.name
                 if target_table is not None
                 and x._table_key() == x.parent.table.key
                 else None)
-             for x in self._elements.values()],
+             for x in self.elements],
             name=self.name,
             onupdate=self.onupdate,
             ondelete=self.ondelete,
@@ -2687,8 +2711,8 @@ class ForeignKeyConstraint(Constraint):
             match=self.match
         )
         for self_fk, other_fk in zip(
-                self._elements.values(),
-                fkc._elements.values()):
+                self.elements,
+                fkc.elements):
             self_fk._schema_item_copy(other_fk)
         return self._schema_item_copy(fkc)
 
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 8198a6733..87029ec2b 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -171,6 +171,79 @@ class Selectable(ClauseElement):
         return self
 
 
+class HasPrefixes(object):
+    _prefixes = ()
+
+    @_generative
+    def prefix_with(self, *expr, **kw):
+        """Add one or more expressions following the statement keyword, i.e.
+        SELECT, INSERT, UPDATE, or DELETE. Generative.
+
+        This is used to support backend-specific prefix keywords such as those
+        provided by MySQL.
+
+        E.g.::
+
+            stmt = table.insert().prefix_with("LOW_PRIORITY", dialect="mysql")
+
+        Multiple prefixes can be specified by multiple calls
+        to :meth:`.prefix_with`.
+
+        :param \*expr: textual or :class:`.ClauseElement` construct which
+         will be rendered following the INSERT, UPDATE, or DELETE
+         keyword.
+        :param \**kw: A single keyword 'dialect' is accepted.  This is an
+         optional string dialect name which will
+         limit rendering of this prefix to only that dialect.
+
+        """
+        dialect = kw.pop('dialect', None)
+        if kw:
+            raise exc.ArgumentError("Unsupported argument(s): %s" %
+                                    ",".join(kw))
+        self._setup_prefixes(expr, dialect)
+
+    def _setup_prefixes(self, prefixes, dialect=None):
+        self._prefixes = self._prefixes + tuple(
+            [(_literal_as_text(p, warn=False), dialect) for p in prefixes])
+
+
+class HasSuffixes(object):
+    _suffixes = ()
+
+    @_generative
+    def suffix_with(self, *expr, **kw):
+        """Add one or more expressions following the statement as a whole.
+
+        This is used to support backend-specific suffix keywords on
+        certain constructs.
+
+        E.g.::
+
+            stmt = select([col1, col2]).cte().suffix_with(
+                "cycle empno set y_cycle to 1 default 0", dialect="oracle")
+
+        Multiple prefixes can be specified by multiple calls
+        to :meth:`.suffix_with`.
+
+        :param \*expr: textual or :class:`.ClauseElement` construct which
+         will be rendered following the target clause.
+        :param \**kw: A single keyword 'dialect' is accepted.  This is an
+         optional string dialect name which will
+         limit rendering of this suffix to only that dialect.
+
+        """
+        dialect = kw.pop('dialect', None)
+        if kw:
+            raise exc.ArgumentError("Unsupported argument(s): %s" %
+                                    ",".join(kw))
+        self._setup_suffixes(expr, dialect)
+
+    def _setup_suffixes(self, suffixes, dialect=None):
+        self._suffixes = self._suffixes + tuple(
+            [(_literal_as_text(p, warn=False), dialect) for p in suffixes])
+
+
 class FromClause(Selectable):
     """Represent an element that can be used within the ``FROM``
     clause of a ``SELECT`` statement.
@@ -1088,7 +1161,7 @@ class Alias(FromClause):
         return self.element.bind
 
 
-class CTE(Alias):
+class CTE(Generative, HasSuffixes, Alias):
     """Represent a Common Table Expression.
 
     The :class:`.CTE` object is obtained using the
@@ -1104,10 +1177,13 @@ class CTE(Alias):
                  name=None,
                  recursive=False,
                  _cte_alias=None,
-                 _restates=frozenset()):
+                 _restates=frozenset(),
+                 _suffixes=None):
         self.recursive = recursive
         self._cte_alias = _cte_alias
         self._restates = _restates
+        if _suffixes:
+            self._suffixes = _suffixes
         super(CTE, self).__init__(selectable, name=name)
 
     def alias(self, name=None, flat=False):
@@ -1116,6 +1192,7 @@ class CTE(Alias):
             name=name,
             recursive=self.recursive,
             _cte_alias=self,
+            _suffixes=self._suffixes
         )
 
     def union(self, other):
@@ -1123,7 +1200,8 @@ class CTE(Alias):
             self.original.union(other),
             name=self.name,
             recursive=self.recursive,
-            _restates=self._restates.union([self])
+            _restates=self._restates.union([self]),
+            _suffixes=self._suffixes
         )
 
     def union_all(self, other):
@@ -1131,7 +1209,8 @@ class CTE(Alias):
             self.original.union_all(other),
             name=self.name,
             recursive=self.recursive,
-            _restates=self._restates.union([self])
+            _restates=self._restates.union([self]),
+            _suffixes=self._suffixes
         )
 
 
@@ -2118,44 +2197,7 @@ class CompoundSelect(GenerativeSelect):
     bind = property(bind, _set_bind)
 
 
-class HasPrefixes(object):
-    _prefixes = ()
-
-    @_generative
-    def prefix_with(self, *expr, **kw):
-        """Add one or more expressions following the statement keyword, i.e.
-        SELECT, INSERT, UPDATE, or DELETE. Generative.
-
-        This is used to support backend-specific prefix keywords such as those
-        provided by MySQL.
-
-        E.g.::
-
-            stmt = table.insert().prefix_with("LOW_PRIORITY", dialect="mysql")
-
-        Multiple prefixes can be specified by multiple calls
-        to :meth:`.prefix_with`.
-
-        :param \*expr: textual or :class:`.ClauseElement` construct which
-         will be rendered following the INSERT, UPDATE, or DELETE
-         keyword.
-        :param \**kw: A single keyword 'dialect' is accepted.  This is an
-         optional string dialect name which will
-         limit rendering of this prefix to only that dialect.
-
-        """
-        dialect = kw.pop('dialect', None)
-        if kw:
-            raise exc.ArgumentError("Unsupported argument(s): %s" %
-                                    ",".join(kw))
-        self._setup_prefixes(expr, dialect)
-
-    def _setup_prefixes(self, prefixes, dialect=None):
-        self._prefixes = self._prefixes + tuple(
-            [(_literal_as_text(p, warn=False), dialect) for p in prefixes])
-
-
-class Select(HasPrefixes, GenerativeSelect):
+class Select(HasPrefixes, HasSuffixes, GenerativeSelect):
     """Represents a ``SELECT`` statement.
 
     """
@@ -2163,6 +2205,7 @@ class Select(HasPrefixes, GenerativeSelect):
     __visit_name__ = 'select'
 
     _prefixes = ()
+    _suffixes = ()
     _hints = util.immutabledict()
     _statement_hints = ()
     _distinct = False
@@ -2180,6 +2223,7 @@ class Select(HasPrefixes, GenerativeSelect):
                  having=None,
                  correlate=True,
                  prefixes=None,
+                 suffixes=None,
                  **kwargs):
         """Construct a new :class:`.Select`.
 
@@ -2425,6 +2469,9 @@ class Select(HasPrefixes, GenerativeSelect):
         if prefixes:
             self._setup_prefixes(prefixes)
 
+        if suffixes:
+            self._setup_suffixes(suffixes)
+
         GenerativeSelect.__init__(self, **kwargs)
 
     @property
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 2729bc83e..9a2de39b4 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -894,7 +894,7 @@ class LargeBinary(_Binary):
 
         :param length: optional, a length for the column for use in
           DDL statements, for those BLOB types that accept a length
-          (i.e. MySQL).  It does *not* produce a small BINARY/VARBINARY
+          (i.e. MySQL).  It does *not* produce a *lengthed* BINARY/VARBINARY
           type - use the BINARY/VARBINARY types specifically for those.
           May be safely omitted if no ``CREATE
           TABLE`` will be issued.  Certain databases may require a
@@ -998,13 +998,11 @@ class SchemaType(SchemaEventTarget):
     def adapt(self, impltype, **kw):
         schema = kw.pop('schema', self.schema)
 
-        # don't associate with MetaData as the hosting type
+        # don't associate with self.metadata as the hosting type
         # is already associated with it, avoid creating event
         # listeners
-        metadata = kw.pop('metadata', None)
         return impltype(name=self.name,
                         schema=schema,
-                        metadata=metadata,
                         inherit_schema=self.inherit_schema,
                         **kw)
 
@@ -1165,7 +1163,8 @@ class Enum(String, SchemaType):
             type_coerce(column, self).in_(self.enums),
             name=_defer_name(self.name),
             _create_rule=util.portable_instancemethod(
-                self._should_create_constraint)
+                self._should_create_constraint),
+            _type_bound=True
         )
         assert e.table is table
 
@@ -1303,7 +1302,8 @@ class Boolean(TypeEngine, SchemaType):
             type_coerce(column, self).in_([0, 1]),
             name=_defer_name(self.name),
             _create_rule=util.portable_instancemethod(
-                self._should_create_constraint)
+                self._should_create_constraint),
+            _type_bound=True
         )
         assert e.table is table
 
@@ -1654,10 +1654,26 @@ class NullType(TypeEngine):
     comparator_factory = Comparator
 
 
+class MatchType(Boolean):
+    """Refers to the return type of the MATCH operator.
+
+    As the :meth:`.Operators.match` is probably the most open-ended
+    operator in generic SQLAlchemy Core, we can't assume the return type
+    at SQL evaluation time, as MySQL returns a floating point, not a boolean,
+    and other backends might do something different.    So this type
+    acts as a placeholder, currently subclassing :class:`.Boolean`.
+    The type allows dialects to inject result-processing functionality
+    if needed, and on MySQL will return floating-point values.
+
+    .. versionadded:: 1.0.0
+
+    """
+
 NULLTYPE = NullType()
 BOOLEANTYPE = Boolean()
 STRINGTYPE = String()
 INTEGERTYPE = Integer()
+MATCHTYPE = MatchType()
 
 _type_map = {
     int: Integer(),
@@ -1685,6 +1701,7 @@ type_api.BOOLEANTYPE = BOOLEANTYPE
 type_api.STRINGTYPE = STRINGTYPE
 type_api.INTEGERTYPE = INTEGERTYPE
 type_api.NULLTYPE = NULLTYPE
+type_api.MATCHTYPE = MATCHTYPE
 type_api._type_map = _type_map
 
 # this one, there's all kinds of ways to play it, but at the EOD
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index 77c6e1b1e..d3e0a008e 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -19,7 +19,7 @@ BOOLEANTYPE = None
 INTEGERTYPE = None
 NULLTYPE = None
 STRINGTYPE = None
-
+MATCHTYPE = None
 
 class TypeEngine(Visitable):
     """The ultimate base class for all SQL datatypes.
diff --git a/lib/sqlalchemy/testing/engines.py b/lib/sqlalchemy/testing/engines.py
index 1284f9c2a..0f6f59401 100644
--- a/lib/sqlalchemy/testing/engines.py
+++ b/lib/sqlalchemy/testing/engines.py
@@ -280,10 +280,10 @@ class DBAPIProxyCursor(object):
 
     """
 
-    def __init__(self, engine, conn):
+    def __init__(self, engine, conn, *args, **kwargs):
         self.engine = engine
         self.connection = conn
-        self.cursor = conn.cursor()
+        self.cursor = conn.cursor(*args, **kwargs)
 
     def execute(self, stmt, parameters=None, **kw):
         if parameters:
@@ -311,8 +311,8 @@ class DBAPIProxyConnection(object):
         self.engine = engine
         self.cursor_cls = cursor_cls
 
-    def cursor(self):
-        return self.cursor_cls(self.engine, self.conn)
+    def cursor(self, *args, **kwargs):
+        return self.cursor_cls(self.engine, self.conn, *args, **kwargs)
 
     def close(self):
         self.conn.close()
diff --git a/lib/sqlalchemy/testing/exclusions.py b/lib/sqlalchemy/testing/exclusions.py
index f94724608..0aff43ae1 100644
--- a/lib/sqlalchemy/testing/exclusions.py
+++ b/lib/sqlalchemy/testing/exclusions.py
@@ -425,7 +425,7 @@ def skip(db, reason=None):
 
 def only_on(dbs, reason=None):
     return only_if(
-        OrPredicate([SpecPredicate(db) for db in util.to_list(dbs)])
+        OrPredicate([Predicate.as_predicate(db) for db in util.to_list(dbs)])
     )
 
 
diff --git a/lib/sqlalchemy/testing/plugin/plugin_base.py b/lib/sqlalchemy/testing/plugin/plugin_base.py
index 6696427dc..614a12133 100644
--- a/lib/sqlalchemy/testing/plugin/plugin_base.py
+++ b/lib/sqlalchemy/testing/plugin/plugin_base.py
@@ -93,7 +93,10 @@ def setup_options(make_option):
                 help="Exclude tests with tag <tag>")
     make_option("--write-profiles", action="store_true",
                 dest="write_profiles", default=False,
-                help="Write/update profiling data.")
+                help="Write/update failing profiling data.")
+    make_option("--force-write-profiles", action="store_true",
+                dest="force_write_profiles", default=False,
+                help="Unconditionally write/update profiling data.")
 
 
 def configure_follower(follower_ident):
diff --git a/lib/sqlalchemy/testing/profiling.py b/lib/sqlalchemy/testing/profiling.py
index fcb888f86..671bbe32d 100644
--- a/lib/sqlalchemy/testing/profiling.py
+++ b/lib/sqlalchemy/testing/profiling.py
@@ -42,7 +42,11 @@ class ProfileStatsFile(object):
     """
 
     def __init__(self, filename):
-        self.write = (
+        self.force_write = (
+            config.options is not None and
+            config.options.force_write_profiles
+        )
+        self.write = self.force_write or (
             config.options is not None and
             config.options.write_profiles
         )
@@ -115,7 +119,11 @@ class ProfileStatsFile(object):
         per_fn = self.data[test_key]
         per_platform = per_fn[self.platform_key]
         counts = per_platform['counts']
-        counts[-1] = callcount
+        current_count = per_platform['current_count']
+        if current_count < len(counts):
+            counts[current_count - 1] = callcount
+        else:
+            counts[-1] = callcount
         if self.write:
             self._write()
 
@@ -235,7 +243,7 @@ def count_functions(variance=0.05):
         deviance = int(callcount * variance)
         failed = abs(callcount - expected_count) > deviance
 
-        if failed:
+        if failed or _profile_stats.force_write:
             if _profile_stats.write:
                 _profile_stats.replace(callcount)
             else:
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index da3e3128a..5744431cb 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -323,6 +323,11 @@ class SuiteRequirements(Requirements):
         return exclusions.closed()
 
     @property
+    def temporary_tables(self):
+        """target database supports temporary tables"""
+        return exclusions.open()
+
+    @property
     def temporary_views(self):
         """target database supports temporary views"""
         return exclusions.closed()
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 08b858b47..3edbdeb8c 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -128,6 +128,10 @@ class ComponentReflectionTest(fixtures.TablesTest):
                 DDL("create temporary view user_tmp_v as "
                     "select * from user_tmp")
             )
+            event.listen(
+                user_tmp, "before_drop",
+                DDL("drop view user_tmp_v")
+            )
 
     @classmethod
     def define_index(cls, metadata, users):
@@ -511,6 +515,8 @@ class ComponentReflectionTest(fixtures.TablesTest):
     def test_get_temp_table_indexes(self):
         insp = inspect(self.metadata.bind)
         indexes = insp.get_indexes('user_tmp')
+        for ind in indexes:
+            ind.pop('dialect_options', None)
         eq_(
             # TODO: we need to add better filtering for indexes/uq constraints
             # that are doubled up
diff --git a/lib/sqlalchemy/types.py b/lib/sqlalchemy/types.py
index b49e389ac..1215bd790 100644
--- a/lib/sqlalchemy/types.py
+++ b/lib/sqlalchemy/types.py
@@ -51,6 +51,7 @@ from .sql.sqltypes import (
     Integer,
     Interval,
     LargeBinary,
+    MatchType,
     NCHAR,
     NVARCHAR,
     NullType,
diff --git a/lib/sqlalchemy/util/_collections.py b/lib/sqlalchemy/util/_collections.py
index a1fbc0fa0..d36852698 100644
--- a/lib/sqlalchemy/util/_collections.py
+++ b/lib/sqlalchemy/util/_collections.py
@@ -10,9 +10,10 @@
 from __future__ import absolute_import
 import weakref
 import operator
-from .compat import threading, itertools_filterfalse
+from .compat import threading, itertools_filterfalse, string_types
 from . import py2k
 import types
+import collections
 
 EMPTY_SET = frozenset()
 
@@ -779,10 +780,12 @@ def coerce_generator_arg(arg):
 def to_list(x, default=None):
     if x is None:
         return default
-    if not isinstance(x, (list, tuple)):
+    if not isinstance(x, collections.Iterable) or isinstance(x, string_types):
         return [x]
-    else:
+    elif isinstance(x, list):
         return x
+    else:
+        return list(x)
 
 
 def to_set(x):