Django中多过滤条件下Count与annotate的最优使用方案

场景背景 #

校园场景下需存储Class、Student、Book等大量数据，所有模型采用软删除（删除操作仅将is_deleted字段设为True），且均包含create_time、update_time、is_deleted字段。当前使用Django annotate+Count实现多条件计数的SQL查询耗时超30秒，需优化以支持同一模型多过滤条件的计数需求。

原低效代码示例：

# 注：各模型已实现软删除管理器查询
School.objects.annotate(
    number_of_class=Count(
        "class", filter=Q(class__is_deleted=False, class__create_time__lte=datetime, ...more filter), distinct=True
    ),
    number_of_student=Count(
        "student", filter=Q(student__is_deleted=False, student__create_time__lte=datetime, ...more filter), distinct=True
    ),
    number_of_public_book=Count(
        "book", filter=Q(book__is_deleted=False, book__create_time__lte=datetime, book__create_time__gte=datetime, book__type=1, ...more filter), distinct=True
    ),
    number_of_private_book=Count(
        "book", filter=Q(book__is_deleted=False, book__create_time__lte=datetime, book__create_time__gte=datetime, book__type__in=[2,3], ...more filter), distinct=True
    )
    # 更多计数字段...
)

优化方案 #

1. 替换Count为Subquery+ValueCount减少关联开销 #

原代码的annotate+Count会触发多表关联，大量数据下生成笛卡尔积导致性能暴跌。改用子查询直接统计关联表的符合条件记录数，避免主表与多张关联表的全量关联。

2. 复用软删除管理器过滤逻辑 #

利用已实现的软删除管理器（如Class.objects.active()）封装is_deleted=False过滤，避免重复写相同条件，保证逻辑统一。

3. 添加联合索引加速过滤查询 #

针对每个计数的过滤字段组合创建联合索引，覆盖过滤、关联、排序的所有字段，让查询走索引扫描而非全表扫描：

• Class模型：(is_deleted, create_time, school_id)
• Student模型：(is_deleted, create_time, school_id)
• Book模型：(is_deleted, school_id, create_time, type)

4. 拆分独立查询（极端大数量场景） #

若子查询仍无法满足性能要求，可拆分多个独立查询分别统计各计数，再将结果映射到主模型对象中，避免单条SQL过于复杂。

优化后的查询代码 #

方案1：Subquery实现高效计数 #

from django.db.models import Subquery, OuterRef, IntegerField, Count

# 班级计数子查询（复用软删除管理器）
class_subquery = Class.objects.active().filter(
    school=OuterRef('pk'),
    create_time__lte=datetime,
    # 其他过滤条件...
).values('school').annotate(count=Count('pk')).values('count')

# 学生计数子查询
student_subquery = Student.objects.active().filter(
    school=OuterRef('pk'),
    create_time__lte=datetime,
    # 其他过滤条件...
).values('school').annotate(count=Count('pk')).values('count')

# 公共书籍计数子查询
public_book_subquery = Book.objects.active().filter(
    school=OuterRef('pk'),
    create_time__range=[start_datetime, end_datetime],
    type=1,
    # 其他过滤条件...
).values('school').annotate(count=Count('pk')).values('count')

# 私人书籍计数子查询
private_book_subquery = Book.objects.active().filter(
    school=OuterRef('pk'),
    create_time__range=[start_datetime, end_datetime],
    type__in=[2,3],
    # 其他过滤条件...
).values('school').annotate(count=Count('pk')).values('count')

# 执行主查询
schools = School.objects.annotate(
    number_of_class=Subquery(class_subquery, output_field=IntegerField()),
    number_of_student=Subquery(student_subquery, output_field=IntegerField()),
    number_of_public_book=Subquery(public_book_subquery, output_field=IntegerField()),
    number_of_private_book=Subquery(private_book_subquery, output_field=IntegerField())
    # 更多计数字段同理扩展
).all()

# 处理无符合条件记录时的None值
for school in schools:
    school.number_of_class = school.number_of_class or 0
    school.number_of_student = school.number_of_student or 0
    school.number_of_public_book = school.number_of_public_book or 0
    school.number_of_private_book = school.number_of_private_book or 0

方案2：拆分查询合并结果（超大数据量场景） #

# 先获取所有学校对象
schools = School.objects.all()
school_ids = [s.pk for s in schools]

# 批量统计各学校班级数
class_counts = (
    Class.objects.active()
    .filter(school_id__in=school_ids, create_time__lte=datetime, ...more filter)
    .values('school_id')
    .annotate(count=Count('pk'))
)
class_count_map = {item['school_id']: item['count'] for item in class_counts}

# 批量统计学生数
student_counts = (
    Student.objects.active()
    .filter(school_id__in=school_ids, create_time__lte=datetime, ...more filter)
    .values('school_id')
    .annotate(count=Count('pk'))
)
student_count_map = {item['school_id']: item['count'] for item in student_counts}

# 批量统计公共书籍数
public_book_counts = (
    Book.objects.active()
    .filter(school_id__in=school_ids, create_time__range=[start_datetime, end_datetime], type=1, ...more filter)
    .values('school_id')
    .annotate(count=Count('pk'))
)
public_book_count_map = {item['school_id']: item['count'] for item in public_book_counts}

# 批量统计私人书籍数
private_book_counts = (
    Book.objects.active()
    .filter(school_id__in=school_ids, create_time__range=[start_datetime, end_datetime], type__in=[2,3], ...more filter)
    .values('school_id')
    .annotate(count=Count('pk'))
)
private_book_count_map = {item['school_id']: item['count'] for item in private_book_counts}

# 将统计结果映射到学校对象
for school in schools:
    school.number_of_class = class_count_map.get(school.pk, 0)
    school.number_of_student = student_count_map.get(school.pk, 0)
    school.number_of_public_book = public_book_count_map.get(school.pk, 0)
    school.number_of_private_book = private_book_count_map.get(school.pk, 0)

方案3：Conditional Aggregation简化同模型多条件计数 #

针对Book这类同一模型多过滤条件的场景，用Case+When+Sum替代多次Count，减少子查询数量：

from django.db.models import Case, When, Sum, IntegerField

# 合并书籍的两个计数为一次子查询
book_subquery = Book.objects.active().filter(
    school=OuterRef('pk'),
    create_time__range=[start_datetime, end_datetime]
).values('school').annotate(
    public_count=Sum(Case(When(type=1, then=1), default=0, output_field=IntegerField())),
    private_count=Sum(Case(When(type__in=[2,3], then=1), default=0, output_field=IntegerField()))
).values('public_count', 'private_count')

# 执行主查询
schools = School.objects.annotate(
    number_of_class=Subquery(class_subquery, output_field=IntegerField()),
    number_of_student=Subquery(student_subquery, output_field=IntegerField()),
    book_stats=Subquery(book_subquery, output_field=IntegerField())
).all()

# 拆分合并的书籍计数
for school in schools:
    school.number_of_public_book = school.book_stats.get('public_count', 0) if school.book_stats else 0
    school.number_of_private_book = school.book_stats.get('private_count', 0) if school.book_stats else 0
    # 其他计数字段处理同上

内容的提问来源于stack exchange，提问作者Chanchhunneng chrea