如何用Django ORM实现Oracle的wm_concat行转列查询

错误原因 #

ORA-00937本质是查询中存在未被分组且未使用聚合函数的字段，或是自定义聚合函数生成的SQL不符合Oracle分组规则。LISTAGG作为Oracle聚合函数，必须配合正确的GROUP BY字段使用，Django ORM的分组逻辑需要严格对应。

正确实现步骤 #

1. 自定义支持Oracle LISTAGG的聚合函数 #

在Django中编写符合Oracle语法的聚合函数，确保生成正确的LISTAGG SQL语句：

from django.db.models import Aggregate, CharField

class OracleGroupConcat(Aggregate):
    function = 'LISTAGG'
    template = '%(function)s(%(expressions)s, %(delimiter)s) WITHIN GROUP (ORDER BY %(order_by)s)'
    allow_distinct = False

    def __init__(self, expression, delimiter=',', order_by=None, output_field=None, **extra):
        output_field = output_field or CharField()
        super().__init__(
            expression,
            delimiter=delimiter,
            order_by=order_by or expression,
            output_field=output_field,
            **extra
        )

    def as_oracle(self, compiler, connection):
        self.template = '%(function)s(%(expressions)s, \'%(delimiter)s\') WITHIN GROUP (ORDER BY %(order_by)s)'
        return super().as_sql(compiler, connection)

2. 正确执行分组查询 #

使用Django ORM的values()指定所有分组字段，再通过annotate()调用自定义聚合函数：

单字段分组（仅NAME）

from yourapp.models import StuTest

# 对应原生SQL: SELECT NAME, LISTAGG(PROJECT, ',') WITHIN GROUP (ORDER BY PROJECT) FROM STU_TEST GROUP BY NAME
result = StuTest.objects.values('name').annotate(
    project_concat=OracleGroupConcat('project')
).values('name', 'project_concat')

多字段分组（比如NAME + CLASS）

# 对应原生SQL: SELECT NAME, CLASS, LISTAGG(PROJECT, ',') WITHIN GROUP (ORDER BY PROJECT) FROM STU_TEST GROUP BY NAME, CLASS
result = StuTest.objects.values('name', 'class').annotate(
    project_concat=OracleGroupConcat('project')
).values('name', 'class', 'project_concat')

关键注意事项 #

• 分组字段必须全部声明：Django ORM会将values()中的字段作为GROUP BY的列，遗漏任何分组字段都会触发ORA-00937错误。
• 自定义参数调整：可根据需求修改分隔符和排序字段，比如OracleGroupConcat('project', delimiter=';', order_by='id')。
• 大文本处理：Oracle默认LISTAGG返回字符串长度受限于VARCHAR2最大值（4000字节），若需拼接超长内容，可改用XMLAGG实现无长度限制拼接：

class OracleGroupConcatLarge(Aggregate):
    function = 'XMLAGG'
    template = 'RTRIM(XMLAGG(XMLELEMENT(E, %(expressions)s, %(delimiter)s).EXTRACT(\'//text()\') ORDER BY %(order_by)s).GETCLOBVAL(), %(delimiter)s)'
    allow_distinct = False

    def __init__(self, expression, delimiter=',', order_by=None, output_field=None, **extra):
        output_field = output_field or CharField()
        super().__init__(
            expression,
            delimiter=delimiter,
            order_by=order_by or expression,
            output_field=output_field,
            **extra
        )

    def as_oracle(self, compiler, connection):
        self.template = "RTRIM(XMLAGG(XMLELEMENT(E, %(expressions)s, '%(delimiter)s').EXTRACT('//text()') ORDER BY %(order_by)s).GETCLOBVAL(), '%(delimiter)s')"
        return super().as_sql(compiler, connection)

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