Fixed #23576 -- Implemented multi-alias fast-path deletion in MySQL backend.

This required moving the entirety of DELETE SQL generation to the
compiler where it should have been in the first place and implementing
a specialized compiler on MySQL/MariaDB.

The MySQL compiler relies on the "DELETE table FROM table JOIN" syntax
for queries spanning over multiple tables.

django/db/backends/mysql/compiler.py

@@ -14,7 +14,25 @@ class SQLInsertCompiler(compiler.SQLInsertCompiler, SQLCompiler):
 
 
 class SQLDeleteCompiler(compiler.SQLDeleteCompiler, SQLCompiler):
-    pass
+    def as_sql(self):
+        if self.single_alias:
+            return super().as_sql()
+        # MySQL and MariaDB < 10.3.2 doesn't support deletion with a subquery
+        # which is what the default implementation of SQLDeleteCompiler uses
+        # when multiple tables are involved. Use the MySQL/MariaDB specific
+        # DELETE table FROM table syntax instead to avoid performing the
+        # operation in two queries.
+        result = [
+            'DELETE %s FROM' % self.quote_name_unless_alias(
+                self.query.get_initial_alias()
+            )
+        ]
+        from_sql, from_params = self.get_from_clause()
+        result.extend(from_sql)
+        where, params = self.compile(self.query.where)
+        if where:
+            result.append('WHERE %s' % where)
+        return ' '.join(result), tuple(from_params) + tuple(params)
 
 
 class SQLUpdateCompiler(compiler.SQLUpdateCompiler, SQLCompiler):

django/db/models/query.py

@@ -744,7 +744,10 @@ class QuerySet:
         Delete objects found from the given queryset in single direct SQL
         query. No signals are sent and there is no protection for cascades.
         """
-        return sql.DeleteQuery(self.model).delete_qs(self, using)
+        query = self.query.clone()
+        query.__class__ = sql.DeleteQuery
+        cursor = query.get_compiler(using).execute_sql(CURSOR)
+        return cursor.rowcount if cursor else 0
     _raw_delete.alters_data = True
 
     def update(self, **kwargs):

django/db/models/sql/compiler.py

@@ -7,13 +7,16 @@ from django.core.exceptions import EmptyResultSet, FieldError
 from django.db.models.constants import LOOKUP_SEP
 from django.db.models.expressions import OrderBy, Random, RawSQL, Ref, Value
 from django.db.models.functions import Cast
-from django.db.models.query_utils import QueryWrapper, select_related_descend
+from django.db.models.query_utils import (
+    Q, QueryWrapper, select_related_descend,
+)
 from django.db.models.sql.constants import (
     CURSOR, GET_ITERATOR_CHUNK_SIZE, MULTI, NO_RESULTS, ORDER_DIR, SINGLE,
 )
 from django.db.models.sql.query import Query, get_order_dir
 from django.db.transaction import TransactionManagementError
 from django.db.utils import DatabaseError, NotSupportedError
+from django.utils.functional import cached_property
 from django.utils.hashable import make_hashable
 
 
@@ -1344,19 +1347,37 @@ class SQLInsertCompiler(SQLCompiler):
 
 
 class SQLDeleteCompiler(SQLCompiler):
+    @cached_property
+    def single_alias(self):
+        return sum(self.query.alias_refcount[t] > 0 for t in self.query.alias_map) == 1
+
+    def _as_sql(self, query):
+        result = [
+            'DELETE FROM %s' % self.quote_name_unless_alias(query.base_table)
+        ]
+        where, params = self.compile(query.where)
+        if where:
+            result.append('WHERE %s' % where)
+        return ' '.join(result), tuple(params)
+
     def as_sql(self):
         """
         Create the SQL for this query. Return the SQL string and list of
         parameters.
         """
-        assert len([t for t in self.query.alias_map if self.query.alias_refcount[t] > 0]) == 1, \
-            "Can only delete from one table at a time."
-        qn = self.quote_name_unless_alias
-        result = ['DELETE FROM %s' % qn(self.query.base_table)]
-        where, params = self.compile(self.query.where)
-        if where:
-            result.append('WHERE %s' % where)
-        return ' '.join(result), tuple(params)
+        if self.single_alias:
+            return self._as_sql(self.query)
+        innerq = self.query.clone()
+        innerq.__class__ = Query
+        innerq.clear_select_clause()
+        pk = self.query.model._meta.pk
+        innerq.select = [
+            pk.get_col(self.query.get_initial_alias())
+        ]
+        outerq = Query(self.query.model)
+        outerq.where = self.query.where_class()
+        outerq.add_q(Q(pk__in=innerq))
+        return self._as_sql(outerq)
 
 
 class SQLUpdateCompiler(SQLCompiler):

django/db/models/sql/subqueries.py

@@ -3,7 +3,6 @@ Query subclasses which provide extra functionality beyond simple data retrieval.
 """
 
 from django.core.exceptions import FieldError
-from django.db import connections
 from django.db.models.query_utils import Q
 from django.db.models.sql.constants import (
     CURSOR, GET_ITERATOR_CHUNK_SIZE, NO_RESULTS,
@@ -41,40 +40,6 @@ class DeleteQuery(Query):
             num_deleted += self.do_query(self.get_meta().db_table, self.where, using=using)
         return num_deleted
 
-    def delete_qs(self, query, using):
-        """
-        Delete the queryset in one SQL query (if possible). For simple queries
-        this is done by copying the query.query.where to self.query, for
-        complex queries by using subquery.
-        """
-        innerq = query.query
-        # Make sure the inner query has at least one table in use.
-        innerq.get_initial_alias()
-        # The same for our new query.
-        self.get_initial_alias()
-        innerq_used_tables = tuple([t for t in innerq.alias_map if innerq.alias_refcount[t]])
-        if not innerq_used_tables or innerq_used_tables == tuple(self.alias_map):
-            # There is only the base table in use in the query.
-            self.where = innerq.where
-        else:
-            pk = query.model._meta.pk
-            if not connections[using].features.update_can_self_select:
-                # We can't do the delete using subquery.
-                values = list(query.values_list('pk', flat=True))
-                if not values:
-                    return 0
-                return self.delete_batch(values, using)
-            else:
-                innerq.clear_select_clause()
-                innerq.select = [
-                    pk.get_col(self.get_initial_alias())
-                ]
-                values = innerq
-            self.where = self.where_class()
-            self.add_q(Q(pk__in=values))
-        cursor = self.get_compiler(using).execute_sql(CURSOR)
-        return cursor.rowcount if cursor else 0
-
 
 class UpdateQuery(Query):
     """An UPDATE SQL query."""

tests/delete/tests.py

@@ -535,9 +535,7 @@ class FastDeleteTests(TestCase):
         a = Avatar.objects.create(desc='a')
         User.objects.create(avatar=a)
         u2 = User.objects.create()
-        expected_queries = 1 if connection.features.update_can_self_select else 2
-        self.assertNumQueries(expected_queries,
-                              User.objects.filter(avatar__desc='a').delete)
+        self.assertNumQueries(1, User.objects.filter(avatar__desc='a').delete)
         self.assertEqual(User.objects.count(), 1)
         self.assertTrue(User.objects.filter(pk=u2.pk).exists())