Django ORM高效标记患者首个PatientJourney的优化方案
高效标记每个患者的首个PatientJourney为Primary
问题背景
我们通过启发式规则对patient_journey查询集排序,patient_journey与patient为外键关联。需要将每个患者对应的首个patient_journey标记为True,其余标记为False。查询集已按patient_id分组排序,但现有实现始终无法将耗时控制在1秒内。
已尝试方案及性能问题
distinct+存在性检查:耗时额外增加1-2秒- 带
[:1]的子查询校验ID:耗时额外增加3-5秒
当前最优实现(仍有优化空间)
当前方案中get_sorted_for_primary_journey_qs耗时可忽略,主要开销来自distinct和Exists操作,核心逻辑是标记每个patient_id首次出现的记录为primary=True:
def annotate_primary( *, qs: QuerySet['PatientJourney'] # noqa ) -> QuerySet['PatientJourney']: # noqa """ 约束条件: -------------- 每个患者恰好有一个非全局的primary journey 注解字段: ------------ primary: Bool, 是否为患者的primary_journey """ from patient_journey.models import PatientJourney sorted_pjs = get_sorted_for_primary_journey_qs(qs=qs) sorted_pjs = sorted_pjs.distinct('patient_id').values('id') pjs = PatientJourney.objects.all().filter(id__in=sorted_pjs) qs = qs.annotate( primary=dm.Exists( pjs.filter( id=dm.OuterRef('id'), ) ) ) return qs
更高效的优化方案:使用窗口函数
利用Django的窗口函数RowNumber(),按patient_id分区并保持原排序,直接判断行号是否为1,仅需一次数据库查询,能大幅降低耗时:
from django.db.models import F, Window from django.db.models.functions import RowNumber def annotate_primary( *, qs: QuerySet['PatientJourney'] # noqa ) -> QuerySet['PatientJourney']: # noqa """ 约束条件: -------------- 每个患者恰好有一个非全局的primary journey 注解字段: ------------ primary: Bool, 是否为患者的primary_journey """ # 先通过启发式规则排序查询集 sorted_pjs = get_sorted_for_primary_journey_qs(qs=qs) # 使用窗口函数按patient_id分区,生成行号 qs = sorted_pjs.annotate( row_num=Window( expression=RowNumber(), partition_by=F('patient_id'), order_by=F('id').asc() # 此处order_by需与启发式排序规则完全匹配 ) ).annotate( primary=F('row_num') == 1 ) return qs
注意事项
order_by参数必须和get_sorted_for_primary_journey_qs中的排序规则一致,确保行号1对应每个患者的首个目标记录。- 窗口函数需要数据库版本支持(如PostgreSQL、MySQL 8.0+等),旧版数据库需调整实现逻辑。
内容的提问来源于stack exchange,提问作者Octavio del Ser
相关产品推荐
相关产品推荐

