- implement kwarg validation and type system for dialect-specific

arguments; [ticket:2866]
- add dialect specific kwarg functionality to ForeignKeyConstraint, ForeignKey

16 files changed, 562 insertions, 81 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 86bd004a3..63f95d242 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -15,6 +15,30 @@
     :version: 0.9.2
 
     .. change::
+        :tags: feature, sql
+        :tickets: 2866
+
+        The system by which schema constructs and certain SQL constructs
+        accept dialect-specific keyword arguments has been enhanced.  This
+        system includes commonly the :class:`.Table` and :class:`.Index` constructs,
+        which accept a wide variety of dialect-specific arguments such as
+        ``mysql_engine`` and ``postgresql_where``, as well as the constructs
+        :class:`.PrimaryKeyConstraint`, :class:`.UniqueConstraint`,
+        :class:`.Update`, :class:`.Insert` and :class:`.Delete`, and also
+        newly added kwarg capability to :class:`.ForeignKeyConstraint`
+        and :class:`.ForeignKey`.  The change is that participating dialects
+        can now specify acceptable argument lists for these constructs, allowing
+        an argument error to be raised if an invalid keyword is specified for
+        a particular dialect.  If the dialect portion of the keyword is unrecognized,
+        a warning is emitted only; while the system will actually make use
+        of setuptools entrypoints in order to locate non-local dialects,
+        the use case where certain dialect-specific arguments are used
+        in an environment where that third-party dialect is uninstalled remains
+        supported.  Dialects also have to explicitly opt-in to this system,
+        so that external dialects which aren't making use of this system
+        will remain unaffected.
+
+    .. change::
         :tags: bug, sql
         :pullreq: bitbucket:11
 
diff --git a/lib/sqlalchemy/dialects/firebird/base.py b/lib/sqlalchemy/dialects/firebird/base.py
index b9af6e580..21db57b68 100644
--- a/lib/sqlalchemy/dialects/firebird/base.py
+++ b/lib/sqlalchemy/dialects/firebird/base.py
@@ -402,6 +402,8 @@ class FBDialect(default.DefaultDialect):
     colspecs = colspecs
     ischema_names = ischema_names
 
+    construct_arguments = []
+
     # defaults to dialect ver. 3,
     # will be autodetected off upon
     # first connect
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 0e779686c..522cb5ce3 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1018,7 +1018,7 @@ class MSDDLCompiler(compiler.DDLCompiler):
             text += "UNIQUE "
 
         # handle clustering option
-        if index.kwargs.get("mssql_clustered"):
+        if index.dialect_options['mssql']['clustered']:
             text += "CLUSTERED "
 
         text += "INDEX %s ON %s (%s)" \
@@ -1033,10 +1033,10 @@ class MSDDLCompiler(compiler.DDLCompiler):
                         )
 
         # handle other included columns
-        if index.kwargs.get("mssql_include"):
+        if index.dialect_options['mssql']['include']:
             inclusions = [index.table.c[col]
                             if isinstance(col, util.string_types) else col
-                          for col in index.kwargs["mssql_include"]]
+                          for col in index.dialect_options['mssql']['include']]
 
             text += " INCLUDE (%s)" \
                 % ', '.join([preparer.quote(c.name)
@@ -1059,8 +1059,7 @@ class MSDDLCompiler(compiler.DDLCompiler):
                     self.preparer.format_constraint(constraint)
         text += "PRIMARY KEY "
 
-        # support clustered option
-        if constraint.kwargs.get("mssql_clustered"):
+        if constraint.dialect_options['mssql']['clustered']:
             text += "CLUSTERED "
 
         text += "(%s)" % ', '.join(self.preparer.quote(c.name)
@@ -1077,8 +1076,7 @@ class MSDDLCompiler(compiler.DDLCompiler):
                     self.preparer.format_constraint(constraint)
         text += "UNIQUE "
 
-        # support clustered option
-        if constraint.kwargs.get("mssql_clustered"):
+        if constraint.dialect_options['mssql']['clustered']:
             text += "CLUSTERED "
 
         text += "(%s)" % ', '.join(self.preparer.quote(c.name)
@@ -1166,6 +1164,19 @@ class MSDialect(default.DefaultDialect):
     type_compiler = MSTypeCompiler
     preparer = MSIdentifierPreparer
 
+    construct_arguments = [
+        (sa_schema.PrimaryKeyConstraint, {
+            "clustered": False
+        }),
+        (sa_schema.UniqueConstraint, {
+            "clustered": False
+        }),
+        (sa_schema.Index, {
+            "clustered": False,
+            "include": None
+        })
+    ]
+
     def __init__(self,
                  query_timeout=None,
                  use_scope_identity=True,
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 22675e592..e45f6ecd8 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1537,9 +1537,9 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
         constraint_string = super(
                         MySQLDDLCompiler, self).create_table_constraints(table)
 
-        engine_key = '%s_engine' % self.dialect.name
-        is_innodb = engine_key in table.kwargs and \
-                    table.kwargs[engine_key].lower() == 'innodb'
+        # why self.dialect.name and not 'mysql'?  because of drizzle
+        is_innodb = 'engine' in table.dialect_options[self.dialect.name] and \
+                    table.dialect_options[self.dialect.name]['engine'].lower() == 'innodb'
 
         auto_inc_column = table._autoincrement_column
 
@@ -1633,8 +1633,8 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
             text += "UNIQUE "
         text += "INDEX %s ON %s " % (name, table)
 
-        if 'mysql_length' in index.kwargs:
-            length = index.kwargs['mysql_length']
+        length = index.dialect_options['mysql']['length']
+        if length is not None:
 
             if isinstance(length, dict):
                 # length value can be a (column_name --> integer value) mapping
@@ -1655,8 +1655,8 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
             columns = ', '.join(columns)
         text += '(%s)' % columns
 
-        if 'mysql_using' in index.kwargs:
-            using = index.kwargs['mysql_using']
+        using = index.dialect_options['mysql']['using']
+        if using is not None:
             text += " USING %s" % (preparer.quote(using))
 
         return text
@@ -1664,8 +1664,8 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
     def visit_primary_key_constraint(self, constraint):
         text = super(MySQLDDLCompiler, self).\
             visit_primary_key_constraint(constraint)
-        if "mysql_using" in constraint.kwargs:
-            using = constraint.kwargs['mysql_using']
+        using = constraint.dialect_options['mysql']['using']
+        if using:
             text += " USING %s" % (self.preparer.quote(using))
         return text
 
@@ -2023,6 +2023,22 @@ class MySQLDialect(default.DefaultDialect):
     _backslash_escapes = True
     _server_ansiquotes = False
 
+    construct_arguments = [
+        (sa_schema.Table, {
+            "*": None
+        }),
+        (sql.Update, {
+            "limit": None
+        }),
+        (sa_schema.PrimaryKeyConstraint, {
+            "using": None
+        }),
+        (sa_schema.Index, {
+            "using": None,
+            "length": None,
+        })
+    ]
+
     def __init__(self, isolation_level=None, **kwargs):
         kwargs.pop('use_ansiquotes', None)   # legacy
         default.DefaultDialect.__init__(self, **kwargs)
diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index e5a160443..74a587d0b 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -754,6 +754,8 @@ class OracleDialect(default.DefaultDialect):
 
     reflection_options = ('oracle_resolve_synonyms', )
 
+    construct_arguments = []
+
     def __init__(self,
                 use_ansi=True,
                 optimize_limits=False,
diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index b7979a3e5..11bd3830d 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -1171,11 +1171,11 @@ class PGDDLCompiler(compiler.DDLCompiler):
                     preparer.format_table(index.table)
                 )
 
-        if 'postgresql_using' in index.kwargs:
-            using = index.kwargs['postgresql_using']
+        using = index.dialect_options['postgresql']['using']
+        if using:
             text += "USING %s " % preparer.quote(using)
 
-        ops = index.kwargs.get('postgresql_ops', {})
+        ops = index.dialect_options["postgresql"]["ops"]
         text += "(%s)" \
                 % (
                     ', '.join([
@@ -1188,10 +1188,7 @@ class PGDDLCompiler(compiler.DDLCompiler):
                         for expr, c in zip(index.expressions, index.columns)])
                     )
 
-        if 'postgresql_where' in index.kwargs:
-            whereclause = index.kwargs['postgresql_where']
-        else:
-            whereclause = None
+        whereclause = index.dialect_options["postgresql"]["where"]
 
         if whereclause is not None:
             where_compiled = self.sql_compiler.process(
@@ -1437,6 +1434,14 @@ class PGDialect(default.DefaultDialect):
     inspector = PGInspector
     isolation_level = None
 
+    construct_arguments = [
+        (schema.Index, {
+            "using": False,
+            "where": None,
+            "ops": {}
+        })
+    ]
+
     _backslash_escapes = True
 
     def __init__(self, isolation_level=None, json_serializer=None,
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index ac644f8df..579a61046 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -130,14 +130,14 @@ for new connections through the usage of events::
 import datetime
 import re
 
-from sqlalchemy import sql, exc
-from sqlalchemy.engine import default, base, reflection
-from sqlalchemy import types as sqltypes
-from sqlalchemy import util
-from sqlalchemy.sql import compiler
-from sqlalchemy import processors
-
-from sqlalchemy.types import BIGINT, BLOB, BOOLEAN, CHAR,\
+from ... import sql, exc
+from ...engine import default, reflection
+from ... import types as sqltypes, schema as sa_schema
+from ... import util
+from ...sql import compiler
+from ... import processors
+
+from ...types import BIGINT, BLOB, BOOLEAN, CHAR,\
     DECIMAL, FLOAT, REAL, INTEGER, NUMERIC, SMALLINT, TEXT,\
     TIMESTAMP, VARCHAR
 
@@ -499,7 +499,7 @@ class SQLiteDDLCompiler(compiler.DDLCompiler):
             colspec += " NOT NULL"
 
         if (column.primary_key and
-            column.table.kwargs.get('sqlite_autoincrement', False) and
+            column.table.dialect_options['sqlite']['autoincrement'] and
             len(column.table.primary_key.columns) == 1 and
             issubclass(column.type._type_affinity, sqltypes.Integer) and
             not column.foreign_keys):
@@ -514,7 +514,7 @@ class SQLiteDDLCompiler(compiler.DDLCompiler):
         if len(constraint.columns) == 1:
             c = list(constraint)[0]
             if c.primary_key and \
-                c.table.kwargs.get('sqlite_autoincrement', False) and \
+                c.table.dialect_options['sqlite']['autoincrement'] and \
                 issubclass(c.type._type_affinity, sqltypes.Integer) and \
                 not c.foreign_keys:
                 return None
@@ -623,6 +623,12 @@ class SQLiteDialect(default.DefaultDialect):
     supports_cast = True
     supports_default_values = True
 
+    construct_arguments = [
+        (sa_schema.Table, {
+            "autoincrement": False
+        })
+    ]
+
     _broken_fk_pragma_quotes = False
 
     def __init__(self, isolation_level=None, native_datetime=False, **kwargs):
diff --git a/lib/sqlalchemy/dialects/sybase/base.py b/lib/sqlalchemy/dialects/sybase/base.py
index 2f58aed97..501270778 100644
--- a/lib/sqlalchemy/dialects/sybase/base.py
+++ b/lib/sqlalchemy/dialects/sybase/base.py
@@ -440,6 +440,8 @@ class SybaseDialect(default.DefaultDialect):
     preparer = SybaseIdentifierPreparer
     inspector = SybaseInspector
 
+    construct_arguments = []
+
     def _get_default_schema_name(self, connection):
         return connection.scalar(
                      text("SELECT user_name() as user_name",
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index c1c012d33..e507885fa 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -111,6 +111,33 @@ class DefaultDialect(interfaces.Dialect):
 
     server_version_info = None
 
+    construct_arguments = None
+    """Optional set of argument specifiers for various SQLAlchemy
+    constructs, typically schema items.
+
+    To
+    implement, establish as a series of tuples, as in::
+
+        construct_arguments = [
+            (schema.Index, {
+                "using": False,
+                "where": None,
+                "ops": None
+            })
+        ]
+
+    If the above construct is established on the Postgresql dialect,
+    the ``Index`` construct will now accept additional keyword arguments
+    such as ``postgresql_using``, ``postgresql_where``, etc.  Any kind of
+    ``postgresql_XYZ`` argument not corresponding to the above template will
+    be rejected with an ``ArgumentError`, for all those SQLAlchemy constructs
+    which implement the :class:`.DialectKWArgs` class.
+
+    The default is ``None``; older dialects which don't implement the argument
+    will have the old behavior of un-validated kwargs to schema/SQL constructs.
+
+    """
+
     # indicates symbol names are
     # UPPERCASEd if they are case insensitive
     # within the database.
@@ -176,6 +203,7 @@ class DefaultDialect(interfaces.Dialect):
         self._decoder = processors.to_unicode_processor_factory(self.encoding)
 
 
+
     @util.memoized_property
     def _type_memos(self):
         return weakref.WeakKeyDictionary()
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index badec84ea..93b66bf0c 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -442,7 +442,7 @@ class Inspector(object):
         # apply table options
         tbl_opts = self.get_table_options(table_name, schema, **table.kwargs)
         if tbl_opts:
-            table.kwargs.update(tbl_opts)
+            table._validate_dialect_kwargs(tbl_opts)
 
         # table.kwargs will need to be passed to each reflection method.  Make
         # sure keywords are strings.
diff --git a/lib/sqlalchemy/exc.py b/lib/sqlalchemy/exc.py
index fe6879b16..68e517e26 100644
--- a/lib/sqlalchemy/exc.py
+++ b/lib/sqlalchemy/exc.py
@@ -26,6 +26,9 @@ class ArgumentError(SQLAlchemyError):
 
     """
 
+class NoSuchModuleError(ArgumentError):
+    """Raised when a dynamically-loaded module (usually a database dialect)
+    of a particular name cannot be located."""
 
 class NoForeignKeysError(ArgumentError):
     """Raised when no foreign keys can be located between two selectables
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index e6c5d6ed7..f4bfe392a 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -12,7 +12,8 @@
 from .. import util, exc
 import itertools
 from .visitors import ClauseVisitor
-
+import re
+import collections
 
 PARSE_AUTOCOMMIT = util.symbol('PARSE_AUTOCOMMIT')
 NO_ARG = util.symbol('NO_ARG')
@@ -43,6 +44,122 @@ def _generative(fn, *args, **kw):
     return self
 
 
+class DialectKWArgs(object):
+    """Establish the ability for a class to have dialect-specific arguments
+    with defaults and validation.
+
+    """
+
+    @util.memoized_property
+    def dialect_kwargs(self):
+        """A collection of keyword arguments specified as dialect-specific
+        options to this construct.
+
+        The arguments are present here in their original ``<dialect>_<kwarg>``
+        format.
+
+        .. versionadded:: 0.9.2
+
+        .. seealso::
+
+            :attr:`.DialectKWArgs.dialect_options` - nested dictionary form
+
+        """
+
+        return util.immutabledict(
+            (
+                "%s_%s" % (dialect_name, kwarg_name),
+                kw_dict[kwarg_name]
+            )
+            for dialect_name, kw_dict in self.dialect_options.items()
+            for kwarg_name in kw_dict if kwarg_name != '*'
+        )
+
+    @property
+    def kwargs(self):
+        """Deprecated; see :attr:`.DialectKWArgs.dialect_kwargs"""
+        return self.dialect_kwargs
+
+    @util.dependencies("sqlalchemy.dialects")
+    def _kw_reg_for_dialect(dialects, dialect_name):
+        dialect_cls = dialects.registry.load(dialect_name)
+        if dialect_cls.construct_arguments is None:
+            return None
+        return dict(dialect_cls.construct_arguments)
+    _kw_registry = util.PopulateDict(_kw_reg_for_dialect)
+
+    def _kw_reg_for_dialect_cls(self, dialect_name):
+        construct_arg_dictionary = DialectKWArgs._kw_registry[dialect_name]
+        if construct_arg_dictionary is None:
+            return {"*": None}
+        else:
+            d = {}
+            for cls in reversed(self.__class__.__mro__):
+                if cls in construct_arg_dictionary:
+                    d.update(construct_arg_dictionary[cls])
+            return d
+
+    @util.memoized_property
+    def dialect_options(self):
+        """A collection of keyword arguments specified as dialect-specific
+        options to this construct.
+
+        This is a two-level nested registry, keyed to ``<dialect_name>``
+        and ``<argument_name>``.  For example, the ``postgresql_where`` argument
+        would be locatable as::
+
+            arg = my_object.dialect_options['postgresql']['where']
+
+        .. versionadded:: 0.9.2
+
+        .. seealso::
+
+            :attr:`.DialectKWArgs.dialect_kwargs` - flat dictionary form
+
+        """
+
+        return util.PopulateDict(
+                    util.portable_instancemethod(self._kw_reg_for_dialect_cls)
+                    )
+
+    def _validate_dialect_kwargs(self, kwargs):
+        # validate remaining kwargs that they all specify DB prefixes
+
+        if not kwargs:
+            return
+
+        self.__dict__.pop('dialect_kwargs', None)
+
+        for k in kwargs:
+            m = re.match('^(.+?)_(.+)$', k)
+            if m is None:
+                raise TypeError("Additional arguments should be "
+                        "named <dialectname>_<argument>, got '%s'" % k)
+            dialect_name, arg_name = m.group(1, 2)
+
+            try:
+                construct_arg_dictionary = self.dialect_options[dialect_name]
+            except exc.NoSuchModuleError:
+                util.warn(
+                        "Can't validate argument %r; can't "
+                        "locate any SQLAlchemy dialect named %r" %
+                        (k, dialect_name))
+                self.dialect_options[dialect_name] = {
+                                            "*": None,
+                                            arg_name: kwargs[k]}
+            else:
+                if "*" not in construct_arg_dictionary and \
+                    arg_name not in construct_arg_dictionary:
+                    raise exc.ArgumentError(
+                            "Argument %r is not accepted by "
+                            "dialect %r on behalf of %r" % (
+                                k,
+                                dialect_name, self.__class__
+                            ))
+                else:
+                    construct_arg_dictionary[arg_name] = kwargs[k]
+
+
 class Generative(object):
     """Allow a ClauseElement to generate itself via the
     @_generative decorator.
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index 22694348b..854b894ee 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -8,13 +8,13 @@ Provide :class:`.Insert`, :class:`.Update` and :class:`.Delete`.
 
 """
 
-from .base import Executable, _generative, _from_objects
+from .base import Executable, _generative, _from_objects, DialectKWArgs
 from .elements import ClauseElement, _literal_as_text, Null, and_, _clone
 from .selectable import _interpret_as_from, _interpret_as_select, HasPrefixes
 from .. import util
 from .. import exc
 
-class UpdateBase(HasPrefixes, Executable, ClauseElement):
+class UpdateBase(DialectKWArgs, HasPrefixes, Executable, ClauseElement):
     """Form the base for ``INSERT``, ``UPDATE``, and ``DELETE`` statements.
 
     """
@@ -23,7 +23,6 @@ class UpdateBase(HasPrefixes, Executable, ClauseElement):
 
     _execution_options = \
         Executable._execution_options.union({'autocommit': True})
-    kwargs = util.immutabledict()
     _hints = util.immutabledict()
     _prefixes = ()
 
@@ -417,7 +416,7 @@ class Insert(ValuesBase):
                 prefixes=None,
                 returning=None,
                 return_defaults=False,
-                **kwargs):
+                **dialect_kw):
         """Construct an :class:`.Insert` object.
 
         Similar functionality is available via the
@@ -462,7 +461,7 @@ class Insert(ValuesBase):
         self.select = self.select_names = None
         self.inline = inline
         self._returning = returning
-        self.kwargs = kwargs
+        self._validate_dialect_kwargs(dialect_kw)
         self._return_defaults = return_defaults
 
     def get_children(self, **kwargs):
@@ -547,7 +546,7 @@ class Update(ValuesBase):
                 prefixes=None,
                 returning=None,
                 return_defaults=False,
-                **kwargs):
+                **dialect_kw):
         """Construct an :class:`.Update` object.
 
         E.g.::
@@ -658,7 +657,7 @@ class Update(ValuesBase):
         else:
             self._whereclause = None
         self.inline = inline
-        self.kwargs = kwargs
+        self._validate_dialect_kwargs(dialect_kw)
         self._return_defaults = return_defaults
 
 
@@ -716,7 +715,7 @@ class Delete(UpdateBase):
             bind=None,
             returning=None,
             prefixes=None,
-            **kwargs):
+            **dialect_kw):
         """Construct :class:`.Delete` object.
 
         Similar functionality is available via the
@@ -746,7 +745,7 @@ class Delete(UpdateBase):
         else:
             self._whereclause = None
 
-        self.kwargs = kwargs
+        self._validate_dialect_kwargs(dialect_kw)
 
     def get_children(self, **kwargs):
         if self._whereclause is not None:
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 6ee92871a..73c2a49c8 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -28,10 +28,9 @@ as components in SQL expressions.
 
 """
 
-import re
 import inspect
 from .. import exc, util, event, inspection
-from .base import SchemaEventTarget
+from .base import SchemaEventTarget, DialectKWArgs
 from . import visitors
 from . import type_api
 from .base import _bind_or_error, ColumnCollection
@@ -53,14 +52,6 @@ def _get_table_key(name, schema):
         return schema + "." + name
 
 
-def _validate_dialect_kwargs(kwargs, name):
-    # validate remaining kwargs that they all specify DB prefixes
-
-    for k in kwargs:
-        m = re.match('^(.+?)_.*', k)
-        if m is None:
-            raise TypeError("Additional arguments should be "
-                    "named <dialectname>_<argument>, got '%s'" % k)
 
 @inspection._self_inspects
 class SchemaItem(SchemaEventTarget, visitors.Visitable):
@@ -115,7 +106,7 @@ class SchemaItem(SchemaEventTarget, visitors.Visitable):
         return schema_item
 
 
-class Table(SchemaItem, TableClause):
+class Table(DialectKWArgs, SchemaItem, TableClause):
     """Represent a table in a database.
 
     e.g.::
@@ -296,9 +287,13 @@ class Table(SchemaItem, TableClause):
         ``quote_schema=True`` to the constructor, or use the :class:`.quoted_name`
         construct to specify the name.
 
-
     :param useexisting: Deprecated.  Use extend_existing.
 
+    :param \**kw: Additional keyword arguments not mentioned above are
+        dialect specific, and passed in the form ``<dialectname>_<argname>``.
+        See the documentation regarding an individual dialect at
+        :ref:`dialect_toplevel` for detail on documented arguments.
+
     """
 
     __visit_name__ = 'table'
@@ -397,7 +392,6 @@ class Table(SchemaItem, TableClause):
         PrimaryKeyConstraint()._set_parent_with_dispatch(self)
         self.foreign_keys = set()
         self._extra_dependencies = set()
-        self.kwargs = {}
         if self.schema is not None:
             self.fullname = "%s.%s" % (self.schema, self.name)
         else:
@@ -502,9 +496,7 @@ class Table(SchemaItem, TableClause):
         self._init_items(*args)
 
     def _extra_kwargs(self, **kwargs):
-        # validate remaining kwargs that they all specify DB prefixes
-        _validate_dialect_kwargs(kwargs, "Table")
-        self.kwargs.update(kwargs)
+        self._validate_dialect_kwargs(kwargs)
 
     def _init_collections(self):
         pass
@@ -1254,7 +1246,7 @@ class Column(SchemaItem, ColumnClause):
             return ColumnClause.get_children(self, **kwargs)
 
 
-class ForeignKey(SchemaItem):
+class ForeignKey(DialectKWArgs, SchemaItem):
     """Defines a dependency between two columns.
 
     ``ForeignKey`` is specified as an argument to a :class:`.Column` object,
@@ -1295,7 +1287,8 @@ class ForeignKey(SchemaItem):
 
     def __init__(self, column, _constraint=None, use_alter=False, name=None,
                     onupdate=None, ondelete=None, deferrable=None,
-                    initially=None, link_to_name=False, match=None):
+                    initially=None, link_to_name=False, match=None,
+                    **dialect_kw):
         """
         Construct a column-level FOREIGN KEY.
 
@@ -1345,6 +1338,14 @@ class ForeignKey(SchemaItem):
             DDL for this constraint. Typical values include SIMPLE, PARTIAL
             and FULL.
 
+        :param \**dialect_kw:  Additional keyword arguments are dialect specific,
+            and passed in the form ``<dialectname>_<argname>``.  The arguments
+            are ultimately handled by a corresponding :class:`.ForeignKeyConstraint`.
+            See the documentation regarding an individual dialect at
+            :ref:`dialect_toplevel` for detail on documented arguments.
+
+            .. versionadded:: 0.9.2
+
         """
 
         self._colspec = column
@@ -1381,6 +1382,7 @@ class ForeignKey(SchemaItem):
         self.initially = initially
         self.link_to_name = link_to_name
         self.match = match
+        self._unvalidated_dialect_kw = dialect_kw
 
     def __repr__(self):
         return "ForeignKey(%r)" % self._get_colspec()
@@ -1410,7 +1412,8 @@ class ForeignKey(SchemaItem):
                 deferrable=self.deferrable,
                 initially=self.initially,
                 link_to_name=self.link_to_name,
-                match=self.match
+                match=self.match,
+                **self._unvalidated_dialect_kw
                 )
         return self._schema_item_copy(fk)
 
@@ -1651,6 +1654,7 @@ class ForeignKey(SchemaItem):
                 onupdate=self.onupdate, ondelete=self.ondelete,
                 deferrable=self.deferrable, initially=self.initially,
                 match=self.match,
+                **self._unvalidated_dialect_kw
                 )
             self.constraint._elements[self.parent] = self
             self.constraint._set_parent_with_dispatch(table)
@@ -2113,14 +2117,14 @@ class PassiveDefault(DefaultClause):
         DefaultClause.__init__(self, *arg, **kw)
 
 
-class Constraint(SchemaItem):
+class Constraint(DialectKWArgs, SchemaItem):
     """A table-level SQL constraint."""
 
     __visit_name__ = 'constraint'
 
     def __init__(self, name=None, deferrable=None, initially=None,
                             _create_rule=None,
-                            **kw):
+                            **dialect_kw):
         """Create a SQL constraint.
 
         :param name:
@@ -2151,9 +2155,10 @@ class Constraint(SchemaItem):
           _create_rule is used by some types to create constraints.
           Currently, its call signature is subject to change at any time.
 
-        :param \**kwargs:
-          Dialect-specific keyword parameters, see the documentation
-          for various dialects and constraints regarding options here.
+        :param \**dialect_kw:  Additional keyword arguments are dialect specific,
+            and passed in the form ``<dialectname>_<argname>``.  See the
+            documentation regarding an individual dialect at :ref:`dialect_toplevel`
+            for detail on documented arguments.
 
         """
 
@@ -2162,8 +2167,7 @@ class Constraint(SchemaItem):
         self.initially = initially
         self._create_rule = _create_rule
         util.set_creation_order(self)
-        _validate_dialect_kwargs(kw, self.__class__.__name__)
-        self.kwargs = kw
+        self._validate_dialect_kwargs(dialect_kw)
 
     @property
     def table(self):
@@ -2237,6 +2241,9 @@ class ColumnCollectionConstraint(ColumnCollectionMixin, Constraint):
           Optional string.  If set, emit INITIALLY <value> when issuing DDL
           for this constraint.
 
+        :param \**kw: other keyword arguments including dialect-specific
+          arguments are propagated to the :class:`.Constraint` superclass.
+
         """
         ColumnCollectionMixin.__init__(self, *columns)
         Constraint.__init__(self, **kw)
@@ -2354,7 +2361,7 @@ class ForeignKeyConstraint(Constraint):
 
     def __init__(self, columns, refcolumns, name=None, onupdate=None,
             ondelete=None, deferrable=None, initially=None, use_alter=False,
-            link_to_name=False, match=None, table=None):
+            link_to_name=False, match=None, table=None, **dialect_kw):
         """Construct a composite-capable FOREIGN KEY.
 
         :param columns: A sequence of local column names. The named columns
@@ -2399,9 +2406,16 @@ class ForeignKeyConstraint(Constraint):
             DDL for this constraint. Typical values include SIMPLE, PARTIAL
             and FULL.
 
+        :param \**dialect_kw:  Additional keyword arguments are dialect specific,
+            and passed in the form ``<dialectname>_<argname>``.  See the
+            documentation regarding an individual dialect at :ref:`dialect_toplevel`
+            for detail on documented arguments.
+
+            .. versionadded:: 0.9.2
+
         """
         super(ForeignKeyConstraint, self).\
-                        __init__(name, deferrable, initially)
+                        __init__(name, deferrable, initially, **dialect_kw)
 
         self.onupdate = onupdate
         self.ondelete = ondelete
@@ -2428,7 +2442,8 @@ class ForeignKeyConstraint(Constraint):
                     link_to_name=self.link_to_name,
                     match=self.match,
                     deferrable=self.deferrable,
-                    initially=self.initially
+                    initially=self.initially,
+                    **self.dialect_kwargs
                 )
 
         if table is not None:
@@ -2552,7 +2567,7 @@ class UniqueConstraint(ColumnCollectionConstraint):
     __visit_name__ = 'unique_constraint'
 
 
-class Index(ColumnCollectionMixin, SchemaItem):
+class Index(DialectKWArgs, ColumnCollectionMixin, SchemaItem):
     """A table-level INDEX.
 
     Defines a composite (one or more column) INDEX.
@@ -2613,11 +2628,18 @@ class Index(ColumnCollectionMixin, SchemaItem):
           be arbitrary SQL expressions which ultmately refer to a
           :class:`.Column`.
 
-        :param unique:
-            Defaults to False: create a unique index.
+        :param unique=False:
+            Keyword only argument; if True, create a unique index.
+
+        :param quote=None:
+            Keyword only argument; whether to apply quoting to the name of
+            the index.  Works in the same manner as that of
+            :paramref:`.Column.quote`.
 
-        :param \**kw:
-            Other keyword arguments may be interpreted by specific dialects.
+        :param \**kw: Additional keyword arguments not mentioned above are
+            dialect specific, and passed in the form ``<dialectname>_<argname>``.
+            See the documentation regarding an individual dialect at
+            :ref:`dialect_toplevel` for detail on documented arguments.
 
         """
         self.table = None
@@ -2637,7 +2659,7 @@ class Index(ColumnCollectionMixin, SchemaItem):
         self.expressions = expressions
         self.name = quoted_name(name, kw.pop("quote", None))
         self.unique = kw.pop('unique', False)
-        self.kwargs = kw
+        self._validate_dialect_kwargs(kw)
 
         # will call _set_parent() if table-bound column
         # objects are present
diff --git a/lib/sqlalchemy/util/langhelpers.py b/lib/sqlalchemy/util/langhelpers.py
index b6ca7eb2a..82e37ce99 100644
--- a/lib/sqlalchemy/util/langhelpers.py
+++ b/lib/sqlalchemy/util/langhelpers.py
@@ -181,7 +181,7 @@ class PluginLoader(object):
                 self.impls[name] = impl.load
                 return impl.load()
 
-        raise exc.ArgumentError(
+        raise exc.NoSuchModuleError(
                 "Can't load plugin: %s:%s" %
                 (self.group, name))
 
diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index c5caa9780..c45001106 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -13,7 +13,8 @@ import sqlalchemy as tsa
 from sqlalchemy.testing import fixtures
 from sqlalchemy import testing
 from sqlalchemy.testing import ComparesTables, AssertsCompiledSQL
-from sqlalchemy.testing import eq_, is_
+from sqlalchemy.testing import eq_, is_, mock
+from contextlib import contextmanager
 
 class MetaDataTest(fixtures.TestBase, ComparesTables):
     def test_metadata_connect(self):
@@ -586,6 +587,8 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
         meta2 = MetaData()
         table_c = table.tometadata(meta2)
 
+        eq_(table.kwargs, {"mysql_engine": "InnoDB"})
+
         eq_(table.kwargs, table_c.kwargs)
 
     def test_tometadata_indexes(self):
@@ -2046,3 +2049,244 @@ class CatchAllEventsTest(fixtures.TestBase):
             ]
         )
 
+class DialectKWArgTest(fixtures.TestBase):
+    @contextmanager
+    def _fixture(self):
+        from sqlalchemy.engine.default import DefaultDialect
+        class ParticipatingDialect(DefaultDialect):
+            construct_arguments = [
+                (schema.Index, {
+                    "x": 5,
+                    "y": False,
+                    "z_one": None
+                }),
+                (schema.ForeignKeyConstraint, {
+                    "foobar": False
+                })
+            ]
+
+        class ParticipatingDialect2(DefaultDialect):
+            construct_arguments = [
+                (schema.Index, {
+                    "x": 9,
+                    "y": True,
+                    "pp": "default"
+                }),
+                (schema.Table, {
+                    "*": None
+                })
+            ]
+
+        class NonParticipatingDialect(DefaultDialect):
+            construct_arguments = None
+
+        def load(dialect_name):
+            if dialect_name == "participating":
+                return ParticipatingDialect
+            elif dialect_name == "participating2":
+                return ParticipatingDialect2
+            elif dialect_name == "nonparticipating":
+                return NonParticipatingDialect
+            else:
+                raise exc.NoSuchModuleError("no dialect %r" % dialect_name)
+        with mock.patch("sqlalchemy.dialects.registry.load", load):
+            yield
+
+    def test_participating(self):
+        with self._fixture():
+            idx = Index('a', 'b', 'c', participating_y=True)
+            eq_(
+                idx.dialect_kwargs,
+                {
+                    'participating_z_one': None,
+                    'participating_y': True,
+                    'participating_x': 5
+                }
+            )
+
+    def test_nonparticipating(self):
+        with self._fixture():
+            idx = Index('a', 'b', 'c', nonparticipating_y=True, nonparticipating_q=5)
+            eq_(
+                idx.dialect_kwargs,
+                {
+                    'nonparticipating_y': True,
+                    'nonparticipating_q': 5
+                }
+            )
+
+    def test_unknown_dialect_warning(self):
+        with self._fixture():
+            assert_raises_message(
+                exc.SAWarning,
+                "Can't validate argument 'unknown_y'; can't locate "
+                "any SQLAlchemy dialect named 'unknown'",
+                Index, 'a', 'b', 'c', unknown_y=True
+            )
+
+    def test_participating_bad_kw(self):
+        with self._fixture():
+            assert_raises_message(
+                exc.ArgumentError,
+                "Argument 'participating_q_p_x' is not accepted by dialect "
+                "'participating' on behalf of "
+                "<class 'sqlalchemy.sql.schema.Index'>",
+                Index, 'a', 'b', 'c', participating_q_p_x=8
+            )
+
+    def test_participating_unknown_schema_item(self):
+        with self._fixture():
+            # the dialect doesn't include UniqueConstraint in
+            # its registry at all.
+            assert_raises_message(
+                exc.ArgumentError,
+                "Argument 'participating_q_p_x' is not accepted by dialect "
+                "'participating' on behalf of "
+                "<class 'sqlalchemy.sql.schema.UniqueConstraint'>",
+                UniqueConstraint, 'a', 'b', participating_q_p_x=8
+            )
+
+    @testing.emits_warning("Can't validate")
+    def test_unknown_dialect_warning_still_populates(self):
+        with self._fixture():
+            idx = Index('a', 'b', 'c', unknown_y=True)
+            eq_(idx.dialect_kwargs, {"unknown_y": True})  # still populates
+
+    @testing.emits_warning("Can't validate")
+    def test_unknown_dialect_warning_still_populates_multiple(self):
+        with self._fixture():
+            idx = Index('a', 'b', 'c', unknown_y=True, unknown_z=5,
+                                otherunknown_foo='bar', participating_y=8)
+            eq_(idx.dialect_kwargs,
+                {'unknown_z': 5, 'participating_y': 8,
+                'unknown_y': True, 'participating_z_one': None,
+                'otherunknown_foo': 'bar', 'participating_x': 5}
+            )  # still populates
+
+    def test_combined(self):
+        with self._fixture():
+            idx = Index('a', 'b', 'c', participating_x=7,
+                                    nonparticipating_y=True)
+            eq_(
+                idx.dialect_kwargs,
+                {
+                    'participating_z_one': None,
+                    'participating_y': False,
+                    'participating_x': 7,
+                    'nonparticipating_y': True,
+                }
+            )
+
+    def test_multiple_participating(self):
+        with self._fixture():
+            idx = Index('a', 'b', 'c',
+                        participating_x=7,
+                        participating2_x=15,
+                        participating2_y="lazy"
+                )
+            eq_(
+                idx.dialect_kwargs,
+                {
+                    'participating_z_one': None,
+                    'participating_x': 7,
+                    'participating_y': False,
+                    'participating2_pp': 'default',
+                    'participating2_x': 15,
+                    'participating2_y': 'lazy'
+                }
+            )
+
+    def test_foreign_key_propagate(self):
+        with self._fixture():
+            m = MetaData()
+            fk = ForeignKey('t2.id', participating_foobar=True)
+            t = Table('t', m, Column('id', Integer, fk))
+            fkc = [c for c in t.constraints if isinstance(c, ForeignKeyConstraint)][0]
+            eq_(
+                fkc.dialect_kwargs,
+                {'participating_foobar': True}
+            )
+
+    def test_foreign_key_propagate_exceptions_delayed(self):
+        with self._fixture():
+            m = MetaData()
+            fk = ForeignKey('t2.id', participating_fake=True)
+            c1 = Column('id', Integer, fk)
+            assert_raises_message(
+                exc.ArgumentError,
+                "Argument 'participating_fake' is not accepted by "
+                "dialect 'participating' on behalf of "
+                "<class 'sqlalchemy.sql.schema.ForeignKeyConstraint'>",
+                Table, 't', m, c1
+            )
+
+    def test_wildcard(self):
+        with self._fixture():
+            m = MetaData()
+            t = Table('x', m, Column('x', Integer),
+                    participating2_xyz='foo',
+                    participating2_engine='InnoDB',
+                )
+            eq_(
+                t.dialect_kwargs,
+                {
+                    'participating2_xyz': 'foo',
+                    'participating2_engine': 'InnoDB'
+                }
+            )
+
+    def test_uninit_wildcard(self):
+        with self._fixture():
+            m = MetaData()
+            t = Table('x', m, Column('x', Integer))
+            eq_(
+                t.dialect_options['participating2'], {'*': None}
+            )
+            eq_(
+                t.dialect_kwargs, {}
+            )
+
+    def test_not_contains_wildcard(self):
+        with self._fixture():
+            m = MetaData()
+            t = Table('x', m, Column('x', Integer))
+            assert 'foobar' not in t.dialect_options['participating2']
+
+    def test_contains_wildcard(self):
+        with self._fixture():
+            m = MetaData()
+            t = Table('x', m, Column('x', Integer), participating2_foobar=5)
+            assert 'foobar' in t.dialect_options['participating2']
+
+
+    def test_update(self):
+        with self._fixture():
+            idx = Index('a', 'b', 'c', participating_x=20)
+            eq_(idx.dialect_kwargs, {
+                        "participating_x": 20,
+                        'participating_z_one': None,
+                        "participating_y": False})
+            idx._validate_dialect_kwargs({
+                        "participating_x": 25,
+                        "participating_z_one": "default"})
+            eq_(idx.dialect_kwargs, {
+                        "participating_x": 25,
+                        'participating_z_one': "default",
+                        "participating_y": False})
+            idx._validate_dialect_kwargs({
+                        "participating_x": 25,
+                        "participating_z_one": "default"})
+            eq_(idx.dialect_kwargs, {
+                        'participating_z_one': 'default',
+                        'participating_y': False,
+                        'participating_x': 25})
+            idx._validate_dialect_kwargs({
+                        "participating_y": True,
+                        'participating2_y': "p2y"})
+            eq_(idx.dialect_kwargs, {
+                        "participating_x": 25,
+                        "participating2_x": 9,
+                        "participating_y": True,
+                        'participating2_y': "p2y",
+                        "participating2_pp": "default",
+                        "participating_z_one": "default"})