Django高性能查询：按状态追溯周期并计算活跃持续天数

针对你千万级数据量、百万级独立reference的场景，结合Django 2.2 + PostgreSQL的特性，最靠谱的方案是用数据库层面的窗口函数定位每个reference的最新状态变更记录，再配合复合索引保障查询性能，最后计算活跃持续天数。

一、核心思路 #

我们需要完成以下几步：

1. 筛选出所有早于/等于查询日期的记录
2. 对每个reference的记录按start_date倒序排列，取最新的一条（即状态变更的最后一次记录）
3. 过滤出is_active符合目标状态的记录
4. 计算该记录的start_date到查询日期的天数差

二、代码实现 #

1. 基础查询代码

首先导入必要的模块：

from datetime import datetime
from django.db.models import Window, F, ExpressionWrapper, IntegerField
from django.db.models.functions import RowNumber, ExtractDay
from rest_framework import serializers, viewsets

然后编写查询逻辑（假设查询日期为2010-12-31，目标活跃状态为True）：

# 定义查询参数
query_date = datetime(2010, 12, 31)
target_active = True

# 步骤1：给每个reference的记录标记行号（按start_date倒序，行号1为最新记录）
annotated_records = SomeModel.objects.filter(start_date__lte=query_date).annotate(
    row_num=Window(
        expression=RowNumber(),
        partition_by=F('reference'),  # 按reference分组
        order_by=F('start_date').desc()  # 每组内按start_date倒序
    )
)

# 步骤2：筛选出每个reference的最新状态记录，且is_active符合要求
latest_status_records = annotated_records.filter(row_num=1, is_active=target_active)

# 步骤3：计算活跃状态持续天数
result = latest_status_records.annotate(
    active_status_days=ExpressionWrapper(
        ExtractDay(query_date - F('start_date')),
        output_field=IntegerField()
    )
).values('reference', 'start_date', 'is_active', 'active_status_days')

2. DRF序列化与视图

如果要在DRF中返回结果，编写对应的序列化器和视图：

# 序列化器
class StatusDurationSerializer(serializers.Serializer):
    reference = serializers.CharField(max_length=24)
    start_date = serializers.DateTimeField()
    is_active = serializers.BooleanField()
    active_status_days = serializers.IntegerField()

# 视图示例
class StatusDurationViewSet(viewsets.ReadOnlyModelViewSet):
    serializer_class = StatusDurationSerializer

    def get_queryset(self):
        query_date = self.request.query_params.get('query_date')
        target_active = self.request.query_params.get('is_active', 'true').lower() == 'true'
        
        # 转换query_date为datetime对象（注意处理时区，这里假设为UTC）
        query_date = datetime.fromisoformat(query_date)
        
        # 复用上面的查询逻辑
        annotated_records = SomeModel.objects.filter(start_date__lte=query_date).annotate(
            row_num=Window(
                expression=RowNumber(),
                partition_by=F('reference'),
                order_by=F('start_date').desc()
            )
        )
        latest_status_records = annotated_records.filter(row_num=1, is_active=target_active)
        return latest_status_records.annotate(
            active_status_days=ExpressionWrapper(
                ExtractDay(query_date - F('start_date')),
                output_field=IntegerField()
            )
        ).values('reference', 'start_date', 'is_active', 'active_status_days')

三、性能优化关键 #

由于数据量达数千万级，必须通过索引优化避免全表扫描：

1. 添加复合索引：在SomeModel的Meta类中添加针对reference和start_date的复合索引，让窗口函数的分组排序操作直接利用索引，无需额外排序：

class SomeModel(models.Model):
    reference = models.CharField(max_length=24, db_index=True)
    start_date = models.DateTimeField()
    is_active = models.BooleanField()

    class Meta:
        indexes = [
            # 按reference分组，start_date倒序的复合索引，完美匹配窗口函数的逻辑
            models.Index(fields=['reference', '-start_date']),
        ]

2. 只返回必要字段：用values()指定需要的字段，减少数据库传输的数据量，提升响应速度。
3. 避免Python层处理：所有逻辑都放在数据库层面执行，千万不要尝试在Python中循环处理每个reference，会导致性能灾难。

四、备选方案：子查询实现 #

如果你觉得窗口函数不够直观，也可以用子查询获取每个reference的最新start_date，再关联原表查询：

from django.db.models import Subquery, OuterRef, Max

# 子查询：获取每个reference在查询日期前的最大start_date
max_date_subquery = SomeModel.objects.filter(
    reference=OuterRef('reference'),
    start_date__lte=query_date
).values('reference').annotate(max_date=Max('start_date')).values('max_date')

# 筛选符合条件的记录并计算天数
result = SomeModel.objects.filter(
    start_date=Subquery(max_date_subquery),
    is_active=target_active
).annotate(
    active_status_days=ExpressionWrapper(
        ExtractDay(query_date - F('start_date')),
        output_field=IntegerField()
    )
).values('reference', 'start_date', 'is_active', 'active_status_days')

这个方案的性能和窗口函数方案相近，同样依赖上面提到的复合索引。

五、注意事项 #

• 时区处理：如果你的start_date是带时区的DateTimeField，确保query_date也使用带时区的datetime对象，避免时区转换错误。
• 边界情况：如果start_date等于查询日期，active_status_days会是0，若业务需要算1天，可以调整为ExtractDay(...) + 1。
• 性能测试：建议用django-debug-toolbar查看查询执行计划，确认索引被正确命中；也可以用EXPLAIN ANALYZE在PostgreSQL中直接测试查询性能。

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