SQLAlchemy中not_in()子查询致应用挂起，单独执行正常

问题根源 #

SQL Server对大结果集的NOT IN语法优化能力极差，当子查询返回几十万行数据时，会触发低效的全表扫描+嵌套循环逻辑，直接导致查询雪崩。而将子查询结果转为Python列表传入，会瞬间占用大量内存（50万ID至少消耗数十MB内存，加上SQLAlchemy的额外处理开销，极易引发内存溢出崩溃）。

可行优化方案 #

1. 改用NOT EXISTS关联查询（首推） #

SQL Server对NOT EXISTS的执行计划优化远优于NOT IN，尤其是关联字段存在索引时，性能提升显著。

示例代码：

from sqlalchemy import exists

# 定义子查询关联逻辑（假设排除表为exclude_table，关联字段为exclude_id）
exclude_subquery = exists().where(exclude_table.c.exclude_id == main_table.c.target_id)
# 主查询用NOT EXISTS替代NOT IN
main_query = select(main_table).where(~exclude_subquery)

# 分批次读取结果，避免一次性加载全量数据到内存
with session.begin():
    for data_chunk in session.execute(main_query).yield_per(1000):
        # 处理单批次数据
        process_data(data_chunk)

关键前提：确保main_table.target_id和exclude_table.exclude_id都已创建非聚集索引，这是关联查询高效运行的核心。

2. 临时表+批量插入+LEFT JOIN排除 #

如果排除ID集合是固定的，可将子查询结果存入SQL Server临时表，再通过LEFT JOIN实现排除逻辑，利用数据库的索引优化能力提升效率。

示例代码：

from sqlalchemy import text

# 1. 创建临时表（SQL Server临时表以#开头，会话结束自动销毁）
session.execute(text("CREATE TABLE #ExcludeIDs (id INT PRIMARY KEY)"))

# 2. 批量插入排除ID到临时表（用参数化批量插入，避免逐行插入的低效）
exclude_ids = session.execute(select(exclude_table.c.exclude_id)).scalars()
session.execute(
    text("INSERT INTO #ExcludeIDs (id) VALUES (:id)"),
    [{"id": item} for item in exclude_ids]
)

# 3. 主查询通过LEFT JOIN排除临时表中的ID
main_query = select(main_table).\
    outerjoin(text("#ExcludeIDs"), main_table.c.target_id == text("#ExcludeIDs.id")).\
    where(text("#ExcludeIDs.id IS NULL"))

# 分批次处理查询结果
for data_chunk in session.execute(main_query).yield_per(1000):
    process_data(data_chunk)

临时表的主键会自动生成索引，关联查询效率极高，同时避免了Python端内存过载问题。

3. 分批次拆分排除ID（备选方案） #

若必须使用NOT IN，可将排除ID拆分为多个小批次，分多次执行主查询后合并结果，降低单批次内存占用。

示例代码：

from sqlalchemy import func

# 获取排除ID总数
total_exclude = session.query(func.count(exclude_table.c.exclude_id)).scalar()
batch_size = 10000  # 每批次处理1万条ID，可根据内存情况调整

final_results = []
for offset in range(0, total_exclude, batch_size):
    # 分批获取排除ID
    exclude_batch = session.execute(
        select(exclude_table.c.exclude_id).offset(offset).limit(batch_size)
    ).scalars().all()
    # 执行当前批次的主查询
    batch_query = select(main_table).where(main_table.c.target_id.not_in(exclude_batch))
    # 收集结果
    final_results.extend(session.execute(batch_query).scalars().all())

此方案适合内存受限场景，但总执行时间会高于前两种，因为需要多次发起查询。

通用优化原则 #

• 索引优先：所有关联字段必须创建索引，这是所有优化的基础。
• 避免全量加载：始终用yield_per()分批次读取结果，禁止一次性将几十万行数据加载到Python内存。
• 数据库端运算：尽量让数据库完成集合运算，不要将大量数据拉到Python端处理，数据库的集合运算效率远高于Python。

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