Support control char reflection in mysql mariadb

Fixed issues regarding reflection of comments for :class:`_schema.Table`
and :class:`_schema.Column` objects, where the comments contained control
characters such as newlines. Additional testing support for these
characters as well as extended Unicode characters in table and column
comments (the latter of which aren't supported by MySQL/MariaDB) added to
testing overall.

Fixes: #9722
Change-Id: Id18bf758fdb6231eb705c61eeaf74bb9fa472601

7 files changed, 119 insertions, 13 deletions

diff --git a/doc/build/changelog/unreleased_20/9722.rst b/doc/build/changelog/unreleased_20/9722.rst
new file mode 100644
index 000000000..ad8fe557a
--- /dev/null
+++ b/doc/build/changelog/unreleased_20/9722.rst
@@ -0,0 +1,10 @@
+.. change::
+    :tags: bug, mysql, mariadb
+    :tickets: 9722
+
+    Fixed issues regarding reflection of comments for :class:`_schema.Table`
+    and :class:`_schema.Column` objects, where the comments contained control
+    characters such as newlines. Additional testing support for these
+    characters as well as extended Unicode characters in table and column
+    comments (the latter of which aren't supported by MySQL/MariaDB) added to
+    testing overall.
diff --git a/lib/sqlalchemy/dialects/mysql/reflection.py b/lib/sqlalchemy/dialects/mysql/reflection.py
index ec3f82a60..ce1b9261d 100644
--- a/lib/sqlalchemy/dialects/mysql/reflection.py
+++ b/lib/sqlalchemy/dialects/mysql/reflection.py
@@ -146,11 +146,8 @@ class MySQLTableDefinitionParser:
 
         options = {}
 
-        if not line or line == ")":
-            pass
-
-        else:
-            rest_of_line = line[:]
+        if line and line != ")":
+            rest_of_line = line
             for regex, cleanup in self._pr_options:
                 m = regex.search(rest_of_line)
                 if not m:
@@ -310,7 +307,7 @@ class MySQLTableDefinitionParser:
         comment = spec.get("comment", None)
 
         if comment is not None:
-            comment = comment.replace("\\\\", "\\").replace("''", "'")
+            comment = cleanup_text(comment)
 
         sqltext = spec.get("generated")
         if sqltext is not None:
@@ -585,11 +582,7 @@ class MySQLTableDefinitionParser:
             re.escape(directive),
             self._optional_equals,
         )
-        self._pr_options.append(
-            _pr_compile(
-                regex, lambda v: v.replace("\\\\", "\\").replace("''", "'")
-            )
-        )
+        self._pr_options.append(_pr_compile(regex, cleanup_text))
 
     def _add_option_word(self, directive):
         regex = r"(?P<directive>%s)%s" r"(?P<val>\w+)" % (
@@ -652,3 +645,28 @@ def _strip_values(values):
             a = a[1:-1].replace(a[0] * 2, a[0])
         strip_values.append(a)
     return strip_values
+
+
+def cleanup_text(raw_text: str) -> str:
+    if "\\" in raw_text:
+        raw_text = re.sub(
+            _control_char_regexp, lambda s: _control_char_map[s[0]], raw_text
+        )
+    return raw_text.replace("''", "'")
+
+
+_control_char_map = {
+    "\\\\": "\\",
+    "\\0": "\0",
+    "\\a": "\a",
+    "\\b": "\b",
+    "\\t": "\t",
+    "\\n": "\n",
+    "\\v": "\v",
+    "\\f": "\f",
+    "\\r": "\r",
+    # '\\e':'\e',
+}
+_control_char_regexp = re.compile(
+    "|".join(re.escape(k) for k in _control_char_map)
+)
diff --git a/lib/sqlalchemy/engine/row.py b/lib/sqlalchemy/engine/row.py
index da781334a..95789ddba 100644
--- a/lib/sqlalchemy/engine/row.py
+++ b/lib/sqlalchemy/engine/row.py
@@ -39,8 +39,8 @@ else:
 
 if TYPE_CHECKING:
     from .result import _KeyType
-    from .result import RMKeyView
     from .result import _ProcessorsType
+    from .result import RMKeyView
 
 _T = TypeVar("_T", bound=Any)
 _TP = TypeVar("_TP", bound=Tuple[Any, ...])
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index b59cce374..ec19e4252 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -656,6 +656,13 @@ class SuiteRequirements(Requirements):
         return exclusions.closed()
 
     @property
+    def comment_reflection_full_unicode(self):
+        """Indicates if the database support table comment reflection in the
+        full unicode range, including emoji etc.
+        """
+        return exclusions.closed()
+
+    @property
     def constraint_comment_reflection(self):
         """indicates if the database support constraint on constraints
         and their reflection"""
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 5927df065..119ef9a36 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -564,6 +564,7 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
                 sa.String(20),
                 comment=r"""Comment types type speedily ' " \ '' Fun!""",
             ),
+            Column("d3", sa.String(42), comment="Comment\nwith\rescapes"),
             schema=schema,
             comment=r"""the test % ' " \ table comment""",
         )
@@ -572,6 +573,7 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
             metadata,
             Column("data", sa.String(20)),
             schema=schema,
+            comment="no\nconstraints\rhas\fescaped\vcomment",
         )
 
         if testing.requires.cross_schema_fk_reflection.enabled:
@@ -831,7 +833,9 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
             (schema, "comment_test"): {
                 "text": r"""the test % ' " \ table comment"""
             },
-            (schema, "no_constraints"): empty,
+            (schema, "no_constraints"): {
+                "text": "no\nconstraints\rhas\fescaped\vcomment"
+            },
             (schema, "local_table"): empty,
             (schema, "remote_table"): empty,
             (schema, "remote_table_2"): empty,
@@ -921,6 +925,7 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
                     "d2",
                     comment=r"""Comment types type speedily ' " \ '' Fun!""",
                 ),
+                col("d3", comment="Comment\nwith\rescapes"),
             ],
             (schema, "no_constraints"): [col("data")],
             (schema, "local_table"): [pk("id"), col("data"), col("remote_id")],
@@ -2271,6 +2276,45 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
             tables = [f"{schema}.{t}" for t in tables]
         eq_(sorted(m.tables), sorted(tables))
 
+    @testing.requires.comment_reflection
+    def test_comments_unicode(self, connection, metadata):
+        Table(
+            "unicode_comments",
+            metadata,
+            Column("unicode", Integer, comment="é試蛇ẟΩ"),
+            Column("emoji", Integer, comment="☁️✨"),
+            comment="試蛇ẟΩ✨",
+        )
+
+        metadata.create_all(connection)
+
+        insp = inspect(connection)
+        tc = insp.get_table_comment("unicode_comments")
+        eq_(tc, {"text": "試蛇ẟΩ✨"})
+
+        cols = insp.get_columns("unicode_comments")
+        value = {c["name"]: c["comment"] for c in cols}
+        exp = {"unicode": "é試蛇ẟΩ", "emoji": "☁️✨"}
+        eq_(value, exp)
+
+    @testing.requires.comment_reflection_full_unicode
+    def test_comments_unicode_full(self, connection, metadata):
+
+        Table(
+            "unicode_comments",
+            metadata,
+            Column("emoji", Integer, comment="🐍🧙🝝🧙‍♂️🧙‍♀️"),
+            comment="🎩🁰🝑🤷‍♀️🤷‍♂️",
+        )
+
+        metadata.create_all(connection)
+
+        insp = inspect(connection)
+        tc = insp.get_table_comment("unicode_comments")
+        eq_(tc, {"text": "🎩🁰🝑🤷‍♀️🤷‍♂️"})
+        c = insp.get_columns("unicode_comments")[0]
+        eq_({c["name"]: c["comment"]}, {"emoji": "🐍🧙🝝🧙‍♂️🧙‍♀️"})
+
 
 class TableNoColumnsTest(fixtures.TestBase):
     __requires__ = ("reflect_tables_no_columns",)
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index f9975a973..a75c05f09 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -1368,6 +1368,29 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
             },
         )
 
+    def test_reflect_comment_escapes(self, connection, metadata):
+        c = "\\ - \\\\ - \\0 - \\a - \\b - \\t - \\n - \\v - \\f - \\r"
+        Table("t", metadata, Column("c", Integer, comment=c), comment=c)
+        metadata.create_all(connection)
+
+        insp = inspect(connection)
+        tc = insp.get_table_comment("t")
+        eq_(tc, {"text": c})
+        col = insp.get_columns("t")[0]
+        eq_({col["name"]: col["comment"]}, {"c": c})
+
+    def test_reflect_comment_unicode(self, connection, metadata):
+        c = "☁️✨🐍🁰🝝"
+        c_exp = "☁️✨???"
+        Table("t", metadata, Column("c", Integer, comment=c), comment=c)
+        metadata.create_all(connection)
+
+        insp = inspect(connection)
+        tc = insp.get_table_comment("t")
+        eq_(tc, {"text": c_exp})
+        col = insp.get_columns("t")[0]
+        eq_({col["name"]: col["comment"]}, {"c": c_exp})
+
 
 class RawReflectionTest(fixtures.TestBase):
     def setup_test(self):
diff --git a/test/requirements.py b/test/requirements.py
index 3c72cd07d..68241330d 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -171,6 +171,10 @@ class DefaultRequirements(SuiteRequirements):
         return only_on(["postgresql", "mysql", "mariadb", "oracle", "mssql"])
 
     @property
+    def comment_reflection_full_unicode(self):
+        return only_on(["postgresql", "oracle", "mssql"])
+
+    @property
     def constraint_comment_reflection(self):
         return only_on(["postgresql"])