Django ORM查询优化：移除注解过滤时的重复子查询

问题背景 #

使用PostgreSQL存储Camp模型，其中dates字段为JSON类型，存储字符串格式的日期数组。当前通过Subquery标注每个Camp的最新日期并过滤，但生成的SQL会重复执行同一子查询，导致性能损耗。需要纯Django ORM方案优化，禁止使用RawSQL。

优化方案1：用聚合函数替换排序取首行 #

将子查询中排序取第一个日期的逻辑，替换为直接用MAX()聚合计算最大日期，减少子查询执行次数，同时提升计算效率。

代码实现 #

from django.db.models import OuterRef, Subquery, Max, F
from django.utils.timezone import now

def get_future_camps():
    today_str = now().date().isoformat()

    # 子查询：聚合计算每个Camp的最大日期
    max_date_subquery = Subquery(
        Camp.objects.filter(id=OuterRef("id"))
        .annotate(extracted_dates=JsonbArrayElementsText(F("dates")))
        .values("id")  # 按Camp ID分组
        .annotate(max_date=Max("extracted_dates"))
        .values("max_date")
    )

    # 标注max_date后过滤符合条件的Camp
    return Camp.objects.annotate(max_date=max_date_subquery)\
                       .filter(max_date__gte=today_str)\
                       .only("id", "pk")

原理说明 #

• 子查询通过values("id")按Camp分组，再用Max("extracted_dates")直接聚合出每个Camp的最新日期，替代原有的排序取首行逻辑，计算更高效。
• 生成的SQL中，子查询只会执行一次，而非两次，彻底避免重复计算。

优化方案2：使用CTE（公共表表达式） #

借助Django的CTE特性，先一次性预计算所有Camp的最新日期，再基于CTE结果进行过滤，彻底避免子查询重复执行的问题，大数量场景下性能更优。

代码实现 #

from django.db.models import Max, F
from django.utils.timezone import now

def get_future_camps():
    today_str = now().date().isoformat()

    # 定义CTE：批量计算所有Camp的max_date
    cte = Camp.objects.annotate(
        extracted_dates=JsonbArrayElementsText(F("dates"))
    ).values("id").annotate(max_date=Max("extracted_dates")).alias()

    # 关联CTE并过滤未来日期的Camp
    return Camp.objects.with_cte(cte)\
                       .filter(id=cte.col.id, cte.col.max_date__gte=today_str)\
                       .only("id", "pk")

原理说明 #

• CTE会先执行一次聚合计算，生成包含所有Camp ID和对应max_date的临时表。
• 主查询直接关联CTE的临时表进行过滤，无需重复计算每个Camp的max_date，性能损耗大幅降低。

关键注意事项 #

• 两种方案均基于PostgreSQL的jsonb_array_elements_text函数，需确保JsonbArrayElementsText自定义Func正确实现。
• CTE方案要求Django版本≥3.2，若使用旧版Django，优先选择聚合Subquery方案。

内容的提问来源于stack exchange，提问作者Rizwan Niaz