多表关联查询中如何避免STRING_AGG生成重复数据

数据库中，一部剧集（show）关联多个演员（actor）和多个流派（genre），属于典型的多对多关系。想要获取ID为1390的剧集全部详情、关联流派名称、演员ID及姓名，最初尝试的SQL因多表关联产生笛卡尔积，导致genres、actor_ids、actors字段出现大量重复值。目前已通过两次查询实现需求，但希望用单条SQL解决重复问题，同时寻求代码优化、多数据处理及文档字符串的建议。

最初尝试的SQL：

SELECT shows.*,
       STRING_AGG(g.name, ', ')                AS genres,
       STRING_AGG(CAST(a.id AS VARCHAR), ', ') AS actor_ids,
       STRING_AGG(a.name, ', ')                AS actors
FROM shows
         LEFT JOIN show_genres sg ON shows.id = sg.show_id
         LEFT JOIN genres g ON g.id = sg.genre_id
         LEFT JOIN show_characters sc ON shows.id = sc.show_id
         LEFT JOIN actors a ON sc.actor_id = a.id
WHERE shows.id = 1390
GROUP BY shows.id

当前使用的两次查询Python代码：

@connection_handler
def get_show_by_id(cursor: 'RealDictCursor', id: int) -> 'RealDictRow':
    """

    Args:
        cursor: a cursor which returns dictionaries (use @connection.connection_handler decorator)
        id: number of items shows on a single page

    Returns:
        All show details in a RealDictRow + genres(as a concatenated string) + actors in a list
        containing the actors ids and names in a RealDictRow

    """
    query = """
    SELECT shows.*,STRING_AGG(g.name,',') AS genres
    FROM shows
    LEFT JOIN show_genres sg ON shows.id = sg.show_id
    LEFT JOIN genres g ON sg.genre_id = g.id
    WHERE shows.id = %s
    GROUP BY shows.id
    """
    val = (id,)
    cursor.execute(query, val)
    shows = cursor.fetchone()

    query = """
    SELECT a.id,a.name
    FROM actors a
    RIGHT JOIN show_characters sc ON a.id = sc.actor_id
    RIGHT JOIN shows s ON sc.show_id = s.id
    WHERE s.id = %s
    GROUP BY a.id
    """
    val = (id,)
    cursor.execute(query, val)
    actors = cursor.fetchall()
    shows["actors"] = actors
    return shows

重复的根源是同时关联流派和演员表产生了笛卡尔积（一个流派对应多个演员，导致数据重复展开）。解决思路是先分别对流派、演员按剧集ID聚合，再与剧集表关联，避免交叉展开：

PostgreSQL 实现（推荐，支持JSON结构化返回） #

WITH show_genres_agg AS (
    -- 先聚合当前剧集的流派字符串
    SELECT sg.show_id, STRING_AGG(g.name, ', ') AS genres
    FROM show_genres sg
    JOIN genres g ON sg.genre_id = g.id
    WHERE sg.show_id = 1390
    GROUP BY sg.show_id
),
show_actors_agg AS (
    -- 聚合当前剧集的演员为JSON对象数组，便于直接转为Python列表
    SELECT sc.show_id, JSON_AGG(JSON_BUILD_OBJECT('id', a.id, 'name', a.name)) AS actors
    FROM show_characters sc
    JOIN actors a ON sc.actor_id = a.id
    WHERE sc.show_id = 1390
    GROUP BY sc.show_id
)
SELECT 
    s.*, 
    COALESCE(sga.genres, '') AS genres, -- 处理无流派的情况
    COALESCE(saa.actors, '[]') AS actors -- 处理无演员的情况
FROM shows s
LEFT JOIN show_genres_agg sga ON s.id = sga.show_id
LEFT JOIN show_actors_agg saa ON s.id = saa.show_id
WHERE s.id = 1390;

MySQL 实现 #

如果使用MySQL，可替换为JSON_OBJECT和JSON_ARRAYAGG：

WITH show_genres_agg AS (
    SELECT sg.show_id, GROUP_CONCAT(g.name SEPARATOR ', ') AS genres
    FROM show_genres sg
    JOIN genres g ON sg.genre_id = g.id
    WHERE sg.show_id = 1390
    GROUP BY sg.show_id
),
show_actors_agg AS (
    SELECT sc.show_id, JSON_ARRAYAGG(JSON_OBJECT('id', a.id, 'name', a.name)) AS actors
    FROM show_characters sc
    JOIN actors a ON sc.actor_id = a.id
    WHERE sc.show_id = 1390
    GROUP BY sc.show_id
)
SELECT 
    s.*, 
    IFNULL(sga.genres, '') AS genres,
    IFNULL(saa.actors, '[]') AS actors
FROM shows s
LEFT JOIN show_genres_agg sga ON s.id = sga.show_id
LEFT JOIN show_actors_agg saa ON s.id = saa.show_id
WHERE s.id = 1390;

这样查询返回的genres无重复，actors直接是结构化的对象数组，无需在Python中二次处理。

1. 修正参数描述错误：原文档字符串中id的描述是“number of items shows on a single page”，完全错误，应改为“ID of the target show to retrieve”。
2. 复用查询参数：两次查询的参数val=(id,)可以复用，避免重复定义。
3. 添加空值处理：如果查询不到对应ID的剧集，shows会是None，直接访问shows["actors"]会报错，需先判断：

   if shows is None:
       return None
   shows["actors"] = actors
   

4. 类型提示优化：如果导入了psycopg2.extras.RealDictCursor和psycopg2.extras.RealDictRow，可以直接用实际类型而非字符串：

   from psycopg2.extras import RealDictCursor, RealDictRow
   
   def get_show_by_id(cursor: RealDictCursor, id: int) -> RealDictRow | None:
   

5. 合并查询（使用单SQL）：如果采用上面的单条SQL，代码可以简化为一次查询，直接返回包含actors结构化数据的结果，无需两次执行：

   @connection_handler
   def get_show_by_id(cursor: RealDictCursor, id: int) -> RealDictRow | None:
       query = """
       WITH show_genres_agg AS (
           SELECT sg.show_id, STRING_AGG(g.name, ', ') AS genres
           FROM show_genres sg
           JOIN genres g ON sg.genre_id = g.id
           WHERE sg.show_id = %s
           GROUP BY sg.show_id
       ),
       show_actors_agg AS (
           SELECT sc.show_id, JSON_AGG(JSON_BUILD_OBJECT('id', a.id, 'name', a.name)) AS actors
           FROM show_characters sc
           JOIN actors a ON sc.actor_id = a.id
           WHERE sc.show_id = %s
           GROUP BY sc.show_id
       )
       SELECT s.*, COALESCE(sga.genres, '') AS genres, COALESCE(saa.actors, '[]') AS actors
       FROM shows s
       LEFT JOIN show_genres_agg sga ON s.id = sga.show_id
       LEFT JOIN show_actors_agg saa ON s.id = saa.show_id
       WHERE s.id = %s
       """
       val = (id, id, id)
       cursor.execute(query, val)
       return cursor.fetchone()
   

1. 批量查询剧集：如果需要获取多个剧集的详情，不要循环调用单条查询，改用IN子句批量聚合：

   WITH show_genres_agg AS (
       SELECT sg.show_id, STRING_AGG(g.name, ', ') AS genres
       FROM show_genres sg
       JOIN genres g ON sg.genre_id = g.id
       WHERE sg.show_id IN (%s, %s, %s) -- 替换为批量ID
       GROUP BY sg.show_id
   ),
   show_actors_agg AS (
       SELECT sc.show_id, JSON_AGG(JSON_BUILD_OBJECT('id', a.id, 'name', a.name)) AS actors
       FROM show_characters sc
       JOIN actors a ON sc.actor_id = a.id
       WHERE sc.show_id IN (%s, %s, %s)
       GROUP BY sc.show_id
   )
   SELECT s.*, COALESCE(sga.genres, '') AS genres, COALESCE(saa.actors, '[]') AS actors
   FROM shows s
   LEFT JOIN show_genres_agg sga ON s.id = sga.show_id
   LEFT JOIN show_actors_agg saa ON s.id = saa.show_id
   WHERE s.id IN (%s, %s, %s);
   

2. 使用ORM简化操作：比如用SQLAlchemy，通过ORM模型定义关联关系，直接查询即可自动处理多对多关联，无需手写复杂SQL：

   from sqlalchemy.orm import Session
   from models import Show, Genre, Actor
   
   def get_show_by_id(db: Session, id: int):
       show = db.query(Show).filter(Show.id == id).first()
       if not show:
           return None
       return {
           **show.__dict__,
           "genres": ", ".join([g.name for g in show.genres]),
           "actors": [{"id": a.id, "name": a.name} for a in show.actors]
       }
   

3. 分页处理大量数据：如果查询结果集过大，添加LIMIT和OFFSET进行分页，避免一次性加载过多数据。

遵循Google风格文档字符串，明确参数、返回值、异常情况，示例清晰：

@connection_handler
def get_show_by_id(cursor: RealDictCursor, show_id: int) -> RealDictRow | None:
    """Retrieve detailed information of a show by its ID, including associated genres and actors.

    Args:
        cursor: Database cursor returning dictionaries (managed by @connection_handler decorator)
        show_id: Unique ID of the show to retrieve

    Returns:
        A dictionary-like object (RealDictRow) containing:
            - All fields from the `shows` table
            - genres: Comma-separated string of associated genre names (empty string if none)
            - actors: List of dictionaries with 'id' and 'name' keys for associated actors
            Returns None if no show matches the given ID.

    Examples:
        >>> result = get_show_by_id(cursor, 1390)
        >>> result["title"]
        "Breaking Bad"
        >>> result["genres"]
        "Crime, Drama, Thriller"
        >>> result["actors"][0]
        {"id": 1, "name": "Bryan Cranston"}
    """

内容的提问来源于stack exchange，提问作者Kevin Németh