如何用Cursor循环更新多行？跨Schema数据迁移实现疑问

需求背景 #

有三个分属不同Schema的表：source_schema.target_table、target_A.target_table、target_B.target_table，需要把source_schema里的数据迁移到target_A和target_B。目标Schema由存储函数get_target_schema根据source_schema.target_table.target_column的值判断（返回A或B）。

之前试过把id和target_col_val复制到临时表循环处理，之后删除临时表顶部行，但因为频繁的SELECT和DELETE操作担心性能问题，就放弃了。现在有个未完成的游标实现代码，需要完善逻辑来实现多行更新和数据迁移。

修正后的游标实现代码 #

下面是完整的存储过程示例（以MySQL为例，不同数据库语法略有差异）：

DELIMITER //

CREATE PROCEDURE migrate_data()
BEGIN
    DECLARE done INT DEFAULT FALSE;
    DECLARE target_schema VARCHAR(100);
    DECLARE target_col_val VARCHAR(100);
    DECLARE tempid VARCHAR(100);
    -- 定义游标，读取源表的id和目标列值
    DECLARE cur CURSOR FOR 
        SELECT id, target_column 
        FROM source_schema.target_table;
    -- 添加循环终止的处理器
    DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = TRUE;

    OPEN cur;

    cur_loop: LOOP
        -- 从游标读取一行数据到变量
        FETCH cur INTO tempid, target_col_val;
        -- 没有数据就退出循环
        IF done THEN
            LEAVE cur_loop;
        END IF;

        -- 调用函数获取目标Schema，拼接成完整的Schema名（比如target_A）
        SET target_schema = CONCAT('target_', get_target_schema('target_table', 'target_column', target_col_val));

        -- 构造动态UPSERT语句，因为Schema是动态的，必须用动态SQL
        SET @sql = CONCAT(
            'INSERT INTO ', target_schema, '.target_table (id, name, etc) ',
            'SELECT id, name, etc FROM source_schema.target_table WHERE id = ? ',
            'ON DUPLICATE KEY UPDATE name = VALUES(name), etc = VALUES(etc)'
        );

        -- 准备并执行动态SQL
        PREPARE stmt FROM @sql;
        SET @tempid = tempid;
        EXECUTE stmt USING @tempid;
        DEALLOCATE PREPARE stmt;
    END LOOP cur_loop;

    CLOSE cur;
END //

DELIMITER ;

关键细节说明 #

• 游标基础修正：补上了FETCH语句读取数据，添加了NOT FOUND处理器，确保没有数据时能正常终止循环。
• 动态SQL必须性：目标Schema是通过函数计算出来的变量，静态SQL无法识别这种动态的Schema名，所以必须用动态SQL来构造执行语句。
• UPSERT适配：示例用的是MySQL的INSERT ... ON DUPLICATE KEY UPDATE实现更新插入，如果你用PostgreSQL，改成INSERT ... ON CONFLICT DO UPDATE；SQL Server用MERGE语句即可。
• 性能注意点：游标是逐行处理的，数据量极大时性能会受限，这种情况下可以考虑下面的批量处理方案。

批量处理优化方案（可选） #

如果数据量很大，逐行游标效率不够，可以改成批量读取处理，减少执行次数：

DELIMITER //

CREATE PROCEDURE migrate_data_batch()
BEGIN
    DECLARE done INT DEFAULT FALSE;
    DECLARE batch_size INT DEFAULT 1000; -- 每次处理1000条
    DECLARE target_schema VARCHAR(100);
    DECLARE temp_ids TEXT;
    DECLARE target_col_val VARCHAR(100);
    -- 按target_column分组批量读取id
    DECLARE cur CURSOR FOR 
        SELECT GROUP_CONCAT(id SEPARATOR ','), target_column 
        FROM source_schema.target_table 
        GROUP BY target_column 
        LIMIT batch_size;
    DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = TRUE;

    OPEN cur;

    cur_loop: LOOP
        FETCH cur INTO temp_ids, target_col_val;
        IF done THEN
            LEAVE cur_loop;
        END IF;

        SET target_schema = CONCAT('target_', get_target_schema('target_table', 'target_column', target_col_val));

        -- 构造批量UPSERT语句
        SET @sql = CONCAT(
            'INSERT INTO ', target_schema, '.target_table (id, name, etc) ',
            'SELECT id, name, etc FROM source_schema.target_table WHERE id IN (', temp_ids, ') ',
            'ON DUPLICATE KEY UPDATE name = VALUES(name), etc = VALUES(etc)'
        );

        PREPARE stmt FROM @sql;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;
    END LOOP cur_loop;

    CLOSE cur;
END //

DELIMITER ;

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