Fixed #34799 -- Made MySQL introspection skip cross-database relations.

2 files changed, 65 insertions, 0 deletions

diff --git a/django/db/backends/mysql/introspection.py b/django/db/backends/mysql/introspection.py
index a1ee51a692..ea1d0aa187 100644
--- a/django/db/backends/mysql/introspection.py
+++ b/django/db/backends/mysql/introspection.py
@@ -198,6 +198,7 @@ class DatabaseIntrospection(BaseDatabaseIntrospection):
             FROM information_schema.key_column_usage
             WHERE table_name = %s
                 AND table_schema = DATABASE()
+                AND referenced_table_schema = DATABASE()
                 AND referenced_table_name IS NOT NULL
                 AND referenced_column_name IS NOT NULL
             """,
@@ -257,6 +258,10 @@ class DatabaseIntrospection(BaseDatabaseIntrospection):
                 information_schema.table_constraints AS c
             WHERE
                 kc.table_schema = DATABASE() AND
+                (
+                    kc.referenced_table_schema = DATABASE() OR
+                    kc.referenced_table_schema IS NULL
+                ) AND
                 c.table_schema = kc.table_schema AND
                 c.constraint_name = kc.constraint_name AND
                 c.constraint_type != 'CHECK' AND
diff --git a/tests/backends/mysql/test_introspection.py b/tests/backends/mysql/test_introspection.py
index c1247de232..a2e2938451 100644
--- a/tests/backends/mysql/test_introspection.py
+++ b/tests/backends/mysql/test_introspection.py
@@ -61,3 +61,63 @@ class StorageEngineTests(TestCase):
                 cursor.execute(drop_sql)
             with other_connection.cursor() as cursor:
                 cursor.execute(drop_sql)
+
+
+@skipUnless(connection.vendor == "mysql", "MySQL specific SQL")
+class TestCrossDatabaseRelations(TestCase):
+    databases = {"default", "other"}
+
+    def test_omit_cross_database_relations(self):
+        default_connection = connections["default"]
+        other_connection = connections["other"]
+        main_table = "cross_schema_get_relations_main_table"
+        main_table_quoted = default_connection.ops.quote_name(main_table)
+        other_schema_quoted = other_connection.ops.quote_name(
+            other_connection.settings_dict["NAME"]
+        )
+        rel_table = "cross_schema_get_relations_rel_table"
+        rel_table_quoted = other_connection.ops.quote_name(rel_table)
+        rel_column = "cross_schema_get_relations_rel_table_id"
+        rel_column_quoted = other_connection.ops.quote_name(rel_column)
+        try:
+            with other_connection.cursor() as other_cursor:
+                other_cursor.execute(
+                    f"""
+                    CREATE TABLE {rel_table_quoted} (
+                        id integer AUTO_INCREMENT,
+                        PRIMARY KEY (id)
+                    )
+                    """
+                )
+            with default_connection.cursor() as default_cursor:
+                # Create table in the default schema with a cross-database
+                # relation.
+                default_cursor.execute(
+                    f"""
+                    CREATE TABLE {main_table_quoted} (
+                        id integer AUTO_INCREMENT,
+                        {rel_column_quoted} integer NOT NULL,
+                        PRIMARY KEY (id),
+                        FOREIGN KEY ({rel_column_quoted})
+                        REFERENCES {other_schema_quoted}.{rel_table_quoted}(id)
+                    )
+                    """
+                )
+                relations = default_connection.introspection.get_relations(
+                    default_cursor, main_table
+                )
+                constraints = default_connection.introspection.get_constraints(
+                    default_cursor, main_table
+                )
+            self.assertEqual(len(relations), 0)
+            rel_column_fk_constraints = [
+                spec
+                for name, spec in constraints.items()
+                if spec["columns"] == [rel_column] and spec["foreign_key"] is not None
+            ]
+            self.assertEqual(len(rel_column_fk_constraints), 0)
+        finally:
+            with default_connection.cursor() as default_cursor:
+                default_cursor.execute(f"DROP TABLE IF EXISTS {main_table_quoted}")
+            with other_connection.cursor() as other_cursor:
+                other_cursor.execute(f"DROP TABLE IF EXISTS {rel_table_quoted}")