Fixed #33017 -- Fixed storage engine introspection on MySQL.

This also improves performance on MySQL instances with a large number
of databases, since querying the information_schema table can be very
slow

django/db/backends/mysql/introspection.py

@@ -180,10 +180,13 @@ class DatabaseIntrospection(BaseDatabaseIntrospection):
         Retrieve the storage engine for a given table. Return the default
         storage engine if the table doesn't exist.
         """
-        cursor.execute(
+        cursor.execute("""
-            "SELECT engine "
+            SELECT engine
-            "FROM information_schema.tables "
+            FROM information_schema.tables
-            "WHERE table_name = %s", [table_name])
+            WHERE
+                table_name = %s AND
+                table_schema = DATABASE()
+        """, [table_name])
         result = cursor.fetchone()
         if not result:
             return self.connection.features._mysql_storage_engine

tests/backends/mysql/test_introspection.py

@@ -1,6 +1,6 @@
 from unittest import skipUnless
 
-from django.db import connection
+from django.db import connection, connections
 from django.test import TestCase
 
 
@@ -27,3 +27,33 @@ class ParsingTests(TestCase):
             with self.subTest(check_clause):
                 check_columns = _parse_constraint_columns(check_clause, table_columns)
                 self.assertEqual(list(check_columns), expected_columns)
+
+
+@skipUnless(connection.vendor == 'mysql', 'MySQL tests')
+class StorageEngineTests(TestCase):
+    databases = {'default', 'other'}
+
+    def test_get_storage_engine(self):
+        table_name = 'test_storage_engine'
+        create_sql = 'CREATE TABLE %s (id INTEGER) ENGINE = %%s' % table_name
+        drop_sql = 'DROP TABLE %s' % table_name
+        default_connection = connections['default']
+        other_connection = connections['other']
+        try:
+            with default_connection.cursor() as cursor:
+                cursor.execute(create_sql % 'InnoDB')
+                self.assertEqual(
+                    default_connection.introspection.get_storage_engine(cursor, table_name),
+                    'InnoDB',
+                )
+            with other_connection.cursor() as cursor:
+                cursor.execute(create_sql % 'MyISAM')
+                self.assertEqual(
+                    other_connection.introspection.get_storage_engine(cursor, table_name),
+                    'MyISAM',
+                )
+        finally:
+            with default_connection.cursor() as cursor:
+                cursor.execute(drop_sql)
+            with other_connection.cursor() as cursor:
+                cursor.execute(drop_sql)