MariaDB中锁定与跳过锁定行问题及技术问询

数据库中有2条符合查询条件的数据行，并行执行两个相同查询时，第一个查询会锁定所有匹配行并返回1条结果，第二个查询则返回null。期望实现仅锁定查询返回的单一行，让第二个查询能正常获取剩下的那条数据。使用TypeORM编写代码，数据库为MariaDB v10.8.4。

当前代码与生成的SQL #

TypeORM代码 #

const queryRunner: QueryRunner = this.connect.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
const identificationProcess = await queryRunner.manager
            .getRepository(IdentificationProcessEntity)
            .createQueryBuilder("identification_process")
            .orderBy("identification_process.updated_at", "ASC")
            .where("identification_process.status = :status", { status: IdentificationStatusEnum.Scanning })
            .useTransaction(true)
            .setLock("pessimistic_write")
            .setOnLocked("skip_locked")
            .take(1)
            .getOne()

生成的SQL语句 #

SELECT `identification_process`.`uid` AS `identification_process_uid`, 
`identification_process`.`user_uid` AS `identification_process_user_uid`, 
`identification_process`.`status` AS `identification_process_status`, 
`identification_process`.`attempts` AS `identification_process_attempts`, 
`identification_process`.`expire_at` AS `identification_process_expire_at`, 
`identification_process`.`created_at` AS `identification_process_created_at`, 
`identification_process`.`updated_at` AS `identification_process_updated_at`, 
`identification_process`.`procedure_uid` AS `identification_process_procedure_uid` 
FROM `identification_process` `identification_process` 
WHERE `identification_process`.`status` = ? 
ORDER BY `identification_process`.`updated_at` ASC 
LIMIT 1 FOR UPDATE SKIP LOCKED 
-- PARAMETERS: ["Scanning"]

问题原因 #

MariaDB 10.8.x中，当FOR UPDATE SKIP LOCKED与ORDER BY、LIMIT配合使用时，数据库会先扫描所有符合WHERE条件的行进行排序，这个过程中会锁定所有扫描到的匹配行，而非仅最终返回的那一行。这就导致第二个并行查询没有可用行，只能返回null。

解决方案 #

通过子查询先定位目标行主键，再基于主键查询并加锁，确保仅锁定最终返回的单一行。

方案1：拆分两步查询 #

const queryRunner: QueryRunner = this.connect.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();

// 第一步：获取目标行的主键uid
const targetUidResult = await queryRunner.manager
    .getRepository(IdentificationProcessEntity)
    .createQueryBuilder("ip")
    .select("ip.uid")
    .where("ip.status = :status", { status: IdentificationStatusEnum.Scanning })
    .orderBy("ip.updated_at", "ASC")
    .take(1)
    .getRawOne();

if (!targetUidResult) {
    await queryRunner.rollbackTransaction();
    await queryRunner.release();
    return null;
}

// 第二步：基于主键查询并加锁，仅锁定该行
const identificationProcess = await queryRunner.manager
    .getRepository(IdentificationProcessEntity)
    .createQueryBuilder("identification_process")
    .where("identification_process.uid = :uid", { uid: targetUidResult.ip_uid })
    .useTransaction(true)
    .setLock("pessimistic_write")
    .setOnLocked("skip_locked")
    .getOne();

// 执行后续事务操作，最后提交/回滚
// await queryRunner.commitTransaction();
// await queryRunner.release();

方案2：嵌套子查询（单语句） #

const queryRunner: QueryRunner = this.connect.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();

const identificationProcess = await queryRunner.manager
    .getRepository(IdentificationProcessEntity)
    .createQueryBuilder("identification_process")
    .where((qb) => {
        const subQuery = qb.subQuery()
            .select("ip.uid")
            .from(IdentificationProcessEntity, "ip")
            .where("ip.status = :status", { status: IdentificationStatusEnum.Scanning })
            .orderBy("ip.updated_at", "ASC")
            .take(1)
            .getQuery();
        return `identification_process.uid IN ${subQuery}`;
    })
    .useTransaction(true)
    .setLock("pessimistic_write")
    .setOnLocked("skip_locked")
    .getOne();

// 执行后续事务操作
// await queryRunner.commitTransaction();
// await queryRunner.release();

生成的SQL类似：

SELECT `identification_process`.`uid` AS `identification_process_uid`, 
`identification_process`.`user_uid` AS `identification_process_user_uid`, 
`identification_process`.`status` AS `identification_process_status`, 
`identification_process`.`attempts` AS `identification_process_attempts`, 
`identification_process`.`expire_at` AS `identification_process_expire_at`, 
`identification_process`.`created_at` AS `identification_process_created_at`, 
`identification_process`.`updated_at` AS `identification_process_updated_at`, 
`identification_process`.`procedure_uid` AS `identification_process_procedure_uid` 
FROM `identification_process` `identification_process` 
WHERE `identification_process`.`uid` IN (
    SELECT `ip`.`uid` 
    FROM `identification_process` `ip` 
    WHERE `ip`.`status` = ? 
    ORDER BY `ip`.`updated_at` ASC 
    LIMIT 1
) 
FOR UPDATE SKIP LOCKED
-- PARAMETERS: ["Scanning"]

这种写法通过子查询先筛选出唯一目标行的主键，外层查询仅针对该主键加锁，避免了锁定所有匹配行的问题，并行查询时第二个请求就能正常获取剩余的那条数据。

内容的提问来源于stack exchange，提问作者Максим