Refs #27149 -- Moved subquery expression resolving to Query.

This makes Subquery a thin wrapper over Query and makes sure it respects
the Expression source expression API by accepting the same number of
expressions as it returns. Refs #30188.

It also makes OuterRef usable in Query without Subquery wrapping. This
should allow Query's internals to more easily perform subquery push downs
during split_exclude(). Refs #21703.

2 files changed, 40 insertions, 3 deletions

diff --git a/django/db/models/sql/query.py b/django/db/models/sql/query.py
index ba4baca2b8..c192530573 100644
--- a/django/db/models/sql/query.py
+++ b/django/db/models/sql/query.py
@@ -21,7 +21,7 @@ from django.core.exceptions import (
 from django.db import DEFAULT_DB_ALIAS, NotSupportedError, connections
 from django.db.models.aggregates import Count
 from django.db.models.constants import LOOKUP_SEP
-from django.db.models.expressions import Col, F, Ref, SimpleCol
+from django.db.models.expressions import BaseExpression, Col, F, Ref, SimpleCol
 from django.db.models.fields import Field
 from django.db.models.fields.related_lookups import MultiColSource
 from django.db.models.lookups import Lookup
@@ -139,7 +139,7 @@ class RawQuery:
         self.cursor.execute(self.sql, params)
 
 
-class Query:
+class Query(BaseExpression):
     """A single SQL query."""
 
     alias_prefix = 'T'
@@ -232,6 +232,13 @@ class Query:
         self.explain_options = {}
 
     @property
+    def output_field(self):
+        if len(self.select) == 1:
+            return self.select[0].field
+        elif len(self.annotation_select) == 1:
+            return next(iter(self.annotation_select.values())).output_field
+
+    @property
     def has_select_fields(self):
         return bool(self.select or self.annotation_select_mask or self.extra_select_mask)
 
@@ -862,7 +869,7 @@ class Query:
             # No clashes between self and outer query should be possible.
             return
 
-        local_recursion_limit = 127  # explicitly avoid infinite loop
+        local_recursion_limit = 67  # explicitly avoid infinite loop
         for pos, prefix in enumerate(prefix_gen()):
             if prefix not in self.subq_aliases:
                 self.alias_prefix = prefix
@@ -997,6 +1004,21 @@ class Query:
                 not self.distinct_fields and
                 not self.select_for_update):
             clone.clear_ordering(True)
+        clone.where.resolve_expression(query, *args, **kwargs)
+        for key, value in clone.annotations.items():
+            resolved = value.resolve_expression(query, *args, **kwargs)
+            if hasattr(resolved, 'external_aliases'):
+                resolved.external_aliases.update(clone.alias_map)
+            clone.annotations[key] = resolved
+        # Outer query's aliases are considered external.
+        clone.external_aliases.update(
+            alias for alias, table in query.alias_map.items()
+            if (
+                isinstance(table, Join) and table.join_field.related_model._meta.db_table != alias
+            ) or (
+                isinstance(table, BaseTable) and table.table_name != table.table_alias
+            )
+        )
         return clone
 
     def as_sql(self, compiler, connection):
diff --git a/django/db/models/sql/where.py b/django/db/models/sql/where.py
index 9d3d6a9366..496822c58b 100644
--- a/django/db/models/sql/where.py
+++ b/django/db/models/sql/where.py
@@ -183,8 +183,23 @@ class WhereNode(tree.Node):
     def is_summary(self):
         return any(child.is_summary for child in self.children)
 
+    @staticmethod
+    def _resolve_rhs(rhs, query, *args, **kwargs):
+        if hasattr(rhs, 'resolve_expression'):
+            rhs = rhs.resolve_expression(query, *args, **kwargs)
+        return rhs
+
+    @classmethod
+    def _resolve_node(cls, node, query, *args, **kwargs):
+        if hasattr(node, 'children'):
+            for child in node.children:
+                cls._resolve_node(child, query, *args, **kwargs)
+        if hasattr(node, 'rhs'):
+            node.rhs = cls._resolve_rhs(node.rhs, query, *args, **kwargs)
+
     def resolve_expression(self, *args, **kwargs):
         clone = self.clone()
+        clone._resolve_node(clone, *args, **kwargs)
         clone.resolved = True
         return clone