Render and reflect MySQL WITH PARSER index options

Support added for the "WITH PARSER" syntax of CREATE FULLTEXT INDEX
in MySQL, using the ``mysql_with_parser`` keyword argument.  Reflection
is also supported, which accommodates MySQL's special comment format
for reporting on this option as well.  Additionally, the "FULLTEXT" and
"SPATIAL" index prefixes are now reflected back into the ``mysql_prefix``
index option.

Change-Id: I0209291978125d8cee1bb5ed386d4f66578697a0
Fixes: #4219

6 files changed, 136 insertions, 5 deletions

diff --git a/doc/build/changelog/unreleased_13/4219.rst b/doc/build/changelog/unreleased_13/4219.rst
new file mode 100644
index 000000000..ad83898bb
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/4219.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: feature, mysql
+    :tickets: 4219
+
+    Support added for the "WITH PARSER" syntax of CREATE FULLTEXT INDEX
+    in MySQL, using the ``mysql_with_parser`` keyword argument.  Reflection
+    is also supported, which accommodates MySQL's special comment format
+    for reporting on this option as well.  Additionally, the "FULLTEXT" and
+    "SPATIAL" index prefixes are now reflected back into the ``mysql_prefix``
+    index option.
+
+
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index b5d4cb4b7..c8a3d3322 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -525,6 +525,19 @@ http://dev.mysql.com/doc/refman/5.0/en/create-index.html
 
 http://dev.mysql.com/doc/refman/5.0/en/create-table.html
 
+Index Parsers
+~~~~~~~~~~~~~
+
+CREATE FULLTEXT INDEX in MySQL also supports a "WITH PARSER" option.  This
+is available using the keyword argument ``mysql_with_parser``::
+
+    Index(
+        'my_index', my_table.c.data,
+        mysql_prefix='FULLTEXT', mysql_with_parser="ngram")
+
+.. versionadded:: 1.3
+
+
 .. _mysql_foreign_keys:
 
 MySQL Foreign Keys
@@ -1276,6 +1289,10 @@ class MySQLDDLCompiler(compiler.DDLCompiler):
             columns = ', '.join(columns)
         text += '(%s)' % columns
 
+        parser = index.dialect_options['mysql']['with_parser']
+        if parser is not None:
+            text += " WITH PARSER %s" % (parser, )
+
         using = index.dialect_options['mysql']['using']
         if using is not None:
             text += " USING %s" % (preparer.quote(using))
@@ -1693,6 +1710,7 @@ class MySQLDialect(default.DefaultDialect):
             "using": None,
             "length": None,
             "prefix": None,
+            "with_parser": None
         })
     ]
 
@@ -2090,20 +2108,31 @@ class MySQLDialect(default.DefaultDialect):
             connection, table_name, schema, **kw)
 
         indexes = []
+
         for spec in parsed_state.keys:
+            dialect_options = {}
             unique = False
             flavor = spec['type']
             if flavor == 'PRIMARY':
                 continue
             if flavor == 'UNIQUE':
                 unique = True
-            elif flavor in (None, 'FULLTEXT', 'SPATIAL'):
+            elif flavor in ('FULLTEXT', 'SPATIAL'):
+                dialect_options["mysql_prefix"] = flavor
+            elif flavor is None:
                 pass
             else:
                 self.logger.info(
                     "Converting unknown KEY type %s to a plain KEY", flavor)
                 pass
+
+            if spec['parser']:
+                dialect_options['mysql_with_parser'] = spec['parser']
+
             index_d = {}
+            if dialect_options:
+                index_d["dialect_options"] = dialect_options
+
             index_d['name'] = spec['name']
             index_d['column_names'] = [s[0] for s in spec['columns']]
             index_d['unique'] = unique
diff --git a/lib/sqlalchemy/dialects/mysql/reflection.py b/lib/sqlalchemy/dialects/mysql/reflection.py
index f9b8c8d3b..e15211044 100644
--- a/lib/sqlalchemy/dialects/mysql/reflection.py
+++ b/lib/sqlalchemy/dialects/mysql/reflection.py
@@ -77,6 +77,13 @@ class MySQLTableDefinitionParser(object):
             spec = m.groupdict()
             # convert columns into name, length pairs
             spec['columns'] = self._parse_keyexprs(spec['columns'])
+            if spec['version_sql']:
+                m2 = self._re_key_version_sql.match(spec['version_sql'])
+                if m2 and m2.groupdict()['parser']:
+                    spec['parser'] = m2.groupdict()['parser']
+            if spec['parser']:
+                spec['parser'] = self.preparer.unformat_identifiers(
+                    spec['parser'])[0]
             return 'key', spec
 
         # FOREIGN KEY CONSTRAINT
@@ -364,7 +371,7 @@ class MySQLTableDefinitionParser(object):
 
         # (PRIMARY|UNIQUE|FULLTEXT|SPATIAL) INDEX `name` (USING (BTREE|HASH))?
         # (`col` (ASC|DESC)?, `col` (ASC|DESC)?)
-        # KEY_BLOCK_SIZE size | WITH PARSER name
+        # KEY_BLOCK_SIZE size | WITH PARSER name  /*!50100 WITH PARSER name */
         self._re_key = _re_compile(
             r'  '
             r'(?:(?P<type>\S+) )?KEY'
@@ -375,10 +382,18 @@ class MySQLTableDefinitionParser(object):
             r'(?: +KEY_BLOCK_SIZE *[ =]? *(?P<keyblock>\S+))?'
             r'(?: +WITH PARSER +(?P<parser>\S+))?'
             r'(?: +COMMENT +(?P<comment>(\x27\x27|\x27([^\x27])*?\x27)+))?'
+            r'(?: +/\*(?P<version_sql>.+)\*/ +)?'
             r',?$'
             % quotes
         )
 
+        # https://forums.mysql.com/read.php?20,567102,567111#msg-567111
+        # It means if the MySQL version >= \d+, execute what's in the comment
+        self._re_key_version_sql = _re_compile(
+            r'\!\d+ '
+            r'(?: *WITH PARSER +(?P<parser>\S+) *)?'
+        )
+
         # CONSTRAINT `name` FOREIGN KEY (`local_col`)
         # REFERENCES `remote` (`remote_col`)
         # MATCH FULL | MATCH PARTIAL | MATCH SIMPLE
diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index fdf799539..9e12e2d4c 100644
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -50,6 +50,16 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
                             'CREATE FULLTEXT INDEX test_idx1 '
                             'ON testtbl (data(10))')
 
+    def test_create_index_with_parser(self):
+        m = MetaData()
+        tbl = Table('testtbl', m, Column('data', String(255)))
+        idx = Index('test_idx1', tbl.c.data, mysql_length=10,
+                    mysql_prefix='FULLTEXT', mysql_with_parser="ngram")
+
+        self.assert_compile(schema.CreateIndex(idx),
+                            'CREATE FULLTEXT INDEX test_idx1 '
+                            'ON testtbl (data(10)) WITH PARSER ngram')
+
     def test_create_index_with_length(self):
         m = MetaData()
         tbl = Table('testtbl', m, Column('data', String(255)))
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index c2cd0dd77..86937cd0d 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -4,14 +4,15 @@ from sqlalchemy.testing import eq_, is_
 from sqlalchemy import Column, Table, DDL, MetaData, TIMESTAMP, \
     DefaultClause, String, Integer, Text, UnicodeText, SmallInteger,\
     NCHAR, LargeBinary, DateTime, select, UniqueConstraint, Unicode,\
-    BigInteger
+    BigInteger, Index
+from sqlalchemy.schema import CreateIndex
 from sqlalchemy import event
 from sqlalchemy import sql
 from sqlalchemy import exc
 from sqlalchemy import inspect
 from sqlalchemy.dialects.mysql import base as mysql
 from sqlalchemy.dialects.mysql import reflection as _reflection
-from sqlalchemy.testing import fixtures, AssertsExecutionResults
+from sqlalchemy.testing import fixtures, AssertsCompiledSQL
 from sqlalchemy import testing
 from sqlalchemy.testing import assert_raises_message, expect_warnings
 import re
@@ -193,7 +194,7 @@ class TypeReflectionTest(fixtures.TestBase):
         self._run_test(specs, ['enums'])
 
 
-class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
+class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
 
     __only_on__ = 'mysql'
     __backend__ = True
@@ -573,6 +574,56 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
         self.assert_(indexes['uc_a'].unique)
         self.assert_('uc_a' not in constraints)
 
+    @testing.provide_metadata
+    def test_reflect_fulltext(self):
+        mt = Table(
+            "mytable", self.metadata,
+            Column("id", Integer, primary_key=True),
+            Column("textdata", String(50)),
+            mysql_engine='InnoDB'
+        )
+        Index("textdata_ix", mt.c.textdata, mysql_prefix="FULLTEXT")
+        self.metadata.create_all(testing.db)
+
+        mt = Table(
+            "mytable", MetaData(), autoload_with=testing.db
+        )
+        idx = list(mt.indexes)[0]
+        eq_(idx.name, "textdata_ix")
+        eq_(idx.dialect_options['mysql']['prefix'], "FULLTEXT")
+        self.assert_compile(
+            CreateIndex(idx),
+            "CREATE FULLTEXT INDEX textdata_ix ON mytable (textdata)"
+        )
+
+    @testing.requires.mysql_ngram_fulltext
+    @testing.provide_metadata
+    def test_reflect_fulltext_comment(self):
+        mt = Table(
+            "mytable", self.metadata,
+            Column("id", Integer, primary_key=True),
+            Column("textdata", String(50)),
+            mysql_engine='InnoDB'
+        )
+        Index(
+            "textdata_ix", mt.c.textdata,
+            mysql_prefix="FULLTEXT", mysql_with_parser="ngram")
+
+        self.metadata.create_all(testing.db)
+
+        mt = Table(
+            "mytable", MetaData(), autoload_with=testing.db
+        )
+        idx = list(mt.indexes)[0]
+        eq_(idx.name, "textdata_ix")
+        eq_(idx.dialect_options['mysql']['prefix'], "FULLTEXT")
+        eq_(idx.dialect_options['mysql']['with_parser'], "ngram")
+        self.assert_compile(
+            CreateIndex(idx),
+            "CREATE FULLTEXT INDEX textdata_ix ON mytable "
+            "(textdata) WITH PARSER ngram"
+        )
+
 
 class RawReflectionTest(fixtures.TestBase):
     __backend__ = True
@@ -611,6 +662,12 @@ class RawReflectionTest(fixtures.TestBase):
             "  KEY (`id`) USING BTREE COMMENT 'prefix''suffix'")
         assert regex.match(
             "  KEY (`id`) USING BTREE COMMENT 'prefix''text''suffix'")
+        # https://forums.mysql.com/read.php?20,567102,567111#msg-567111
+        # "It means if the MySQL version >= 501, execute what's in the comment"
+        assert regex.match(
+            "  FULLTEXT KEY `ix_fulltext_oi_g_name` (`oi_g_name`) "
+            "/*!50100 WITH PARSER `ngram` */ "
+        )
 
     def test_fk_reflection(self):
         regex = self.parser._re_fk_constraint
diff --git a/test/requirements.py b/test/requirements.py
index 395344041..1b6559cf9 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -1075,6 +1075,14 @@ class DefaultRequirements(SuiteRequirements):
 
         return only_if(check)
 
+    @property
+    def mysql_ngram_fulltext(self):
+        def check(config):
+            return against(config, "mysql") and \
+                not config.db.dialect._is_mariadb and \
+                config.db.dialect.server_version_info >= (5, 7)
+        return only_if(check)
+
     def _mariadb_102(self, config):
         return against(config, "mysql") and \
                 config.db.dialect._is_mariadb and \