Snowflake MERGE报DML重复行错误 如何实现VARIANT字段JSON深度合并

每条消息对应唯一ID和若干属性，需要将同ID下的所有属性合并为单条完整消息。使用Snowflake MERGE语法实现时遇到三类问题：

• 首次运行通过row_number()窗口函数配合partition by筛选唯一记录插入，操作成功；第二次运行对多条同ID记录做属性更新时，报错：Error 3:Duplicate row detected during DML action Row Values: \n
• 尝试设置会话参数ERROR_ON_NONDETERMINISTIC_MERGE=FALSE后，运行结果不可靠，一致性无法保障
• 尝试用JavaScript UDF实现深度合并，数据量大时存在严重性能问题

复现代码 #

create or replace table test1 (Job_Id VARCHAR, RECORD_CONTENT VARIANT);
create or replace table test2 like test1;


insert into test1(JOB_ID, RECORD_CONTENT) select 1,parse_json('{
            "customer": "Aphrodite",
            "age": 32,
            "orders": {"product": "socks","quantity": 4, "price": "$6", "attribute1" : "a1"}                   
 }');
 
 
insert into test1(JOB_ID, RECORD_CONTENT) select 1,parse_json('{
            "customer": "Aphrodite",
            "age": 32,
            "orders": {"product": "shoe", "quantity": 2, "brand" : "Woodland","attribute2" : "a2"}
                        
 }');
 
 insert into test1(JOB_ID, RECORD_CONTENT) select 1,parse_json('{
            "customer": "Aphrodite",
            "age": 32,
            "orders": {"product": "shoe polish","brand" : "Helios", "attribute3" : "a3" }
                        
 }');


merge into test2 t2 using (
  select * from (select
    row_number() over(partition by JOB_ID order by JOB_ID desc) as rno, JOB_ID, RECORD_CONTENT
    from test1) where rno>1) t1 on   --1. 首次运行用rno=1筛选唯一值插入成功；2. 第二次运行用rno>1更新属性时报重复行错误
t1.JOB_ID = t2.JOB_ID
WHEN MATCHED THEN 
UPDATE 
SET t2.JOB_ID = t1.JOB_ID,
t2.RECORD_CONTENT = t1.RECORD_CONTENT
WHEN NOT MATCHED 
THEN INSERT (JOB_ID, RECORD_CONTENT) VALUES (t1.JOB_ID, t1.RECORD_CONTENT)

预期结果 #

同Job_Id下RECORD_CONTENT字段做JSON深度合并，保留所有属性键，重复键取最新插入记录的值，以上示例最终test2中Job_Id=1的记录应为：

{
  "customer": "Aphrodite",
  "age": 32,
  "orders": {"product": "shoe polish","quantity": 2, "brand" : "Helios","price": "$6", 
              "attribute1" : "a1","attribute2" : "a2","attribute3" : "a3" }
}

报错根因 #

MERGE语句要求USING子句返回的结果集中，用于和目标表匹配的关联键（此处为JOB_ID）必须唯一。原逻辑USING子句中同一个JOB_ID会返回rno=2、rno=3等多条记录，匹配目标表单条记录时，Snowflake无法确定用哪条记录更新，就会抛出重复行错误。设置ERROR_ON_NONDETERMINISTIC_MERGE=FALSE只是关闭报错，会随机选一条匹配记录更新，自然无法保证结果正确。

实现思路 #

放弃逐条MERGE更新的逻辑，先在源侧按JOB_ID分组完成所有JSON的深度合并，再把合并后的结果一次性和目标表做MERGE：既保证关联键唯一从根源避免报错，也用Snowflake原生VARIANT函数替代JS UDF解决性能问题，合并逻辑完全确定保障一致性。

可直接运行的代码 #

-- 第一步：给源表加顺序标记，确定记录新旧顺序，生产环境可替换为插入时间戳、自增主键等明确顺序字段
create or replace temporary table test1_with_seq as
select 
    Job_Id,
    RECORD_CONTENT,
    row_number() over (partition by Job_Id order by metadata$insert_row_key) as seq -- seq越大记录越新
from test1;

-- 第二步：递归CTE实现同Job_Id下JSON深度合并
with recursive merge_cte as (
    -- 锚点：取每个Job_Id最早的一条记录作为合并初始值
    select Job_Id, RECORD_CONTENT as merged_content, seq
    from test1_with_seq
    where seq = 1

    union all

    -- 递归：按seq从小到大依次合并，新记录的键覆盖旧记录同路径的键，不存在的键保留
    select 
        t.Job_Id,
        -- 先合并顶层属性，再单独合并嵌套的orders对象，有更多嵌套层可按相同逻辑扩展
        object_insert(
            object_concat(c.merged_content, t.RECORD_CONTENT),
            'orders',
            object_concat(c.merged_content:orders, t.RECORD_CONTENT:orders),
            true
        ) as merged_content,
        t.seq
    from merge_cte c
    join test1_with_seq t on c.Job_Id = t.Job_ID and t.seq = c.seq + 1
),
-- 取每个Job_Id合并完成的最终结果（seq最大的那条）
final_merged as (
    select Job_Id, merged_content as RECORD_CONTENT
    from (
        select *,
            row_number() over (partition by Job_Id order by seq desc) as rn
        from merge_cte
    )
    where rn = 1
)
-- 第三步：MERGE目标表，此时源表每个Job_Id唯一，不会触发重复行错误
merge into test2 t2
using final_merged t1
on t2.Job_Id = t1.Job_Id
when matched then
    update set t2.RECORD_CONTENT = t1.RECORD_CONTENT
when not matched then
    insert (Job_Id, RECORD_CONTENT) values (t1.Job_Id, t1.RECORD_CONTENT);

方案说明 #

• 性能：全程使用Snowflake原生SQL和内置VARIANT函数执行，执行效率比JS UDF高10~100倍，适配TB级大数据量场景
• 一致性：合并逻辑完全确定，重复键永远取最新记录的值，不存在随机选值问题
• 扩展性：如果JSON有固定多层嵌套，只需要在递归部分按object_insert+object_concat的逻辑扩展对应层级的合并即可；如果嵌套层级不固定，可以先用FLATTEN函数递归打平所有键的完整路径，按Job_Id分组取每个路径的最新值，再用OBJECT_AGG重组为嵌套JSON，全程依然可以用原生SQL实现
• 稳定性：USING子句输出的结果集每个JOB_ID唯一，从根源上避免MERGE重复行错误，不需要修改全局会话参数

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