SQL Server Merge仅更新变更数据并追踪变更的实现方案咨询

我们正在构建一套追踪Active Directory数据源随时间变更的解决方案，每小时获取数据快照并与基线对比，识别变更的同时将现有基线更新为新基线。计划用MERGE实现UPSERT，但不知道如何仅更新发生变更的数据，同时记录变更内容。数据规模约60列、数千行。

数据示例 #

快照数据 #

基线数据 #

更新后的新基线 #

预期变更记录 #

• 变更行：2（SN字段）、3（名称、其他字段）
• 新增行：4

当前尝试的问题 #

曾考虑逐行读取基线与快照对比再编写更新语句，但太繁琐。尝试在MERGE中加字段判断逻辑，但无法实现，也不知道如何记录变更内容。现有MERGE语句如下：

MERGE BASELINE AS TARGET
USING SNAPSHOT AS SOURCE
    ON (TARGET.[id] = SOURCE.[id])

WHEN MATCHED 
    THEN
        UPDATE
        --IF columns are not the same then update, else skip this row, but how?
        --This will always update the row, and cannot use multple matches
        SET TARGET.[shortName] = SOURCE.[shortName]

WHEN NOT MATCHED BY TARGET
    THEN
        INSERT (
            [Id]
            ,[Name]
            ,[shortName]
            ,[other]
            )
        VALUES (
            SOURCE.[Id]
            ,SOURCE.[Name]
            ,SOURCE.[shortName]
            ,source.[other]
            );

解决方案思路 #

1. 仅更新发生变更的行 #

在MERGE的WHEN MATCHED条件后添加字段差异判断，只有当任意字段不相等时才执行更新，避免无意义的行更新。

针对60列的场景，用EXCEPT简化判断逻辑，无需逐个列写条件：

MERGE BASELINE AS TARGET
USING SNAPSHOT AS SOURCE
    ON (TARGET.[id] = SOURCE.[id])

WHEN MATCHED 
    -- 判断当前行是否存在字段差异
    AND EXISTS (
        SELECT TARGET.* 
        EXCEPT 
        SELECT SOURCE.*
    )
    THEN
        UPDATE
        SET 
            TARGET.[名称] = SOURCE.[名称],
            TARGET.[SN] = SOURCE.[SN],
            TARGET.[其他] = SOURCE.[其他]
            -- 依次列出所有需要更新的60个字段

WHEN NOT MATCHED BY TARGET
    THEN
        INSERT (
            [Id], [名称], [SN], [其他]
            -- 对应所有60个字段
        )
        VALUES (
            SOURCE.[Id], SOURCE.[名称], SOURCE.[SN], SOURCE.[其他]
            -- 对应所有60个字段
        );

注：EXCEPT会自动处理NULL值的对比（NULL=NULL视为相等），比逐个写TARGET.col <> SOURCE.col OR (TARGET.col IS NULL AND SOURCE.col IS NOT NULL)更简洁。

如果需要排除某些不追踪变更的字段（比如最后更新时间），可以在SELECT中去掉这些字段：

AND EXISTS (
    SELECT TARGET.[Id], TARGET.[名称], TARGET.[SN], TARGET.[其他]
    EXCEPT
    SELECT SOURCE.[Id], SOURCE.[名称], SOURCE.[SN], SOURCE.[其他]
)

2. 记录变更内容 #

先创建一个变更记录表，用于存储每次同步的变更详情：

CREATE TABLE ChangeLog (
    LogId INT IDENTITY(1,1) PRIMARY KEY,
    ChangeType VARCHAR(10), -- 'UPDATE'或'INSERT'
    RecordId INT, -- 关联BASELINE的ID
    ChangedColumns NVARCHAR(MAX), -- 变更的字段列表
    OldValues NVARCHAR(MAX), -- 旧值（JSON格式）
    NewValues NVARCHAR(MAX), -- 新值（JSON格式）
    ChangeTime DATETIME DEFAULT GETDATE()
);

然后在MERGE语句中使用OUTPUT子句捕获变更，并插入到ChangeLog表中：

DECLARE @ChangeResults TABLE (
    ActionType NVARCHAR(10),
    TargetId INT,
    TargetName NVARCHAR(50),
    TargetSN NVARCHAR(50),
    TargetOther NVARCHAR(50),
    SourceName NVARCHAR(50),
    SourceSN NVARCHAR(50),
    SourceOther NVARCHAR(50)
);

MERGE BASELINE AS TARGET
USING SNAPSHOT AS SOURCE
    ON (TARGET.[id] = SOURCE.[id])

WHEN MATCHED 
    AND EXISTS (
        SELECT TARGET.* EXCEPT SELECT SOURCE.*
    )
    THEN
        UPDATE
        SET 
            TARGET.[名称] = SOURCE.[名称],
            TARGET.[SN] = SOURCE.[SN],
            TARGET.[其他] = SOURCE.[其他]
WHEN NOT MATCHED BY TARGET
    THEN
        INSERT ([Id], [名称], [SN], [其他])
        VALUES (SOURCE.[Id], SOURCE.[名称], SOURCE.[SN], SOURCE.[其他])
-- 捕获MERGE的操作结果
OUTPUT 
    $action AS ActionType,
    ISNULL(TARGET.[Id], SOURCE.[Id]) AS TargetId,
    TARGET.[名称] AS TargetName,
    TARGET.[SN] AS TargetSN,
    TARGET.[其他] AS TargetOther,
    SOURCE.[名称] AS SourceName,
    SOURCE.[SN] AS SourceSN,
    SOURCE.[其他] AS SourceOther
INTO @ChangeResults;

-- 将变更结果插入到ChangeLog表
INSERT INTO ChangeLog (ChangeType, RecordId, ChangedColumns, OldValues, NewValues)
SELECT
    ActionType,
    TargetId,
    -- 生成变更字段列表
    STRING_AGG(ColName, ', ') AS ChangedColumns,
    -- 旧值转为JSON
    (SELECT TargetName AS [名称], TargetSN AS [SN], TargetOther AS [其他] FOR JSON PATH, WITHOUT_ARRAY_WRAPPER) AS OldValues,
    -- 新值转为JSON
    (SELECT SourceName AS [名称], SourceSN AS [SN], SourceOther AS [其他] FOR JSON PATH, WITHOUT_ARRAY_WRAPPER) AS NewValues
FROM @ChangeResults
CROSS APPLY (
    VALUES
        ('名称', TargetName, SourceName),
        ('SN', TargetSN, SourceSN),
        ('其他', TargetOther, SourceOther)
) AS ChangedCols(ColName, OldVal, NewVal)
WHERE 
    (ActionType = 'UPDATE' AND OldVal <> NewVal)
    OR (ActionType = 'INSERT')
GROUP BY ActionType, TargetId, TargetName, TargetSN, TargetOther, SourceName, SourceSN, SourceOther;

针对60列的场景，可以通过动态SQL生成VALUES中的字段对比项，避免手动写60行。

3. 动态SQL优化（适配60列场景） #

如果字段较多，手动写所有字段的对比和更新会很繁琐，用动态SQL自动生成MERGE语句和变更字段对比逻辑：

DECLARE @Columns NVARCHAR(MAX), @UpdateSet NVARCHAR(MAX), @ChangeCols NVARCHAR(MAX);

-- 获取所有需要同步的字段（排除ID）
SELECT 
    @Columns = STRING_AGG(QUOTENAME(COLUMN_NAME), ', '),
    @UpdateSet = STRING_AGG('TARGET.' + QUOTENAME(COLUMN_NAME) + ' = SOURCE.' + QUOTENAME(COLUMN_NAME), ', '),
    @ChangeCols = STRING_AGG('(''' + COLUMN_NAME + ''', TARGET.' + QUOTENAME(COLUMN_NAME) + ', SOURCE.' + QUOTENAME(COLUMN_NAME) + ')', ', ')
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_NAME = 'BASELINE' AND COLUMN_NAME <> 'Id';

-- 生成MERGE语句
DECLARE @MergeSQL NVARCHAR(MAX) = N'
DECLARE @ChangeResults TABLE (
    ActionType NVARCHAR(10),
    TargetId INT,
    ' + REPLACE(@Columns, ', ', ' NVARCHAR(MAX), ') + ' NVARCHAR(MAX),
    Source_' + REPLACE(@Columns, ', ', ' NVARCHAR(MAX), Source_') + ' NVARCHAR(MAX)
);

MERGE BASELINE AS TARGET
USING SNAPSHOT AS SOURCE
    ON (TARGET.[Id] = SOURCE.[Id])

WHEN MATCHED 
    AND EXISTS (
        SELECT TARGET.* EXCEPT SELECT SOURCE.*
    )
    THEN
        UPDATE
        SET ' + @UpdateSet + '
WHEN NOT MATCHED BY TARGET
    THEN
        INSERT (Id, ' + @Columns + ')
        VALUES (SOURCE.Id, SOURCE.' + REPLACE(@Columns, ', ', ', SOURCE.') + ')
OUTPUT 
    $action AS ActionType,
    ISNULL(TARGET.[Id], SOURCE.[Id]) AS TargetId,
    TARGET.' + REPLACE(@Columns, ', ', ', TARGET.') + ',
    SOURCE.' + REPLACE(@Columns, ', ', ', SOURCE.') + '
INTO @ChangeResults;

INSERT INTO ChangeLog (ChangeType, RecordId, ChangedColumns, OldValues, NewValues)
SELECT
    ActionType,
    TargetId,
    STRING_AGG(ColName, '', '') AS ChangedColumns,
    (SELECT ' + REPLACE(@Columns, ', ', ' AS ' + QUOTENAME(COLUMN_NAME) + ', ') + ' AS ' + QUOTENAME(COLUMN_NAME) + ' FOR JSON PATH, WITHOUT_ARRAY_WRAPPER) AS OldValues,
    (SELECT Source_' + REPLACE(@Columns, ', ', ' AS ' + QUOTENAME(COLUMN_NAME) + ', Source_') + ' AS ' + QUOTENAME(COLUMN_NAME) + ' FOR JSON PATH, WITHOUT_ARRAY_WRAPPER) AS NewValues
FROM @ChangeResults
CROSS APPLY (
    VALUES ' + @ChangeCols + '
) AS ChangedCols(ColName, OldVal, NewVal)
WHERE 
    (ActionType = ''UPDATE'' AND OldVal <> NewVal)
    OR (ActionType = ''INSERT'')
GROUP BY ActionType, TargetId, TARGET.' + REPLACE(@Columns, ', ', ', TARGET.') + ', SOURCE.' + REPLACE(@Columns, ', ', ', SOURCE.') + ';
';

EXEC sp_executesql @MergeSQL;

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