set identifier length for MySQL constraints to 64

The rule to limit index names to 64 also applies to all
DDL names, such as those coming from naming conventions.
Add another limiting variable for constraint names and
create test cases against all constraint types.

Additionally, codified in the test suite MySQL's lack of
support for naming of a FOREIGN KEY constraint after
the name was given, which apparently assigns the name to an
associated KEY but not the constraint itself, until MySQL 8
and MariaDB 10.5 which appear to have resolved the
behavior.  However it's not clear how Alembic hasn't had
issues reported with this so far.

Fixed long-lived bug in MySQL dialect where the maximum identifier length
of 255 was too long for names of all types of constraints, not just
indexes, all of which have a size limit of 64. As metadata naming
conventions can create too-long names in this area, apply the limit to the
identifier generator within the DDL compiler.

Fixes: #5898
Change-Id: I79549474845dc29922275cf13321c07598dcea08
(cherry picked from commit fc1e419d139711117b0d2451d6b6e11045afeeb3)

9 files changed, 313 insertions, 10 deletions

diff --git a/doc/build/changelog/unreleased_13/5898.rst b/doc/build/changelog/unreleased_13/5898.rst
new file mode 100644
index 000000000..a0758dad8
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/5898.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, mysql
+    :tickets: 5898
+
+    Fixed long-lived bug in MySQL dialect where the maximum identifier length
+    of 255 was too long for names of all types of constraints, not just
+    indexes, all of which have a size limit of 64. As metadata naming
+    conventions can create too-long names in this area, apply the limit to the
+    identifier generator within the DDL compiler.
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 57a4ab8ce..c41d6acf7 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2337,6 +2337,7 @@ class MySQLDialect(default.DefaultDialect):
     # identifiers are 64, however aliases can be 255...
     max_identifier_length = 255
     max_index_name_length = 64
+    max_constraint_name_length = 64
 
     supports_native_enum = True
 
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 8a087a39c..1c0a87b4a 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -114,12 +114,11 @@ class DefaultDialect(interfaces.Dialect):
     max_identifier_length = 9999
     _user_defined_max_identifier_length = None
 
-    # length at which to truncate
-    # the name of an index.
-    # Usually None to indicate
-    # 'use max_identifier_length'.
-    # thanks to MySQL, sigh
+    # sub-categories of max_identifier_length.
+    # currently these accommodate for MySQL which allows alias names
+    # of 255 but DDL names only of 64.
     max_index_name_length = None
+    max_constraint_name_length = None
 
     supports_sane_rowcount = True
     supports_sane_multi_rowcount = True
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 91e87eed2..e8b5bdb25 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -3757,13 +3757,19 @@ class IdentifierPreparer(object):
             name = constraint.name
 
         if isinstance(name, elements._truncated_label):
+            # calculate these at format time so that ad-hoc changes
+            # to dialect.max_identifier_length etc. can be reflected
+            # as IdentifierPreparer is long lived
             if constraint.__visit_name__ == "index":
                 max_ = (
                     self.dialect.max_index_name_length
                     or self.dialect.max_identifier_length
                 )
             else:
-                max_ = self.dialect.max_identifier_length
+                max_ = (
+                    self.dialect.max_constraint_name_length
+                    or self.dialect.max_identifier_length
+                )
             if len(name) > max_:
                 name = name[0 : max_ - 8] + "_" + util.md5_hex(name)[-4:]
         else:
diff --git a/lib/sqlalchemy/testing/exclusions.py b/lib/sqlalchemy/testing/exclusions.py
index a6e198c0e..1ddfcbb49 100644
--- a/lib/sqlalchemy/testing/exclusions.py
+++ b/lib/sqlalchemy/testing/exclusions.py
@@ -40,6 +40,13 @@ class compound(object):
     def __add__(self, other):
         return self.add(other)
 
+    def as_skips(self):
+        rule = compound()
+        rule.skips.update(self.skips)
+        rule.skips.update(self.fails)
+        rule.tags.update(self.tags)
+        return rule
+
     def add(self, *others):
         copy = compound()
         copy.fails.update(self.fails)
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index d012b1b30..b55f2f37d 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -386,6 +386,18 @@ class SuiteRequirements(Requirements):
         return exclusions.closed()
 
     @property
+    def foreign_key_constraint_name_reflection(self):
+        """Target supports refleciton of FOREIGN KEY constraints and
+        will return the name of the constraint that was used in the
+        "CONSTRANT <name> FOREIGN KEY" DDL.
+
+        MySQL prior to version 8 and MariaDB prior to version 10.5
+        don't support this.
+
+        """
+        return exclusions.closed()
+
+    @property
     def implicit_default_schema(self):
         """target system has a strong concept of 'default' schema that can
         be referred to implicitly.
diff --git a/lib/sqlalchemy/testing/suite/test_ddl.py b/lib/sqlalchemy/testing/suite/test_ddl.py
index 9028fe8b1..3a0ac1bcf 100644
--- a/lib/sqlalchemy/testing/suite/test_ddl.py
+++ b/lib/sqlalchemy/testing/suite/test_ddl.py
@@ -1,14 +1,21 @@
+import random
+
+from . import testing
 from .. import config
 from .. import fixtures
 from .. import util
 from ..assertions import eq_
 from ..config import requirements
+from ..schema import Table
+from ... import CheckConstraint
 from ... import Column
+from ... import ForeignKeyConstraint
+from ... import Index
 from ... import inspect
 from ... import Integer
 from ... import schema
 from ... import String
-from ... import Table
+from ... import UniqueConstraint
 
 
 class TableDDLTest(fixtures.TestBase):
@@ -89,4 +96,197 @@ class TableDDLTest(fixtures.TestBase):
         eq_(inspect(config.db).get_table_comment("test_table"), {"text": None})
 
 
-__all__ = ("TableDDLTest",)
+class LongNameBlowoutTest(fixtures.TestBase):
+    """test the creation of a variety of DDL structures and ensure
+    label length limits pass on backends
+
+    """
+
+    __backend__ = True
+
+    def fk(self, metadata, connection):
+        convention = {
+            "fk": "foreign_key_%(table_name)s_"
+            "%(column_0_N_name)s_"
+            "%(referred_table_name)s_"
+            + (
+                "_".join(
+                    "".join(random.choice("abcdef") for j in range(20))
+                    for i in range(10)
+                )
+            ),
+        }
+        metadata.naming_convention = convention
+
+        Table(
+            "a_things_with_stuff",
+            metadata,
+            Column("id_long_column_name", Integer, primary_key=True),
+            test_needs_fk=True,
+        )
+
+        cons = ForeignKeyConstraint(
+            ["aid"], ["a_things_with_stuff.id_long_column_name"]
+        )
+        Table(
+            "b_related_things_of_value",
+            metadata,
+            Column(
+                "aid",
+            ),
+            cons,
+            test_needs_fk=True,
+        )
+        actual_name = cons.name
+
+        metadata.create_all(connection)
+
+        if testing.requires.foreign_key_constraint_name_reflection.enabled:
+            insp = inspect(connection)
+            fks = insp.get_foreign_keys("b_related_things_of_value")
+            reflected_name = fks[0]["name"]
+
+            return actual_name, reflected_name
+        else:
+            return actual_name, None
+
+    def pk(self, metadata, connection):
+        convention = {
+            "pk": "primary_key_%(table_name)s_"
+            "%(column_0_N_name)s"
+            + (
+                "_".join(
+                    "".join(random.choice("abcdef") for j in range(30))
+                    for i in range(10)
+                )
+            ),
+        }
+        metadata.naming_convention = convention
+
+        a = Table(
+            "a_things_with_stuff",
+            metadata,
+            Column("id_long_column_name", Integer, primary_key=True),
+            Column("id_another_long_name", Integer, primary_key=True),
+        )
+        cons = a.primary_key
+        actual_name = cons.name
+
+        metadata.create_all(connection)
+        insp = inspect(connection)
+        pk = insp.get_pk_constraint("a_things_with_stuff")
+        reflected_name = pk["name"]
+        return actual_name, reflected_name
+
+    def ix(self, metadata, connection):
+        convention = {
+            "ix": "index_%(table_name)s_"
+            "%(column_0_N_name)s"
+            + (
+                "_".join(
+                    "".join(random.choice("abcdef") for j in range(30))
+                    for i in range(10)
+                )
+            ),
+        }
+        metadata.naming_convention = convention
+
+        a = Table(
+            "a_things_with_stuff",
+            metadata,
+            Column("id_long_column_name", Integer, primary_key=True),
+            Column("id_another_long_name", Integer),
+        )
+        cons = Index(None, a.c.id_long_column_name, a.c.id_another_long_name)
+        actual_name = cons.name
+
+        metadata.create_all(connection)
+        insp = inspect(connection)
+        ix = insp.get_indexes("a_things_with_stuff")
+        reflected_name = ix[0]["name"]
+        return actual_name, reflected_name
+
+    def uq(self, metadata, connection):
+        convention = {
+            "uq": "unique_constraint_%(table_name)s_"
+            "%(column_0_N_name)s"
+            + (
+                "_".join(
+                    "".join(random.choice("abcdef") for j in range(30))
+                    for i in range(10)
+                )
+            ),
+        }
+        metadata.naming_convention = convention
+
+        cons = UniqueConstraint("id_long_column_name", "id_another_long_name")
+        Table(
+            "a_things_with_stuff",
+            metadata,
+            Column("id_long_column_name", Integer, primary_key=True),
+            Column("id_another_long_name", Integer),
+            cons,
+        )
+        actual_name = cons.name
+
+        metadata.create_all(connection)
+        insp = inspect(connection)
+        uq = insp.get_unique_constraints("a_things_with_stuff")
+        reflected_name = uq[0]["name"]
+        return actual_name, reflected_name
+
+    def ck(self, metadata, connection):
+        convention = {
+            "ck": "check_constraint_%(table_name)s"
+            + (
+                "_".join(
+                    "".join(random.choice("abcdef") for j in range(30))
+                    for i in range(10)
+                )
+            ),
+        }
+        metadata.naming_convention = convention
+
+        cons = CheckConstraint("some_long_column_name > 5")
+        Table(
+            "a_things_with_stuff",
+            metadata,
+            Column("id_long_column_name", Integer, primary_key=True),
+            Column("some_long_column_name", Integer),
+            cons,
+        )
+        actual_name = cons.name
+
+        metadata.create_all(connection)
+        insp = inspect(connection)
+        ck = insp.get_check_constraints("a_things_with_stuff")
+        reflected_name = ck[0]["name"]
+        return actual_name, reflected_name
+
+    @testing.combinations(
+        ("fk",),
+        ("pk",),
+        ("ix",),
+        ("ck", testing.requires.check_constraint_reflection.as_skips()),
+        ("uq", testing.requires.unique_constraint_reflection.as_skips()),
+        argnames="type_",
+    )
+    @testing.provide_metadata
+    def test_long_convention_name(self, type_, connection):
+        metadata = self.metadata
+
+        actual_name, reflected_name = getattr(self, type_)(
+            metadata, connection
+        )
+
+        assert len(actual_name) > 255
+
+        if reflected_name is not None:
+            overlap = actual_name[0 : len(reflected_name)]
+            if len(overlap) < len(actual_name):
+                eq_(overlap[0:-5], reflected_name[0 : len(overlap) - 5])
+            else:
+                eq_(overlap, reflected_name)
+
+
+__all__ = ("TableDDLTest", "LongNameBlowoutTest")
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index 276de0a60..f32341ebe 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -10,6 +10,7 @@ from sqlalchemy import DefaultClause
 from sqlalchemy import event
 from sqlalchemy import exc
 from sqlalchemy import ForeignKey
+from sqlalchemy import ForeignKeyConstraint
 from sqlalchemy import Index
 from sqlalchemy import inspect
 from sqlalchemy import Integer
@@ -960,6 +961,52 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
                 ],
             )
 
+    @testing.provide_metadata
+    def test_get_foreign_key_name_w_foreign_key_in_name(self, connection):
+        metadata = self.metadata
+        Table(
+            "a",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            mysql_engine="InnoDB",
+        )
+
+        cons = ForeignKeyConstraint(
+            ["aid"], ["a.id"], name="foreign_key_thing_with_stuff"
+        )
+        Table(
+            "b",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column(
+                "aid",
+            ),
+            cons,
+            mysql_engine="InnoDB",
+        )
+        actual_name = cons.name
+
+        metadata.create_all(connection)
+
+        if testing.requires.foreign_key_constraint_name_reflection.enabled:
+            expected_name = actual_name
+        else:
+            expected_name = "b_ibfk_1"
+
+        eq_(
+            inspect(connection).get_foreign_keys("b"),
+            [
+                {
+                    "name": expected_name,
+                    "constrained_columns": ["aid"],
+                    "referred_schema": None,
+                    "referred_table": "a",
+                    "referred_columns": ["id"],
+                    "options": {},
+                }
+            ],
+        )
+
     @testing.requires.mysql_fully_case_sensitive
     @testing.provide_metadata
     def test_case_sensitive_reflection_dual_case_references(self):
diff --git a/test/requirements.py b/test/requirements.py
index 49a19a53b..791f6500e 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -78,6 +78,14 @@ class DefaultRequirements(SuiteRequirements):
         return skip_if(no_support("sqlite", "not supported by database"))
 
     @property
+    def foreign_key_constraint_name_reflection(self):
+        return fails_if(
+            lambda config: against(config, ["mysql", "mariadb"])
+            and not self._mysql_80(config)
+            and not self._mariadb_105(config)
+        )
+
+    @property
     def on_update_cascade(self):
         """target database must support ON UPDATE..CASCADE behavior in
         foreign keys."""
@@ -1448,11 +1456,25 @@ class DefaultRequirements(SuiteRequirements):
 
         return only_if(check)
 
-    def _mariadb_102(self, config):
+    def _mysql_80(self, config):
         return (
             against(config, "mysql")
+            and config.db.dialect._is_mysql
+            and config.db.dialect.server_version_info >= (8,)
+        )
+
+    def _mariadb_102(self, config):
+        return (
+            against(config, ["mysql", "mariadb"])
+            and config.db.dialect._is_mariadb
+            and config.db.dialect._mariadb_normalized_version_info >= (10, 2)
+        )
+
+    def _mariadb_105(self, config):
+        return (
+            against(config, ["mysql", "mariadb"])
             and config.db.dialect._is_mariadb
-            and config.db.dialect._mariadb_normalized_version_info > (10, 2)
+            and config.db.dialect._mariadb_normalized_version_info >= (10, 5)
         )
 
     def _mysql_and_check_constraints_exist(self, config):