基于Change Tracking的增量复制存在执行延迟问题求助

问题描述 #

利用Change Tracking实现数据库A（源端）到数据库B（目标端）的check_exist_Jan_2024表增量复制，已在两库创建同名表，并通过table_store_ChangeTracking_version存储最后同步的Change Tracking版本号，采用触发器+存储过程实现自动化同步。

出现异常：首次向源表插入数据，目标表无变化；第二次插入时，目标表出现第一次插入的行；第三次插入时，出现第二次的行，table_store_ChangeTracking_version的版本更新同样存在滞后。

相关SQL代码：

-- CREATE PROCEDURE for updating data in table_store_ChangeTracking_version
CREATE PROCEDURE Update_ChangeTracking_Version (@TableName varchar(50))
AS
BEGIN 
    DECLARE @Current_ChangeTracking_version BIGINT;
    SET @Current_ChangeTracking_version = (SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion)

    UPDATE table_store_ChangeTracking_version
    SET [SYS_CHANGE_VERSION] = @Current_ChangeTracking_version
    WHERE [TableName] = @TableName
END


-- CREATE PROCEDURE for Incremental copy data from source to another table
ALTER PROCEDURE Incremental_Copy_check_exist_Jan_2024 
AS
BEGIN
    DECLARE @Last_ChangeTracking_version BIGINT, @Current_ChangeTracking_version BIGINT;
    SET @Current_ChangeTracking_version = (SELECT CHANGE_TRACKING_CURRENT_VERSION() as CurrentChangeTrackingVersion);
    SET @Last_ChangeTracking_version = (select max(SYS_CHANGE_VERSION) as last_version
                                        from table_store_ChangeTracking_version
                                        where TableName = 'dbo.check_exist_Jan_2024');

    -- Inserted data or updated data
    SET IDENTITY_INSERT B.dbo.check_exist_Jan_2024 ON
    Insert into B.dbo.check_exist_Jan_2024 (LocationID, Name, CostRate, Availability)

    SELECT source_tbl.LocationID, source_tbl.Name, source_tbl.CostRate, source_tbl.Availability
    FROM A.dbo.check_exist_Jan_2024 AS source_tbl
    RIGHT OUTER JOIN changetable(changes A.dbo.check_exist_Jan_2024, @Last_ChangeTracking_version) AS ct
        ON ct.LocationID = source_tbl.LocationID
    WHERE ct.SYS_CHANGE_VERSION <= @Current_ChangeTracking_version 
      AND ct.SYS_CHANGE_OPERATION IN ('I', 'U')

    SET IDENTITY_INSERT B.dbo.check_exist_Jan_2024 OFF;

    -- Deleted data
    DELETE FROM B.dbo.check_exist_Jan_2024 
    WHERE LocationID IN (SELECT LocationID
                         FROM changetable(changes A.dbo.check_exist_Jan_2024, @Last_ChangeTracking_version) AS ct
                         WHERE ct.SYS_CHANGE_OPERATION = 'D');
END;

-- Create trigger when source table has been changed
ALTER TRIGGER Incremental_ChangeTracking_Trigger
ON A.dbo.check_exist_Jan_2024
AFTER INSERT, UPDATE, DELETE
AS 
BEGIN
    EXECUTE Incremental_Copy_check_exist_Jan_2024;

    EXECUTE Update_ChangeTracking_Version
            @TableName = 'dbo.check_exist_Jan_2024';
END

问题原因 #

1. 事务提交顺序与Change Tracking可见性冲突
   SQL Server的Change Tracking变更记录仅在事务提交后才会被changetable()查询到。而AFTER触发器是在源表操作的事务内部执行的，此时事务尚未提交，本次操作产生的变更记录还未写入Change Tracking存储，因此changetable()无法获取到本次的变更：

   • 首次插入时，触发器内的同步逻辑查不到本次插入的行，版本号也更新为旧值；事务提交后，变更记录才生成，但同步已经完成，导致目标表无数据。
   • 第二次插入时，触发器执行时，第一次的事务已提交，changetable()能查到第一次的变更，同步到目标表，版本号更新为第一次的版本，以此类推形成滞后。

2. 同步逻辑的版本范围限制冗余
   代码中WHERE ct.SYS_CHANGE_VERSION <= @Current_ChangeTracking_version是多余的，changetable(changes 表名, @LastVersion)本身就只会返回版本大于@LastVersion的变更，加上这个限制会导致如果@Current_ChangeTracking_version是旧版本（事务未提交时的版本），会漏掉部分变更。

3. 插入/更新逻辑存在主键冲突风险
   对于UPDATE操作，当前用INSERT语句会导致目标表中已存在的行重复插入，引发主键冲突。

解决方案 #

方案一：改用异步定时同步（推荐） #

放弃触发器实时同步，改用SQL Server代理作业定期执行同步存储过程，确保事务已提交，Change Tracking变更记录已生成：

1. 删除Incremental_ChangeTracking_Trigger触发器。
2. 创建SQL Server代理作业，定期执行Incremental_Copy_check_exist_Jan_2024存储过程，执行频率根据业务需求设置（如每分钟、每5分钟）。

方案二：调整触发器逻辑（实时同步场景） #

如果必须实时同步，通过Service Broker实现异步触发，确保同步逻辑在事务提交后执行：

1. 配置Service Broker，创建消息类型、契约、队列和服务。
2. 修改触发器，不再直接执行同步存储过程，而是向Service Broker发送同步消息。
3. 创建激活存储过程，当队列收到消息时，执行同步和版本更新逻辑。

代码优化（针对同步逻辑本身） #

无论采用哪种方案，都需要优化同步存储过程的逻辑：

-- 优化后的增量复制存储过程
ALTER PROCEDURE Incremental_Copy_check_exist_Jan_2024 
AS
BEGIN
    SET NOCOUNT ON;
    DECLARE @Last_ChangeTracking_version BIGINT, @Current_ChangeTracking_version BIGINT;

    -- 获取当前已提交的最新版本（确保事务已提交）
    SET @Current_ChangeTracking_version = CHANGE_TRACKING_CURRENT_VERSION();
    -- 获取上次同步的版本
    SET @Last_ChangeTracking_version = (SELECT MAX(SYS_CHANGE_VERSION) 
                                        FROM table_store_ChangeTracking_version
                                        WHERE TableName = 'dbo.check_exist_Jan_2024');

    -- 处理插入/更新：用MERGE避免主键冲突
    SET IDENTITY_INSERT B.dbo.check_exist_Jan_2024 ON;
    MERGE INTO B.dbo.check_exist_Jan_2024 AS target_tbl
    USING (
        SELECT source_tbl.LocationID, source_tbl.Name, source_tbl.CostRate, source_tbl.Availability
        FROM A.dbo.check_exist_Jan_2024 AS source_tbl
        INNER JOIN changetable(changes A.dbo.check_exist_Jan_2024, @Last_ChangeTracking_version) AS ct
            ON source_tbl.LocationID = ct.LocationID
        WHERE ct.SYS_CHANGE_OPERATION IN ('I', 'U')
    ) AS source_tbl
    ON target_tbl.LocationID = source_tbl.LocationID
    WHEN MATCHED THEN
        UPDATE SET Name = source_tbl.Name, CostRate = source_tbl.CostRate, Availability = source_tbl.Availability
    WHEN NOT MATCHED THEN
        INSERT (LocationID, Name, CostRate, Availability)
        VALUES (source_tbl.LocationID, source_tbl.Name, source_tbl.CostRate, source_tbl.Availability);
    SET IDENTITY_INSERT B.dbo.check_exist_Jan_2024 OFF;

    -- 处理删除
    DELETE FROM B.dbo.check_exist_Jan_2024 
    WHERE LocationID IN (
        SELECT LocationID
        FROM changetable(changes A.dbo.check_exist_Jan_2024, @Last_ChangeTracking_version) AS ct
        WHERE ct.SYS_CHANGE_OPERATION = 'D'
    );

    -- 更新同步版本号
    UPDATE table_store_ChangeTracking_version
    SET [SYS_CHANGE_VERSION] = @Current_ChangeTracking_version
    WHERE [TableName] = 'dbo.check_exist_Jan_2024';
END;

关键优化点：

• 用MERGE语句替代单纯的INSERT，同时处理更新和插入场景，避免主键冲突。
• 移除多余的ct.SYS_CHANGE_VERSION <= @Current_ChangeTracking_version限制，因为changetable()本身已经过滤出大于@Last_ChangeTracking_version的变更。
• 将版本更新逻辑整合到同步存储过程中，减少不必要的存储过程调用。

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