PostgreSQL带FILTER的COUNT子查询转SQLAlchemy性能优化问题

你性能劣化的根本原因是当前SQLAlchemy代码生成的SQL逻辑和你手写的原生SQL完全不同：

• 原生SQL仅对user_model做1次扫描，先通过外层WHERE过滤出符合条件的行，再通过FILTER语法在同一批行内统计不同条件的计数，全程仅扫描一次表。
• 你现有代码生成的SQL每个计数都是独立的子查询，每个子查询都会单独扫描一次user_model表，有N个统计字段就会扫N次表，性能自然会大幅下降。

优化方案 #

SQLAlchemy 1.4及以上版本原生支持PostgreSQL的聚合函数FILTER修饰符，无需额外编写子查询，直接给count函数追加.filter()方法即可，生成的SQL和你手写的原生SQL完全一致：

from sqlalchemy import func

search_term = "你的搜索词"
user_uuid = "20d7c90d-ebfa-4b04-9ee7-4fdedabc6c0b"

# 定义要统计的向量列和对应的别名
vector_config = [
    (user_model.email_vector, "email_vector"),
    (user_model.first_name_vector, "first_name_vector"),
    # 新增其他统计字段直接追加到列表即可
]

# 构造带FILTER的计数表达式
count_exprs = []
for vec_col, alias in vector_config:
    count_expr = func.count(user_model.uuid).filter(
        vec_col.op("@@")(func.parse_websearch(search_term))
    ).label(alias)
    count_exprs.append(count_expr)

# 构造主查询
query = db.session.query(*count_exprs).filter(
    user_model.uuid == user_uuid,
    user_model.all_vectors.op("@@")(func.parse_websearch(search_term))
)

# 执行查询
result = query.first()

如果你使用的是1.3及更早版本的SQLAlchemy，没有内置FILTER支持，可以用CASE表达式手动实现等价逻辑，执行效率和FILTER完全一致：

count_expr = func.count(
    func.case(
        [(vec_col.op("@@")(func.parse_websearch(search_term)), user_model.uuid)],
        else_=None
    )
).label(alias)

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