优化多对多关系基于行数过滤的Django查询性能

模型定义如下：

class RecipeTag(models.Model):
    tag = models.CharField(max_length=300, blank=True, null=True, default=None, db_index=True)
    recipes = models.ManyToManyField(Recipe, blank=True)
    description = models.TextField(blank=True, null=True, default=None)
    language_code = models.CharField(
        max_length=5,
        choices=[(lang[0], lang[1]) for lang in settings.LANGUAGES],
        default='en',
        db_index=True  # Index added here
    )
    slug = models.CharField(max_length=300, blank=True, null=True, default=None, db_index=True)  # Index added here
    is_visible = models.BooleanField(blank=True, null=True, default=True, db_index=True)  # Index added here

    def save(self, *args, **kwargs):
        if not self.slug:
            self.slug = slugify(self.tag)
        super(NewRecipeTag, self).save(*args, **kwargs)

    def __str__(self):
        tag = self.tag if self.tag else "Tag"
        return f"{tag} - Visible - {self.is_visible}"

需要筛选出关联Recipe数量≥2的RecipeTag，当前实现代码：

tags = NewRecipeTag.objects.filter(language_code=meta['language'],
    is_visible=True).annotate(num_recipes=Count('recipes')).filter(num_recipes__gte=2)

分页代码：

paginator = Paginator(recipe_tags, 60)
page = paginator.page(page_number)
recipe_tags = page.object_list

该查询耗时约20秒，移除annotate后查询几乎瞬间完成。现有数据量为4391166行，希望找到更高效的分页展示方案。

更新内容 #

实际模型名为NewRecipeTag，Recipe模型定义：

class Recipe(models.Model):
    author = models.ForeignKey(CustomUser, blank=True, null=True, on_delete=models.CASCADE, related_name='author', default=None)
    category = models.ForeignKey(Category, related_name="dish_category", blank=True, null=True, on_delete=models.SET_DEFAULT, default=None)
    level = models.ForeignKey(Category, related_name="category_level", blank=True, null=True, on_delete=models.SET_DEFAULT, default=None)
    time = models.ForeignKey(Category, related_name="time", blank=True, null=True, on_delete=models.SET_DEFAULT, default=None)
    website = models.ForeignKey(Website, blank=True, null=True, default=None, on_delete=models.CASCADE, related_name='recipes')
    language = models.CharField(
        max_length=5,
        choices=[('en', 'en'), ('uk', 'uk')],
        default='uk'
    )
    name = models.CharField(max_length=250, blank=True, null=True, default="")
    description = models.TextField(blank=True, null=True, default="")
    image = models.TextField(blank=True, null=True, default=None)
    filed_image = models.ImageField(blank=True, null=True, default=None, upload_to="recipes_photos/")
    url = models.URLField(blank=True, null=True, default="")
    created_at = models.DateTimeField(blank=True, null=True, default=timezone.now)
    updated_at = models.DateTimeField(auto_now=True)
    views = models.IntegerField(blank=True, null=True, default=0)
    slug = models.CharField(max_length=300, blank=True, null=True, default=None)
    sitemap_loc_pk = models.IntegerField(blank=True, null=True, default=None)
    is_active = models.BooleanField(blank=True, null=True, default=True)
    kitchen = models.ForeignKey(Category, related_name="kitchen", blank=True, null=True, on_delete=models.SET_DEFAULT, default=None)
    cook_methods = models.ForeignKey(Category, related_name="cook_methods", blank=True, null=True, on_delete=models.SET_DEFAULT, default=None)
    vegetarian = models.BooleanField(blank=True, null=True, default=False)
    hot = models.BooleanField(blank=True, null=True, default=False)
    dietary = models.ForeignKey(Category, related_name="category_dietary", blank=True, null=True,
                                  on_delete=models.SET_DEFAULT, default=None)

    type_of_food = models.ForeignKey(Category, related_name="category_type_of_food", blank=True, null=True,
                                  on_delete=models.SET_DEFAULT, default=None)

    season = models.ForeignKey(Category, related_name="category_season", blank=True, null=True,
                                  on_delete=models.SET_DEFAULT, default=None)

    food_group = models.ForeignKey(Category, related_name="category_food_group", blank=True, null=True,
                               on_delete=models.SET_DEFAULT, default=None)
    new_tags_generated = models.BooleanField(blank=True, null=True, default=False)

生成的SQL查询语句：

SELECT "recipe_newrecipetag"."id", "recipe_newrecipetag"."tag", "recipe_newrecipetag"."description", "recipe_newrecipetag"."seo_h1", "recipe_newrecipetag"."seo_title", "recipe_newrecipetag"."seo_description", "recipe_newrecipetag"."seo_text", "recipe_newrecipetag"."language_code", "recipe_newrecipetag"."slug", "recipe_newrecipetag"."is_visible", "recipe_newrecipetag"."sitemap_loc_pk", COUNT("recipe_newrecipetag_recipes"."recipe_id") AS "num_recipes"
FROM "recipe_newrecipetag" LEFT OUTER JOIN "recipe_newrecipetag_recipes" ON ("recipe_newrecipetag"."id" = "recipe_newrecipetag_recipes"."newrecipetag_id")
WHERE ("recipe_newrecipetag"."is_visible" AND "recipe_newrecipetag"."language_code" = uk)
GROUP BY "recipe_newrecipetag"."id"
HAVING COUNT("recipe_newrecipetag_recipes"."recipe_id") >= 2
ORDER BY "recipe_newrecipetag"."tag" ASC

EXPLAIN输出结果：

Sort  (cost=3706458.92..3710089.69 rows=1452305 width=1555)
  Sort Key: recipe_newrecipetag.tag
  ->  GroupAggregate  (cost=4576.12..609237.64 rows=1452305 width=1555)
        Group Key: recipe_newrecipetag.id
        Filter: (count(recipe_newrecipetag_recipes.recipe_id) >= 2)
        ->  Merge Left Join  (cost=4576.12..525178.10 rows=5919621 width=1555)
              Merge Cond: (recipe_newrecipetag.id = recipe_newrecipetag_recipes.newrecipetag_id)
              ->  Index Scan using recipe_newrecipetag_pkey on recipe_newrecipetag  (cost=0.43..228510.04 rows=4356915 width=1547)
                    Filter: (is_visible AND ((language_code)::text = 'uk'::text))
              ->  Index Only Scan using recipe_newrecipetag_reci_newrecipetag_id_recipe_i_87147e82_uniq on recipe_newrecipetag_recipes  (cost=0.43..211664.16 rows=5966157 width=16)

目前考虑使用请求缓存，但希望找到更优的性能优化方案。

1. 优化索引策略 #

• 添加复合索引：针对language_code、is_visible、tag字段创建复合索引，直接覆盖过滤+排序逻辑，消除高成本的Sort操作：

  class NewRecipeTag(models.Model):
      # ... 现有字段 ...
      class Meta:
          indexes = [
              models.Index(fields=['language_code', 'is_visible', 'tag']),
          ]
  

• 优化中间表索引：给中间表单独添加newrecipetag_id的普通索引，加速分组统计：

  CREATE INDEX idx_ntr_newrecipetag_id ON recipe_newrecipetag_recipes(newrecipetag_id);
  

2. 重写查询逻辑，减少全量扫描 #

先从中间表筛选出符合关联数要求的Tag ID，再关联主表查询，避免全量Left Join：

from django.db.models import Subquery, Count

# 先获取关联Recipe≥2的Tag ID列表
valid_tag_ids = NewRecipeTag.recipes.through.objects.values('newrecipetag_id').annotate(
    cnt=Count('recipe_id')
).filter(cnt__gte=2).values_list('newrecipetag_id', flat=True)

# 再筛选主表数据并排序
tags = NewRecipeTag.objects.filter(
    language_code=meta['language'],
    is_visible=True,
    id__in=Subquery(valid_tag_ids)
).order_by('tag')

3. 预计算关联数量，彻底消除分组统计 #

• 新增存储字段：在NewRecipeTag模型中添加recipe_count字段，存储关联Recipe的数量：

  class NewRecipeTag(models.Model):
      # ... 现有字段 ...
      recipe_count = models.IntegerField(default=0, db_index=True)
  

• 用信号维护字段值：在Tag与Recipe的关联关系变更时，自动更新recipe_count：

  from django.db.models.signals import m2m_changed
  from django.dispatch import receiver
  
  @receiver(m2m_changed, sender=NewRecipeTag.recipes.through)
  def update_recipe_count(sender, instance, action, **kwargs):
      if action in ['post_add', 'post_remove', 'post_clear']:
          instance.recipe_count = instance.recipes.count()
          instance.save(update_fields=['recipe_count'])
  

• 简化查询：直接用预计算字段过滤，查询效率大幅提升：

  tags = NewRecipeTag.objects.filter(
      language_code=meta['language'],
      is_visible=True,
      recipe_count__gte=2
  ).order_by('tag')
  

4. 分页优化，避免全量计数 #

如果不需要跳转指定页码，改用keyset分页（游标分页），避免Paginator的全量COUNT操作：

# 以上一页最后一条的tag和id作为游标
tags = NewRecipeTag.objects.filter(
    language_code=meta['language'],
    is_visible=True,
    recipe_count__gte=2,
    # 组合条件保证排序唯一性
    (Q(tag__gt=last_tag) | Q(tag=last_tag, id__gt=last_id))
).order_by('tag', 'id')[:60]

内容的提问来源于stack exchange，提问作者Artem Mashin