Peewee中EXISTS子查询在更新语句中未生效问题排查

我在Python应用中使用Peewee ORM操作SQLite数据库（采用APSW扩展），尝试实现批量更新记录的逻辑。单独执行子查询时能正确返回2条granule_id结果，但将其放入update语句的fn.EXISTS(sub_query)条件中时，数据库中所有记录都被更新，该条件未起到过滤作用。另外，由于存在相同granule_id的行，我需要获取最多batch_size个granule_id对应的记录，可能需要GROUP BY。

代码示例 #

def batch_logic(self, id_1, path_1, batch_size=1000, **kwargs):
    sub_query = (self.select(ModelClass.granule_id).distinct().where(
        (ModelClass.status == 'old_status') &
        (ModelClass.collection_id == collection_id) &
        (ModelClass.name.contains(provider_path))
    ).order_by(ModelClass.discovered_date.asc()).limit(batch_size)).limit(batch_size))
    print(f'len(sub_query): {len(sub_query)}')
    fb_st_2 = time.time()
    updated_records= list(
        (self.update(status='new_status').where(fn.EXISTS(sub_query)).returning(ModelClass))
    )
    print(f'update {len(updated_records)}: {time.time() - fb_st_2}')

    db.close()
    return updated_records

本地测试输出 #

id_1: id_1_1676475997_PQXYEQGJWR
len(sub_query): 2
update 20000: 1.0583274364471436
fetch_batch 20000: 1.1167597770690918
count_things 0: 0.02147078514099121
processed_things: 20000

解决方案 #

问题根源 #

你当前的EXISTS子查询是独立执行的，没有和主更新语句的表做关联。只要子查询返回非空结果，EXISTS就会一直返回True，导致所有行都被更新，完全起不到过滤作用。

修正方案 #

针对你的需求，推荐用in_替代EXISTS来匹配子查询返回的granule_id集合，写法更直观且符合业务逻辑。同时用GROUP BY确保获取最多batch_size个唯一的granule_id：

def batch_logic(self, id_1, path_1, batch_size=1000, **kwargs):
    # 获取最多batch_size个符合条件的唯一granule_id
    sub_query = (self.select(ModelClass.granule_id)
                 .where(
                     (ModelClass.status == 'old_status') &
                     (ModelClass.collection_id == collection_id) &
                     (ModelClass.name.contains(provider_path))
                 )
                 .group_by(ModelClass.granule_id)  # 按granule_id分组去重
                 .order_by(ModelClass.discovered_date.asc())
                 .limit(batch_size))
    
    print(f'符合条件的granule_id数量: {len(sub_query)}')
    
    fb_st_2 = time.time()
    # 仅更新granule_id在子查询结果中的记录
    updated_records = list(
        self.update(status='new_status')
            .where(ModelClass.granule_id.in_(sub_query))
            .returning(ModelClass)
    )
    
    print(f'更新记录数: {len(updated_records)}，耗时: {time.time() - fb_st_2}')
    
    db.close()
    return updated_records

若坚持使用EXISTS的写法 #

如果一定要用EXISTS，需要给主表创建别名，在子查询中添加关联条件，确保只检查当前行的granule_id是否在目标集合中：

def batch_logic(self, id_1, path_1, batch_size=1000, **kwargs):
    # 给主表创建别名用于关联
    main_alias = ModelClass.alias()
    
    # 子查询添加关联条件，匹配主表的granule_id
    sub_query = (self.select(ModelClass.granule_id)
                 .where(
                     (ModelClass.status == 'old_status') &
                     (ModelClass.collection_id == collection_id) &
                     (ModelClass.name.contains(provider_path)) &
                     (ModelClass.granule_id == main_alias.granule_id)
                 )
                 .group_by(ModelClass.granule_id)
                 .order_by(ModelClass.discovered_date.asc())
                 .limit(batch_size))
    
    print(f'符合条件的granule_id数量: {len(sub_query)}')
    
    fb_st_2 = time.time()
    updated_records = list(
        self.update(status='new_status')
            .from_(main_alias)  # 指定主表别名
            .where(fn.EXISTS(sub_query))
            .returning(ModelClass)
    )
    
    print(f'更新记录数: {len(updated_records)}，耗时: {time.time() - fb_st_2}')
    
    db.close()
    return updated_records

内容的提问来源于stack exchange，提问作者Cogito Ergo Sum