SQLAlchemy关联查询返回重复数据问题排查与解决

需求说明 #

我有component和component_transform两张表，是一对多关联关系（一个Component对应多条ComponentTransform）。需要根据序列号搜索，可匹配component或component_transform表中的任一序列号，但无论匹配到多少条关联记录，仅返回component表中的唯一对应记录。

当前实现的search_components函数会返回重复记录：比如component表中序列号为"A"的记录，关联的component_transform表有序列号"B""C""D"，搜索其中任意序列号都应只返回A对应的Component记录，但实际返回多条重复的A记录。

实体类定义 #

class Component(BaseModel):
    __tablename__ = "component"

    component_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    component_serial_number: Mapped[str] = mapped_column(String(250), unique=True)
    component_transform: Mapped[List["ComponentTransform"]] = relationship(
    "ComponentTransform", back_populates="component"
)

class ComponentTransform(BaseModel):
    __tablename__ = "component_transform"

    transform_id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    component_id: Mapped[Optional[int]] = mapped_column(
        ForeignKey("component.component_id")
    )
    component_serial_number: Mapped[Optional[str]] = mapped_column(String(250))
    component: Mapped["Component"] = relationship(
    "Component", back_populates="component_transform"
)

当前查询函数代码 #

async def search_components(
    session: AsyncSession,
    component_serial_number: Optional[str] = None,
    component_name: Optional[str] = None,
    component_status: Optional[list[str]] = None,
    limit: int = 20,
    offset: int = 0,
) -> Sequence[ComponentSearchModel]:
    async with session:
        subquery_service_hrs = (
            select(
                DBMotorComponent.component_id,
                func.sum(DBMotorComponent.drilling_hrs).label("service_hrs"),
                func.sum(DBMotorComponent.drilling_hrs).label("life_hrs"),
            ).group_by(DBMotorComponent.component_id)
        ).subquery()

        statement = (
            select(DBComponent)
            .options(joinedload(DBComponent.part))
            .outerjoin(
                subquery_service_hrs,
                DBComponent.component_id == subquery_service_hrs.c.component_id,
            )
            .outerjoin(
                DBComponentTransform,
                DBComponent.component_id == DBComponentTransform.component_id,
            )
            .add_columns(
                subquery_service_hrs.c.service_hrs, subquery_service_hrs.c.life_hrs
            )
        )

        if component_serial_number is not None:
            statement = statement.where(
                or_(
                    DBComponent.component_serial_number.ilike(
                        f"%{component_serial_number}%"
                    ),
                    DBComponentTransform.component_serial_number.ilike(
                        f"%{component_serial_number}%"
                    ),
                )
            )

        if component_name is not None:
            statement = statement.where(
                DBComponent.component_name.ilike(f"%{component_name}%")
            )

        if component_status is not None:
            statement = statement.where(
                DBComponent.component_status.in_(component_status)
            )

        statement = statement.limit(limit).offset(offset)
        result = await session.execute(statement)
        components = result.fetchall()

        component_model = [
            ComponentSearchModel.model_validate(
                {
                    **component.__dict__,
                    "service_hrs": service_hrs or 0,
                    "life_hrs": life_hrs or 0,
                }
            )
            for component, service_hrs, life_hrs in components
        ]

    return component_model

尝试的解决方案及报错 #

• 使用distinct()去重时，Oracle返回错误：ORA-00932: inconsistent datatypes: expected - got CLOB
• 对DBComponent进行group_by时，返回错误：Not a group by expression

正确解决方案 #

问题根源在于直接outerjoin一对多关联的component_transform表，会导致每条Component记录被重复返回（对应关联的每条ComponentTransform）。而Oracle的distinct无法处理CLOB类型字段，group_by需要包含所有非聚合字段，操作成本极高。

修改方案：用EXISTS子句替代关联查询 #

通过EXISTS子句判断当前Component是否自身匹配序列号，或者关联的ComponentTransform中有匹配的序列号，无需实际关联表，避免产生重复记录。

具体修改如下：

1. 移除不必要的outerjoin(DBComponentTransform)
   不需要关联component_transform表，仅用EXISTS判断匹配条件即可。

2. 修改序列号查询条件
   将原来的or_条件替换为包含EXISTS子查询的逻辑：

if component_serial_number is not None:
    statement = statement.where(
        or_(
            DBComponent.component_serial_number.ilike(f"%{component_serial_number}%"),
            exists(
                select(1)
                .where(DBComponentTransform.component_id == DBComponent.component_id)
                .where(DBComponentTransform.component_serial_number.ilike(f"%{component_serial_number}%"))
            )
        )
    )

3. 完整修改后的查询语句部分

statement = (
    select(DBComponent)
    .options(joinedload(DBComponent.part))
    .outerjoin(
        subquery_service_hrs,
        DBComponent.component_id == subquery_service_hrs.c.component_id,
    )
    .add_columns(
        subquery_service_hrs.c.service_hrs, subquery_service_hrs.c.life_hrs
    )
)

原理说明 #

• EXISTS子句仅返回布尔结果，不会引入额外的记录行，因此查询结果中每条Component只会出现一次。
• 避免了join带来的笛卡尔积问题，同时绕过了Oracle对CLOB字段的distinct限制。

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