如何检测随时间变化的表中删除行并同步更新目标表

我们有source和destination两张数据库表，source表会随时间执行DML操作发生变化。为便于演示，我们创建source_before和source_after两张表模拟source表的前后状态。

步骤1：搭建测试环境 #

创建并初始化模拟源表 #

-- 创建source_before，模拟source表初始状态
create table source_before (
id int,
name varchar2(40),
creation_time date
);

-- 插入初始数据
insert all 
into source_before values (1,'bola','01-Jan-20') 
into source_before values (2,'gol','02-Jan-21') 
into source_before values (3,'cav','02-Jan-23')
into source_before values (4,'bhf','02-Jan-25')
select * from dual;

-- 查询source_before数据
select * from source_before;

执行结果：

1   bola 01-JAN-20
2   gol 02-JAN-21
3   cav 02-JAN-23
4   bhf 02-JAN-25

-- 创建source_after，模拟source表变更后的状态
create table source_after (
id int,
name varchar2(40),
creation_time date
);

-- 插入变更后的数据
insert all 
into source_after values (1,'bola','01-Jan-20') 
into source_after values (2,'gol','02-Jan-21') 
into source_after values (5,'zzz','02-Jan-28')
into source_after values (6,'sss','02-Jan-25')
select * from dual;

-- 查询source_after数据
select * from source_after;

执行结果：

1   bola 01-JAN-20
2   gol 02-JAN-21
5   zzz 02-JAN-28
6   sss 02-JAN-25

创建并初始化目标表 #

-- 创建destination表，初始状态与source_before完全一致
create table destination (
id int,
name varchar2(40),
creation_time date
);

-- 插入初始数据
insert all 
into destination values (1,'bola','01-Jan-20') 
into destination values (2,'gol','02-Jan-21') 
into destination values (3,'cav','02-Jan-23')
into destination values (4,'bhf','02-Jan-25')
select * from dual;

-- 查询destination初始数据
select * from destination;

执行结果：

1   bola 01-JAN-20
2   gol 02-JAN-21
3   cav 02-JAN-23
4   bhf 02-JAN-25

步骤2：普通MERGE语句的局限 #

使用常规MERGE仅能处理新增和更新操作，无法自动删除destination中source_after不存在的行：

merge into destination d
using (select * from source_after) sa on (d.id = sa.id)
when matched then update
set 
d.name = sa.name,
d.creation_time = sa.creation_time 
when not matched then
insert (
d.id,
d.name,
d.creation_time
)
values
(
sa.id,
sa.name,
sa.creation_time
);

-- 查询执行后的destination数据
select * from destination;

执行结果：

1   bola 01-JAN-20
2   gol 02-JAN-21
3   cav 02-JAN-23
4   bhf 02-JAN-25
6   sss 02-JAN-25
5   zzz 02-JAN-28

可见新增行已插入，但id=3、4的冗余行未被删除，无法实现全量同步。

步骤3：实现全量同步并捕获变更记录 #

要完成完全同步，需同时处理更新、新增、删除操作，并捕获所有变更行。我们采用「MERGE处理新增/更新 + DELETE清理冗余行」的组合方案，配合RETURNING子句记录变更。

3.1 创建变更日志表 #

用于存储新增、删除的行记录：

create table change_log (
change_type varchar2(10), -- 标记变更类型：INSERT/DELETE
id int,
name varchar2(40),
creation_time date,
change_timestamp timestamp default systimestamp -- 记录变更时间
);

3.2 执行MERGE处理新增与更新，捕获新增行 #

declare
  type change_rec is record (
    id int,
    name varchar2(40),
    creation_time date
  );
  type change_tab is table of change_rec;
  v_changes change_tab;
begin
  merge into destination d
  using (select * from source_after) sa on (d.id = sa.id)
  when matched then update
  set d.name = sa.name, d.creation_time = sa.creation_time
  when not matched then
  insert (d.id, d.name, d.creation_time)
  values (sa.id, sa.name, sa.creation_time)
  returning d.id, d.name, d.creation_time bulk collect into v_changes;
  
  -- 将新增行写入变更日志
  forall i in 1..v_changes.count
    insert into change_log (change_type, id, name, creation_time)
    values ('INSERT', v_changes(i).id, v_changes(i).name, v_changes(i).creation_time);
    
  commit;
end;
/

3.3 删除冗余行，捕获删除记录 #

declare
  type change_rec is record (
    id int,
    name varchar2(40),
    creation_time date
  );
  type change_tab is table of change_rec;
  v_changes change_tab;
begin
  delete from destination d
  where not exists (select 1 from source_after sa where sa.id = d.id)
  returning d.id, d.name, d.creation_time bulk collect into v_changes;
  
  -- 将删除行写入变更日志
  forall i in 1..v_changes.count
    insert into change_log (change_type, id, name, creation_time)
    values ('DELETE', v_changes(i).id, v_changes(i).name, v_changes(i).creation_time);
    
  commit;
end;
/

3.4 验证同步结果 #

查询destination表，确认与source_after完全一致：

select * from destination;

结果：

1   bola 01-JAN-20
2   gol 02-JAN-21
5   zzz 02-JAN-28
6   sss 02-JAN-25

查询变更日志，查看捕获的记录：

select * from change_log;

结果示例：

INSERT  5   zzz 02-JAN-28  2024-05-20 15:30:00.123
INSERT  6   sss 02-JAN-25  2024-05-20 15:30:00.123
DELETE  3   cav 02-JAN-23  2024-05-20 15:31:00.456
DELETE  4   bhf 02-JAN-25  2024-05-20 15:31:00.456

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