如何合并多个PostgreSQL查询实现多实体标签搜索优化

我实现了一个通过标签查找关联实体的查询（用于带推荐功能的搜索栏），函数接收["partition","biography","masterclass"]数组作为搜索范围参数以避免硬编码。当前查询可正常运行，但每次需要构建三个独立查询，我希望将它们合并为一个以节省查询时间。

尝试过PostgreSQL的UNION ALL，但该语法要求每个SELECT语句的结果集列数一致，而我的三个实体表字段结构不同，无法满足这一要求。请问有没有更优的实现方式？

实体配置字典 #

objects = {
    "biography": {
        "table": biography_table,
        "object_id": "biography_id",
        "tag_table": biography_tag_table,
        "entity": Biography,
    },
    "masterclass": {
        "table": masterclass_table,
        "object_id": "masterclass_id",
        "tag_table": masterclass_tag_table,
        "entity": Masterclass,
    },
    "partition": {
        "table": partition_table,
        "object_id": "partition_id",
        "tag_table": partition_tag_table,
        "entity": Partition,
    },
}

当前实现代码 #

cte_query = []
for table in tables:
    object_table = objects[table]["table"]
    object_tag_table = objects[table]["tag_table"]
    object_id = objects[table]["object_id"]

    cte_query.append(
        sa.select(object_table, func.array_agg(tag_table.c.content).label("tags"))
        .select_from(
            object_table.join(
                object_tag_table,
                object_table.c.id == object_tag_table.c[object_id],
            ).join(tag_table, tag_table.c.id == object_tag_table.c.tag_id)
        )
        .group_by(object_table.c.id)
        .cte(f"{table}_cte")
    )

cte_partition, cte_biography, cte_masterclass = cte_query
print(cte_partition)
print(cte_biography)
print(cte_masterclass)

partition = sa.select(cte_partition).where(
    func.lower(func.array_to_string(cte_partition.c.tags, ",")).like(
        func.lower(f"%{search}%")
    )
)
masterclass = sa.select(cte_masterclass).where(
    func.lower(func.array_to_string(cte_masterclass.c.tags, ",")).like(
        func.lower(f"%{search}%")
    )
)
biography = sa.select(cte_biography).where(
    func.lower(func.array_to_string(cte_biography.c.tags, ",")).like(
        func.lower(f"%{search}%")
    )
)
result = []
result.append(conn.execute(partition).fetchall())
result.append(conn.execute(masterclass).fetchall())
result.append(conn.execute(biography).fetchall())

print(result)

结果示例 #

[[],
[(UUID('61fa9287-fd5b-49fc-8e91-037f6343b46d'), UUID('12345648-1234-1234-1234-123456789123'), 'masterclass', None, None, None, None, None, ['violon', 'flute'], 'created', datetime.datetime(2023, 7, 19, 21, 41, 55, 556527), None, UUID('12345648-1234-1234-1234-123456789123'), None, ['masterclass', 'violon', 'flute'])],
[(UUID('9bb75325-ad82-4170-908b-179e6fadd8b6'), 'flute', 'Boby', ['statement'], 'Française', 'http://www.gaines-johnson.com/', ['bag'], 'Number five region no power look. Energy government financial. Leave present nice.', 'compositor', 'created', None, datetime.datetime(2023, 7, 21, 14, 45, 19, 220057), None, UUID('12345648-1234-1234-1234-123456789123'), None, ['flute', 'boby', 'flute boby', 'statement'])]]

核心思路是统一结果集的列结构：给每个实体添加类型标识，将不同实体的字段打包为JSON对象，这样就能用UNION ALL合并查询，同时优化标签搜索的效率。

具体实现步骤 #

1. 统一列结构：每个子查询返回三列：
   • entity_type：字符串，标记实体类型（如'partition'）
   • entity_data：JSON对象，存储对应实体的所有字段
   • tags：标签数组（与原逻辑保持一致）
2. 优化标签搜索：用func.any()直接匹配数组中的标签，比array_to_string加模糊搜索效率更高；若需模糊匹配，可结合全文搜索实现。
3. 合并查询：用UNION ALL合并所有子查询，统一添加过滤条件，只需执行一次查询即可获取所有结果。

改写后的代码 #

from sqlalchemy import func, cast, JSON

# 构建每个实体的查询片段
union_queries = []
for entity_type, config in objects.items():
    object_table = config["table"]
    object_tag_table = config["tag_table"]
    object_id = config["object_id"]

    # 查询：实体类型 + 实体字段转JSON + 标签数组
    sub_query = sa.select(
        func.literal(entity_type).label("entity_type"),
        cast(object_table, JSON).label("entity_data"),
        func.array_agg(tag_table.c.content).label("tags")
    ).select_from(
        object_table.join(
            object_tag_table,
            object_table.c.id == object_tag_table.c[object_id]
        ).join(tag_table, tag_table.c.id == object_tag_table.c.tag_id)
    ).group_by(object_table.c.id)

    union_queries.append(sub_query)

# 合并所有子查询
combined_query = union_queries[0].union_all(*union_queries[1:])

# 添加标签过滤条件（精确匹配，如需模糊可调整）
filtered_query = sa.select(combined_query).where(
    func.lower(func.any(combined_query.c.tags)).like(func.lower(f"%{search}%"))
)

# 执行一次查询获取所有结果
result = conn.execute(filtered_query).fetchall()

# 根据entity_type映射回对应的实体类
mapped_result = []
for row in result:
    entity_cls = objects[row.entity_type]["entity"]
    mapped_result.append(entity_cls(**row.entity_data))

print(mapped_result)

关键说明 #

• JSON打包字段：PostgreSQL支持将整行记录转为JSON，cast(object_table, JSON)会自动将行字段转为键值对格式，完美解决列数不一致问题。
• 标签搜索优化：func.any(tags)直接检查关键词是否存在于标签数组中，可利用标签数组的GIN索引提升效率；若需前缀模糊匹配，可改用：

  func.exists(
      sa.select(1).where(func.lower(tag_table.c.content).like(func.lower(f"%{search}%")))
  )
  

• 结果映射：通过entity_type字段可区分不同实体，将JSON数据转回ORM类实例，方便后续业务处理。

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