SQL Server Mentalist

Hi Geeks,

In this article we are going to cover How to get information from Change Data Capture (CDC) Tables.

In this post we are going to discuss following points:

• Problem Statement of SQL Script:
• Description of SQL Script:
• SQL Script Code
• User Level to execute

Problem Statement of SQL Script:

Find How to get information from Change Data Capture (CDC) Tables.

Description of SQL Script:

This Transact-SQL Server code goes through all enabled CDC tables and read the data from there.

SQL Script Code

EXEC sys.sp_cdc_enable_db

DECLARE @SchemaId INT
    ,@SchemaName VARCHAR(128)
    ,@TableId INT
    ,@TableName VARCHAR(128)
    ,@ColumnId INT
    ,@ColumnName VARCHAR(128)
DECLARE @strSQL NVARCHAR(4000)
    ,@CDCTableLoopNo SMALLINT
    ,@CountCDCTables SMALLINT
DECLARE @CDCTableName VARCHAR(128)
    ,@CDCSchemaName VARCHAR(128)
    ,@CDCCaptureInstance VARCHAR(128)
DECLARE @PrimaryKeyFields VARCHAR(Max)
    ,@PrimaryKeyFieldsForValues VARCHAR(Max)

SET @CDCTableLoopNo = 0

DECLARE @CDCTables TABLE (
    CDCObjectId INT
    ,TableObjectId INT
    ,Capture_Instance VARCHAR(128)
    ,Role_Name VARCHAR(128)
    ,Index_Name VARCHAR(128)
    ,TableName VARCHAR(128)
    ,SchemaId INT
    ,SchemaName VARCHAR(128)
    ,RN SMALLINT
    )

INSERT INTO @CDCTables
SELECT CDC_CT.Object_Id AS CDCObjectId
    ,CDC_CT.Source_Object_Id AS TableObjectId
    ,QuoteName(CDC_CT.Capture_Instance + '_CT') AS 
    Capture_Instance
    ,CDC_CT.Role_Name
    ,CDC_CT.Index_Name
    ,QuoteName(SysTbl.NAME) AS TableName
    ,SysSchem.schema_id
    ,QuoteName(SysSchem.NAME) AS SchemaName
    ,ROW_NUMBER() OVER (
        ORDER BY CDC_CT.Object_Id
        ) AS RN
FROM CDC.change_tables CDC_CT
INNER JOIN Sys.tables AS SysTbl ON CDC_CT.Source_Object_Id = SysTbl
    .object_id
INNER JOIN Sys.schemas AS SysSchem ON SysTbl.schema_id = SysSchem.
    schema_id
WHERE 1 = 1

SET @CountCDCTables = @@ROWCOUNT

DECLARE @IndexColumns TABLE (
    TableObjectId INT
    ,TableName VARCHAR(128)
    ,SchemaId INT
    ,SchemaName VARCHAR(128)
    ,IndexId INT
    ,ColumnId INT
    ,ColumnName VARCHAR(128)
    )

INSERT INTO @IndexColumns
SELECT sysTbls.object_id AS TableObjectId
    ,QuoteName(sysTbls.NAME) AS TableName
    ,SysSchem.schema_id AS SchemaId
    ,QuoteName(SysSchem.NAME) AS SchemaName
    ,SysIndex.index_id
    ,SysIndexCols.column_id
    ,QuoteName(SysCols.NAME)
FROM sys.tables AS sysTbls
INNER JOIN sys.schemas AS SysSchem ON sysTbls.schema_id = SysSchem.
    schema_id
INNER JOIN sys.indexes AS SysIndex ON sysTbls.object_id = SysIndex.
    object_id
INNER JOIN sys.index_columns AS SysIndexCols ON SysIndex.
    object_id = SysIndexCols.object_id
    AND SysIndex.index_id = SysIndexCols.index_id
INNER JOIN sys.columns AS SysCols ON SysIndexCols.column_id = 
    SysCols.column_id
    AND SysIndexCols.object_id = SysCols.object_id
WHERE SysIndex.type = 1
    AND SysSchem.NAME <> 'CDC'
ORDER BY sysTbls.object_id

WHILE @CDCTableLoopNo < @CountCDCTables
BEGIN
    SELECT @CDCTableName = ''
        ,@CDCSchemaName = ''
        ,@CDCCaptureInstance = ''
        ,@PrimaryKeyFields = ''
        ,@PrimaryKeyFieldsForValues = ''

    SELECT @CDCTableName = TableName
        ,@CDCSchemaName = SchemaName
        ,@CDCCaptureInstance = Capture_Instance
    FROM @CDCTables
    WHERE RN = @CDCTableLoopNo + 1

    SELECT @PrimaryKeyFields = STUFF((
                SELECT ',' + SubQry.ColumnName
                FROM @IndexColumns AS SubQry
                WHERE SubQry.TableObjectId = MainQry.
                    TableObjectId
                    AND SubQry.SchemaId = MainQry.SchemaId
                FOR XML Path('')
                ), 1, 1, '')
        ,@PrimaryKeyFieldsForValues = STUFF((
                SELECT ', Convert(Varchar(Max), SubQry.' + 
                    SubQry.ColumnName + ') +' + ''','''
                FROM @IndexColumns AS SubQry
                WHERE SubQry.TableObjectId = MainQry.
                    TableObjectId
                    AND SubQry.SchemaId = MainQry.SchemaId
                FOR XML Path('')
                ), 1, 1, '')
    FROM @IndexColumns AS MainQry
    WHERE MainQry.SchemaName = @CDCSchemaName
        AND MainQry.TableName = @CDCTableName
    GROUP BY MainQry.SchemaId
        ,MainQry.TableObjectId
        ,MainQry.TableName

    IF @PrimaryKeyFields = ''
    BEGIN
        SET @PrimaryKeyFields = '''-'''
        SET @PrimaryKeyFieldsForValues = '''-'''
    END

    DECLARE mySysCursor CURSOR
    FOR
    SELECT SysSchem.schema_id AS SchemaId
        ,QuoteName(SysSchem.NAME) AS SchemaName
        ,SysTbls.object_id AS TableId
        ,QuoteName(SysTbls.NAME) AS TableName
        ,SysCols.column_id AS ColumnId
        ,QuoteName(SysCols.NAME) AS ColumnName
    FROM sys.schemas AS SysSchem
    INNER JOIN sys.tables AS SysTbls ON SysSchem.schema_id = 
        SysTbls.schema_id
    INNER JOIN Sys.columns AS SysCols ON SysTbls.object_id = 
        SysCols.object_id
    WHERE QuoteName(SysTbls.NAME) = @CDCTableName
        AND QuoteName(SysSchem.NAME) = @CDCSchemaName

    OPEN mySysCursor;

    FETCH NEXT
    FROM mySysCursor
    INTO @SchemaId
        ,@SchemaName
        ,@TableId
        ,@TableName
        ,@ColumnId
        ,@ColumnName

    WHILE @@FETCH_STATUS = 0
    BEGIN
        BEGIN TRY
            SET @strSQL = 
                ' 
            Select  
                Convert(DateTime, TimeMapping.tran_end_time, 101) As AuditDate 
                ,(Case MAX(CDC_Tbl.__$operation) When 1 Then ' 
                + '''D''' + ' When 2 Then ' + '''I''' + 
                ' When 3 Then ' + '''U''' + ' When 4 Then ' + 
                '''U''' + 
                ' End) As AuditType 
                ,(''' + 
                @SchemaName + 
                ''') As SchemaName  
                ,(''' + 
                @TableName + 
                ''') As TableName  
            '

            IF @PrimaryKeyFields = '''-'''
            BEGIN
                SET @strSQL = @strSQL + 
                    '    ,''-'' As PrimaryKeyFields'
                SET @strSQL = @strSQL + 
                    '    ,''-'' As PrimaryKeyValues'
            END
            ELSE
            BEGIN
                SET @strSQL = @strSQL + '    ,''' + 
                    @PrimaryKeyFields + 
                    ''' As PrimaryKeyFields'
                SET @strSQL = @strSQL + 
                    '   ,(Select Stuff((Select Distinct '','' + ' 
                    + @PrimaryKeyFieldsForValues + ' From cdc.' 
                    + @CDCCaptureInstance + 
                    ' As SubQry Where SubQry.__$start_lsn = 

CDC_Tbl.__$start_lsn And SubQry.__$seqval = CDC_Tbl.__$seqval 

For XML Path('''')), 1, 1, '''')) As PrimaryKeyValues '
            END

            SET @strSQL = @strSQL + 
                '             
                ,(''' + 
                @ColumnName + 
                ''') As FieldName  
                ,Max(Case CDC_Tbl.__$operation When 2 Then '''' When 3 

Then Convert(Varchar(Max), ' 
                + @ColumnName + 
                ') When 4 Then '''' When 1 Then Convert(Varchar(Max), ' 
                + @ColumnName + 
                ') End) As OldValue  
                ,Max(Case CDC_Tbl.__$operation When 2 Then Convert

(Varchar(Max), ' 
                + @ColumnName + 
                ') When 3 Then '''' When 4 Then Convert(Varchar(Max), ' 
                + @ColumnName + 
                ') When 1 Then '''' End) As NewValue              
            From  
                cdc.' 
                + @CDCCaptureInstance + 
                ' As CDC_Tbl  
                Left Outer Join cdc.lsn_time_mapping As TimeMapping On 

CDC_Tbl.__$start_lsn = TimeMapping.start_lsn  
            Where  
                TimeMapping.tran_end_time Between DATEADD(Day, -1, 

GetDate()) And DATEADD(Day, 0, GetDate())  
            Group By  
                Convert(DateTime, TimeMapping.tran_end_time, 101) 
                ,CDC_Tbl.__$start_lsn 
                ,CDC_Tbl.__$seqval 
                '

            IF @PrimaryKeyFields <> '''-'''
            BEGIN
                SET @strSQL = @strSQL + ',' + @PrimaryKeyFields + ''
            END

            SET @strSQL = @strSQL + 
                '             
            Having  
                Max(Case CDC_Tbl.__$operation When 2 Then '''' When 3 

Then Convert(Varchar(Max), ' 
                + @ColumnName + 
                ') When 4 Then '''' When 1 Then Convert(Varchar(Max), ' 
                + @ColumnName + 
                ') End) 
                <> Max(Case CDC_Tbl.__$operation When 2 

Then Convert(Varchar(Max), ' 
                + @ColumnName + 
                ') When 3 Then '''' When 4 Then Convert(Varchar(Max), ' 
                + @ColumnName + 
                ') When 1 Then '''' End) 
            Order By  
                __$start_lsn 
            '

            EXEC sp_ExecuteSQL @strSQL

            PRINT (@TableName)
        END TRY

        BEGIN CATCH
            --catch errors which are happening  
            SELECT GETDATE() AS ErrorDate
                ,@SchemaName
                ,@TableName
                ,@ColumnName
                ,ERROR_NUMBER() AS ErrorNumber
                ,ERROR_SEVERITY() AS ErrorSeverity
                ,ERROR_STATE() AS ErrorState
                ,ERROR_PROCEDURE() AS ErrorProcedure
                ,ERROR_LINE() AS ErrorLine
                ,ERROR_MESSAGE() AS ErrorMessage;
        END CATCH

        FETCH NEXT
        FROM mySysCursor
        INTO @SchemaId
            ,@SchemaName
            ,@TableId
            ,@TableName
            ,@ColumnId
            ,@ColumnName
    END

    CLOSE mySysCursor

    DEALLOCATE mySysCursor

    SET @CDCTableLoopNo = @CDCTableLoopNo + 1
END

User Level to execute

300

Hope you will like How to get information from Change Data Capture (CDC) Tables.

If you really like reading my blog and understood at least few thing then please don’t forget to subscribe my blog.

If you want daily link and analysis or interesting link go to following website which will give @ your inbox please subscribe our following link resource blog :

Link Resource Website

For More information related to BI World visit my Mentalist Blog

SQL Server Mentalist >> SQL Learning Blog

Business Intelligence Mentalist >> BI World

Infographic Mentalist >> Image worth explaining thousand Words

Microsoft Mentalist >> MVC,ASP.NET, WCF & LinQ

DBA Mentalist >>Advance SQL Server Blog

Microsoft BI Mentalist >> MS BI Development Update

Connect With me on

| FaceBook |Twitter | linkedIn| Google+ | WordPress | RSS |