Django ORM annotate中如何对两个Rank窗口函数值求和解决SQLite报错

SQLite 对窗口函数的使用有严格限制：不允许在同一个SELECT子句中引用另一个窗口函数的计算结果，也不允许将窗口函数的输出作为参数，用于同一层级的其他计算或是另一个窗口函数的排序逻辑。
你之前的写法将所有字段都放在同一个annotate中，Django 会将它们生成到同一个SELECT子句里，因此触发了SQLite的限制抛出错误。

将查询拆分为两个层级：第一层先计算基础聚合字段和两个单项排名，第二层基于第一层的结果计算总排名分和总排名。可以通过子查询的方式实现，无需引入第三方依赖：

from django.db.models import Subquery, OuterRef, F, Count, Avg, Window, Rank, Cast

class ChampionKillManager(BaseManager):
    def get_queryset(self):
        # 第一层查询：计算基础聚合和两个单项排名
        base_queryset = super().get_queryset().annotate(
            length=Count('sequence'),
            avg_damage_contribution=Avg('sequence__damage_contribution'),
            avg_interval=Cast(Max('sequence__time') - Min('sequence__time'), models.FloatField()) / F('length'),
            rank_damage_contribution=Window(Rank(), order_by=F('avg_damage_contribution').desc()),
            rank_interval=Window(Rank(), order_by=F('avg_interval').asc()),
        )
        
        # 第二层查询：基于第一层的结果计算总得分和总排名
        return super().get_queryset().annotate(
            # 从子查询关联获取第一层计算的字段
            length=Subquery(base_queryset.filter(pk=OuterRef('pk')).values('length')[:1]),
            avg_damage_contribution=Subquery(base_queryset.filter(pk=OuterRef('pk')).values('avg_damage_contribution')[:1]),
            avg_interval=Subquery(base_queryset.filter(pk=OuterRef('pk')).values('avg_interval')[:1]),
            rank_damage_contribution=Subquery(base_queryset.filter(pk=OuterRef('pk')).values('rank_damage_contribution')[:1]),
            rank_interval=Subquery(base_queryset.filter(pk=OuterRef('pk')).values('rank_interval')[:1]),
            # 计算总排名分
            total_rank_score=F('rank_damage_contribution') + F('rank_interval'),
            # 生成总排名
            rank_total=Window(Rank(), order_by=F('total_rank_score').asc()),
        )

如果你使用的Django版本 >= 3.2，也可以使用CTE（公共表表达式）实现，性能比子查询更优：

from django.db.models import Count, Avg, Window, Rank, Cast, F
from django.db.models.expressions import CTE

class ChampionKillManager(BaseManager):
    def get_queryset(self):
        # 定义CTE存储基础聚合和两个单项排名
        base_cte = CTE(
            super().get_queryset().annotate(
                length=Count('sequence'),
                avg_damage_contribution=Avg('sequence__damage_contribution'),
                avg_interval=Cast(Max('sequence__time') - Min('sequence__time'), models.FloatField()) / F('length'),
                rank_damage_contribution=Window(Rank(), order_by=F('avg_damage_contribution').desc()),
                rank_interval=Window(Rank(), order_by=F('avg_interval').asc()),
            )
        )
        
        # 关联CTE计算总得分和总排名
        return super().get_queryset().join(
            base_cte, on=base_cte.col.pk == OuterRef('pk')
        ).annotate(
            length=base_cte.col.length,
            avg_damage_contribution=base_cte.col.avg_damage_contribution,
            avg_interval=base_cte.col.avg_interval,
            rank_damage_contribution=base_cte.col.rank_damage_contribution,
            rank_interval=base_cte.col.rank_interval,
            total_rank_score=F('rank_damage_contribution') + F('rank_interval'),
            rank_total=Window(Rank(), order_by=F('total_rank_score').asc()),
        ).with_cte(base_cte)

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