Snowflake中如何按顺序对同一主键执行多DML操作？

在Snowflake中搭建持续数据复制任务时，需要按源系统事务的原顺序（以binlogkey排序）执行DML操作。使用MERGE语句时，同一主键存在多笔操作会出现异常：要么遗漏操作，要么抛出“DML操作期间检测到重复行”错误。

关键要求：事务必须严格按顺序执行，不能仅保留主键的最新事务（例如记录先INSERT后UPDATE，即使INSERT是临时状态，也必须先执行INSERT再执行UPDATE）。

示例数据表与MERGE代码：

create or replace table employee_source (
id int,
first_name varchar(255),
last_name varchar(255),
operation_name varchar(255),
binlogkey integer
);

create or replace table employee_destination ( id int, first_name varchar(255), last_name varchar(255) );

insert into employee_source values (1,'Wayne','Bells','INSERT',11);
insert into employee_source values (1,'Wayne','BellsT','UPDATE',12);
insert into employee_source values (2,'Anthony','Allen','INSERT',13);
insert into employee_source values (3,'Eric','Henderson','INSERT',14);
insert into employee_source values (4,'Jimmy','Smith','INSERT',15);
insert into employee_source values (1,'Wayne','Bellsa','UPDATE',16);
insert into employee_source values (1,'Wayner','Bellsat','UPDATE',17);
insert into employee_source values (2,'Anthony','Allen','DELETE',18); 

MERGE into employee_destination as T using (select * from employee_source order by binlogkey) 
AS S
ON T.id = s.id
when not matched
And S.operation_name = 'INSERT' THEN
INSERT (id,
first_name,
last_name)
VALUES (
S.id,    
S.first_name,
S.last_name)
when matched AND S.operation_name = 'UPDATE'
THEN
update set T.first_name = S.first_name, T.last_name = S.last_name
When matched
And S.operation_name = 'DELETE' THEN DELETE;

预期结果：employee_destination表中ID为1的员工姓氏应为Bellsat，ID为2的员工不应存在。

请问除MERGE外，是否有其他方法可实现按binlogkey顺序逐行执行每笔DML操作？

1. 生成逐行DML语句批量执行 #

通过拼接SQL语句，将employee_source的每条记录转换为对应INSERT/UPDATE/DELETE语句，按binlogkey排序后运行：

-- 生成DML语句
SELECT 
    CASE operation_name
        WHEN 'INSERT' THEN CONCAT('INSERT INTO employee_destination (id, first_name, last_name) VALUES (', id, ', ''', first_name, ''', ''', last_name, ''');')
        WHEN 'UPDATE' THEN CONCAT('UPDATE employee_destination SET first_name = ''', first_name, ''', last_name = ''', last_name, ''' WHERE id = ', id, ';')
        WHEN 'DELETE' THEN CONCAT('DELETE FROM employee_destination WHERE id = ', id, ';')
    END AS dml_statement
FROM employee_source
ORDER BY binlogkey;

将生成的语句直接批量执行，或结合存储过程自动执行。注意：若记录包含单引号等特殊字符，需手动或通过逻辑转义。

2. 存储过程循环逐行处理 #

编写Snowflake存储过程，按binlogkey顺序遍历源表记录，执行对应DML操作：

CREATE OR REPLACE PROCEDURE process_source_records()
RETURNS VARCHAR
LANGUAGE JAVASCRIPT
AS
$$
    var stmt = snowflake.createStatement({
        sqlText: "SELECT id, first_name, last_name, operation_name FROM employee_source ORDER BY binlogkey"
    });
    var rs = stmt.execute();
    
    while (rs.next()) {
        var id = rs.getColumnValue(1);
        var firstName = rs.getColumnValue(2).replace(/'/g, "''"); -- 转义单引号
        var lastName = rs.getColumnValue(3).replace(/'/g, "''");
        var op = rs.getColumnValue(4);
        
        var dmlStmt;
        if (op === 'INSERT') {
            dmlStmt = `INSERT INTO employee_destination (id, first_name, last_name) VALUES (${id}, '${firstName}', '${lastName}')`;
        } else if (op === 'UPDATE') {
            dmlStmt = `UPDATE employee_destination SET first_name = '${firstName}', last_name = '${lastName}' WHERE id = ${id}`;
        } else if (op === 'DELETE') {
            dmlStmt = `DELETE FROM employee_destination WHERE id = ${id}`;
        }
        
        snowflake.createStatement({sqlText: dmlStmt}).execute();
    }
    
    return '所有记录处理完成';
$$;

-- 调用存储过程
CALL process_source_records();

3. 流+任务实现持续增量复制 #

针对持续复制场景，用Snowflake流捕获源表新增记录，再通过定时任务按binlogkey顺序处理：

1. 创建流捕获employee_source的变更：

CREATE OR REPLACE STREAM employee_source_stream ON TABLE employee_source APPEND_ONLY = TRUE;

2. 创建任务，按顺序处理流中记录并执行DML：

CREATE OR REPLACE TASK process_stream_task
WAREHOUSE = YOUR_WAREHOUSE_NAME
SCHEDULE = 'USING CRON 0 * * * * UTC' -- 按需求设置调度频率
AS
CALL process_source_records_from_stream(); -- 需编写对应存储过程处理流数据

这种方式可自动处理增量数据，避免重复执行历史操作。

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