Reflect ASC/DESC in MySQL index columns

Fixed bug in index reflection where on MySQL 8.0 an index that includes
ASC or DESC in an indexed column specfication would not be correctly
reflected, as MySQL 8.0 introduces support for returning this information
in a table definition string.

Change-Id: I21f64984ade690aac8c87dbe3aad0c1ee8e9727f
Fixes: #4293

3 files changed, 74 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_12/4293.rst b/doc/build/changelog/unreleased_12/4293.rst
new file mode 100644
index 000000000..51fac2033
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/4293.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, mysql
+    :tickets: 4293
+
+    Fixed bug in index reflection where on MySQL 8.0 an index that includes
+    ASC or DESC in an indexed column specfication would not be correctly
+    reflected, as MySQL 8.0 introduces support for returning this information
+    in a table definition string.
diff --git a/lib/sqlalchemy/dialects/mysql/reflection.py b/lib/sqlalchemy/dialects/mysql/reflection.py
index e15211044..e88bc3f42 100644
--- a/lib/sqlalchemy/dialects/mysql/reflection.py
+++ b/lib/sqlalchemy/dialects/mysql/reflection.py
@@ -76,6 +76,8 @@ class MySQLTableDefinitionParser(object):
         if m:
             spec = m.groupdict()
             # convert columns into name, length pairs
+            # NOTE: we may want to consider SHOW INDEX as the
+            # format of indexes in MySQL becomes more complex
             spec['columns'] = self._parse_keyexprs(spec['columns'])
             if spec['version_sql']:
                 m2 = self._re_key_version_sql.match(spec['version_sql'])
@@ -317,11 +319,10 @@ class MySQLTableDefinitionParser(object):
 
         # `col`,`col2`(32),`col3`(15) DESC
         #
-        # Note: ASC and DESC aren't reflected, so we'll punt...
         self._re_keyexprs = _re_compile(
             r'(?:'
             r'(?:%(iq)s((?:%(esc_fq)s|[^%(fq)s])+)%(fq)s)'
-            r'(?:\((\d+)\))?(?=\,|$))+' % quotes)
+            r'(?:\((\d+)\))?(?: +(ASC|DESC))?(?=\,|$))+' % quotes)
 
         # 'foo' or 'foo','bar' or 'fo,o','ba''a''r'
         self._re_csv_str = _re_compile(r'\x27(?:\x27\x27|[^\x27])*\x27')
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index 76b6954ae..c244ba972 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -633,6 +633,20 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
             "(textdata) WITH PARSER ngram"
         )
 
+    @testing.provide_metadata
+    def test_non_column_index(self):
+        m1 = self.metadata
+        t1 = Table(
+            'add_ix', m1, Column('x', String(50)), mysql_engine='InnoDB')
+        Index('foo_idx', t1.c.x.desc())
+        m1.create_all()
+
+        insp = inspect(testing.db)
+        eq_(
+            insp.get_indexes("add_ix"),
+            [{'name': 'foo_idx', 'column_names': ['x'], 'unique': False}]
+        )
+
 
 class RawReflectionTest(fixtures.TestBase):
     __backend__ = True
@@ -678,6 +692,55 @@ class RawReflectionTest(fixtures.TestBase):
             "/*!50100 WITH PARSER `ngram` */ "
         )
 
+    def test_key_reflection_columns(self):
+        regex = self.parser._re_key
+        exprs = self.parser._re_keyexprs
+        m = regex.match(
+            "  KEY (`id`) USING BTREE COMMENT '''comment'")
+        eq_(m.group("columns"), '`id`')
+
+        m = regex.match(
+            "  KEY (`x`, `y`) USING BTREE")
+        eq_(m.group("columns"), '`x`, `y`')
+
+        eq_(
+            exprs.findall(m.group("columns")),
+            [("x", "", ""), ("y", "", "")]
+        )
+
+        m = regex.match(
+            "  KEY (`x`(25), `y`(15)) USING BTREE")
+        eq_(m.group("columns"), '`x`(25), `y`(15)')
+        eq_(
+            exprs.findall(m.group("columns")),
+            [("x", "25", ""), ("y", "15", "")]
+        )
+
+        m = regex.match(
+            "  KEY (`x`(25) DESC, `y`(15) ASC) USING BTREE")
+        eq_(m.group("columns"), '`x`(25) DESC, `y`(15) ASC')
+        eq_(
+            exprs.findall(m.group("columns")),
+            [("x", "25", "DESC"), ("y", "15", "ASC")]
+        )
+
+        m = regex.match(
+            "  KEY `foo_idx` (`x` DESC)")
+        eq_(m.group("columns"), '`x` DESC')
+        eq_(
+            exprs.findall(m.group("columns")),
+            [("x", "", "DESC")]
+        )
+
+        eq_(
+            exprs.findall(m.group("columns")),
+            [("x", "", "DESC")]
+        )
+
+        m = regex.match(
+            "  KEY `foo_idx` (`x` DESC, `y` ASC)")
+        eq_(m.group("columns"), '`x` DESC, `y` ASC')
+
     def test_fk_reflection(self):
         regex = self.parser._re_fk_constraint