如何在多分组统计中复用同一基础查询？（SQLAlchemy/PostgreSQL）

针对你想复用id in (1,2)的行匹配逻辑、避免PostgreSQL重复执行过滤操作的需求，我整理了两种实用方案，不管是原生SQL还是SQLAlchemy都能高效实现：

一、原生PostgreSQL SQL方案：用CTE复用过滤逻辑 #

最直接的方式是使用公共表表达式（CTE），它会先执行一次基础过滤查询并暂存结果，后续所有统计都基于这个临时数据集，彻底避免重复执行行匹配逻辑。

示例代码：

WITH filtered_stats AS (
    -- 只执行一次的基础过滤逻辑
    SELECT status, category FROM stats WHERE id IN (1, 2)
)
-- 统计status列的频次
SELECT 'status' AS stat_column, status AS value, COUNT(*) AS count
FROM filtered_stats
GROUP BY status
UNION ALL
-- 统计category列的频次
SELECT 'category' AS stat_column, category AS value, COUNT(*) AS count
FROM filtered_stats
GROUP BY category
ORDER BY stat_column, count DESC;

核心优势 #

• PostgreSQL仅执行一次filtered_stats中的过滤查询，后续统计直接基于已过滤好的结果集计算，大幅降低数据库负载
• 用UNION ALL将两个统计结果合并为统一输出，方便客户端处理；如果需要分开结果，也可以保留两个独立查询但共享同一个CTE

二、SQLAlchemy实现方案 #

不管你用SQLAlchemy Core还是ORM，都能轻松实现CTE复用逻辑：

1. SQLAlchemy Core版本 #

from sqlalchemy import create_engine, select, func, union_all
from sqlalchemy.schema import Table, MetaData

engine = create_engine("postgresql://user:pass@host/db")
metadata = MetaData()
stats = Table("stats", metadata, autoload_with=engine)

# 定义基础过滤CTE
filtered_cte = (
    select(stats.c.status, stats.c.category)
    .where(stats.c.id.in_([1, 2]))
    .cte("filtered_stats")
)

# 构建status统计查询
status_stats = select(
    func.literal("status").label("stat_column"),
    filtered_cte.c.status.label("value"),
    func.count().label("count")
).group_by(filtered_cte.c.status)

# 构建category统计查询
category_stats = select(
    func.literal("category").label("stat_column"),
    filtered_cte.c.category.label("value"),
    func.count().label("count")
).group_by(filtered_cte.c.category)

# 合并查询并执行
combined_stats = union_all(status_stats, category_stats).order_by("stat_column", "count desc")

with engine.connect() as conn:
    result = conn.execute(combined_stats)
    for row in result:
        print(row)

2. SQLAlchemy ORM版本（假设已有Stats模型） #

from sqlalchemy.orm import sessionmaker
from sqlalchemy import func, union_all
from models import Stats  # 替换为你的ORM模型路径

Session = sessionmaker(bind=engine)
session = Session()

# 定义CTE
filtered_cte = (
    session.query(Stats.status, Stats.category)
    .filter(Stats.id.in_([1, 2]))
    .cte("filtered_stats")
)

# 统计status列
status_q = session.query(
    func.literal("status").label("stat_column"),
    filtered_cte.c.status.label("value"),
    func.count().label("count")
).group_by(filtered_cte.c.status)

# 统计category列
category_q = session.query(
    func.literal("category").label("stat_column"),
    filtered_cte.c.category.label("value"),
    func.count().label("count")
).group_by(filtered_cte.c.category)

# 合并并执行查询
combined_q = union_all(status_q, category_q).order_by("stat_column", "count desc")
results = session.execute(combined_q).all()

for res in results:
    print(f"统计列: {res.stat_column}, 值: {res.value}, 频次: {res.count}")

额外优化建议 #

• 如果基础过滤结果集非常大，可以给CTE加上MATERIALIZED关键字（WITH filtered_stats AS MATERIALIZED (...)），强制将结果写入临时表，进一步提升大数据集下的统计性能
• 若不需要统一结果集，也可以基于同一个CTE执行多个独立查询，性能同样高效，因为CTE只会被执行一次

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