Django中含窗口函数的PostgreSQL SQL转Queryset问题咨询

1. 是否可以对窗口表达式结果进行过滤？ #

可以，但不能直接在窗口函数所在的查询层的filter中调用。SQL执行顺序为FROM/JOIN → WHERE → GROUP BY → 聚合函数 → HAVING → 窗口函数 → SELECT → DISTINCT → ORDER BY → LIMIT，窗口函数的计算在WHERE执行之后，直接写WHERE rn=1数据库无法识别rn字段，这也是Django抛出NotSupportedError的根本原因。过滤窗口结果必须把带窗口函数的查询封装成子查询/临时表，在外层查询完成过滤。

2. 将窗口查询改写为子查询时需要遵循哪些最佳实践？ #

• 不要在窗口子查询中添加和外层查询的主键关联条件：你之前写的qs2.filter(pk=OuterRef('pk'))会强制子查询每次仅计算外层当前匹配的单行数据，partition by和排序逻辑完全失效，所以所有行的rn都是1，这是结果错误的核心原因。
• 窗口子查询必须是独立的完整查询：要完整包含JOIN、过滤条件、窗口注解，计算出所有符合条件的行的rn值，不要提前做行级的外层关联限制。
• 字段对齐：子查询暴露的字段要和外层查询需要的字段完全匹配，涉及union操作时，两边查询的字段数量、字段类型、字段顺序要严格对应。
• 优先用.values()明确指定子查询返回的字段，避免隐式字段带来的类型不匹配问题。

3. 合理的Queryset实现 #

你之前的Queryset还遗漏了原SQL中与User表的INNER JOIN逻辑，也是结果不符合预期的原因之一，正确写法如下：

from django.db.models import Case, When, Value, IntegerField, Window, RowNumber

# 构造qs1：Column_C为空的部分
qs1 = Table.objects.filter(
    Column_C__isnull=True,
    Column_B__in=["UserA", "UserB", "UserC"]  # 替换为实际用户列表值
).annotate(
    rn=Value(0, output_field=IntegerField())
).values("Column_A", "Column_B", "Column_C", "rn")

# 构造带窗口函数的子查询：计算所有Column_C不为空的行的rn
window_subquery = Table.objects.filter(
    Column_C__isnull=False,
    Column_B__in=["UserA", "UserB", "UserC"]
).annotate(
    rn=Window(
        expression=RowNumber(),
        partition_by=["Column_C"],
        order_by=[
            Case(When(Column_B="UserA", then=Value(0)), default=Value(1), output_field=IntegerField()),
            "user__time_created"  # 替换为Table到User的实际外键字段名
        ]
    )
).values("pk", "rn")

# 构造qs2：关联子查询过滤rn=1的行
qs2 = Table.objects.filter(
    pk__in=window_subquery.filter(rn=1).values("pk")
).annotate(
    rn=Value(1, output_field=IntegerField())
).values("Column_A", "Column_B", "Column_C", "rn")

# 合并结果，all=True对应原生SQL的UNION ALL
return qs1.union(qs2, all=True)

如果使用Django 4.2及以上版本，也可以用QuerySet.from_queryset()直接把带窗口的查询作为临时表，写法更贴近原生SQL逻辑：

# 带窗口的完整查询
window_qs = Table.objects.filter(
    Column_C__isnull=False,
    Column_B__in=["UserA", "UserB", "UserC"]
).annotate(
    rn=Window(
        expression=RowNumber(),
        partition_by=["Column_C"],
        order_by=[
            Case(When(Column_B="UserA", then=0), default=1, output_field=IntegerField()),
            "user__time_created"
        ]
    )
).values("Column_A", "Column_B", "Column_C", "rn")

# 从临时表过滤rn=1的行
qs2 = Table.objects.from_queryset(window_qs, model=Table).filter(rn=1).values("Column_A", "Column_B", "Column_C", "rn")

return qs1.union(qs2, all=True)

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