请求协助：将SQL Server CDC捕获的X至Y时段数据变更合并至启用CDC前的数据库备份

Got it, let’s walk through exactly how to set up CDC for all your tables and merge the post-backup changes into your time-X backup. I’ll focus on SQL Server since it has native CDC support—if you’re using a different database, just let me know and I can adjust the steps.

1. Create Database Backup at Time Point X #

First, take a consistent backup of your database at time X. Use COPY_ONLY if you don’t want this backup to interfere with your regular backup sequence (it won’t break the backup chain).

BACKUP DATABASE YourDatabaseName
TO DISK = 'C:\Backup\YourDB_Backup_X.bak'
WITH COPY_ONLY, COMPRESSION, INIT; -- INIT overwrites existing backup files on the disk

Critical: Verify the backup is valid immediately after creation to avoid headaches later:

RESTORE VERIFYONLY FROM DISK = 'C:\Backup\YourDB_Backup_X.bak';

2. Enable CDC for Database and All Tables #

CDC works at both the database and table level. First, enable it for the entire database, then loop through all user tables to enable CDC on each.

Enable Database-Level CDC #

Run this in your target database:

USE YourDatabaseName;
EXEC sys.sp_cdc_enable_db;

Check if it’s enabled successfully:

SELECT name, is_cdc_enabled FROM sys.databases WHERE name = 'YourDatabaseName';

Enable CDC for All User Tables #

Use a cursor to automate enabling CDC on every non-system table. Note: Tables need a primary key (or unique index) for CDC to work—if any tables don’t have one, you’ll need to adjust the script to specify an index.

USE YourDatabaseName;
DECLARE @TableName NVARCHAR(128);
DECLARE TableCursor CURSOR FOR
SELECT t.name
FROM sys.tables t
JOIN sys.schemas s ON t.schema_id = s.schema_id
WHERE t.is_ms_shipped = 0 -- Skip system tables
AND NOT EXISTS (
    SELECT 1 FROM sys.change_tracking_tables ctt WHERE ctt.object_id = t.object_id
);

OPEN TableCursor;
FETCH NEXT FROM TableCursor INTO @TableName;

WHILE @@FETCH_STATUS = 0
BEGIN
    DECLARE @EnableCmd NVARCHAR(MAX);
    SET @EnableCmd = N'EXEC sys.sp_cdc_enable_table 
        @source_schema = N''dbo'', -- Update schema if your tables use non-dbo schemas
        @source_name = N''' + @TableName + N''',
        @role_name = NULL, -- Leave NULL to allow all users with access to view CDC data
        @supports_net_changes = 1; -- Enables net change tracking for simpler merges';

    EXEC sp_executesql @EnableCmd;
    PRINT 'Enabled CDC for table: ' + @TableName;

    FETCH NEXT FROM TableCursor INTO @TableName;
END

CLOSE TableCursor;
DEALLOCATE TableCursor;

3. Merge Changes (Time X to Y) into the Backup #

Now you’ll restore the time-X backup to a staging database, then use CDC’s change tables to apply all changes between X and Y.

3.1 Restore the Backup to a Staging Database #

Never restore the backup over your production database. Create a staging instance instead:

RESTORE DATABASE YourDB_Staging
FROM DISK = 'C:\Backup\YourDB_Backup_X.bak'
WITH REPLACE,
MOVE 'YourDB_Data' TO 'C:\Data\YourDB_Staging.mdf', -- Adjust paths to your environment
MOVE 'YourDB_Log' TO 'C:\Logs\YourDB_Staging.ldf';

3.2 Map Time Range to CDC LSNs #

CDC tracks changes using Log Sequence Numbers (LSNs), not timestamps directly. Convert your X and Y time points to LSNs first:

USE YourDatabaseName;
DECLARE @StartLSN BINARY(10), @EndLSN BINARY(10);

-- Replace with your actual time X (backup creation time)
SET @StartLSN = sys.fn_cdc_map_time_to_lsn('smallest greater than or equal', '2024-01-01 00:00:00');

-- Replace with your actual time Y (when you want to stop capturing changes)
SET @EndLSN = sys.fn_cdc_map_time_to_lsn('largest less than or equal', '2024-01-07 17:00:00');

3.3 Apply Changes to Staging #

For each table, use CDC’s built-in function to get all changes between X and Y, then merge them into the staging table. Here’s an example for a Customers table:

USE YourDB_Staging;

MERGE INTO Customers AS Target
USING (
    SELECT 
        __$operation, -- 1=Delete, 2=Insert, 3=Update (before), 4=Update (after)
        CustomerID, Name, Email, Phone -- Include all columns from your table
    FROM YourDatabaseName.cdc.fn_cdc_get_all_changes_dbo_Customers(@StartLSN, @EndLSN, 'all')
) AS Source
ON Target.CustomerID = Source.CustomerID
WHEN MATCHED AND Source.__$operation = 1 THEN DELETE
WHEN MATCHED AND Source.__$operation = 4 THEN UPDATE SET
    Target.Name = Source.Name,
    Target.Email = Source.Email,
    Target.Phone = Source.Phone
WHEN NOT MATCHED AND Source.__$operation IN (2,4) THEN INSERT (CustomerID, Name, Email, Phone)
VALUES (Source.CustomerID, Source.Name, Source.Email, Source.Phone);

Automate Merge for All Tables

To avoid writing this for every table, generate dynamic SQL with a cursor:

USE YourDatabaseName;
DECLARE @SchemaName NVARCHAR(128), @TableName NVARCHAR(128);
DECLARE MergeCursor CURSOR FOR
SELECT s.name, t.name
FROM sys.tables t
JOIN sys.schemas s ON t.schema_id = s.schema_id
JOIN sys.change_tracking_tables ctt ON t.object_id = ctt.object_id
WHERE t.is_ms_shipped = 0;

OPEN MergeCursor;
FETCH NEXT FROM MergeCursor INTO @SchemaName, @TableName;

WHILE @@FETCH_STATUS = 0
BEGIN
    DECLARE @CaptureInstance NVARCHAR(128) = @SchemaName + '_' + @TableName;
    DECLARE @PKCols NVARCHAR(MAX), @AllCols NVARCHAR(MAX), @UpdateSet NVARCHAR(MAX);

    -- Get primary key columns for the table
    SELECT @PKCols = STRING_AGG(c.name, ', ')
    FROM sys.key_constraints kc
    JOIN sys.columns c ON kc.parent_object_id = c.object_id
    WHERE kc.type = 'PK' AND kc.parent_object_id = OBJECT_ID(@SchemaName + '.' + @TableName);

    -- Get all non-CDC system columns
    SELECT @AllCols = STRING_AGG(c.name, ', ')
    FROM sys.columns c
    WHERE c.object_id = OBJECT_ID(@SchemaName + '.' + @TableName)
    AND c.name NOT IN ('__$operation', '__$start_lsn', '__$end_lsn', '__$seqval', '__$update_mask');

    -- Build UPDATE SET clause (exclude PK columns)
    SELECT @UpdateSet = STRING_AGG('Target.' + c.name + ' = Source.' + c.name, ', ')
    FROM sys.columns c
    WHERE c.object_id = OBJECT_ID(@SchemaName + '.' + @TableName)
    AND c.name NOT IN (SELECT value FROM STRING_SPLIT(@PKCols, ', '))
    AND c.name NOT IN ('__$operation', '__$start_lsn', '__$end_lsn', '__$seqval', '__$update_mask');

    -- Generate full MERGE command
    DECLARE @MergeCmd NVARCHAR(MAX) = N'USE YourDB_Staging;
MERGE INTO ' + @SchemaName + '.' + @TableName + ' AS Target
USING (
    SELECT 
        __$operation,
        ' + @PKCols + ',
        ' + @AllCols + '
    FROM YourDatabaseName.cdc.fn_cdc_get_all_changes_' + @CaptureInstance + '(@StartLSN, @EndLSN, ''all'')
) AS Source
ON Target.' + REPLACE(@PKCols, ', ', ' = Source.') + '
WHEN MATCHED AND Source.__$operation = 1 THEN DELETE
WHEN MATCHED AND Source.__$operation = 4 THEN UPDATE SET
    ' + @UpdateSet + '
WHEN NOT MATCHED AND Source.__$operation IN (2,4) THEN INSERT (' + @AllCols + ')
VALUES (' + REPLACE(@AllCols, ', ', ', Source.') + ');';

    PRINT @MergeCmd; -- Print to review, or uncomment below to execute directly
    -- EXEC sp_executesql @MergeCmd, N'@StartLSN BINARY(10), @EndLSN BINARY(10)', @StartLSN, @EndLSN;

    FETCH NEXT FROM MergeCursor INTO @SchemaName, @TableName;
END

CLOSE MergeCursor;
DEALLOCATE MergeCursor;

Key Notes & Troubleshooting #

• SQL Server Agent: Must be running before enabling CDC—CDC uses agent jobs to capture and process changes in the background.
• Permissions: You need ALTER ANY DATABASE rights to enable database CDC, and ALTER rights on each table to enable table CDC.
• LSN Validation: Use sys.fn_cdc_map_lsn_to_time to double-check that your LSNs correspond to the correct time points.
• Conflict Handling: Test merge logic thoroughly—for example, if a row was deleted and reinserted between X and Y, ensure the merge handles it correctly.

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