如何将目标SQL转换为Django ORM？当前ORM结果与原SQL不符

原SQL查询 #

SELECT
    * 
FROM
    ( 
        SELECT
            T_JOB_TASK.ID
            , T_JOB_TASK.JOB_DETAIL_ID
            , T_JOB_DETAIL.JOB_ID
            , ( 
                SELECT
                    T_INVOICE.ISSUE_FLAG 
                FROM
                    T_CASE 
                    INNER JOIN T_JOB 
                        ON T_JOB.ID = T_CASE.JOB_ID 
                    INNER JOIN T_CASE_REPORT_RELATION 
                        ON T_CASE_REPORT_RELATION.CASE_ID = T_CASE.ID 
                    INNER JOIN T_INVOICE 
                        ON T_INVOICE.ID = T_CASE_REPORT_RELATION.INVOICE_ID 
                WHERE
                    T_CASE.JOB_ID = T_JOB_DETAIL.JOB_ID
            ) AS INVOICE_ISSUE_FLAG
            , ( 
                SELECT
                    BOOL_OR(T_INVOICE_DATA_EXPORT_RESULT.INVOICE_ISSUE_FLAG) 
                FROM
                    T_INVOICE_DATA_EXPORT_RESULT_DETAIL 
                    INNER JOIN T_INVOICE_DATA_EXPORT_RESULT 
                        ON T_INVOICE_DATA_EXPORT_RESULT.ID =    T_INVOICE_DATA_EXPORT_RESULT_DETAIL.INVOICE_DATA_EXPORT_RESULT_ID
                WHERE
                    T_INVOICE_DATA_EXPORT_RESULT_DETAIL.JOB_ID = T_JOB_DETAIL.JOB_ID
            ) AS INVOICE_ISSUE_FLAG2 
        FROM
            T_JOB_TASK 
            INNER JOIN T_JOB_DETAIL 
                ON T_JOB_DETAIL.ID = T_JOB_TASK.ID 
        WHERE
            T_JOB_TASK.INSPECTION_FLAG = TRUE
    ) TMP 
WHERE
    NOT ( 
        COALESCE(INVOICE_ISSUE_FLAG, FALSE) 
        OR COALESCE(INVOICE_ISSUE_FLAG2, FALSE)
    );

原ORM代码问题分析 #

1. 子查询逻辑错误：第二个子查询未实现原SQL的BOOL_OR聚合逻辑，错误使用annotate和Coalesce，且添加了不必要的[:1]限制。
2. 过滤条件完全反转：原SQL是排除两个标志任意为True的记录，原ORM却保留这些记录。
3. 冗余数据处理：将查询结果转为列表循环提取ID，多此一举且可能引入数据不一致。
4. 空值处理缺失：未正确实现COALESCE(..., FALSE)的空值转False逻辑。

修正后的ORM代码 #

from django.db.models import Subquery, OuterRef, Q, Value, Coalesce, BoolOr

# 第一个子查询：匹配原SQL的INVOICE_ISSUE_FLAG子查询（模型关联名称需与实际定义一致）
invoice_issue_flag_subquery = Invoice.objects.filter(
    case_report_invoice__case__job__id=OuterRef("job_detail__job_id")
).values("issue_flag")[:1]  # 原SQL子查询若返回多行会报错，此处限制取第一条保持逻辑一致

# 第二个子查询：实现BOOL_OR聚合逻辑
invoice_issue_flag2_subquery = Subquery(
    InvoiceDataExportResultDetail.objects.filter(
        job_id=OuterRef("job_detail__job_id")
    )
    .values("invoice_data_export_result")  # 按关联的Result记录分组
    .annotate(flag=BoolOr("invoice_data_export_result__invoice_issue_flag"))
    .values("flag")[:1]
)

# 主查询：匹配原SQL的嵌套查询+过滤逻辑
filtered_job_tasks = JobTask.objects.filter(inspection_flag=True).annotate(
    # 注入两个子查询结果
    invoice_issue_flag=Subquery(invoice_issue_flag_subquery),
    invoice_issue_flag2=Subquery(invoice_issue_flag2_subquery)
).annotate(
    # 处理空值，转为False
    invoice_issue_flag=Coalesce("invoice_issue_flag", Value(False)),
    invoice_issue_flag2=Coalesce("invoice_issue_flag2", Value(False))
).filter(
    # 过滤掉任意标志为True的记录，等价于原SQL的NOT(A OR B)
    ~Q(invoice_issue_flag=True) & ~Q(invoice_issue_flag2=True)
)

# 获取对应的JobDetail记录
queryset = JobDetail.objects.filter(id__in=filtered_job_tasks.values("job_detail_id"))

关键修正点说明 #

• 聚合逻辑实现：使用Django 3.2+支持的BoolOr聚合函数，实现原SQL中BOOL_OR的统计功能，判断关联记录中是否存在任意True值。
• 过滤逻辑对齐：通过~Q反转条件，用NOT A AND NOT B等价实现原SQL的NOT(A OR B)逻辑。
• 空值处理：用Coalesce将子查询返回的空值转为False，与原SQL的COALESCE(..., FALSE)完全匹配。
• 简化数据处理：直接用values("job_detail_id")作为id__in的数据源，避免冗余循环操作。

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