多对多关系查询性能优化及SQLAlchemy实现咨询

表结构关系 #

三个表的关联逻辑如下：

------------     ----------------     -------------
| app_user |----<| user_comment |>----| user_post |
------------     ----------------     -------------

即app_user与user_post通过user_comment形成多对多关联：一个用户可评论多个帖子，一个帖子可被多个用户评论。

需求说明 #

给定时间戳，获取所有曾在该时间戳之前创建的帖子下发表过评论的用户的全部user_comment记录。

现有查询问题 #

当前使用四层嵌套子查询的SQL，性能冗余，语句如下：

SELECT *
FROM user_post
JOIN (
  SELECT * 
  FROM user_comment
  WHERE user_comment.app_user_id IN (  
    SELECT user_comment.app_user_id
    FROM user_comment
    WHERE user_comment.user_post_id IN (
      SELECT user_post.id
      FROM user_post
      WHERE user_post.created_at < '2022-01-02 00:00.00'
    )
  )
) AS fpr ON fpr.user_post_id = user_post.id;

优化后的SQL查询 #

方案1：使用EXISTS子查询（性能最优） #

直接定位符合条件的用户，再获取其所有评论，避免无意义的表关联：

SELECT uc.*
FROM user_comment uc
WHERE EXISTS (
    SELECT 1
    FROM user_comment uc2
    JOIN user_post up ON uc2.user_post_id = up.id
    WHERE up.created_at < '2022-01-02 00:00:00'
      AND uc2.app_user_id = uc.app_user_id
);

方案2：先筛选目标用户再关联评论 #

先通过关联筛选出符合条件的用户ID（去重），再关联获取评论，减少后续数据处理量：

SELECT uc.*
FROM user_comment uc
JOIN (
    SELECT DISTINCT uc.app_user_id
    FROM user_comment uc
    JOIN user_post up ON uc.user_post_id = up.id
    WHERE up.created_at < '2022-01-02 00:00:00'
) AS target_users ON uc.app_user_id = target_users.app_user_id;

SQLAlchemy ORM实现 #

假设已定义如下ORM模型：

from sqlalchemy import Column, Integer, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class AppUser(Base):
    __tablename__ = 'app_user'
    id = Column(Integer, primary_key=True)
    comments = relationship("UserComment", back_populates="user")

class UserPost(Base):
    __tablename__ = 'user_post'
    id = Column(Integer, primary_key=True)
    created_at = Column(DateTime)
    comments = relationship("UserComment", back_populates="post")

class UserComment(Base):
    __tablename__ = 'user_comment'
    id = Column(Integer, primary_key=True)
    app_user_id = Column(Integer, ForeignKey('app_user.id'))
    user_post_id = Column(Integer, ForeignKey('user_post.id'))
    user = relationship("AppUser", back_populates="comments")
    post = relationship("UserPost", back_populates="comments")

实现方案1（EXISTS方式） #

from sqlalchemy import exists, select
from datetime import datetime

target_date = datetime(2022, 1, 2)

# 子查询：筛选在目标日期前的帖子下评论过的用户ID
subquery = select(UserComment.app_user_id).join(UserPost).filter(UserPost.created_at < target_date)

# 主查询：获取这些用户的所有评论
query = select(UserComment).where(exists(subquery.where(UserComment.app_user_id == subquery.c.app_user_id)))

# 执行查询（需提前创建好session）
results = session.execute(query).scalars().all()

实现方案2（先筛选用户再关联） #

from sqlalchemy import distinct
from datetime import datetime

target_date = datetime(2022, 1, 2)

# 子查询：获取符合条件的去重用户ID
target_users_subquery = select(distinct(UserComment.app_user_id)).join(UserPost).filter(UserPost.created_at < target_date)

# 主查询：关联获取用户的所有评论
query = select(UserComment).join(
    target_users_subquery, 
    UserComment.app_user_id == target_users_subquery.c.app_user_id
)

# 执行查询
results = session.execute(query).scalars().all()

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