子查询求和返回重复值：Django注解reserved_quantity总和的性能优化

开发者阿泽

2026-7-6

Django Product模型注解reserved_quantity总和去重优化问题

我需要为Product模型基于指定过滤条件注解reserved_quantity字段的总和。使用以下代码时，查询返回重复数据导致求和结果虚高，推测是反向外键关联条件listproduct__productionrequest__writtenoffwarehouseproducts=None的过滤引发的问题。

def annotate_total_reserved_by_opened_production_lists_quantity(self):
    return self.annotate(
        total_reserved_by_opened_production_lists_quantity=Coalesce(
            Subquery(
                WarehouseProductReservationMapping.objects.filter(
                    Q(
                        warehouseproduct__current_technology_id=Case(
                            When(
                                warehouseproduct__nomenclature__type__in=(
                                    product_structure_api.models.NomenclatureType.ASSEMBLY,
                                    product_structure_api.models.NomenclatureType.MANUFACTURED,
                                ),
                                then=OuterRef("current_technology_id"),
                            ),
                        )
                    )
                    | Q(
                        warehouseproduct__current_technology_id=None,
                        warehouseproduct__nomenclature__type=product_structure_api.models.NomenclatureType.PURCHASED,
                    ),
                    listproduct__for_list__is_completed=False,
                    warehouseproduct__nomenclature_id=OuterRef("nomenclature_id"),
                    listproduct__productionrequest__writtenoffwarehouseproducts=None,
                    warehouseproduct__quantity__gt=0,
                )
                .order_by()
                .values("warehouseproduct__nomenclature_id")
                .annotate(total_reserved_quantity=Sum("reserved_quantity"))
                .values("total_reserved_quantity")[:1]
            ),
            0,
        )
    )

我尝试过使用Sum(field, distinct=True)，以及在子查询的不同位置添加.distinct()和.distinct("id")，但均未解决问题。目前实现了以下代码（无重复值，结果正确），但性能下降约20倍，无法用于生产环境。

def annotate_total_reserved_by_opened_production_lists_quantity(self):
    # Get reservation ids to filter only unique entries.
    reservation_ids = WarehouseProductReservationMapping.objects.filter(
        Q(
            warehouseproduct__current_technology_id=Case(
                When(
                    warehouseproduct__nomenclature__type__in=(
                        product_structure_api.models.NomenclatureType.ASSEMBLY,
                        product_structure_api.models.NomenclatureType.MANUFACTURED,
                    ),
                    then=OuterRef(OuterRef("current_technology_id")),
                ),
            )
        )
        | Q(
            warehouseproduct__current_technology_id=None,
            warehouseproduct__nomenclature__type=product_structure_api.models.NomenclatureType.PURCHASED,
        ),
        Q(
            listproduct__for_list__is_completed=False,
            listproduct__productionrequest__writtenoffwarehouseproducts=None,
        )
        | Q(for_order_product__isnull=False),
        warehouseproduct__nomenclature_id=OuterRef(OuterRef("nomenclature_id")),
        warehouseproduct__quantity__gt=0,
    ).values("id")
    subquery = Subquery(
        WarehouseProductReservationMapping.objects.filter(id__in=reservation_ids)
        .order_by()
        .values("warehouseproduct__nomenclature_id")
        .annotate(total_reserved_quantity=Sum("reserved_quantity"))
        .values("total_reserved_quantity")[:1]
    )
    return self.annotate(
        total_reserved_by_opened_production_lists_quantity=Coalesce(
            subquery,
            0,
        )
    )

涉及的模型（仅保留使用到的字段）

class Product(models.Model):  # Base model to which I need to annotate sum.
    nomenclature = models.ForeignKey(
        product_structure_api.models.Nomenclature,
        on_delete=models.CASCADE,
        null=True,
        related_name="warehouse_product",
    )
    current_technology = models.ForeignKey(
        product_structure_api.models.Technology,
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
    )
    quantity = models.PositiveIntegerField(null=True, db_index=True)

class NomenclatureType(TextChoices):
    PURCHASED = "0", "Purchased"
    ASSEMBLY = "1", "Assembly"
    MANUFACTURED = "2", "Manufactured"


class Nomenclature(models.Model):
    type = models.CharField(
        max_length=1,
        choices=NomenclatureType.choices,
        default=NomenclatureType.ASSEMBLY,
        help_text="""
        "0" = "Purchased"
        "1" = "Assembly"
        "2" = "Manufactured"
        """,
        db_index=True,
    )

from mptt.models import MPTTModel


class ListProduct(MPTTModel):
    for_list = models.ForeignKey(
        List,
        on_delete=models.CASCADE,
        null=True,
        blank=True,
        related_name="forlistproduct_set",
    )

class List(models.Model):
    is_completed = models.BooleanField(default=False, db_index=True)

class ProductionRequest(models.Model):
    list_product = models.ForeignKey(
        ListProduct,
        on_delete=models.CASCADE,
        null=True,
        blank=True,
    )

class WrittenOffWarehouseProducts(models.Model):
    listproduct = models.ForeignKey(
        "ListProduct", on_delete=models.DO_NOTHING, null=True, blank=True
    )
    productionrequest = models.ForeignKey(
        "ProductionRequest", on_delete=models.SET_NULL, null=True, blank=True
    )
    warehouseproduct = models.ForeignKey(
        "Product",
        on_delete=models.CASCADE,
        null=True,
        blank=True,
    )

class WarehouseProductReservationMapping(models.Model):
    warehouseproduct = models.ForeignKey(
        "Product", on_delete=models.CASCADE, null=True, blank=True
    )
    reserved_quantity = models.PositiveIntegerField(
        default=1, db_index=True, validators=[MinValueValidator(1)]
    )

优化方案

问题根源

多表关联查询时产生了笛卡尔积，同一条WarehouseProductReservationMapping记录被多次匹配，导致求和时重复计算，结果虚高。

优化代码1：直接在子查询中去重后聚合

在子查询中先对WarehouseProductReservationMapping的id去重，确保每条记录只被计算一次，再执行求和聚合：

def annotate_total_reserved_by_opened_production_lists_quantity(self):
    return self.annotate(
        total_reserved_by_opened_production_lists_quantity=Coalesce(
            Subquery(
                WarehouseProductReservationMapping.objects.filter(
                    Q(
                        warehouseproduct__current_technology_id=Case(
                            When(
                                warehouseproduct__nomenclature__type__in=(
                                    product_structure_api.models.NomenclatureType.ASSEMBLY,
                                    product_structure_api.models.NomenclatureType.MANUFACTURED,
                                ),
                                then=OuterRef("current_technology_id"),
                            ),
                        )
                    )
                    | Q(
                        warehouseproduct__current_technology_id=None,
                        warehouseproduct__nomenclature__type=product_structure_api.models.NomenclatureType.PURCHASED,
                    ),
                    listproduct__for_list__is_completed=False,
                    warehouseproduct__nomenclature_id=OuterRef("nomenclature_id"),
                    listproduct__productionrequest__writtenoffwarehouseproducts=None,
                    warehouseproduct__quantity__gt=0,
                )
                # 去重，确保每条Reservation记录仅被统计一次
                .distinct("id")
                .order_by()
                .values("warehouseproduct__nomenclature_id")
                .annotate(total_reserved_quantity=Sum("reserved_quantity"))
                .values("total_reserved_quantity")[:1]
            ),
            0,
        )
    )

优化代码2：用Exists子查询筛选唯一记录

通过Exists子查询先筛选出符合条件的唯一WarehouseProductReservationMapping记录，再在外层聚合，避免关联产生的重复：

from django.db.models import Exists, OuterRef

def annotate_total_reserved_by_opened_production_lists_quantity(self):
    # 定义筛选符合条件的Reservation子查询
    valid_reservations = WarehouseProductReservationMapping.objects.filter(
        id=OuterRef("id"),
        Q(
            warehouseproduct__current_technology_id=Case(
                When(
                    warehouseproduct__nomenclature__type__in=(
                        product_structure_api.models.NomenclatureType.ASSEMBLY,
                        product_structure_api.models.NomenclatureType.MANUFACTURED,
                    ),
                    then=OuterRef(OuterRef("current_technology_id")),
                ),
            )
        )
        | Q(
            warehouseproduct__current_technology_id=None,
            warehouseproduct__nomenclature__type=product_structure_api.models.NomenclatureType.PURCHASED,
        ),
        listproduct__for_list__is_completed=False,
        warehouseproduct__nomenclature_id=OuterRef(OuterRef("nomenclature_id")),
        listproduct__productionrequest__writtenoffwarehouseproducts=None,
        warehouseproduct__quantity__gt=0,
    )

    return self.annotate(
        total_reserved_by_opened_production_lists_quantity=Coalesce(
            Subquery(
                WarehouseProductReservationMapping.objects.filter(Exists(valid_reservations))
                .order_by()
                .values("warehouseproduct__nomenclature_id")
                .annotate(total_reserved_quantity=Sum("reserved_quantity"))
                .values("total_reserved_quantity")[:1]
            ),
            0,
        )
    )

额外性能优化建议

给关联字段添加数据库索引，能大幅提升查询效率：

WarehouseProductReservationMapping的warehouseproduct、listproduct字段
ListProduct的for_list、productionrequest字段
WrittenOffWarehouseProducts的listproduct、productionrequest字段

内容的提问来源于stack exchange，提问作者Viktor Mironov

子查询求和返回重复值：Django注解reserved_quantity总和的性能优化

涉及的模型（仅保留使用到的字段） #

优化方案 #

问题根源 #

优化代码1：直接在子查询中去重后聚合 #

优化代码2：用Exists子查询筛选唯一记录 #